wormhole-metrics-mcp

Instructions

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

Implementation Reference

• main.py:72-140 (handler)

  Handler function decorated with @mcp.tool() that implements the get_cross_chain_activity tool. Fetches data from Wormholescan API, validates params, processes into a pivot DataFrame of source-to-dest chain volumes, 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 convert chain ID to name using WORMHOLE_CHAINS dict, used in get_cross_chain_activity for labeling chains.

  def id2name(id) -> str:
      id = str(id)
      return WORMHOLE_CHAINS.get(id, id)

• main.py:73-88 (schema)

  Input schema defined by annotated parameters and docstring describing valid values and purpose.

  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
      """

• main.py:72-72 (registration)

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

  @mcp.tool()