validate_query
Validate SQL queries for safety before execution to identify potential issues without running them.
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 |
Implementation Reference
- src/cockroachdb_mcp/server.py:84-98 (registration)MCP tool registration for 'validate_query' using @mcp.tool() decorator. Thin wrapper that delegates to the core implementation in query.py.@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 handler function implementing the validation logic for SQL queries: checks for empty queries, blocked commands, read-only compliance, determines query type, and returns validation results.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), }
- Helper function to detect blocked SQL commands using regex patterns from settings.def _is_blocked_command(query: str) -> tuple[bool, str | None]: """Check if a query contains blocked commands. Args: query: SQL query to check. Returns: Tuple of (is_blocked, blocked_command). """ query_upper = query.upper().strip() for cmd in settings.blocked_commands_list: # Match command at start of query or after whitespace pattern = rf"(^|\s){re.escape(cmd)}(\s|$)" if re.search(pattern, query_upper): return True, cmd return False, None
- Helper function to check if a query is read-only based on starting prefixes.def _is_read_only_query(query: str) -> bool: """Check if query is read-only (SELECT, SHOW, EXPLAIN, etc.).""" query_stripped = query.strip().upper() read_only_prefixes = ( "SELECT", "SHOW", "EXPLAIN", "WITH", "DESCRIBE", "DESC", ) return query_stripped.startswith(read_only_prefixes)