query_latest_withdrawals
Retrieve recent Tornado Cash withdrawal records in a formatted table, displaying key details such as amount, timestamp, recipient, and block number.
Instructions
Query the most recent withdrawals from Tornado Cash Subgraph and return results as a formatted table.
Parameters:
limits (int): The maximum number of withdrawal records to return. Must be positive. Default is 10.
Returns:
A string containing a tabulated representation of withdrawal records with columns: id, amount, timestamp, to, blockNumber.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limits | No |
Implementation Reference
- main.py:89-134 (handler)The primary handler function for the 'query_latest_withdrawals' MCP tool. Decorated with @mcp.tool() for automatic registration and schema inference from types and docstring. Executes GraphQL query to fetch latest withdrawals from Tornado Cash subgraph, validates input, formats output as ASCII table using tabulate library.
@mcp.tool() async def query_latest_withdrawals(limits: int = 10, ctx: Context = None) -> str: """ Query the most recent withdrawals from Tornado Cash Subgraph and return results as a formatted table. Parameters: limits (int): The maximum number of withdrawal records to return. Must be positive. Default is 10. Returns: A string containing a tabulated representation of withdrawal records with columns: id, amount, timestamp, to, blockNumber. """ if limits <= 0: raise ValueError("limits must be positive") query = """ query LatestWithdrawals($first: Int, $orderBy: String, $orderDirection: String) { withdrawals(first: $first, orderBy: $orderBy, orderDirection: $orderDirection) { to amount blockNumber timestamp } } """ variables = { "first": limits, "orderBy": "timestamp", "orderDirection": "desc" } result = await query_subgraph(query, variables) withdrawals = result["data"]["withdrawals"] # Format withdrawals as a table table_data = [ [ withdrawal["to"], withdrawal["amount"], withdrawal["blockNumber"], datetime.fromtimestamp(int(withdrawal["timestamp"])) ] for withdrawal in withdrawals ] headers = ["to", "amount", "blockNumber", "time"] table = tabulate(table_data, headers=headers, tablefmt="grid") return table - main.py:24-38 (helper)Utility helper function called by the tool handler to perform authenticated POST requests to the Tornado Cash subgraph GraphQL endpoint.
async def query_subgraph(query: str, variables: Dict = None) -> Dict: """Helper function to query the Tornado Cash Subgraph with API key.""" headers = { "Authorization": f"Bearer {THEGRAPH_API_KEY}", "Content-Type": "application/json" } async with httpx.AsyncClient() as client: response = await client.post( SUBGRAPH_URL, json={"query": query, "variables": variables or {}}, headers=headers ) response.raise_for_status() return response.json()