wormhole-metrics-mcp

Instructions

Fetch transaction count and volume data from Wormholescan API for a specific period.

Args:
    timespan: Time span for data (1h, 1d, 1mo, 1y). Default: 1d
    from_date: From date in ISO 8601 format (e.g., 2024-01-01T15:04:05Z). Default: empty
    to_date: To date in ISO 8601 format (e.g., 2024-01-01T15:04:05Z). Default: empty
    appId: Application ID to filter results. Default: empty
    sourceChain: Source chain ID to filter results. Default: empty
    targetChain: Target chain ID to filter results. Default: empty

Returns:
    String representation of a pandas DataFrame containing transaction count and volume data

Implementation Reference

• main.py:142-222 (handler)

  Full implementation of the 'get_money_flow' tool handler. This async function validates input parameters, queries the Wormholescan API endpoint '/api/v1/x-chain-activity/tops', transforms the JSON response into a pandas DataFrame with columns for timestamps, source chain (using id2name helper), volume, and count, sorts by 'from' date, and returns a markdown table string.

  @mcp.tool()
  async def get_money_flow(
      timespan: str = "1d",
      from_date: str = "",
      to_date: str = "",
      appId: str = "",
      sourceChain: str = "",
      targetChain: str = ""
  ) -> str:
      """
      Fetch transaction count and volume data from Wormholescan API for a specific period.
      
      Args:
          timespan: Time span for data (1h, 1d, 1mo, 1y). Default: 1d
          from_date: From date in ISO 8601 format (e.g., 2024-01-01T15:04:05Z). Default: empty
          to_date: To date in ISO 8601 format (e.g., 2024-01-01T15:04:05Z). Default: empty
          appId: Application ID to filter results. Default: empty
          sourceChain: Source chain ID to filter results. Default: empty
          targetChain: Target chain ID to filter results. Default: empty
      
      Returns:
          String representation of a pandas DataFrame containing transaction count and volume data
      """
      try:
          # Validate parameters
          valid_timespans = {"1h", "1d", "1mo", "1y"}
          
          if timespan not in valid_timespans:
              raise ValueError(f"Invalid timespan. Must be one of {valid_timespans}")
          
          # Construct query parameters
          params = {"timespan": timespan}
          if from_date:
              params["from"] = from_date
          if to_date:
              params["to"] = to_date
          if appId:
              params["appId"] = appId
          if sourceChain:
              params["sourceChain"] = sourceChain
          if targetChain:
              params["targetChain"] = targetChain
          
          # Make API request
          async with httpx.AsyncClient() as client:
              response = await client.get(
                  f"{API_BASE}/api/v1/x-chain-activity/tops",
                  params=params
              )
              response.raise_for_status()
              
              # Parse JSON response
              data = response.json()
              
              # Transform data for DataFrame
              rows = [
                  {
                      "from": item.get("from"),
                      "to": item.get("to"),
                      "source_chain": id2name(item.get("emitter_chain")),
                      "volume": item.get("volume"),
                      "count": item.get("count")
                  }
                  for item in data
              ]
              
              # Create DataFrame
              df = pd.DataFrame(rows)
              
              # Convert numeric columns
              df["volume"] = pd.to_numeric(df["volume"], errors="coerce")
              df["count"] = pd.to_numeric(df["count"], errors="coerce")
              
              # Sort by 'from' date for readability
              df = df.sort_values("from")
              
              return df.to_markdown(index=False)
              
      except Exception as e:
          return str(e)        

• main.py:67-69 (helper)

  Helper utility function 'id2name' that converts Wormhole chain IDs to human-readable names using the WORMHOLE_CHAINS dictionary, called within the get_money_flow handler to label source chains.

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

• main.py:143-170 (schema)

  Schema definition via type hints on parameters, comprehensive docstring describing inputs/outputs, and runtime validation for 'timespan' parameter.

  async def get_money_flow(
      timespan: str = "1d",
      from_date: str = "",
      to_date: str = "",
      appId: str = "",
      sourceChain: str = "",
      targetChain: str = ""
  ) -> str:
      """
      Fetch transaction count and volume data from Wormholescan API for a specific period.
      
      Args:
          timespan: Time span for data (1h, 1d, 1mo, 1y). Default: 1d
          from_date: From date in ISO 8601 format (e.g., 2024-01-01T15:04:05Z). Default: empty
          to_date: To date in ISO 8601 format (e.g., 2024-01-01T15:04:05Z). Default: empty
          appId: Application ID to filter results. Default: empty
          sourceChain: Source chain ID to filter results. Default: empty
          targetChain: Target chain ID to filter results. Default: empty
      
      Returns:
          String representation of a pandas DataFrame containing transaction count and volume data
      """
      try:
          # Validate parameters
          valid_timespans = {"1h", "1d", "1mo", "1y"}
          
          if timespan not in valid_timespans:
              raise ValueError(f"Invalid timespan. Must be one of {valid_timespans}")

• main.py:142-142 (registration)

  MCP tool registration decorator applied to the get_money_flow function, registering it with the FastMCP server instance.

  @mcp.tool()

• main.py:15-64 (helper)

  Global dictionary mapping Wormhole chain IDs to names, used by the id2name helper function in the tool implementation.

  WORMHOLE_CHAINS = {
      "1": "Solana",
      "2": "Ethereum",
      "4": "BNB Smart Chain",
      "5": "Polygon",
      "6": "Avalanche",
      "8": "Algorand",
      "10": "Fantom",
      "13": "Kaia",
      "14": "Celo",
      "15": "NEAR",
      "16": "Moonbeam",
      "17": "Neon",
      "18": "Terra 2.0",
      "19": "Injective",
      "20": "Osmosis",
      "21": "Sui",
      "22": "Aptos",
      "23": "Arbitrum",
      "24": "Optimism",
      "25": "Gnosis",
      "26": "Pythnet",
      "30": "Base",
      "32": "Sei",
      "34": "Scroll",
      "35": "Mantle",
      "36": "Blast",
      "37": "X Layer",
      "38": "Linea",
      "39": "Berachain",
      "40": "Seievm",
      "43": "SNAXchain",
      "44": "Unichain",
      "45": "World Chain",
      "46": "Ink",
      "47": "HyperEVM",
      "48": "Monad",
      "50": "Mezo",
      "52": "Sonic",
      "53": "Converge",
      "4000": "Cosmos Hub",
      "4001": "Evmos",
      "4002": "Kujira",
      "4003": "Neutron",
      "4004": "Celestia",
      "4005": "Stargaze",
      "4006": "SEDA",
      "4007": "Dymension",
      "4008": "Provenance",
      "4009": "Noble"