wormhole-metrics-mcp

Instructions

Fetch top assets by volume from Wormholescan API.

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

Returns:
    String representation of a pandas DataFrame containing top assets by volume

Implementation Reference

• main.py:223-282 (handler)

  The handler function decorated with @mcp.tool() that implements the get_top_assets_by_volume tool. It validates the timeSpan parameter, fetches data from the Wormholescan API, transforms it into a pandas DataFrame sorted by volume, and returns a markdown representation.

  # Define the get_top_assets_by_volume tool
  @mcp.tool()
  async def get_top_assets_by_volume(
      timeSpan: str = "7d"
  ) -> str:
      """
      Fetch top assets by volume from Wormholescan API.
      
      Args:
          timeSpan: Time span for data (7d, 15d, 30d). Default: 7d
      
      Returns:
          String representation of a pandas DataFrame containing top assets by volume
      """
      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-assets-by-volume",
                  params=params
              )
              response.raise_for_status()
              
              # Parse JSON response
              data = response.json()
              
              # Transform data for DataFrame
              rows = [
                  {
                      "emitter_chain": id2name(item.get("emitterChain")),
                      "symbol": item.get("symbol"),
                      "token_chain": id2name(item.get("tokenChain")),
                      "token_address": item.get("tokenAddress"),
                      "volume": item.get("volume")
                  }
                  for item in data.get("assets", [])
              ]
              
              # Create DataFrame
              df = pd.DataFrame(rows)
              
              # Convert volume to numeric
              df["volume"] = pd.to_numeric(df["volume"], errors="coerce")
              
              # Sort by volume descending for readability
              df = df.sort_values("volume", ascending=False)
              
              return df.to_markdown(index=False)
              
      except Exception as e:
          return str(e)