pumpfun-wallets-mcp

Instructions

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