get_total_wallets
Retrieve the total number of wallets on Pumpfun and Pumpswap by querying Dune Analytics. Returns the wallet count as an integer or 0 if the query fails. Handles API errors during data retrieval.
Instructions
Retrieve the total number of wallets on Pumpfun and Pumpswap platforms.
This function queries Dune Analytics (query ID: 5239155) to fetch the total wallet count.
Returns:
int: The total number of wallets, or 0 if the query fails.
Raises:
Exception: If the API request or data retrieval encounters an error.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- main.py:46-63 (handler)The main handler function for the 'get_total_wallets' tool. It is decorated with @mcp.tool() for registration and implements the core logic by calling the helper get_latest_result on Dune query 5239155 to retrieve and return the total number of wallets, defaulting to 0 on error.
@mcp.tool() def get_total_wallets() -> int: """ Retrieve the total number of wallets on Pumpfun and Pumpswap platforms. This function queries Dune Analytics (query ID: 5239155) to fetch the total wallet count. Returns: int: The total number of wallets, or 0 if the query fails. Raises: Exception: If the API request or data retrieval encounters an error. """ try: data = get_latest_result(5239155) return data[0].get("total_wallets", 0) except: return 0 - main.py:22-44 (helper)Supporting helper function used by get_total_wallets (and others) to fetch the latest execution results from a specified Dune Analytics query.
def get_latest_result(query_id: int, limit: int = 1000): """ Fetch the latest results from a Dune Analytics query. Args: query_id (int): The ID of the Dune query to fetch results from. limit (int, optional): Maximum number of rows to return. Defaults to 1000. Returns: list: A list of dictionaries containing the query results, or an empty list if the request fails. Raises: httpx.HTTPStatusError: If the API request fails due to a client or server error. """ url = f"{BASE_URL}/query/{query_id}/results" params = {"limit": limit} with httpx.Client() as client: response = client.get(url, params=params, headers=HEADERS, timeout=300) response.raise_for_status() data = response.json() result_data = data.get("result", {}).get("rows", []) return result_data