autogen_agentchat

High-level Python API for building conversational multi-agent applications

The autogen_agentchat package provides high-level abstractions for building conversational agents and teams.

Agent Classes

AssistantAgent

AI-powered assistant agent with tool use and handoff capabilities.

from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

assistant = AssistantAgent(
    name="assistant",
    model_client=OpenAIChatCompletionClient(model="gpt-4"),
    tools=[calculator_tool],
    handoffs=["supervisor"],
    system_message="You are a helpful assistant.",
    description="General purpose assistant"
)

# Run the agent
result = await assistant.run(task="Calculate 25 * 4")
print(result.messages)

name · str · required

Unique name for the agent

model_client · ChatCompletionClient · required

LLM client for generating responses

tools · List[Tool] | None

Tools the agent can use

handoffs · List[Handoff | str] | None

Other agents this agent can transfer control to

system_message · str | None

System prompt for the agent

description · str · required

Description of the agent’s capabilities

model_context · ChatCompletionContext | None

Context manager for conversation history

memory · List[Memory] | None

Memory modules for the agent

reflect_on_tool_use · bool

Whether to reflect on tool execution results (default: False)

max_tool_iterations · int

Maximum tool call iterations per turn (default: 1)

Methods

on_messages · async method

Process messages and return a response

response = await assistant.on_messages(
    messages=[TextMessage(content="Hello", source="user")],
    cancellation_token=token
)

messages · Sequence[ChatMessage] · required

Conversation messages

cancellation_token · CancellationToken | None

Token to cancel the operation

Returns: Response with the agent’s reply

on_messages_stream · async generator

Stream messages and events as they’re generated

async for message in assistant.on_messages_stream(messages):
    if isinstance(message, TextMessage):
        print(message.content)

run · async method

Run the agent with a task

result = await assistant.run(
    task="Write a poem",
    termination_condition=MaxMessageTermination(10)
)

task · str | ChatMessage · required

Initial task or message

termination_condition · TerminationCondition | None

Condition to stop execution

Returns: TaskResult with conversation history

UserProxyAgent

Agent representing a human user with input capabilities.

from autogen_agentchat.agents import UserProxyAgent

user = UserProxyAgent(
    name="user",
    description="Human user"
)

# Agent will prompt for input during execution
result = await user.run(task="Enter your message")

name · str · required

Agent name

description · str · required

Agent description

input_func · Callable | None

Custom function for getting user input

CodeExecutorAgent

Agent that executes code in a sandboxed environment.

from autogen_agentchat.agents import CodeExecutorAgent
from autogen_core.code_executor import DockerCommandLineCodeExecutor

executor_agent = CodeExecutorAgent(
    name="executor",
    code_executor=DockerCommandLineCodeExecutor(),
    description="Executes Python code"
)

name · str · required

Agent name

code_executor · CodeExecutor · required

Code execution backend

approval_func · ApprovalFuncType | None

Function to approve code before execution

description · str · required

Agent description

SocietyOfMindAgent

Meta-agent that manages an internal team of agents.

from autogen_agentchat.agents import SocietyOfMindAgent
from autogen_agentchat.teams import RoundRobinGroupChat

inner_team = RoundRobinGroupChat([agent1, agent2])

som_agent = SocietyOfMindAgent(
    name="team",
    group_chat=inner_team,
    description="Coordinated team of specialists"
)

name · str · required

Agent name

group_chat · BaseGroupChat · required

Internal team configuration

description · str · required

Agent description

MessageFilterAgent

Agent that filters messages based on configurable criteria.

from autogen_agentchat.agents import MessageFilterAgent, PerSourceFilter

filter_agent = MessageFilterAgent(
    name="filter",
    filter_config=PerSourceFilter(max_messages_per_source=3),
    description="Limits messages per source"
)

name · str · required

Agent name

filter_config · MessageFilterConfig · required

Filter configuration

description · str · required

Agent description

BaseChatAgent

Base class for all chat agents in the framework.

from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.messages import ChatMessage, Response

class CustomAgent(BaseChatAgent):
    async def on_messages(
        self,
        messages: Sequence[ChatMessage],
        cancellation_token: CancellationToken | None = None
    ) -> Response:
        # Custom message handling
        return Response(
            chat_message=TextMessage(content="Response", source=self.name)
        )

name · str · required

Agent name

description · str · required

Agent description

Team Classes

RoundRobinGroupChat

Team where agents take turns in a fixed order.

from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination

team = RoundRobinGroupChat(
    participants=[agent1, agent2, agent3],
    termination_condition=MaxMessageTermination(20)
)

result = await team.run(task="Solve this problem collaboratively")

participants · List[BaseChatAgent] · required

List of agents in the team

termination_condition · TerminationCondition | None

Condition to stop the conversation

SelectorGroupChat

Team with dynamic agent selection using an LLM.

from autogen_agentchat.teams import SelectorGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient

team = SelectorGroupChat(
    participants=[researcher, writer, reviewer],
    model_client=OpenAIChatCompletionClient(model="gpt-4"),
    termination_condition=MaxMessageTermination(30)
)

result = await team.run(task="Research and write an article")

participants · List[BaseChatAgent] · required

Agents available for selection

model_client · ChatCompletionClient · required

LLM for selecting the next speaker

selector_prompt · str | None

Custom prompt for agent selection

allow_repeated_speaker · bool

Whether the same agent can speak twice in a row (default: False)

Swarm

Dynamic team with agent handoffs and context transfer.

from autogen_agentchat.teams import Swarm

team = Swarm(
    participants=[triage_agent, specialist1, specialist2],
    termination_condition=TextMentionTermination("TERMINATE")
)

# Agents can transfer control using handoffs
result = await team.run(task="Handle customer request")

