Implementation:Microsoft Autogen DatabaseManager Operations

Knowledge Sources	Microsoft AutoGen
Domains	Agent Frameworks, Database Management, CRUD Operations, Configuration Persistence
Last Updated	2026-02-11 00:00 GMT

Overview

Concrete tool for persisting and retrieving multi-agent team configurations in a relational database provided by the autogenstudio package.

Description

The DatabaseManager class provides CRUD operations for AutoGen Studio entities (teams, sessions, runs, messages, settings) backed by SQLModel/SQLAlchemy. It supports both SQLite and PostgreSQL database engines. The upsert() method creates or updates any entity based on ID existence. The get() method retrieves entities with optional filters and ordering. The import_team() method loads team configurations from JSON/YAML files or dictionaries and stores them in the database with user association. The delete() method removes entities with integrity constraint checking. A SchemaManager handles Alembic-based database migrations.

Usage

Use DatabaseManager when you need to store, retrieve, or manage team configurations and associated data (sessions, runs, messages) in AutoGen Studio. It is the primary data access layer for the studio web application.

Code Reference

Source Location

• Repository: Microsoft AutoGen
• File: python/packages/autogen-studio/autogenstudio/database/db_manager.py (L165-L310)

Signature

class DatabaseManager:

    def __init__(self, engine_uri: str, base_dir: Optional[Union[str, Path]] = None) -> None: ...

    def upsert(self, model: BaseDBModel, return_json: bool = True) -> Response: ...

    def get(
        self,
        model_class: type[BaseDBModel],
        filters: dict | None = None,
        return_json: bool = False,
        order: str = "desc",
    ) -> Response: ...

    def delete(self, model_class: type[BaseDBModel], filters: dict | None = None) -> Response: ...

    async def import_team(
        self, team_config: Union[str, Path, dict], user_id: str, check_exists: bool = False
    ) -> Response: ...

    async def import_teams_from_directory(
        self, directory: Union[str, Path], user_id: str, check_exists: bool = False
    ) -> Response: ...

    def initialize_database(self, auto_upgrade: bool = False, force_init_alembic: bool = True) -> Response: ...

    def reset_db(self, recreate_tables: bool = True) -> Response: ...

Import

from autogenstudio.database.db_manager import DatabaseManager

I/O Contract

Inputs (constructor)

Name	Type	Required	Description
engine_uri	str	Yes	Database connection URI (e.g., "sqlite:///path/to/db.sqlite3" or "postgresql://user:pass@host/db")
base_dir	Optional[Union[str, Path]]	No (default None)	Base directory for Alembic migration files

Inputs (upsert)

Name	Type	Required	Description
model	BaseDBModel	Yes	The SQLModel instance to create or update
return_json	bool	No (default True)	If True, returns the model as a dictionary; if False, returns the SQLModel instance

Inputs (get)

Name	Type	Required	Description
model_class	type[BaseDBModel]	Yes	The SQLModel class to query (e.g., Team, Session, Run)
filters	dict or None	No (default None)	Dictionary of column-value pairs to filter results (e.g., {"user_id": "user123"})
return_json	bool	No (default False)	If True, converts results to dictionaries; if False, returns SQLModel instances
order	str	No (default "desc")	Sort order for created_at column ("asc" or "desc")

Inputs (import_team)

Name	Type	Required	Description
team_config	Union[str, Path, dict]	Yes	Team configuration as a file path (JSON/YAML) or a dictionary
user_id	str	Yes	User ID to associate the imported team with
check_exists	bool	No (default False)	If True, checks for an identical existing team before importing

Outputs

Name	Type	Description
Response	Response	Contains status (bool), message (str), and data (dict, list, or SQLModel instance depending on operation)

Usage Examples

Basic Example

import asyncio
from autogenstudio.database.db_manager import DatabaseManager
from autogenstudio.datamodel import Team

# Initialize database manager with SQLite
db = DatabaseManager(engine_uri="sqlite:///autogenstudio.db")
db.initialize_database(auto_upgrade=True)

# Import a team from a JSON file
async def import_example():
    result = await db.import_team(
        team_config="./my_team.json",
        user_id="user_001",
        check_exists=True,
    )
    print(f"Import status: {result.status}, message: {result.message}")
    if result.data:
        print(f"Team ID: {result.data.get('id')}")

asyncio.run(import_example())

# Retrieve all teams for a user
response = db.get(Team, filters={"user_id": "user_001"}, return_json=True, order="desc")
if response.status:
    for team in response.data:
        print(f"Team: {team['id']}, created: {team['created_at']}")

# Update a team (upsert)
team_instance = db.get(Team, filters={"user_id": "user_001"}, return_json=False).data[0]
team_instance.component = {"provider": "updated_config", "config": {}}
update_response = db.upsert(team_instance)
print(f"Update: {update_response.message}")

# Delete a team
delete_response = db.delete(Team, filters={"id": team_instance.id})
print(f"Delete: {delete_response.message}")

Related Pages

Implements Principle

• Principle:Microsoft_Autogen_Team_Persistence
• Environment:Microsoft_Autogen_Studio_Server_Environment