get_query_details

Retrieve SQL code and parameters for a specific Dune Analytics query ID to analyze blockchain data.

Instructions

Get SQL and parameters for a query ID.

Input Schema

TableJSON Schema

Name	Required	Description	Default
`query_id`	Yes

Implementation Reference

src/main.py:112-127 (handler)

The primary MCP tool handler for 'get_query_details'. Decorated with @mcp.tool() for automatic registration and schema generation from signature/docstring. Executes the tool logic by calling DuneService.get_query and formatting the response.

@mcp.tool()
def get_query_details(query_id: int) -> str:
    """
    Get SQL and parameters for a query ID.
    """
    try:
        details = dune_service.get_query(query_id)
        return (
            f"Query ID: {details['id']}\n"
            f"Name: {details['name']}\n"
            f"Description: {details['description']}\n"
            f"Parameters: {details['parameters']}\n"
            f"SQL:\n{details['sql']}"
        )
    except Exception as e:
        return f"Error fetching details: {str(e)}"

src/services/dune_client.py:432-462 (helper)

Supporting utility method in DuneService that implements the actual API calls to retrieve query details, including caching, SDK usage, and GraphQL fallback for restricted queries.

def get_query(self, query_id: int) -> Dict[str, Any]:
    cache_key = str(query_id)
    cached = self.cache.get("query", cache_key)
    if cached:
        return cached

    try:
        query = self.client.get_query(query_id)
        # Serialize
        data = {
            "id": query.base.query_id,
            "name": query.base.name,
            "description": query.base.description or "",
            "sql": query.sql,
            "parameters": [p.to_dict() for p in query.base.parameters] if query.base.parameters else []
        }
        self.cache.set("query", cache_key, data)
        return data
    except Exception as e:
        # Check for 403 Forbidden (common for public-but-not-published queries)
        is_forbidden = "403" in str(e) or "Forbidden" in str(e)
        
        if is_forbidden:
            logger.info(f"Access Forbidden via SDK for Query {query_id}. Attempting GraphQL fallback...")
            fallback_data = self._get_query_graphql(query_id)
            if fallback_data:
                self.cache.set("query", cache_key, fallback_data)
                return fallback_data
        
        logger.error(f"Error fetching query {query_id}: {e}")
        raise

dune-mcp