analyze_query_error
Analyze failed Dune Analytics queries to identify errors and provide specific fixes for issues like 'Column not found'.
Instructions
Analyze failed queries and suggest fixes (e.g. for 'Column not found').
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| error_message | Yes | ||
| query_sql | Yes |
Implementation Reference
- src/main.py:33-46 (handler)The main handler function for the 'analyze_query_error' tool, decorated with @mcp.tool(). Delegates analysis to ErrorAnalyzer and formats the response with error type and suggestion.@mcp.tool() def analyze_query_error(error_message: str, query_sql: str) -> str: """ Analyze failed queries and suggest fixes (e.g. for 'Column not found'). """ analysis = error_analyzer.analyze(error_message, query_sql) response = [f"Error Type: {analysis['error_type']}"] if analysis['suggestion']: response.append(f"Suggestion: {analysis['suggestion']}") else: response.append("No specific suggestion found. Please check the SQL syntax and schema manually.") return "\n".join(response)
- src/services/error_analyzer.py:12-55 (helper)Core helper method implementing the error analysis logic using regex to detect common issues like ColumnNotFound or TableNotFound and generate targeted suggestions.def analyze(self, error_message: str, sql: str) -> Dict[str, Any]: """ Analyze a query error and SQL to provide suggestions. """ analysis = { "error_type": "Unknown", "original_error": error_message, "suggestion": None } # 1. Check for "Column not found" # Example error: "Column 'block_time' cannot be resolved" or "Column 'block_time' not found" col_match = re.search(r"Column '(\w+)'", error_message, re.IGNORECASE) if col_match: bad_col = col_match.group(1) analysis["error_type"] = "ColumnNotFound" # Heuristic: Check for common renames if bad_col == "block_time": analysis["suggestion"] = "In many raw decoded tables, 'block_time' is named 'evt_block_time'. Try using 'evt_block_time'." elif bad_col == "address": analysis["suggestion"] = "Try 'contract_address' or 'wallet_address' depending on context." else: # Advanced: Search spellbook for table usage # Extract table name from SQL (very basic regex) table_match = re.search(r"FROM\s+([\w\.]+)", sql, re.IGNORECASE) if table_match: table_name = table_match.group(1) analysis["suggestion"] = f"Column '{bad_col}' not found in '{table_name}'. Try searching Spellbook for '{table_name}' to see valid columns." else: analysis["suggestion"] = f"Column '{bad_col}' appears incorrect. Check schema." # 2. Check for "Table not found" elif "Table" in error_message and ("not found" in error_message or "cannot be resolved" in error_message): analysis["error_type"] = "TableNotFound" # Extract likely table table_match = re.search(r"Table '([\w\.]+)'", error_message, re.IGNORECASE) if table_match: bad_table = table_match.group(1) # Suggest searching spellbook analysis["suggestion"] = f"Table '{bad_table}' does not exist. Use 'search_spellbook' to find the correct table name." return analysis