#!/usr/bin/env python3 """CryptoBro MCP server (stdio, stdlib only). Exposes the CryptoBro REST API as Model Context Protocol tools so agent harnesses (opencode, Claude Code, Hermes, and any MCP client) can pull liquidation, open interest, funding, long/short, order flow, order book, live-series and docs data directly. Configuration (env): CRYPTOBRO_API_KEY API key (cb_...) created in Account settings. CRYPTOBRO_API_URL Base endpoint, default https://crypto-bro.lol/api Run (a harness launches this itself): CRYPTOBRO_API_KEY=cb_... python3 mcp_server.py Install snippets are in the "MCP" section of the API docs (/docs and /docs.md). """ from __future__ import annotations import json import os import sys import urllib.error import urllib.parse import urllib.request PROTOCOL_VERSION = "2024-11-05" SERVER_NAME = "cryptobro" SERVER_VERSION = "1.0.0" DEFAULT_API = "https://crypto-bro.lol/api" TOOLS = [ { "name": "cryptobro_catalog", "description": "List every CryptoBro data view the configured API key can access " "(aggregated endpoints, direct exchange sources and native models).", "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}, }, { "name": "cryptobro_data", "description": "Fetch a CryptoBro data view as JSON. Use route 'ep' for aggregated " "endpoints (e.g. key 'oi_all', 'ls_person', 'fr_current'), 'src' for a " "direct exchange source (e.g. key 'binance_oi_hist'), or 'liq' for a native " "liquidation model (key 'event'|'estimate'|'hybrid').", "inputSchema": { "type": "object", "properties": { "route": {"type": "string", "enum": ["ep", "src", "liq"]}, "key": {"type": "string"}, "params": { "type": "object", "description": "Query parameters, e.g. {\"symbol\":\"BTCUSDT\",\"interval\":\"1h\"}.", "additionalProperties": True, }, }, "required": ["route", "key"], "additionalProperties": False, }, }, { "name": "cryptobro_series_me", "description": "Your Live Series access: tier, unlocked views and remaining daily refresh quota.", "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}, }, { "name": "cryptobro_series", "description": "Fetch a Live Series frame (tiered and quota-limited by your plan). " "view: unified | deviation_base | deviation_voldv.", "inputSchema": { "type": "object", "properties": { "view": {"type": "string", "default": "unified"}, "asset": {"type": "string", "default": "SOLUSDT"}, "at": {"type": "string", "description": "Optional epoch seconds for an archived frame."}, }, "additionalProperties": False, }, }, { "name": "cryptobro_docs", "description": "Return the full CryptoBro API reference as Markdown.", "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}, }, ] def _api_base() -> str: return os.environ.get("CRYPTOBRO_API_URL", DEFAULT_API).rstrip("/") def _api_key() -> str: return os.environ.get("CRYPTOBRO_API_KEY", "").strip() def _http_get(query: dict) -> tuple[bool, str]: url = _api_base() + "?" + urllib.parse.urlencode(query) req = urllib.request.Request(url, headers={ "Accept": "application/json", "User-Agent": "CryptoBro-MCP/%s (+https://crypto-bro.lol)" % SERVER_VERSION, }) key = _api_key() if key: req.add_header("Authorization", "Bearer " + key) req.add_header("X-API-Key", key) try: with urllib.request.urlopen(req, timeout=120) as resp: body = resp.read().decode("utf-8", "replace") return True, body except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", "replace") if exc.fp else "" return False, "HTTP %s: %s" % (exc.code, detail or exc.reason) except Exception as exc: # noqa: BLE001 return False, "request failed: %s" % exc def _json_get(query: dict) -> str: ok, body = _http_get(query) if not ok: raise RuntimeError(body) return body def call_tool(name: str, args: dict) -> str: if name == "cryptobro_catalog": return _json_get({"action": "catalog"}) if name == "cryptobro_data": route = str(args.get("route") or "ep") key = str(args.get("key") or "") if route not in ("ep", "src", "liq") or not key: raise RuntimeError("route must be ep|src|liq and key is required") params = args.get("params") if isinstance(args.get("params"), dict) else {} query = {route: key} for k, v in params.items(): query[str(k)] = "" if v is None else str(v) return _json_get(query) if name == "cryptobro_series_me": return _json_get({"action": "series_me"}) if name == "cryptobro_series": q = {"action": "series", "view": str(args.get("view") or "unified"), "asset": str(args.get("asset") or "SOLUSDT")} if args.get("at") not in (None, ""): q["at"] = str(args["at"]) return _json_get(q) if name == "cryptobro_docs": ok, body = _http_get({"action": "docs_md"}) return body if ok else ("# CryptoBro API\n\n" + body) raise RuntimeError("unknown tool: " + name) def _result(rid, result): return {"jsonrpc": "2.0", "id": rid, "result": result} def _error(rid, code, message): return {"jsonrpc": "2.0", "id": rid, "error": {"code": code, "message": message}} def handle(message: dict): method = message.get("method") rid = message.get("id") if method == "initialize": return _result(rid, { "protocolVersion": PROTOCOL_VERSION, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION}, }) if method in ("notifications/initialized", "initialized"): return None if method == "ping": return _result(rid, {}) if method == "tools/list": return _result(rid, {"tools": TOOLS}) if method == "tools/call": params = message.get("params") or {} name = params.get("name") args = params.get("arguments") or {} try: text = call_tool(name, args) return _result(rid, {"content": [{"type": "text", "text": text}], "isError": False}) except Exception as exc: # noqa: BLE001 return _result(rid, {"content": [{"type": "text", "text": str(exc)}], "isError": True}) if rid is None: return None return _error(rid, -32601, "method not found: " + str(method)) def serve() -> None: for line in sys.stdin: line = line.strip() if not line: continue try: message = json.loads(line) except ValueError: continue response = handle(message) if response is not None: sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() if __name__ == "__main__": serve()