get_top_chain_pairs_by_num_transfers
Analyze and retrieve top chain pairs by the number of transfers on the Wormhole protocol, within specified time spans, to identify high-activity cross-chain routes.
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
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| timeSpan | No | 7d |
Implementation Reference
- main.py:285-341 (handler)The handler function for the 'get_top_chain_pairs_by_num_transfers' tool. Decorated with @mcp.tool() for registration and schema inference. Fetches data from Wormholescan API, transforms into a pandas DataFrame sorted by number of transfers, and returns a markdown table.@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)
- main.py:67-70 (helper)Helper function used by the tool to convert chain IDs to human-readable names.def id2name(id) -> str: id = str(id) return WORMHOLE_CHAINS.get(id, id)
- main.py:15-64 (helper)Global dictionary mapping chain IDs to names, used by id2name helper in the tool.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"