Building Custom Team Tools with TeamTool for Complex Workflows in AutoGen

TeamTool wraps a BaseGroupChat as a callable instrument, allowing an autonomous agent to orchestrate entire multi-agent teams through a single function call while maintaining state persistence and streaming capabilities.

Building sophisticated multi-agent systems in the microsoft/autogen repository often requires hierarchical delegation patterns. By implementing custom team tools with TeamTool, you can encapsulate complex agent orchestration—such as review-revise loops or specialized processing pipelines—into reusable, serializable components that higher-level agents invoke like standard functions.

Understanding the TeamTool Architecture

TeamTool is implemented as a specialized subclass within the agentchat tool hierarchy. It transforms a group chat instance into a discrete tool that conforms to the component model, enabling declarative configuration and runtime execution.

Core Components and Source Files

The architecture follows a layered composition pattern across three primary source files:

• TaskRunnerTool (_task_runner_tool.py): The abstract base class residing in python/packages/autogen-agentchat/src/autogen_agentchat/tools/_task_runner_tool.py provides the foundation for tools that execute tasks on BaseGroupChat or BaseChatAgent instances. It handles argument parsing, streaming infrastructure, and conversion of TaskResult objects to string outputs.

• TeamTool (_team.py): Located in python/packages/autogen-agentchat/src/autogen_agentchat/tools/_team.py, this concrete implementation binds a BaseGroupChat instance to the tool interface. It overrides component_provider_override to register the provider string "autogen_agentchat.tools.TeamTool" for the configuration system.

• RoundRobinGroupChat (_round_robin_group_chat.py): The default team implementation in python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_round_robin_group_chat.py manages participant scheduling and termination conditions, serving as the execution engine wrapped by TeamTool.

Configuration Schema and Serialization

TeamTool leverages pydantic models for declarative configuration through TeamToolConfig:

class TeamToolConfig(BaseModel):
    name: str
    description: str
    team: ComponentModel               # serialized BaseGroupChat

    return_value_as_last_message: bool = False

When serializing via dump_component(), the underlying team is recursively dumped using team.dump_component() (lines 22-23 of _team.py). The inverse operation, load_component, reconstructs the team through BaseGroupChat.load_component, enabling full configuration-driven instantiation without imperative setup code.

Critical Runtime Constraints

Both TeamTool and its sibling AgentTool enforce a strict concurrency rule (lines 28-33 of _team.py): parallel tool calls must be disabled in the model client configuration. Concurrent execution would corrupt the internal state of the wrapped group chat.

Configure your model client explicitly:

from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(
    model="gpt-4o", 
    parallel_tool_calls=False  # Required for TeamTool

)

Implementing State Persistence and Streaming

TeamTool inherits robust lifecycle management capabilities from TaskRunnerTool, supporting both checkpointing and real-time output consumption.

Checkpointing Workflow State

The tool delegates state operations directly to the wrapped team implementation:

async def save_state_json(self) -> Mapping[str, Any]:
    return await self._task_runner.save_state()

This delegation means the entire group chat message thread, speaker index, and termination state can be serialized via save_state_json() and later restored through load_state_json(). This capability is essential for fault-tolerant workflows and long-running processes that must survive system restarts.

Real-Time Streaming Output

TaskRunnerTool.run_stream yields every BaseAgentEvent and BaseChatMessage produced by the nested team, concluding with a TaskResult. TeamTool adds no intermediate buffering, meaning any streaming consumer—whether a FastAPI endpoint, Chainlit interface, or the built-in Console UI—receives incremental updates as the internal team generates them.

Step-by-Step: Building a Complex Multi-Agent Workflow

The following implementation demonstrates a write-review-summarize pipeline wrapped as a single callable tool. This pattern illustrates how to compose specialist agents into a RoundRobinGroupChat, enforce termination conditions, and expose the entire workflow through TeamTool.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import SourceMatchTermination
from autogen_agentchat.tools import TeamTool
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.ui import Console
import asyncio

# 1. Configure specialist agents

base_client = OpenAIChatCompletionClient(model="gpt-4o")

writer = AssistantAgent(
    name="Writer",
    model_client=base_client,
    system_message="You are a creative writer. Produce vivid short stories.",
)

reviewer = AssistantAgent(
    name="Reviewer",
    model_client=base_client,
    system_message="Critique stories for factual and logical consistency.",
)

