CockroachDB MCP Server

Instructions

Args:
    sql: SQL statement to validate.

Returns:
    Validation result with any issues found.

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)}

• src/cockroachdb_mcp/tools/query.py:52-102 (helper)

  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),
      }