pumpfun-wallets-mcp

Instructions

Retrieve the top wallets by all-time trading volume on Pumpfun and Pumpswap.

This function queries Dune Analytics (query ID: 5232018) to fetch a ranked list of wallets
based on their total trading volume, formatted as a tabulated string.

Args:
    limit (int, optional): Maximum number of wallets to return. Defaults to 10.

Returns:
    str: A tabulated string containing the rank, wallet address, trade count, and total
         trading volume (in USD) for each wallet, or an empty string if the query fails.

Raises:
    Exception: If the API request or data retrieval encounters an error.

Implementation Reference

• main.py:94-121 (handler)

  The main handler function for the get_trading_wallets tool. It is decorated with @mcp.tool() for registration. Fetches data from Dune Analytics query ID 5232018 using the helper get_latest_result, processes it into a tabulated string of top trading wallets by volume.

  @mcp.tool()
  def get_trading_wallets(limit: int = 10) -> str:
      """
      Retrieve the top wallets by all-time trading volume on Pumpfun and Pumpswap.
  
      This function queries Dune Analytics (query ID: 5232018) to fetch a ranked list of wallets
      based on their total trading volume, formatted as a tabulated string.
  
      Args:
          limit (int, optional): Maximum number of wallets to return. Defaults to 10.
  
      Returns:
          str: A tabulated string containing the rank, wallet address, trade count, and total
               trading volume (in USD) for each wallet, 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(5232018, limit)
          rows = [
              [row["rank"], row["wallet"], row["trade_count"], f'${row["total_volume_usd"]:.0f}']
              for row in data
          ]
          headers = ["Rank", "Wallet", "Trade Count", "Total Volume"]
          return tabulate(rows, headers=headers)
      except:
          return ""

• main.py:22-45 (helper)

  Helper function used by get_trading_wallets (and other tools) to fetch latest 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