summarizer = AssistantAgent(
    name="Summarizer",
    model_client=base_client,
    system_message="Synthesize the story and critique into a polished final version.",
)

# 2. Compose the team with termination logic

termination = SourceMatchTermination(sources=["Summarizer"])
team = RoundRobinGroupChat(
    participants=[writer, reviewer, summarizer],
    termination_condition=termination,
)

# 3. Wrap as a TeamTool

writing_tool = TeamTool(
    team=team,
    name="WritingPipeline",
    description="Executes write-review-summarize workflow and returns final story",
    return_value_as_last_message=True,
)

# 4. Create top-level agent with disabled parallel tool calls

main_client = OpenAIChatCompletionClient(
    model="gpt-4o", 
    parallel_tool_calls=False
)

main_assistant = AssistantAgent(
    name="MainAssistant",
    model_client=main_client,
    tools=[writing_tool],
    system_message="Delegate story requests to the WritingPipeline tool.",
)

# 5. Execute

async def demo():
    await Console(
        main_assistant.run_stream(
            task="Write a sci-fi story about a robot learning to love."
        )
    )

if __name__ == "__main__":
    asyncio.run(demo())

State Save and Load Pattern

Preserve workflow progress across sessions:

# After execution

state = await writing_tool.save_state_json()

# Later restoration

new_team = RoundRobinGroupChat(
    participants=[writer, reviewer, summarizer],
    termination_condition=termination
)
restored_tool = TeamTool(
    team=new_team,
    name="WritingPipeline",
    description="Restored workflow",
    return_value_as_last_message=True
)
await restored_tool.load_state_json(state)

Production Patterns for TeamTool

FastAPI Streaming Endpoints

Expose TeamTool workflows through HTTP APIs with Server-Sent Events:

from fastapi import FastAPI
from autogen_agentchat.ui import StreamingResponseAdapter

app = FastAPI()

@app.post("/story")
async def generate_story(prompt: str):
    async_gen = main_assistant.run_stream(task=prompt)
    return StreamingResponseAdapter(async_gen)

This pattern leverages the native run_stream generator to push incremental updates to clients without buffering the entire multi-agent conversation in memory.

Summary

• Component Architecture: TeamTool inherits from TaskRunnerTool and resides in python/packages/autogen-agentchat/src/autogen_agentchat/tools/_team.py, providing a thin wrapper around BaseGroupChat instances.

• Configuration-Driven: Uses TeamToolConfig for serialization, supporting dump_component() and load_component() for declarative pipeline reconstruction.

• Concurrency Safety: Requires parallel_tool_calls=False in the model client to prevent state corruption during tool execution.

• State Management: Implements save_state_json() and load_state_json() to checkpoint entire group chat histories and speaker states.

• Streaming Native: Supports run_stream() for real-time event consumption, compatible with FastAPI and async UI frameworks.

Frequently Asked Questions

What is TeamTool in AutoGen?

TeamTool is a component-based utility class in the microsoft/autogen repository that treats an entire multi-agent team (a BaseGroupChat implementation) as a single callable tool. According to the source code in _team.py, it enables agents to invoke complex workflows—like round-robin discussions with multiple specialists—through standard function-calling interfaces.

Why must parallel tool calls be disabled when using TeamTool?

The source code explicitly guards against parallel execution (lines 28-33 of _team.py) because concurrent invocations would corrupt the internal message buffer and speaker state of the wrapped group chat. You must instantiate your OpenAIChatCompletionClient or equivalent with parallel_tool_calls=False to ensure sequential, state-safe execution.

How does TeamTool handle state persistence?

TeamTool inherits save_state_json and load_state_json from TaskRunnerTool, which delegate to the underlying team's state management methods. This allows you to serialize the complete conversation history, termination conditions, and round-robin position to JSON, then restore exact execution context later—critical for resuming long-running workflows.

Can TeamTool stream intermediate results?

Yes. TeamTool utilizes the run_stream method from its parent class TaskRunnerTool (defined in _task_runner_tool.py), yielding every BaseAgentEvent and BaseChatMessage as the internal team generates them. This streaming capability integrates natively with FastAPI's StreamingResponse or the built-in Console UI for real-time monitoring of complex multi-agent workflows.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}