SQL Server MCP

Instructions

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

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}