sql-query-mcp

Instructions

Implementation Reference

• sql_query_mcp/introspection.py:28-64 (handler)

  The primary handler for the `list_schemas` tool, which orchestrates connection retrieval, adapter usage, and auditing.

  def list_schemas(self, connection_id: str) -> Dict[str, object]:
      started = time.perf_counter()
      config = None
      try:
          config = self._registry.get_connection_config(connection_id)
          require_engine(config, "postgres", "list_schemas")
          with self._registry.connection_from_config(config) as (conn, adapter):
              _apply_statement_timeout(
                  adapter, conn, self._settings.statement_timeout_ms
              )
              schemas = adapter.list_schemas(conn)
          duration_ms = _elapsed_ms(started)
          self._audit.log(
              tool="list_schemas",
              connection_id=connection_id,
              success=True,
              duration_ms=duration_ms,
              row_count=len(schemas),
              extra={"engine": config.engine},
          )
          return {
              "connection_id": connection_id,
              "engine": "postgres",
              "schemas": schemas,
          }
      except Exception as exc:
          duration_ms = _elapsed_ms(started)
          sanitized = sanitize_error_message(str(exc))
          self._audit.log(
              tool="list_schemas",
              connection_id=connection_id,
              success=False,
              duration_ms=duration_ms,
              error=sanitized,
              extra=_build_audit_extra(config),
          )
          raise QueryExecutionError(sanitized) from exc