validate_query
Check SQL queries for safety before execution by validating statement types, blocked commands, read-only compliance, and potential issues like missing WHERE clauses.
Instructions
Check if a query is safe to execute without running it.
Validates the query against:
- Statement type (SELECT, INSERT, UPDATE, DELETE, DDL, EXEC)
- Blocked commands list
- Read-only mode compliance
- Potential issues (missing WHERE clause, unbounded SELECT)
Args:
query: SQL statement to validate
Returns:
Dictionary with:
- query: The original query
- valid: Whether the query is valid
- statement_type: Type of SQL statement
- warnings: List of warning messages
- suggestions: List of suggested improvements
- error: Error message if invalidInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/mssql_mcp/tools/query.py:84-149 (handler)The main handler function for the 'validate_query' MCP tool. It is decorated with @mcp.tool(), which registers it as a tool. The function validates SQL queries using SQLValidator, checks statement types, blocked commands, and provides warnings and suggestions.
@mcp.tool() def validate_query(query: str) -> dict[str, Any]: """Check if a query is safe to execute without running it. Validates the query against: - Statement type (SELECT, INSERT, UPDATE, DELETE, DDL, EXEC) - Blocked commands list - Read-only mode compliance - Potential issues (missing WHERE clause, unbounded SELECT) Args: query: SQL statement to validate Returns: Dictionary with: - query: The original query - valid: Whether the query is valid - statement_type: Type of SQL statement - warnings: List of warning messages - suggestions: List of suggested improvements - error: Error message if invalid """ try: manager = get_connection_manager() config = manager.config # Create validator validator = SQLValidator( blocked_commands=config.blocked_commands, read_only=config.read_only, allowed_schemas=config.allowed_schemas if config.allowed_schemas else None, ) # Detect statement type stmt_type = validator.detect_statement_type(query) # Validate is_valid, error = validator.validate(query) # Get warnings warnings = validator.get_warnings(query) # Build suggestions suggestions: list[str] = [] if stmt_type.value == "SELECT" and "TOP" not in query.upper(): suggestions.append("Consider using TOP clause to limit results") if stmt_type.value in ("UPDATE", "DELETE") and "WHERE" not in query.upper(): suggestions.append("Add WHERE clause to target specific rows") result: dict[str, Any] = { "query": query, "valid": is_valid, "statement_type": stmt_type.value, "warnings": warnings, "suggestions": suggestions, } if not is_valid: result["error"] = error return result except Exception as e: logger.error(f"Error validating query: {e}") return {"error": str(e), "query": query}