wormhole-metrics-mcp

Instructions

Fetch top chain pairs by number of transfers from Wormholescan API.

Args:
    timeSpan: Time span for data (7d, 15d, 30d). Default: 7d

Returns:
    String representation of a pandas DataFrame containing top chain pairs by number of transfers

Implementation Reference

• main.py:284-341 (handler)

  The handler function for the 'get_top_chain_pairs_by_num_transfers' tool, decorated with @mcp.tool() for registration. It validates the timeSpan parameter, fetches data from the Wormholescan API endpoint '/api/v1/top-chain-pairs-by-num-transfers', transforms the JSON response into a pandas DataFrame using the id2name helper for chain names, sorts by number of transfers, and returns a markdown representation of the table.

  # Define the get_top_chain_pairs_by_num_transfers tool
  @mcp.tool()
  async def get_top_chain_pairs_by_num_transfers(
      timeSpan: str = "7d"
  ) -> str:
      """
      Fetch top chain pairs by number of transfers from Wormholescan API.
      
      Args:
          timeSpan: Time span for data (7d, 15d, 30d). Default: 7d
      
      Returns:
          String representation of a pandas DataFrame containing top chain pairs by number of transfers
      """
      try:
          # Validate parameters
          valid_time_spans = {"7d", "15d", "30d"}
          
          if timeSpan not in valid_time_spans:
              raise ValueError(f"Invalid timeSpan. Must be one of {valid_time_spans}")
          
          # Construct query parameters
          params = {"timeSpan": timeSpan}
          
          # Make API request
          async with httpx.AsyncClient() as client:
              response = await client.get(
                  f"{API_BASE}/api/v1/top-chain-pairs-by-num-transfers",
                  params=params
              )
              response.raise_for_status()
              
              # Parse JSON response
              data = response.json()
              
              # Transform data for DataFrame
              rows = [
                  {
                      "source_chain": id2name(item.get("emitterChain")),
                      "destination_chain": id2name(item.get("destinationChain")),
                      "number_of_transfers": item.get("numberOfTransfers")
                  }
                  for item in data.get("chainPairs", [])
              ]
              
              # Create DataFrame
              df = pd.DataFrame(rows)
              
              # Convert number_of_transfers to numeric
              df["number_of_transfers"] = pd.to_numeric(df["number_of_transfers"], errors="coerce")
              
              # Sort by number_of_transfers descending for readability
              df = df.sort_values("number_of_transfers", ascending=False)
              
              return df.to_markdown(index=False)
              
      except Exception as e:
          return str(e)