BigQuery Validator

Instructions

Implementation Reference

• src/mcp_bigquery/server.py:562-586 (handler)

  Primary handler function for the bq_analyze_query_structure tool. Instantiates SQLAnalyzer and delegates to its analyze_structure method to perform the analysis.
  async def analyze_query_structure(sql: str, params: dict[str, Any] | None = None) -> dict[str, Any]: """ Analyze BigQuery SQL query structure and complexity. Args: sql: The SQL query to analyze params: Optional query parameters (not used in static analysis) Returns: Dict with structure analysis including query type, complexity score, and feature detection """ try: analyzer = SQLAnalyzer(sql) result = analyzer.analyze_structure() return result except Exception as e: return { "error": { "code": "ANALYSIS_ERROR", "message": f"Failed to analyze query structure: {str(e)}", } }
• src/mcp_bigquery/server.py:116-134 (schema)

  Input schema definition for the bq_analyze_query_structure tool, registered in the list_tools handler.
  types.Tool( name="bq_analyze_query_structure", description=("Analyze BigQuery SQL query structure and complexity"), inputSchema={ "type": "object", "properties": { "sql": { "type": "string", "description": "The SQL query to analyze", }, "params": { "type": "object", "description": ("Optional query parameters (key-value pairs)"), "additionalProperties": True, }, }, "required": ["sql"], }, ),
• src/mcp_bigquery/server.py:342-344 (registration)

  Tool dispatch registration in the call_tool handler, routing requests to the analyze_query_structure function.
  elif name == "bq_analyze_query_structure": result = await analyze_query_structure(sql=arguments["sql"], params=arguments.get("params")) return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
• src/mcp_bigquery/sql_analyzer.py:23-60 (helper)

  Core helper method in SQLAnalyzer class that performs static analysis of SQL query structure, detecting features like joins, subqueries, CTEs, aggregations, window functions, and calculating complexity score.
  def analyze_structure(self) -> dict[str, Any]: """Analyze the SQL query structure. Returns: Dictionary containing structure analysis including: - query_type: Type of SQL query (SELECT, INSERT, etc.) - has_joins: Whether query contains JOINs - has_subqueries: Whether query contains subqueries - has_cte: Whether query uses CTEs (WITH clause) - has_aggregations: Whether query uses GROUP BY/aggregation functions - has_window_functions: Whether query uses window functions - complexity_score: Numerical complexity score """ if not self.parsed: return {"error": "Unable to parse SQL query"} result = { "query_type": self._get_query_type(), "has_joins": self._has_joins(), "has_subqueries": self._has_subqueries(), "has_cte": self._has_cte(), "has_aggregations": self._has_aggregations(), "has_window_functions": self._has_window_functions(), "has_union": self._has_union(), "table_count": len(self._extract_tables()), "complexity_score": self._calculate_complexity_score(), } # Add JOIN details if present if result["has_joins"]: result["join_types"] = self._get_join_types() # Add function usage analysis functions_used = self._get_functions_used() if functions_used: result["functions_used"] = functions_used return result
• src/mcp_bigquery/sql_analyzer.py:9-22 (helper)

  SQLAnalyzer class providing the static analysis capabilities used by the tool handler, including parsing and caching for tables and columns.
  class SQLAnalyzer: """Analyzer for BigQuery SQL queries providing static analysis capabilities.""" def __init__(self, sql: str): """Initialize the SQL analyzer with a query string. Args: sql: The SQL query string to analyze """ self.sql = sql self.parsed = sqlparse.parse(sql)[0] if sql else None self._tables_cache: list[dict[str, str]] | None = None self._columns_cache: list[str] | None = None