get_top_assets_by_volume
Fetch top assets by volume from the Wormholescan API for specified time spans, returning a pandas DataFrame for cross-chain activity insights.
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
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| timeSpan | No | 7d |
Implementation Reference
- main.py:225-282 (handler)The async handler function implementing the tool logic: validates timeSpan parameter, queries Wormholescan API for top assets by volume, transforms data with chain name mapping, creates sorted pandas DataFrame, returns markdown representation.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)
- main.py:224-224 (registration)The @mcp.tool() decorator registers the get_top_assets_by_volume function as an MCP-compatible tool.@mcp.tool()
- main.py:228-236 (schema)Docstring providing input/output schema description and parameter details used by MCP for tool schema generation, along with type hints in function signature.""" 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 """
- main.py:67-69 (helper)Helper function to map Wormhole chain IDs to human-readable names, used in data transformation for emitter_chain and token_chain fields.def id2name(id) -> str: id = str(id) return WORMHOLE_CHAINS.get(id, id)