etf-flow-mcp

Instructions

Implementation Reference

• src/etf_flow_mcp/cli.py:106-132 (handler)

  The primary handler function for the 'get_etf_flow' tool. It handles input validation, API calls to CoinGlass, data processing, and returns formatted Markdown table. Registered via @mcp.tool() decorator.
  @mcp.tool() async def get_etf_flow(coin: str, ctx: Context = None) -> str: """ Fetch historical ETF flow data for BTC or ETH from CoinGlass API and return as a Markdown table. Parameters: coin (str): Cryptocurrency to query ('BTC' or 'ETH'). Returns: str: Markdown table with ETF flow data (tickers as columns, dates as rows, with total column). """ coin = coin.upper() if coin not in ["BTC", "ETH"]: return "Invalid coin specified. Please use 'BTC' or 'ETH'." ctx.info(f"Fetching {coin} ETF flow data") endpoint = f"/api/etf/{'bitcoin' if coin == 'BTC' else 'ethereum'}/flow-history" try: data = await fetch_coinglass_data(endpoint) if data.get("code") == "0" and data.get("data"): return format_to_markdown_table(data["data"], coin) else: return f"No {coin} ETF flow data available" except Exception as e: return f"Error fetching {coin} ETF flow: {str(e)}"
• src/etf_flow_mcp/cli.py:23-47 (helper)

  Helper function to make authenticated HTTP requests to the CoinGlass API, used by get_etf_flow.
  async def fetch_coinglass_data(endpoint: str) -> Dict: """ Make an HTTP GET request to the CoinGlass API. Args: endpoint (str): API endpoint (e.g., '/api/etf/bitcoin/flow-history') Returns: Dict: JSON response from the API """ headers = { "accept": "application/json", "CG-API-KEY": COINGLASS_API_KEY } async with httpx.AsyncClient() as client: try: response = await client.get( f"{COINGLASS_API_BASE}{endpoint}", headers=headers ) response.raise_for_status() return response.json() except httpx.HTTPError as e: raise Exception(f"API request failed: {str(e)}")
• src/etf_flow_mcp/cli.py:49-104 (helper)

  Helper function to process raw ETF flow data into a pivoted Markdown table with dates as rows, tickers as columns, and totals, using pandas. Called by get_etf_flow.
  def format_to_markdown_table(data: List[Dict], coin: str) -> str: """ Format ETF flow data into a Markdown table using pandas pivot table. Args: data (List[Dict]): List of ETF flow data entries coin (str): Cryptocurrency ('BTC' or 'ETH') Returns: str: Markdown table string """ if not data: return f"No {coin} ETF flow data available" # Prepare data for pandas records = [] for entry in data: timestamp = entry.get("timestamp") if not timestamp: continue date_str = datetime.fromtimestamp(timestamp / 1000).strftime("%Y-%m-%d") for etf in entry.get("etf_flows", []): ticker = etf.get("etf_ticker") flow = etf.get("change_usd") if ticker: records.append({ "Date": date_str, "Ticker": ticker, "Flow": flow }) if not records: return f"No {coin} ETF flow data available" # Create DataFrame df = pd.DataFrame(records) # Create pivot table pivot = df.pivot_table( values="Flow", index="Date", columns="Ticker", aggfunc="sum", fill_value=0 ) # Sort dates in descending order pivot = pivot.sort_index(ascending=False) # Calculate total column pivot["Total"] = pivot.sum(axis=1) # Convert to Markdown table markdown = pivot.to_markdown(floatfmt=".0f") return markdown
• src/etf_flow_mcp/cli.py:106-106 (registration)

  The @mcp.tool() decorator registers the get_etf_flow function as an MCP tool.
  @mcp.tool()