validate_query
Validate SQL queries for safety before execution to identify potential issues without running the query.
Instructions
Check if a query is safe to execute without running it.
Args:
sql: SQL statement to validate.
Returns:
Validation result with any issues found.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/cockroachdb_mcp/server.py:84-98 (handler)MCP tool handler for 'validate_query' that wraps and delegates to the core validation logic in the query module.
@mcp.tool() async def validate_query(sql: str) -> dict[str, Any]: """Check if a query is safe to execute without running it. Args: sql: SQL statement to validate. Returns: Validation result with any issues found. """ try: return await query.validate_query(sql) except Exception as e: return {"status": "error", "error": str(e)} - Core implementation of query validation logic, including checks for blocked commands, read-only mode compliance, query type classification, and issue reporting.
async def validate_query(query: str) -> dict[str, Any]: """Validate a SQL query without executing it. Args: query: SQL query to validate. Returns: Validation result with is_valid and any issues. """ issues: list[str] = [] # Check for empty query if not query or not query.strip(): return { "is_valid": False, "issues": ["Query is empty"], "query_type": None, } # Check for blocked commands is_blocked, blocked_cmd = _is_blocked_command(query) if is_blocked: issues.append(f"Blocked command: {blocked_cmd}") # Check read-only mode if settings.read_only and not _is_read_only_query(query): issues.append("Server is in read-only mode; only SELECT/SHOW/EXPLAIN allowed") # Determine query type query_upper = query.strip().upper() if query_upper.startswith("SELECT") or query_upper.startswith("WITH"): query_type = "SELECT" elif query_upper.startswith("INSERT"): query_type = "INSERT" elif query_upper.startswith("UPDATE"): query_type = "UPDATE" elif query_upper.startswith("DELETE"): query_type = "DELETE" elif query_upper.startswith("SHOW"): query_type = "SHOW" elif query_upper.startswith("EXPLAIN"): query_type = "EXPLAIN" else: query_type = "OTHER" return { "is_valid": len(issues) == 0, "issues": issues, "query_type": query_type, "is_read_only": _is_read_only_query(query), }