participants · List[BaseChatAgent] · required

Agents in the swarm

termination_condition · TerminationCondition | None

Condition to stop execution

MagenticOneGroupChat

Specialized team for the Magentic-One architecture.

from autogen_agentchat.teams import MagenticOneGroupChat

team = MagenticOneGroupChat(
    participants=[orchestrator, web_surfer, file_surfer, coder, executor],
    max_turns=50
)

participants · List[BaseChatAgent] · required

Specialized agents for the Magentic-One pattern

max_turns · int

Maximum conversation turns

GraphFlow

Graph-based team with custom agent flow.

from autogen_agentchat.teams import DiGraphBuilder, GraphFlow

# Build agent graph
builder = DiGraphBuilder()
builder.add_node("start", agent1)
builder.add_node("process", agent2)
builder.add_node("end", agent3)
builder.add_edge("start", "process")
builder.add_edge("process", "end")

graph = builder.build()
team = GraphFlow(graph=graph, entry_point="start")

result = await team.run(task="Process through pipeline")

graph · DiGraph · required

Directed graph of agents

entry_point · str · required

Starting node identifier

BaseGroupChat

Base class for all team implementations.

from autogen_agentchat.teams import BaseGroupChat

class CustomTeam(BaseGroupChat):
    async def run(
        self,
        task: str | ChatMessage,
        termination_condition: TerminationCondition | None = None
    ) -> TaskResult:
        # Custom team logic
        pass

Message Types

TextMessage

Plain text message.

from autogen_agentchat.messages import TextMessage

message = TextMessage(
    content="Hello, how can I help?",
    source="assistant"
)

content · str · required

Message text content

source · str · required

Name of the agent that created the message

HandoffMessage

Message indicating agent handoff.

from autogen_agentchat.messages import HandoffMessage

handoff = HandoffMessage(
    target="specialist",
    content="Transferring to specialist for details",
    source="triage"
)

target · str · required

Name of the agent to transfer to

content · str · required

Context for the handoff

ToolCallSummaryMessage

Summary of tool execution results.

from autogen_agentchat.messages import ToolCallSummaryMessage

summary = ToolCallSummaryMessage(
    content="Executed calculator: 25 * 4 = 100",
    source="assistant"
)

StructuredMessage

Message with structured data.

from autogen_agentchat.messages import StructuredMessage
from pydantic import BaseModel

class Report(BaseModel):
    title: str
    summary: str
    score: float

message = StructuredMessage[
    content=Report(title="Analysis", summary="...", score=0.95),
    source="analyst"
]

Events

ToolCallRequestEvent

Event emitted when a tool is called.

from autogen_agentchat.messages import ToolCallRequestEvent

# Emitted during streaming
event = ToolCallRequestEvent(
    source="assistant",
    tool_calls=[FunctionCall(id="1", name="calc", arguments={})]
)

ToolCallExecutionEvent

Event emitted after tool execution.

from autogen_agentchat.messages import ToolCallExecutionEvent

event = ToolCallExecutionEvent(
    source="assistant",
    tool_call_id="call_123",
    result="100"
)

ModelClientStreamingChunkEvent

Streaming chunk from the model.

from autogen_agentchat.messages import ModelClientStreamingChunkEvent

# Emitted during streaming responses
event = ModelClientStreamingChunkEvent(
    source="assistant",
    content="Hello"
)

ThoughtEvent

Agent’s internal reasoning.

from autogen_agentchat.messages import ThoughtEvent

event = ThoughtEvent(
    source="assistant",
    thought="I should use the calculator tool for this"
)

Response Types

Response

Agent response to messages.

from autogen_agentchat.base import Response

response = Response(
    chat_message=TextMessage(content="Done", source="assistant"),
    inner_messages=[event1, event2]
)

chat_message · ChatMessage

The final response message

inner_messages · List[BaseAgentEvent]

Intermediate events and messages

TaskResult

Result of running an agent or team.

result = await agent.run(task="Write code")

print(result.messages)  # All messages
print(result.stop_reason)  # Why it stopped

messages · List[ChatMessage]

Complete conversation history

stop_reason · str | None

Reason for termination

Termination Conditions

MaxMessageTermination

Stop after a maximum number of messages.

from autogen_agentchat.conditions import MaxMessageTermination

condition = MaxMessageTermination(max_messages=20)

TextMentionTermination

Stop when specific text is mentioned.

from autogen_agentchat.conditions import TextMentionTermination

condition = TextMentionTermination("TERMINATE")

StopMessageTermination

Stop on a specific message type.

from autogen_agentchat.conditions import StopMessageTermination

condition = StopMessageTermination()

TimeoutTermination

Stop after a time limit.

from autogen_agentchat.conditions import TimeoutTermination

condition = TimeoutTermination(timeout_seconds=300)

TokenUsageTermination

Stop after token budget is exhausted.

from autogen_agentchat.conditions import TokenUsageTermination

condition = TokenUsageTermination(max_tokens=10000)

State Management

AssistantAgentState

Serializable state for assistant agents.

# Save state
state = await assistant.save_state()

# Load state
await assistant.load_state(state)

Logging

from autogen_agentchat import TRACE_LOGGER_NAME, EVENT_LOGGER_NAME
import logging

# Configure logging
logging.getLogger(TRACE_LOGGER_NAME).setLevel(logging.DEBUG)
logging.getLogger(EVENT_LOGGER_NAME).setLevel(logging.INFO)

TRACE_LOGGER_NAME · str

Logger name: "autogen_agentchat"

EVENT_LOGGER_NAME · str

Logger name: "autogen_agentchat.events"

See Also

• autogen_core - Core agent runtime and messaging
• autogen_ext - Model clients and extensions