explain_query

Instructions

Implementation Reference

• sql_query_mcp/app.py:75-79 (registration)

  Registration of 'explain_query' as an MCP tool via @mcp.tool() decorator. Delegates to executor.explain_query().

  @mcp.tool()
  def explain_query(connection_id: str, sql: str, analyze: bool = False) -> dict:
      """Run EXPLAIN on a read-only SELECT or CTE query."""
  
      return _run_tool(lambda: executor.explain_query(connection_id, sql, analyze))

• sql_query_mcp/executor.py:102-152 (handler)

  Core handler that validates SQL, builds EXPLAIN via adapter, executes it, extracts plan, and returns results with audit logging.

  def explain_query(
      self, connection_id: str, sql_text: str, analyze: bool = False
  ) -> Dict[str, object]:
      started = time.perf_counter()
      config = None
      try:
          config = self._registry.get_connection_config(connection_id)
          cleaned_sql = validate_select_sql(sql_text, config.engine)
          sql_summary = summarize_sql(cleaned_sql)
          adapter = self._registry.get_adapter(config)
          explain_sql = adapter.build_explain_query(cleaned_sql, analyze=analyze)
          with self._registry.connection_from_config(config) as (conn, adapter):
              _apply_statement_timeout(
                  adapter, conn, self._settings.statement_timeout_ms
              )
              with conn.cursor() as cur:
                  cur.execute(explain_sql)
                  rows = cur.fetchall()
  
              duration_ms = _elapsed_ms(started)
              plan = adapter.extract_plan(rows)
              self._audit.log(
                  tool="explain_query",
                  connection_id=connection_id,
                  success=True,
                  duration_ms=duration_ms,
                  row_count=1 if rows else 0,
                  sql_summary=sql_summary,
                  extra={"engine": config.engine, "analyze": analyze},
              )
              return {
                  "connection_id": connection_id,
                  "engine": config.engine,
                  "plan": plan,
                  "duration_ms": duration_ms,
                  "analyze": analyze,
              }
      except Exception as exc:
          duration_ms = _elapsed_ms(started)
          sql_summary = summarize_sql(sql_text)
          sanitized = sanitize_error_message(str(exc))
          self._audit.log(
              tool="explain_query",
              connection_id=connection_id,
              success=False,
              duration_ms=duration_ms,
              sql_summary=sql_summary,
              error=sanitized,
              extra=_build_audit_extra(config, analyze=analyze),
          )
          raise QueryExecutionError(sanitized) from exc

• sql_query_mcp/validator.py:55-66 (schema)

  Validation function that ensures the SQL is a SELECT/WITH query and rejects raw EXPLAIN statements (which would conflict with the tool).

  def validate_select_sql(sql: str, engine: str) -> str:
      cleaned = _clean_sql(sql)
      lowered = cleaned.lstrip().lower()
      if lowered.startswith("explain"):
          raise SecurityError(
              "explain_query 会自动包装 SQL，请直接传 SELECT 或 WITH 查询。"
          )
      if not (lowered.startswith("select") or lowered.startswith("with")):
          raise SecurityError("仅允许 SELECT 或 WITH ... SELECT 语句。")
      statement = _parse_statement(cleaned, engine)
      _ensure_read_only_statement(statement)
      return cleaned

• sql_query_mcp/adapters/postgres.py:168-169 (helper)

  Postgres adapter: builds EXPLAIN (FORMAT JSON, ANALYZE ...) SQL.

  def build_explain_query(self, sql_text: str, analyze: bool = False) -> str:
      return f"EXPLAIN (FORMAT JSON, ANALYZE {'TRUE' if analyze else 'FALSE'}) {sql_text}"

• sql_query_mcp/executor.py:102-152 (helper)

  Core handler that validates SQL, builds EXPLAIN via adapter, executes it, extracts plan, and returns results with audit logging.

  def explain_query(
      self, connection_id: str, sql_text: str, analyze: bool = False
  ) -> Dict[str, object]:
      started = time.perf_counter()
      config = None
      try:
          config = self._registry.get_connection_config(connection_id)
          cleaned_sql = validate_select_sql(sql_text, config.engine)
          sql_summary = summarize_sql(cleaned_sql)
          adapter = self._registry.get_adapter(config)
          explain_sql = adapter.build_explain_query(cleaned_sql, analyze=analyze)
          with self._registry.connection_from_config(config) as (conn, adapter):
              _apply_statement_timeout(
                  adapter, conn, self._settings.statement_timeout_ms
              )
              with conn.cursor() as cur:
                  cur.execute(explain_sql)
                  rows = cur.fetchall()
  
              duration_ms = _elapsed_ms(started)
              plan = adapter.extract_plan(rows)
              self._audit.log(
                  tool="explain_query",
                  connection_id=connection_id,
                  success=True,
                  duration_ms=duration_ms,
                  row_count=1 if rows else 0,
                  sql_summary=sql_summary,
                  extra={"engine": config.engine, "analyze": analyze},
              )
              return {
                  "connection_id": connection_id,
                  "engine": config.engine,
                  "plan": plan,
                  "duration_ms": duration_ms,
                  "analyze": analyze,
              }
      except Exception as exc:
          duration_ms = _elapsed_ms(started)
          sql_summary = summarize_sql(sql_text)
          sanitized = sanitize_error_message(str(exc))
          self._audit.log(
              tool="explain_query",
              connection_id=connection_id,
              success=False,
              duration_ms=duration_ms,
              sql_summary=sql_summary,
              error=sanitized,
              extra=_build_audit_extra(config, analyze=analyze),
          )
          raise QueryExecutionError(sanitized) from exc