bq_validate_sql
Validate BigQuery SQL syntax and check for errors without executing queries. Get syntax validation and error detection to ensure your SQL is correct before running it.
Instructions
Validate BigQuery SQL syntax without executing the query
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql | Yes | The SQL query to validate | |
| params | No | Optional query parameters (key-value pairs) |
Implementation Reference
- src/mcp_bigquery/server.py:407-459 (handler)The core implementation of the bq_validate_sql tool. Uses BigQuery client with dry_run=True to validate SQL syntax, supports parameters, extracts error locations and details.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:73-87 (schema)Input schema defining the parameters for the bq_validate_sql tool: required 'sql' string and optional 'params' object.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:70-88 (registration)Registration of the bq_validate_sql tool in the @server.list_tools() handler, including name, description, and schema.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:330-332 (registration)Dispatch/registration in @server.call_tool() handler: routes calls to bq_validate_sql to the validate_sql function.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:45-63 (helper)Helper function used by validate_sql to build query parameters for the dry-run job.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() ]