Implementation:Microsoft Autogen Studio Web Dependencies

Knowledge Sources	Microsoft_Autogen
Domains	Dependency Injection, FastAPI, Web Services, Authentication
Last Updated	2026-02-11 17:00 GMT

Overview

Studio Web Dependencies module provides FastAPI dependency injection for database, WebSocket, team, and authentication managers with support for lite mode and full application modes.

Description

This module implements the dependency injection pattern for AutoGen Studio's web API, managing the lifecycle of core service managers including DatabaseManager, WebSocketManager, TeamManager, and AuthManager. It provides async dependency functions compatible with FastAPI's dependency injection system, handles initialization and cleanup of all managers, and supports two operational modes: full mode with database persistence and migrations, and lite mode with in-memory database and simplified configuration. The module also includes authentication integration, environment-based configuration detection, error handling for manager operations, and utilities for checking manager initialization status. It serves as the central coordination point for all service dependencies in the web application.

Usage

Use this module when setting up FastAPI routes that need access to database operations, WebSocket connections, team management, or authentication services, or when initializing the AutoGen Studio web application with proper dependency management.

Code Reference

Source Location

• Repository: Microsoft_Autogen
• File: python/packages/autogen-studio/autogenstudio/web/deps.py
• Lines: 1-300

Signature

# Dependency Providers
async def get_db() -> DatabaseManager
async def get_websocket_manager() -> WebSocketManager
async def get_team_manager() -> TeamManager
async def get_current_user(request: Request) -> str
async def get_managers() -> dict

# Manager Lifecycle
async def init_managers(
    database_uri: str,
    config_dir: str | Path,
    app_root: str | Path
) -> None

async def cleanup_managers() -> None

# Authentication
def init_auth_manager(config_dir: Path) -> AuthManager
async def register_auth_dependencies(
    app: FastAPI,
    auth_manager: AuthManager
) -> None

# Utilities
def is_lite_mode() -> bool
def get_manager_status() -> dict
def require_managers(*manager_names: str)

# Context Manager
@contextmanager
def get_db_context()

Import

from autogenstudio.web.deps import (
    get_db,
    get_websocket_manager,
    get_team_manager,
    get_current_user,
    init_managers,
    cleanup_managers
)
from fastapi import Depends

I/O Contract

Inputs

Name	Type	Required	Description
database_uri	str	Yes	Database connection URI (e.g., "sqlite:///./database.db")
config_dir	str or Path	Yes	Directory containing configuration files
app_root	str or Path	Yes	Root directory of the application
config_dir	Path	Yes	Configuration directory for authentication
app	FastAPI	Yes	FastAPI application instance for auth registration
auth_manager	AuthManager	Yes	Authentication manager to register
request	Request	Yes	FastAPI request object for user extraction
manager_names	str	Yes	Names of required managers (variadic)

Outputs

Name	Type	Description
get_db	DatabaseManager	Database manager instance for database operations
get_websocket_manager	WebSocketManager	WebSocket manager for real-time connections
get_team_manager	TeamManager	Team manager for agent team operations
get_current_user	str	Current authenticated user ID
get_managers	dict	Dictionary with all managers (db, connection, team)
is_lite_mode	bool	True if lite mode is enabled via environment variable
get_manager_status	dict	Dictionary showing initialization status of all managers
init_auth_manager	AuthManager	Initialized authentication manager instance
require_managers	Depends	FastAPI dependency that validates manager availability

Usage Examples

Basic FastAPI Route with Dependencies

from fastapi import APIRouter, Depends
from autogenstudio.web.deps import get_db, get_current_user
from autogenstudio.database import DatabaseManager

router = APIRouter()

@router.get("/teams")
async def list_teams(
    db: DatabaseManager = Depends(get_db),
    user_id: str = Depends(get_current_user)
):
    """List all teams for the current user."""
    response = db.get(TeamDB, filters={"user_id": user_id})
    if response.status:
        return {"teams": response.data}
    return {"error": response.message}

@router.get("/sessions")
async def list_sessions(
    db: DatabaseManager = Depends(get_db),
    user_id: str = Depends(get_current_user)
):
    """List all sessions for the current user."""
    response = db.get(SessionDB, filters={"user_id": user_id})
    return {"sessions": response.data if response.status else []}

Application Initialization

from fastapi import FastAPI
from pathlib import Path
from autogenstudio.web.deps import (
    init_managers,
    cleanup_managers,
    init_auth_manager,
    register_auth_dependencies
)

app = FastAPI(title="AutoGen Studio")

