dune-mcp

Instructions

Implementation Reference

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

  MCP tool handler for 'list_user_queries'. Decorated with @mcp.tool(), handles input parameters, fetches user ID, calls the Dune service to list queries, and formats the output as a string list.
  @mcp.tool() 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:233-288 (helper)

  Core implementation in DuneClient service that performs GraphQL query to list user's queries from Dune API, handles errors, and returns formatted list of query info.
  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 []