wormhole-metrics-mcp

Instructions

Implementation Reference

• main.py:72-140 (handler)

  The @mcp.tool() decorator registers the tool, and the async function implements the core logic: validates params, fetches data from Wormholescan API /api/v1/x-chain-activity, processes into a pivot table of source-to-dest chain volumes using pandas, returns markdown table.
  @mcp.tool() async def get_cross_chain_activity( timeSpan: str = "7d", by: str = "notional", app: str = "" ) -> str: """ Fetch cross-chain activity data from Wormholescan API and return as a pandas DataFrame. Args: timeSpan: Time span for data (7d, 30d, 90d, 1y, all-time). Default: 7d by: Render results by notional or tx count. Default: notional app: Comma-separated list of apps. Default: all apps Returns: String representation of a pandas DataFrame containing cross-chain activity data """ try: # Validate parameters valid_time_spans = {"7d", "30d", "90d", "1y", "all-time"} valid_by = {"notional", "tx count"} if timeSpan not in valid_time_spans: raise ValueError(f"Invalid timeSpan. Must be one of {valid_time_spans}") if by not in valid_by: raise ValueError(f"Invalid 'by' parameter. Must be one of {valid_by}") # Construct query parameters params = { "timeSpan": timeSpan, "by": by } if app: params["apps"] = app # API expects 'apps' as query param name # Make API request async with httpx.AsyncClient() as client: response = await client.get( f"{API_BASE}/api/v1/x-chain-activity", params=params ) response.raise_for_status() # Parse JSON response data = response.json() # Flatten the nested data for DataFrame rows = [] for tx in data.get("txs", []): source_chain = tx.get("chain") for dest in tx.get("destinations", []): rows.append({ "source_chain": id2name(source_chain), "dest_chain": id2name(dest.get("chain")), "volume": dest.get("volume"), }) # Create DataFrame df = pd.DataFrame(rows) pivot_df = df.pivot(index="source_chain", columns="dest_chain", values="volume") # Convert volume to numeric and fill NaN with empty string pivot_df = pivot_df.apply(pd.to_numeric, errors="coerce").fillna("") return pivot_df.to_markdown() except Exception as e: return str(e)
• main.py:67-70 (helper)

  Helper function to map chain ID to name, used in get_cross_chain_activity to label chains in the output DataFrame.
  def id2name(id) -> str: id = str(id) return WORMHOLE_CHAINS.get(id, id)
• main.py:72-72 (registration)

  The @mcp.tool() decorator on the handler function registers 'get_cross_chain_activity' as an MCP tool.
  @mcp.tool()