Jump to content

Connect Leeroopedia MCP: Equip your AI agents to search best practices, build plans, verify code, diagnose failures, and look up hyperparameter defaults.

Implementation:Microsoft Autogen HandoffTermination Init

From Leeroopedia
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())

Related Pages

Implements Principle

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment