pumpfun-wallets-mcp

Instructions

Implementation Reference

• main.py:123-148 (handler)

  The handler function for the 'get_trading_wallet_distribution' tool. It is decorated with @mcp.tool() for registration and fetches the wallet distribution data from Dune Analytics query ID 5239138, processes it into a tabulated string using tabulate, or returns empty string on error.
  @mcp.tool() def get_trading_wallet_distribution() -> str: """ Retrieve the distribution of wallets by trading volume on Pumpfun and Pumpswap, excluding bots. This function queries Dune Analytics (query ID: 5239138) to fetch the number of wallets grouped by trading volume tiers, formatted as a tabulated string. Returns: str: A tabulated string containing the volume tier and number of wallets in each tier, or an empty string if the query fails. Raises: Exception: If the API request or data retrieval encounters an error. """ try: data = get_latest_result(5239138) rows = [ [row["volume_tier"], row["num_wallets"]] for row in data ] headers = ["Volume Tier", "Number of Wallets"] return tabulate(rows, headers=headers) except: return ""
• main.py:22-45 (helper)

  Supporting helper function used by the tool 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