From 792258f4969517d7770eb2213ce53bb6233aafde Mon Sep 17 00:00:00 2001 From: Oxygen <1391083091@qq.com> Date: Fri, 5 Jun 2026 23:36:34 +0800 Subject: [PATCH] feat: add automatic back-handoff support for orchestrating agents When an agent is reached via a handoff, it can now automatically hand back to the originating agent upon completion. Controlled by the auto_handoff_back parameter on the Handoff/handoff() config. This enables orchestration patterns where a main agent delegates to specialist agents and expects results back without needing explicit circular handoff configurations. Changes: - Add auto_handoff_back: bool = False to Handoff class - Pass through auto_handoff_back in handoff() factory function - Extend NextStepHandoff to carry auto_handoff_back and originating_agent - Track handoff chain in both streaming and non-streaming run loops - Auto-handoff back when child agent produces final output - Add comprehensive tests for the feature Closes #847 Co-Authored-By: Claude Opus 4.8 --- src/agents/handoffs/__init__.py | 18 ++ src/agents/run.py | 39 +++- src/agents/run_internal/run_loop.py | 56 +++++- src/agents/run_internal/run_steps.py | 7 + src/agents/run_internal/turn_resolution.py | 6 +- tests/test_auto_handoff_back.py | 207 +++++++++++++++++++++ 6 files changed, 328 insertions(+), 5 deletions(-) create mode 100644 tests/test_auto_handoff_back.py diff --git a/src/agents/handoffs/__init__.py b/src/agents/handoffs/__init__.py index b318414ce8..f4f603e469 100644 --- a/src/agents/handoffs/__init__.py +++ b/src/agents/handoffs/__init__.py @@ -160,6 +160,16 @@ class Handoff(Generic[TContext, TAgent]): context or state. """ + auto_handoff_back: bool = False + """Whether the target agent should automatically handoff back to the originating agent when it + finishes its task. + + When True, after the child agent completes its work (produces a final output), control + automatically returns to the agent that initiated the handoff. This enables orchestration + patterns where a main orchestrator agent delegates to specialist agents and expects results + back without needing explicit circular handoff configurations. + """ + _agent_ref: weakref.ReferenceType[AgentBase[Any]] | None = field( default=None, init=False, repr=False ) @@ -188,6 +198,7 @@ def handoff( tool_description_override: str | None = None, input_filter: Callable[[HandoffInputData], HandoffInputData] | None = None, nest_handoff_history: bool | None = None, + auto_handoff_back: bool = False, is_enabled: bool | Callable[[RunContextWrapper[Any], Agent[Any]], MaybeAwaitable[bool]] = True, ) -> Handoff[TContext, Agent[TContext]]: ... @@ -202,6 +213,7 @@ def handoff( tool_name_override: str | None = None, input_filter: Callable[[HandoffInputData], HandoffInputData] | None = None, nest_handoff_history: bool | None = None, + auto_handoff_back: bool = False, is_enabled: bool | Callable[[RunContextWrapper[Any], Agent[Any]], MaybeAwaitable[bool]] = True, ) -> Handoff[TContext, Agent[TContext]]: ... @@ -215,6 +227,7 @@ def handoff( tool_name_override: str | None = None, input_filter: Callable[[HandoffInputData], HandoffInputData] | None = None, nest_handoff_history: bool | None = None, + auto_handoff_back: bool = False, is_enabled: bool | Callable[[RunContextWrapper[Any], Agent[Any]], MaybeAwaitable[bool]] = True, ) -> Handoff[TContext, Agent[TContext]]: ... @@ -227,6 +240,7 @@ def handoff( input_type: type[THandoffInput] | None = None, input_filter: Callable[[HandoffInputData], HandoffInputData] | None = None, nest_handoff_history: bool | None = None, + auto_handoff_back: bool = False, is_enabled: bool | Callable[[RunContextWrapper[Any], Agent[TContext]], MaybeAwaitable[bool]] = True, ) -> Handoff[TContext, Agent[TContext]]: @@ -247,6 +261,9 @@ def handoff( input_filter: A function that filters the inputs that are passed to the next agent. nest_handoff_history: Optional override for the RunConfig-level ``nest_handoff_history`` flag. If ``None`` we fall back to the run's configuration. + auto_handoff_back: Whether the target agent should automatically handoff back to the + originating agent when it finishes its task. Useful for orchestration patterns where a + main agent delegates to specialist agents and expects results back. is_enabled: Whether the handoff is enabled. Can be a bool or a callable that takes the run context and agent and returns whether the handoff is enabled. Disabled handoffs are hidden from the LLM at runtime. @@ -327,6 +344,7 @@ async def _is_enabled(ctx: RunContextWrapper[Any], agent_base: AgentBase[Any]) - input_filter=input_filter, nest_handoff_history=nest_handoff_history, agent_name=agent.name, + auto_handoff_back=auto_handoff_back, is_enabled=_is_enabled if callable(is_enabled) else is_enabled, ) handoff_obj._agent_ref = weakref.ref(agent) diff --git a/src/agents/run.py b/src/agents/run.py index 014271a5ea..76bae34a50 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -765,6 +765,7 @@ def _finalize_result(result: RunResult) -> RunResult: raise try: + handoff_chain: list[Agent[Any]] = [] while True: resuming_turn = is_resumed_state all_input_guardrails = ( @@ -959,6 +960,20 @@ def _finalize_result(result: RunResult) -> RunResult: ) if isinstance(turn_result.next_step, NextStepFinalOutput): + if handoff_chain: + parent_agent = handoff_chain.pop() + child_output_text = format_final_output_text( + current_agent, turn_result.next_step.output + ) + back_msg = create_message_output_item( + current_agent, child_output_text + ) + generated_items.append(back_msg) + session_items.append(back_msg) + current_agent = cast(Agent[TContext], parent_agent) + if run_state is not None: + run_state._current_agent = current_agent + continue output_guardrail_results = await run_output_guardrails( current_agent.output_guardrails + (run_config.output_guardrails or []), @@ -1009,8 +1024,11 @@ def _finalize_result(result: RunResult) -> RunResult: result._original_input = copy_input_items(original_input) return _finalize_result(result) elif isinstance(turn_result.next_step, NextStepHandoff): + next_step = turn_result.next_step + if next_step.auto_handoff_back and next_step.originating_agent is not None: + handoff_chain.append(next_step.originating_agent) current_agent = cast( - Agent[TContext], turn_result.next_step.new_agent + Agent[TContext], next_step.new_agent ) if run_state is not None: run_state._current_agent = current_agent @@ -1365,6 +1383,20 @@ def _finalize_result(result: RunResult) -> RunResult: try: if isinstance(turn_result.next_step, NextStepFinalOutput): + if handoff_chain: + parent_agent = handoff_chain.pop() + child_output_text = format_final_output_text( + current_agent, turn_result.next_step.output + ) + back_msg = create_message_output_item( + current_agent, child_output_text + ) + generated_items.append(back_msg) + session_items.append(back_msg) + current_agent = cast(Agent[TContext], parent_agent) + if run_state is not None: + run_state._current_agent = current_agent + continue output_guardrail_results = await run_output_guardrails( current_agent.output_guardrails + (run_config.output_guardrails or []), @@ -1470,7 +1502,10 @@ def _finalize_result(result: RunResult) -> RunResult: ) return _finalize_result(result) elif isinstance(turn_result.next_step, NextStepHandoff): - current_agent = cast(Agent[TContext], turn_result.next_step.new_agent) + next_step = turn_result.next_step + if next_step.auto_handoff_back and next_step.originating_agent is not None: + handoff_chain.append(next_step.originating_agent) + current_agent = cast(Agent[TContext], next_step.new_agent) if run_state is not None: run_state._current_agent = current_agent # Next agent starts with the nested/filtered input. diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index 45f09c0fa0..e53589648a 100644 --- a/src/agents/run_internal/run_loop.py +++ b/src/agents/run_internal/run_loop.py @@ -668,6 +668,7 @@ async def _save_stream_items_without_count( raise try: + handoff_chain: list[Agent[Any]] = [] while True: all_input_guardrails = ( starting_agent.input_guardrails + (run_config.input_guardrails or []) @@ -801,7 +802,10 @@ async def _save_stream_items_without_count( break if isinstance(turn_result.next_step, NextStepHandoff): - current_agent = turn_result.next_step.new_agent + next_step = turn_result.next_step + if next_step.auto_handoff_back and next_step.originating_agent is not None: + handoff_chain.append(next_step.originating_agent) + current_agent = next_step.new_agent if run_state is not None: run_state._current_agent = current_agent if current_span: @@ -815,6 +819,29 @@ async def _save_stream_items_without_count( continue if isinstance(turn_result.next_step, NextStepFinalOutput): + if handoff_chain: + parent_agent = handoff_chain.pop() + child_output_text = format_final_output_text( + current_agent, turn_result.next_step.output + ) + back_msg = create_message_output_item( + current_agent, child_output_text + ) + turn_session_items.append(back_msg) + streamed_result.new_items.append(back_msg) + current_agent = parent_agent + if run_state is not None: + run_state._current_agent = current_agent + streamed_result._event_queue.put_nowait( + AgentUpdatedStreamEvent(new_agent=current_agent) + ) + await _save_resumed_items( + list(turn_session_items), + turn_result.model_response.response_id, + store_setting, + ) + run_state._current_step = NextStepRunAgain() # type: ignore[assignment] + continue await _finalize_streamed_final_output( streamed_result=streamed_result, agent=current_agent, @@ -1089,12 +1116,15 @@ async def _save_stream_items_without_count( server_conversation_tracker.track_server_items(turn_result.model_response) if isinstance(turn_result.next_step, NextStepHandoff): + next_step = turn_result.next_step + if next_step.auto_handoff_back and next_step.originating_agent is not None: + handoff_chain.append(next_step.originating_agent) await _save_stream_items_without_count( turn_session_items, turn_result.model_response.response_id, store_setting, ) - current_agent = turn_result.next_step.new_agent + current_agent = next_step.new_agent if run_state is not None: run_state._current_agent = current_agent current_span.finish(reset_current=True) @@ -1111,6 +1141,28 @@ async def _save_stream_items_without_count( streamed_result._event_queue.put_nowait(QueueCompleteSentinel()) break elif isinstance(turn_result.next_step, NextStepFinalOutput): + if handoff_chain: + parent_agent = handoff_chain.pop() + child_output_text = format_final_output_text( + current_agent, turn_result.next_step.output + ) + back_msg = create_message_output_item( + current_agent, child_output_text + ) + turn_session_items.append(back_msg) + streamed_result.new_items.append(back_msg) + current_agent = parent_agent + if run_state is not None: + run_state._current_agent = current_agent + streamed_result._event_queue.put_nowait( + AgentUpdatedStreamEvent(new_agent=current_agent) + ) + await _save_stream_items_with_count( + turn_session_items, + turn_result.model_response.response_id, + store_setting, + ) + continue await _finalize_streamed_final_output( streamed_result=streamed_result, agent=current_agent, diff --git a/src/agents/run_internal/run_steps.py b/src/agents/run_internal/run_steps.py index ec64b760a6..08b2db6ed0 100644 --- a/src/agents/run_internal/run_steps.py +++ b/src/agents/run_internal/run_steps.py @@ -154,6 +154,13 @@ def has_interruptions(self) -> bool: @dataclass class NextStepHandoff: new_agent: Agent[Any] + auto_handoff_back: bool = False + """Whether the new agent should automatically handoff back to the originating agent when it + finishes its task.""" + + originating_agent: Agent[Any] | None = None + """The agent that initiated the handoff. When auto_handoff_back is True, control will return + to this agent after the new agent completes its task.""" @dataclass diff --git a/src/agents/run_internal/turn_resolution.py b/src/agents/run_internal/turn_resolution.py index 2c95cf2e13..bfeb7a7cc6 100644 --- a/src/agents/run_internal/turn_resolution.py +++ b/src/agents/run_internal/turn_resolution.py @@ -584,7 +584,11 @@ def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffIn model_response=new_response, pre_step_items=pre_step_items, new_step_items=new_step_items, - next_step=NextStepHandoff(new_agent), + next_step=NextStepHandoff( + new_agent, + auto_handoff_back=handoff.auto_handoff_back, + originating_agent=public_agent if handoff.auto_handoff_back else None, + ), tool_input_guardrail_results=list(tool_input_guardrail_results or []), tool_output_guardrail_results=list(tool_output_guardrail_results or []), session_step_items=session_step_items, diff --git a/tests/test_auto_handoff_back.py b/tests/test_auto_handoff_back.py new file mode 100644 index 0000000000..a2819ec273 --- /dev/null +++ b/tests/test_auto_handoff_back.py @@ -0,0 +1,207 @@ +"""Tests for automatic back-handoff support (#847). + +These tests verify that when a handoff is configured with auto_handoff_back=True, +the child agent automatically hands control back to the originating agent upon +completion, enabling orchestration patterns without circular references. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from agents import Agent, RunConfig, Runner, handoff +from agents.handoffs import Handoff + +from .fake_model import FakeModel +from .test_responses import get_handoff_tool_call, get_text_message + + +@pytest.mark.asyncio +async def test_auto_handoff_back_returns_to_originating_agent() -> None: + """Verify that an agent with auto_handoff_back returns control to the + originating agent after the child completes its task.""" + orchestrator_model = FakeModel() + specialist_model = FakeModel() + + specialist = Agent(name="specialist", model=specialist_model) + orchestrator = Agent( + name="orchestrator", + model=orchestrator_model, + handoffs=[ + handoff( + specialist, + auto_handoff_back=True, + ) + ], + ) + + # Turn 1: orchestrator hands off to specialist + orchestrator_model.add_multiple_turn_outputs( + [[get_text_message("let me delegate"), get_handoff_tool_call(specialist)]] + ) + # Turn 2: specialist finishes its task + specialist_model.add_multiple_turn_outputs( + [[get_text_message("specialist analysis complete")]] + ) + # Turn 3: orchestrator produces final output after regaining control + orchestrator_model.add_multiple_turn_outputs( + [[get_text_message("orchestrator final response")]] + ) + + result = await Runner.run( + orchestrator, + input="analyze this and give me a final answer", + ) + + # The last agent should be the orchestrator, not the specialist + assert result.last_agent.name == "orchestrator" + assert result.final_output == "orchestrator final response" + + +@pytest.mark.asyncio +async def test_auto_handoff_back_disabled_by_default() -> None: + """Verify that by default (auto_handoff_back=False), control does NOT + return to the originating agent.""" + orchestrator_model = FakeModel() + specialist_model = FakeModel() + + specialist = Agent(name="specialist", model=specialist_model) + orchestrator = Agent( + name="orchestrator", + model=orchestrator_model, + handoffs=[specialist], # default auto_handoff_back=False + ) + + # Turn 1: orchestrator hands off to specialist + orchestrator_model.add_multiple_turn_outputs( + [[get_text_message("let me delegate"), get_handoff_tool_call(specialist)]] + ) + # Turn 2: specialist finishes its task + specialist_model.add_multiple_turn_outputs( + [[get_text_message("specialist analysis complete")]] + ) + + result = await Runner.run( + orchestrator, + input="analyze this", + ) + + # The last agent should be the specialist (no auto-handoff-back) + assert result.last_agent.name == "specialist" + assert result.final_output == "specialist analysis complete" + + +@pytest.mark.asyncio +async def test_auto_handoff_back_with_nested_handoffs() -> None: + """Verify that nested auto_handoff_back chains work correctly. + + orchestrator -> specialist_a -> specialist_b + Each handoff has auto_handoff_back=True, so control should return + through the chain: specialist_b -> specialist_a -> orchestrator. + """ + orch_model = FakeModel() + model_a = FakeModel() + model_b = FakeModel() + + specialist_b = Agent(name="specialist_b", model=model_b) + specialist_a = Agent( + name="specialist_a", + model=model_a, + handoffs=[handoff(specialist_b, auto_handoff_back=True)], + ) + orchestrator = Agent( + name="orchestrator", + model=orch_model, + handoffs=[handoff(specialist_a, auto_handoff_back=True)], + ) + + # Turn 1: orchestrator hands off to specialist_a + orch_model.add_multiple_turn_outputs( + [[get_text_message("delegating to a"), get_handoff_tool_call(specialist_a)]] + ) + # Turn 2: specialist_a hands off to specialist_b + model_a.add_multiple_turn_outputs( + [[get_text_message("delegating to b"), get_handoff_tool_call(specialist_b)]] + ) + # Turn 3: specialist_b finishes + model_b.add_multiple_turn_outputs( + [[get_text_message("b's analysis complete")]] + ) + # Turn 4: specialist_a finishes after regaining control + model_a.add_multiple_turn_outputs( + [[get_text_message("a's combined analysis")]] + ) + # Turn 5: orchestrator finishes after regaining control + orch_model.add_multiple_turn_outputs( + [[get_text_message("orchestrator final response")]] + ) + + result = await Runner.run( + orchestrator, + input="analyze this deeply", + ) + + assert result.last_agent.name == "orchestrator" + assert result.final_output == "orchestrator final response" + + +@pytest.mark.asyncio +async def test_auto_handoff_back_handoff_config() -> None: + """Verify the Handoff object correctly stores auto_handoff_back.""" + agent_b = Agent(name="agent_b") + agent_a = Agent(name="agent_a") + + # Default: auto_handoff_back=False + h_default = Handoff.default_tool_name(agent_b) + assert isinstance(h_default, str) + + # Explicitly set auto_handoff_back=True via handoff() + h_auto = handoff(agent_b, auto_handoff_back=True) + assert h_auto.auto_handoff_back is True + assert h_auto.agent_name == "agent_b" + + # Explicitly set auto_handoff_back=False (default) via handoff() + h_no_auto = handoff(agent_b, auto_handoff_back=False) + assert h_no_auto.auto_handoff_back is False + + +@pytest.mark.asyncio +async def test_auto_handoff_back_without_final_orchestrator_turn() -> None: + """Verify that the orchestrator receives the specialist's output as context + when control returns. The final result should be the specialist output + if the orchestrator has no more turns.""" + orchestrator_model = FakeModel() + specialist_model = FakeModel() + + specialist = Agent(name="specialist", model=specialist_model) + orchestrator = Agent( + name="orchestrator", + model=orchestrator_model, + handoffs=[ + handoff(specialist, auto_handoff_back=True) + ], + ) + + # Turn 1: orchestrator hands off to specialist + orchestrator_model.add_multiple_turn_outputs( + [[get_text_message("delegating"), get_handoff_tool_call(specialist)]] + ) + # Turn 2: specialist finishes + specialist_model.add_multiple_turn_outputs( + [[get_text_message("specialist final answer")]] + ) + # Turn 3: orchestrator runs again after regaining control and produces output + orchestrator_model.add_multiple_turn_outputs( + [[get_text_message("based on specialist input, here is my answer")]] + ) + + result = await Runner.run( + orchestrator, + input="help me with this", + max_turns=10, + ) + + assert result.last_agent.name == "orchestrator" + assert result.final_output == "based on specialist input, here is my answer"