# Building Custom Team Tools with TeamTool for Complex Workflows in AutoGen

> Build custom team tools with TeamTool for complex workflows in AutoGen. Orchestrate multi-agent teams with state persistence and streaming via a single function call.

- Repository: [Microsoft/autogen](https://github.com/microsoft/autogen)
- Tags: how-to-guide
- Published: 2026-03-07

---

**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](https://github.com/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`](https://github.com/microsoft/autogen/blob/main/_task_runner_tool.py)): The abstract base class residing in [`python/packages/autogen-agentchat/src/autogen_agentchat/tools/_task_runner_tool.py`](https://github.com/microsoft/autogen/blob/main/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`](https://github.com/microsoft/autogen/blob/main/_team.py)): Located in [`python/packages/autogen-agentchat/src/autogen_agentchat/tools/_team.py`](https://github.com/microsoft/autogen/blob/main/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`](https://github.com/microsoft/autogen/blob/main/_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`](https://github.com/microsoft/autogen/blob/main/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`:

```python
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`](https://github.com/microsoft/autogen/blob/main/_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`](https://github.com/microsoft/autogen/blob/main/_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:

```python
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:

```python
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`.

```python
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:

```python

# 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:

```python
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`](https://github.com/microsoft/autogen/blob/main/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`](https://github.com/microsoft/autogen/blob/main/_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`](https://github.com/microsoft/autogen/blob/main/_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`](https://github.com/microsoft/autogen/blob/main/_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.