sql-query-mcp

Instructions

Implementation Reference

• sql_query_mcp/introspection.py:66-102 (handler)

  The handler for the 'list_databases' tool, which manages connection configuration, invokes the specific adapter's database listing logic, and logs the outcome.

  def list_databases(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, "mysql", "list_databases")
          with self._registry.connection_from_config(config) as (conn, adapter):
              _apply_statement_timeout(
                  adapter, conn, self._settings.statement_timeout_ms
              )
              databases = adapter.list_databases(conn)
          duration_ms = _elapsed_ms(started)
          self._audit.log(
              tool="list_databases",
              connection_id=connection_id,
              success=True,
              duration_ms=duration_ms,
              row_count=len(databases),
              extra={"engine": config.engine},
          )
          return {
              "connection_id": connection_id,
              "engine": "mysql",
              "databases": databases,
          }
      except Exception as exc:
          duration_ms = _elapsed_ms(started)
          sanitized = sanitize_error_message(str(exc))
          self._audit.log(
              tool="list_databases",
              connection_id=connection_id,
              success=False,
              duration_ms=duration_ms,
              error=sanitized,
              extra=_build_audit_extra(config),
          )
          raise QueryExecutionError(sanitized) from exc