sql-query-mcp

Instructions

Implementation Reference

• sql_query_mcp/executor.py:98-138 (handler)

  The main handler function for the explain_query tool, which validates the SQL, builds the explain query using an adapter, and executes it on the database.

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

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

  The registration point of the explain_query tool in the MCP app, wrapping the execution in _run_tool.

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