@app.on_event("startup")
async def startup_event():
    """Initialize all managers on application startup."""
    database_uri = "sqlite:///./autogen_studio.db"
    config_dir = Path("./config")
    app_root = Path("./")

    # Initialize core managers
    await init_managers(
        database_uri=database_uri,
        config_dir=config_dir,
        app_root=app_root
    )

    # Initialize authentication
    auth_manager = init_auth_manager(config_dir)
    await register_auth_dependencies(app, auth_manager)

    print("AutoGen Studio initialized successfully")

@app.on_event("shutdown")
async def shutdown_event():
    """Cleanup all managers on application shutdown."""
    await cleanup_managers()
    print("AutoGen Studio shutdown complete")

Using Multiple Managers

from fastapi import APIRouter, Depends, WebSocket
from autogenstudio.web.deps import (
    get_db,
    get_websocket_manager,
    get_team_manager,
    get_current_user
)

router = APIRouter()

@router.post("/run-team/{team_id}")
async def run_team(
    team_id: int,
    db=Depends(get_db),
    ws_manager=Depends(get_websocket_manager),
    team_manager=Depends(get_team_manager),
    user_id=Depends(get_current_user)
):
    """Execute a team with all necessary managers."""
    # Get team configuration
    team_response = db.get(TeamDB, filters={"id": team_id, "user_id": user_id})
    if not team_response.status:
        return {"error": "Team not found"}

    team_config = team_response.data[0]

    # Run team
    result = await team_manager.run_team(team_config)

    # Broadcast result via WebSocket
    await ws_manager.broadcast({"type": "team_result", "data": result})

    return {"result": result}

Context Manager for Database Operations

from autogenstudio.web.deps import get_db_context
from autogenstudio.datamodel.db import TeamDB

def create_multiple_teams(teams_data):
    """Create multiple teams in a transactional context."""
    with get_db_context() as db:
        created_teams = []
        for team_data in teams_data:
            team = TeamDB(**team_data)
            response = db.upsert(team)
            if response.status:
                created_teams.append(response.data)
            else:
                raise Exception(f"Failed to create team: {response.message}")
        return created_teams

Lite Mode Configuration

import os
from pathlib import Path
from autogenstudio.web.deps import init_managers, is_lite_mode

async def initialize_app():
    """Initialize application with environment-based configuration."""
    # Set lite mode via environment
    os.environ["AUTOGENSTUDIO_LITE_MODE"] = "true"
    os.environ["LITE_TEAM_FILE"] = "/path/to/team.json"
    os.environ["LITE_SESSION_NAME"] = "My Lite Session"

    # Check mode
    if is_lite_mode():
        print("Running in lite mode")
        database_uri = "sqlite:///:memory:"  # In-memory database
    else:
        print("Running in full mode")
        database_uri = "sqlite:///./autogen_studio.db"

    # Initialize with appropriate configuration
    await init_managers(
        database_uri=database_uri,
        config_dir=Path("./config"),
        app_root=Path("./")
    )

Manager Status Checking

from fastapi import APIRouter
from autogenstudio.web.deps import get_manager_status, require_managers
from fastapi import Depends

router = APIRouter()

@router.get("/health")
async def health_check():
    """Check the health status of all managers."""
    status = get_manager_status()

    all_healthy = all(status.values())

    return {
        "status": "healthy" if all_healthy else "degraded",
        "managers": status,
        "details": {
            "database": "✓" if status["database_manager"] else "✗",
            "websocket": "✓" if status["websocket_manager"] else "✗",
            "team": "✓" if status["team_manager"] else "✗",
            "auth": "✓" if status["auth_manager"] else "✗",
        }
    }

@router.get("/protected-route", dependencies=[require_managers("database", "team")])
async def protected_route():
    """Route that requires specific managers to be initialized."""
    return {"message": "All required managers are available"}

Custom Authentication Setup

import os
from pathlib import Path
from autogenstudio.web.deps import init_auth_manager
from fastapi import FastAPI, Request

# Disable authentication via environment
os.environ["AUTOGENSTUDIO_AUTH_DISABLED"] = "true"

# Or provide auth config file
os.environ["AUTOGENSTUDIO_AUTH_CONFIG"] = "/path/to/auth_config.yaml"

# Initialize auth manager
config_dir = Path("./config")
auth_manager = init_auth_manager(config_dir)

# Check auth configuration
print(f"Auth type: {auth_manager.config.type}")

# Use in FastAPI app
app = FastAPI()

@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    """Custom authentication middleware."""
    # Auth manager is available via app.state
    if hasattr(request.app.state, "auth_manager"):
        # Authentication logic here
        pass
    response = await call_next(request)
    return response

Related Pages

• DatabaseManager - Database operations manager
• WebSocketManager - WebSocket connection manager
• TeamManager - Agent team management
• AuthManager - Authentication management
• FastAPI Routes - API endpoint definitions