Implementation:Microsoft Autogen TeamManager LiteStudio

Knowledge Sources	Microsoft AutoGen
Domains	Agent Frameworks, Deployment, Streaming Execution, Environment Configuration
Last Updated	2026-02-11 00:00 GMT

Overview

Concrete tool for deploying and executing multi-agent teams programmatically and as lightweight web services provided by the autogenstudio package.

Description

The TeamManager class manages team operations including loading configurations from files and running teams with streaming output. The run_stream() method accepts a task and team configuration, creates the team via BaseGroupChat.load_component(), wires up UserProxyAgent input functions, sets up an LLM event logger, and yields agent messages and a final TeamResult through an async generator. The LiteStudio class provides a lightweight web interface for single-team deployment, handling team serialization from multiple input formats, environment setup (in-memory database, disabled auth), and uvicorn server management in both foreground and background modes. It supports context manager usage for automatic server lifecycle management.

Usage

Use TeamManager when you need to execute teams programmatically with streaming output, such as in application backends, scripts, or integration tests. Use LiteStudio when you need a minimal web interface for demonstrating or testing a single team configuration.

Code Reference

Source Location

Repository: Microsoft AutoGen
File: python/packages/autogen-studio/autogenstudio/teammanager/teammanager.py (L109-L147 for run_stream)
File: python/packages/autogen-studio/autogenstudio/lite/studio.py (L171-L208 for start)

Signature

# TeamManager.run_stream (teammanager.py L109-L147)
class TeamManager:
    def __init__(self) -> None: ...

    @staticmethod
    async def load_from_file(path: Union[str, Path]) -> Any: ...

    @staticmethod
    async def load_from_directory(directory: Union[str, Path]) -> List[Any]: ...

    async def run_stream(
        self,
        task: str | BaseChatMessage | Sequence[BaseChatMessage] | None,
        team_config: Union[str, Path, Dict[str, Any], ComponentModel],
        input_func: Optional[InputFuncType] = None,
        cancellation_token: Optional[CancellationToken] = None,
        env_vars: Optional[List[EnvironmentVariable]] = None,
    ) -> AsyncGenerator[
        Union[BaseAgentEvent | BaseChatMessage | LLMCallEvent, BaseChatMessage, TeamResult],
        None,
    ]: ...

    async def run(
        self,
        task: str | BaseChatMessage | Sequence[BaseChatMessage] | None,
        team_config: Union[str, Path, Dict[str, Any], ComponentModel],
        input_func: Optional[InputFuncType] = None,
        cancellation_token: Optional[CancellationToken] = None,
        env_vars: Optional[List[EnvironmentVariable]] = None,
    ) -> TeamResult: ...

# LiteStudio.start (studio.py L171-L208)
class LiteStudio:
    def __init__(
        self,
        team: Union[str, Path, Dict[str, Any], ComponentModel, None] = None,
        host: str = "127.0.0.1",
        port: int = 8080,
        session_name: str = "Lite Session",
        auto_open: bool = True,
    ) -> None: ...

    def start(self, background: bool = False) -> None: ...

    def stop(self) -> None: ...

    def __enter__(self) -> "LiteStudio": ...

    def __exit__(self, exc_type, exc_val, exc_tb) -> None: ...

    @classmethod
    def shutdown_port(cls, port: int) -> None: ...

Import

from autogenstudio.teammanager import TeamManager
from autogenstudio.lite import LiteStudio

I/O Contract

Inputs (TeamManager.run_stream)

Name	Type	Required	Description
task	BaseChatMessage \| Sequence[BaseChatMessage] \| None	Yes	The task to execute (text prompt or structured chat message(s))
team_config	Union[str, Path, Dict[str, Any], ComponentModel]	Yes	Team configuration as file path (JSON/YAML), dictionary, or ComponentModel
input_func	Optional[InputFuncType]	No (default None)	Callback function for human-in-the-loop input; wired to UserProxyAgent if present
cancellation_token	Optional[CancellationToken]	No (default None)	Token for cooperative cancellation of the running team
env_vars	Optional[List[EnvironmentVariable]]	No (default None)	List of environment variables to inject (e.g., API keys) before team instantiation

Inputs (LiteStudio constructor)

Name	Type	Required	Description
team	Union[str, Path, Dict, ComponentModel, None]	No (default None)	Team configuration; None creates a default team with calculator tool
host	str	No (default "127.0.0.1")	Host address to bind the server to
port	int	No (default 8080)	Port number for the server
session_name	str	No (default "Lite Session")	Name for the auto-created session
auto_open	bool	No (default True)	Whether to auto-open the browser to /lite endpoint

Inputs (LiteStudio.start)

Name	Type	Required	Description
background	bool	No (default False)	If True, run server in a daemon background thread; if False, run in foreground (blocking)

Outputs (TeamManager.run_stream)

Name	Type	Description
AsyncGenerator	AsyncGenerator[Union[BaseAgentEvent, BaseChatMessage, LLMCallEvent, TeamResult], None]	Yields individual agent messages, LLM call events, and a final TeamResult containing task_result, usage, and duration

Outputs (LiteStudio.start)

Name	Type	Description
(side effect)	None	Starts uvicorn server; blocks if background=False, returns immediately if background=True

Usage Examples

Basic Example: TeamManager Streaming

import asyncio
from autogenstudio.teammanager import TeamManager
from autogenstudio.datamodel.types import EnvironmentVariable

async def run_team():
    manager = TeamManager()

    # Run with streaming output
    async for message in manager.run_stream(
        task="What is the square root of 144?",
        team_config="./my_team.json",
        env_vars=[
            EnvironmentVariable(name="OPENAI_API_KEY", value="sk-..."),
        ],
    ):
        # Process each message as it arrives
        if hasattr(message, "content"):
            print(f"[{getattr(message, 'source', 'system')}]: {message.content}")
        elif hasattr(message, "task_result"):
            print(f"Completed in {message.duration:.2f}s")
            print(f"Stop reason: {message.task_result.stop_reason}")

asyncio.run(run_team())

Basic Example: LiteStudio Deployment

from autogenstudio.lite import LiteStudio

# Start with a team file (blocking, for scripts)
studio = LiteStudio(team="./research_team.json", port=8085)
studio.start()  # Blocks until Ctrl+C

# Start in background (for notebooks or applications)
studio = LiteStudio(team="./research_team.json", port=8085, auto_open=False)
studio.start(background=True)
# ... do other work ...
studio.stop()

# Use as context manager
with LiteStudio(team="./my_team.json", port=8090) as studio:
    # Server is running in background
    import time
    time.sleep(60)  # Server available for 60 seconds
# Server automatically stopped

# Start with default team (no arguments needed)
studio = LiteStudio()
studio.start()

Related Pages

Implements Principle

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment