dune-mcp

Instructions

Implementation Reference

• src/main.py:316-350 (handler)

  The MCP tool handler for 'list_user_queries', decorated with @mcp.tool(). Resolves user handle to ID, fetches query list from DuneService, handles errors, and returns formatted string summary.

  def list_user_queries(handle: Optional[str] = None, limit: int = 10) -> str:
      """
      List queries by user handle.
      """
      target_handle = handle or config.DUNE_USER_HANDLE
      
      if not target_handle:
          return (
              "Error: No user handle provided. "
              "Please provide a 'handle' argument or set DUNE_USER_HANDLE in your .env file."
          )
  
      user_id = dune_service.get_user_id_by_handle(target_handle)
      
      # Check for sentinel -1 (Blocked)
      if user_id == -1:
          return "Error: Public search is currently blocked by Dune's security. Please use 'search_spellbook' or 'get_query_details' instead."
          
      if not user_id:
          return f"Error: Could not find user with handle '{target_handle}'."
          
      results = dune_service.list_user_queries(user_id, limit)
      
      # Handle WAF/Error in results
      if isinstance(results, dict) and "error" in results:
          return f"Error: {results['error']}"
  
      if not results:
          return f"No queries found for user '{target_handle}'."
  
      summary = []
      for q in results:
          summary.append(f"ID: {q.get('id')} | Name: {q.get('name')} | Owner: {q.get('owner')}")
      
      return "\n".join(summary)

• src/services/dune_client.py:243-299 (helper)

  Helper method in DuneService/DuneClient class that performs the GraphQL query to Dune API to retrieve the list of queries for a given user ID, processes the response, and returns formatted results.

  def list_user_queries(self, user_id: int, limit: int = 10) -> Any: # Changed return hint
      """
      List queries for a given user ID using Dune's GraphQL endpoint.
      """
      if user_id == -1: # WAF Block sentinel
           return {"error": "User lookup failed due to Cloudflare block. Cannot list queries."}
  
      url = "https://core-api.dune.com/public/graphql" # Redundant, but good for clarity.
      
      payload = {
          "operationName": "ListUserQueries",
          "variables": {"userId": user_id, "limit": limit},
          "query": """
              query ListUserQueries($userId: Int!, $limit: Int!) {
                  queries(
                      filters: { userId: { equals: $userId } }
                      pagination: { first: $limit }
                  ) {
                      edges {
                          node {
                              id
                              name
                              description
                              user {
                                  name
                                  handle
                              }
                          }
                      }
                  }
              }
          """
      }
  
      response_data = self._get_graphql_response(payload)
      
      if response_data and "error" in response_data and response_data["error"] == "WAF_BLOCK":
           return {"error": "Public search is currently blocked by Dune's security."}
  
      if response_data:
          edges = response_data.get("data", {}).get("queries", {}).get("edges", [])
          
          results = []
          for edge in edges:
              node = edge.get("node", {})
              if not node:
                  continue
                  
              results.append({
                  "id": node.get("id"),
                  "name": node.get("name"),
                  "owner": node.get("user", {}).get("handle", "unknown"),
                  "description": node.get("description", "")
              })
          return results
      return []