get_money_flow
Fetch transaction count and volume data from Wormholescan API for specific time periods and filter by application, source chain, or target chain to analyze cross-chain activity.
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 dataInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| timespan | No | 1d | |
| from_date | No | ||
| to_date | No | ||
| appId | No | ||
| sourceChain | No | ||
| targetChain | No |
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"