bq_validate_sql
Verify BigQuery SQL syntax and structure before execution. Use this tool to check query validity, analyze schema, and estimate costs without running the query.
Instructions
Validate BigQuery SQL syntax without executing the query
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| params | No | Optional query parameters (key-value pairs) | |
| sql | Yes | The SQL query to validate |
Implementation Reference
- src/mcp_bigquery/server.py:412-464 (handler)Core implementation of bq_validate_sql tool: performs dry-run BigQuery query to validate SQL syntax, catches errors, extracts line/column info, and returns structured validation result.async def validate_sql(sql: str, params: dict[str, Any] | None = None) -> dict[str, Any]: """ Validate BigQuery SQL syntax using dry-run. Args: sql: The SQL query to validate params: Optional query parameters Returns: Dict with 'isValid' boolean and optional 'error' details """ logger.debug( f"Validating SQL query: {sql[:100]}..." if len(sql) > 100 else f"Validating SQL query: {sql}" ) try: client = get_bigquery_client() job_config = bigquery.QueryJobConfig( dry_run=True, use_query_cache=False, query_parameters=build_query_parameters(params), ) client.query(sql, job_config=job_config) logger.info("SQL validation successful") return {"isValid": True} except BadRequest as e: error_msg = str(e) logger.warning(f"SQL validation failed: {error_msg}") error_result = { "isValid": False, "error": {"code": "INVALID_SQL", "message": error_msg}, } location = extract_error_location(error_msg) if location: error_result["error"]["location"] = location if hasattr(e, "errors") and e.errors: error_result["error"]["details"] = e.errors return error_result except Exception as e: return { "isValid": False, "error": {"code": "UNKNOWN_ERROR", "message": str(e)}, }
- src/mcp_bigquery/server.py:75-93 (schema)Input schema definition for bq_validate_sql tool, specifying required 'sql' string and optional 'params' object.types.Tool( name="bq_validate_sql", description=("Validate BigQuery SQL syntax without executing the query"), inputSchema={ "type": "object", "properties": { "sql": { "type": "string", "description": "The SQL query to validate", }, "params": { "type": "object", "description": ("Optional query parameters (key-value pairs)"), "additionalProperties": True, }, }, "required": ["sql"], }, ),
- src/mcp_bigquery/server.py:335-337 (registration)Tool dispatch registration in @server.call_tool(): maps 'bq_validate_sql' name to validate_sql handler execution.if name == "bq_validate_sql": result = await validate_sql(sql=arguments["sql"], params=arguments.get("params")) return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
- src/mcp_bigquery/server.py:74-94 (registration)Tool registration in @server.list_tools(): includes bq_validate_sql in the returned list of available tools.return [ types.Tool( name="bq_validate_sql", description=("Validate BigQuery SQL syntax without executing the query"), inputSchema={ "type": "object", "properties": { "sql": { "type": "string", "description": "The SQL query to validate", }, "params": { "type": "object", "description": ("Optional query parameters (key-value pairs)"), "additionalProperties": True, }, }, "required": ["sql"], }, ), types.Tool(
- src/mcp_bigquery/server.py:50-68 (helper)Helper function used by validate_sql to construct BigQuery ScalarQueryParameter list from input params dict.def build_query_parameters(params: dict[str, Any] | None) -> list[bigquery.ScalarQueryParameter]: """ Build BigQuery query parameters from a dictionary. Initial implementation treats all values as STRING type. Args: params: Dictionary of parameter names to values Returns: List of ScalarQueryParameter objects """ if not params: return [] return [ bigquery.ScalarQueryParameter(name, "STRING", str(value)) for name, value in params.items() ]