list_databases

Instructions

Implementation Reference

• sql_query_mcp/app.py:42-46 (handler)

  MCP tool handler that exposes list_databases as a FastMCP tool, delegating to MetadataService.list_databases.

  @mcp.tool()
  def list_databases(connection_id: str) -> dict:
      """List visible databases for a MySQL or Hive connection."""
  
      return _run_tool(lambda: metadata.list_databases(connection_id))

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

  MetadataService.list_databases: validates engine (mysql/hive), gets connection, delegates to adapter, and returns databases with auditing.

  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)
          if config.engine not in {"mysql", "hive"}:
              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": config.engine,
              "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

• sql_query_mcp/app.py:42-43 (schema)

  Input schema: accepts a single string parameter 'connection_id'. Returns a dict.

  @mcp.tool()
  def list_databases(connection_id: str) -> dict:

• sql_query_mcp/adapters/hive.py:58-61 (helper)

  Hive adapter implementation: executes 'SHOW DATABASES'.

  def list_databases(self, conn: object) -> List[str]:
      with conn.cursor() as cur:
          cur.execute("SHOW DATABASES")
          return [self._first_value(row) for row in cur.fetchall()]

• sql_query_mcp/adapters/mysql.py:45-55 (helper)

  MySQL adapter implementation: queries information_schema.schemata, excluding system databases.

  def list_databases(self, conn: object) -> List[str]:
      with conn.cursor() as cur:
          cur.execute(
              """
              SELECT schema_name AS database_name
              FROM information_schema.schemata
              WHERE schema_name NOT IN ('information_schema', 'mysql', 'performance_schema', 'sys')
              ORDER BY schema_name
              """
          )
          return [row["database_name"] for row in cur.fetchall()]