Implementation:Microsoft Autogen HandoffTermination Init
| Knowledge Sources | |
|---|---|
| Domains | Multi-Agent Systems, Termination Conditions, Human-in-the-Loop, Workflow Control |
| Last Updated | 2026-02-11 00:00 GMT |
Overview
Concrete tool for stopping swarm execution when a specific handoff target is detected, provided by Microsoft AutoGen.
Description
The HandoffTermination class implements the TerminationCondition protocol and monitors the message stream for HandoffMessage instances whose target field matches the configured target name. When a match is found, it returns a StopMessage containing a description of the detected handoff, including the source agent and target.
Internally, the class maintains a _terminated boolean flag. Once triggered, any subsequent call to the termination condition raises a TerminatedException. The reset() method clears the flag, allowing the condition to be reused in subsequent swarm runs. This is essential for the common pattern of pausing a swarm for human input and then resuming.
The class also implements the Component protocol with HandoffTerminationConfig, enabling declarative serialization and deserialization of the termination condition.
Usage
Import and instantiate HandoffTermination when building a swarm that needs to pause execution upon reaching a specific handoff target. The most common pattern is detecting handoffs to a "user" pseudo-agent for human-in-the-loop workflows. Combine with other termination conditions using the | operator for compound stopping logic.
Code Reference
Source Location
- Repository: Microsoft AutoGen
- File:
python/packages/autogen-agentchat/src/autogen_agentchat/conditions/_terminations.py(Lines 309-351)
Signature
class HandoffTermination(TerminationCondition, Component[HandoffTerminationConfig]):
def __init__(self, target: str) -> None: ...
@property
def terminated(self) -> bool: ...
async def __call__(
self, messages: Sequence[BaseAgentEvent | BaseChatMessage]
) -> StopMessage | None: ...
async def reset(self) -> None: ...
Import
from autogen_agentchat.conditions import HandoffTermination
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| target | str | Yes | The name of the agent whose HandoffMessage should trigger termination. Typically "user" for human-in-the-loop patterns, but can be any string matching a HandoffMessage target. |
Outputs
| Name | Type | Description |
|---|---|---|
| HandoffTermination instance | HandoffTermination | A termination condition that monitors for handoffs to the specified target. |
| __call__ result | None | Returns a StopMessage with content "Handoff to {target} from {source} detected." when triggered, or None if no matching handoff is found. |
| terminated | bool | Property indicating whether the condition has been triggered. |
Usage Examples
Basic Example
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import Swarm
from autogen_agentchat.conditions import HandoffTermination, MaxMessageTermination
from autogen_agentchat.messages import HandoffMessage
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
async def main() -> None:
model_client = OpenAIChatCompletionClient(model="gpt-4o")
agent = AssistantAgent(
"assistant",
model_client=model_client,
handoffs=["user"],
system_message=(
"You are a helpful assistant. If you need more information "
"from the user, hand off to user."
),
)
# Terminate when the agent hands off to "user" OR after 10 messages
termination = HandoffTermination(target="user") | MaxMessageTermination(10)
team = Swarm([agent], termination_condition=termination)
# First run: agent will hand off to user when it needs input
result = await Console(team.run_stream(task="Help me plan a vacation."))
print(f"Stop reason: {result.stop_reason}")
# Resume with user input after the handoff pause
result = await Console(
team.run_stream(
task=HandoffMessage(
source="user",
target="assistant",
content="I want to go to Japan for 2 weeks in April.",
)
)
)
asyncio.run(main())