etf-flow-mcp

Instructions

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).

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()