[BREAKING] Python: Add factory pattern to handoff orchestration builder (#2844)

* WIP: Factory pattern to handoff

* Add factory pattern to concurrent orchestration builder; Next: tests and sample verification

* Add tests and improve comments

* Fix mypy

* Simplify handoff_simple.py

* Simplify handoff_autonoumous.py and bug fix

* Update readme

* Address Copilot comments

python/packages/core/agent_framework/_workflows/_agent_executor.py

@@ -105,6 +105,11 @@ class AgentExecutor(Executor):
             return [AgentRunResponse]
         return []
 
+    @property
+    def description(self) -> str | None:
+        """Get the description of the underlying agent."""
+        return self._agent.description
+
     @handler
     async def run(
         self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse, AgentRunResponse]

python/packages/core/agent_framework/_workflows/_participant_utils.py

@@ -47,15 +47,13 @@ def wrap_participant(participant: AgentProtocol | Executor, *, executor_id: str
     """Represent `participant` as an `Executor`."""
     if isinstance(participant, Executor):
         return participant
+
     if not isinstance(participant, AgentProtocol):
         raise TypeError(
             f"Participants must implement AgentProtocol or be Executor instances. Got {type(participant).__name__}."
         )
-    name = getattr(participant, "name", None)
-    if executor_id is None:
-        if not name:
-            raise ValueError("Agent participants must expose a stable 'name' attribute.")
-        executor_id = str(name)
+
+    executor_id = executor_id or participant.display_name
     return AgentExecutor(participant, id=executor_id)
 
 

python/packages/core/agent_framework/_workflows/_sequential.py

@@ -154,6 +154,9 @@ class SequentialBuilder:
                 "Cannot mix .participants([...]) and .register_participants() in the same builder instance."
             )
 
+        if self._participant_factories:
+            raise ValueError("register_participants() has already been called on this builder instance.")
+
         if not participant_factories:
             raise ValueError("participant_factories cannot be empty")
 
@@ -171,6 +174,9 @@ class SequentialBuilder:
                 "Cannot mix .participants([...]) and .register_participants() in the same builder instance."
             )
 
+        if self._participants:
+            raise ValueError("participants() has already been called on this builder instance.")
+
         if not participants:
             raise ValueError("participants cannot be empty")
 

python/packages/core/agent_framework/openai/_chat_client.py

@@ -162,6 +162,7 @@ class OpenAIBaseChatClient(OpenAIBase, BaseChatClient):
             exclude={
                 "type",
                 "instructions",  # included as system message
+                "allow_multiple_tool_calls",  # handled separately
             }
         )
 
@@ -174,6 +175,8 @@ class OpenAIBaseChatClient(OpenAIBase, BaseChatClient):
             if web_search_options:
                 options_dict["web_search_options"] = web_search_options
             options_dict["tools"] = self._chat_to_tool_spec(chat_options.tools)
+        if chat_options.allow_multiple_tool_calls is not None:
+            options_dict["parallel_tool_calls"] = chat_options.allow_multiple_tool_calls
         if not options_dict.get("tools", None):
             options_dict.pop("tools", None)
             options_dict.pop("parallel_tool_calls", None)

python/packages/core/tests/workflow/test_handoff.py

@@ -223,7 +223,7 @@ async def test_handoff_preserves_complex_additional_properties(complex_metadata:
 
     workflow = (
         HandoffBuilder(participants=[triage, specialist])
-        .set_coordinator("triage")
+        .set_coordinator(triage)
         .with_termination_condition(lambda conv: sum(1 for msg in conv if msg.role == Role.USER) >= 2)
         .build()
     )
@@ -286,7 +286,7 @@ async def test_tool_call_handoff_detection_with_text_hint():
     triage = _RecordingAgent(name="triage", handoff_to="specialist", text_handoff=True)
     specialist = _RecordingAgent(name="specialist")
 
-    workflow = HandoffBuilder(participants=[triage, specialist]).set_coordinator("triage").build()
+    workflow = HandoffBuilder(participants=[triage, specialist]).set_coordinator(triage).build()
 
     await _drain(workflow.run_stream("Package arrived broken"))
 
@@ -301,7 +301,7 @@ async def test_autonomous_interaction_mode_yields_output_without_user_request():
 
     workflow = (
         HandoffBuilder(participants=[triage, specialist])
-        .set_coordinator("triage")
+        .set_coordinator(triage)
         .with_interaction_mode("autonomous", autonomous_turn_limit=1)
         .build()
     )
@@ -433,13 +433,13 @@ def test_build_fails_without_coordinator():
     triage = _RecordingAgent(name="triage")
     specialist = _RecordingAgent(name="specialist")
 
-    with pytest.raises(ValueError, match="coordinator must be defined before build"):
+    with pytest.raises(ValueError, match=r"Must call set_coordinator\(...\) before building the workflow."):
         HandoffBuilder(participants=[triage, specialist]).build()
 
 
 def test_build_fails_without_participants():
     """Verify that build() raises ValueError when no participants are provided."""
-    with pytest.raises(ValueError, match="No participants provided"):
+    with pytest.raises(ValueError, match="No participants or participant_factories have been configured."):
         HandoffBuilder().build()
 
 
@@ -610,7 +610,7 @@ async def test_return_to_previous_enabled():
 
     workflow = (
         HandoffBuilder(participants=[triage, specialist_a, specialist_b])
-        .set_coordinator("triage")
+        .set_coordinator(triage)
         .enable_return_to_previous(True)
         .with_termination_condition(lambda conv: sum(1 for m in conv if m.role == Role.USER) >= 3)
         .build()
@@ -643,7 +643,7 @@ def test_handoff_builder_sets_start_executor_once(monkeypatch: pytest.MonkeyPatc
 
     workflow = (
         HandoffBuilder(participants=[coordinator, specialist])
-        .set_coordinator("coordinator")
+        .set_coordinator(coordinator)
         .with_termination_condition(lambda conv: len(conv) > 0)
         .build()
     )
@@ -703,7 +703,7 @@ async def test_handoff_builder_with_request_info():
     # Build workflow with request info enabled
     workflow = (
         HandoffBuilder(participants=[coordinator, specialist])
-        .set_coordinator("coordinator")
+        .set_coordinator(coordinator)
         .with_termination_condition(lambda conv: len([m for m in conv if m.role == Role.USER]) >= 1)
         .with_request_info()
         .build()
@@ -782,6 +782,497 @@ async def test_return_to_previous_state_serialization():
     assert coordinator2._current_agent_id == "specialist_a", "Current agent should be restored from checkpoint"  # type: ignore[reportPrivateUsage]
 
 
+# region Participant Factory Tests
+
+
+def test_handoff_builder_rejects_empty_participant_factories():
+    """Test that HandoffBuilder rejects empty participant_factories dictionary."""
+    # Empty factories are rejected immediately when calling participant_factories()
+    with pytest.raises(ValueError, match=r"participant_factories cannot be empty"):
+        HandoffBuilder().participant_factories({})
+
+    with pytest.raises(ValueError, match=r"No participants or participant_factories have been configured"):
+        HandoffBuilder(participant_factories={}).build()
+
+
+def test_handoff_builder_rejects_mixing_participants_and_factories():
+    """Test that mixing participants and participant_factories in __init__ raises an error."""
+    triage = _RecordingAgent(name="triage")
+    with pytest.raises(ValueError, match="Cannot mix .participants"):
+        HandoffBuilder(participants=[triage], participant_factories={"triage": lambda: triage})
+
+
+def test_handoff_builder_rejects_mixing_participants_and_participant_factories_methods():
+    """Test that mixing .participants() and .participant_factories() raises an error."""
+    triage = _RecordingAgent(name="triage")
+
+    # Case 1: participants first, then participant_factories
+    with pytest.raises(ValueError, match="Cannot mix .participants"):
+        HandoffBuilder(participants=[triage]).participant_factories({
+            "specialist": lambda: _RecordingAgent(name="specialist")
+        })
+
+    # Case 2: participant_factories first, then participants
+    with pytest.raises(ValueError, match="Cannot mix .participants"):
+        HandoffBuilder(participant_factories={"triage": lambda: triage}).participants([
+            _RecordingAgent(name="specialist")
+        ])
+
+    # Case 3: participants(), then participant_factories()
+    with pytest.raises(ValueError, match="Cannot mix .participants"):
+        HandoffBuilder().participants([triage]).participant_factories({
+            "specialist": lambda: _RecordingAgent(name="specialist")
+        })
+
+    # Case 4: participant_factories(), then participants()
+    with pytest.raises(ValueError, match="Cannot mix .participants"):
+        HandoffBuilder().participant_factories({"triage": lambda: triage}).participants([
+            _RecordingAgent(name="specialist")
+        ])
+
+    # Case 5: mix during initialization
+    with pytest.raises(ValueError, match="Cannot mix .participants"):
+        HandoffBuilder(
+            participants=[triage], participant_factories={"specialist": lambda: _RecordingAgent(name="specialist")}
+        )
+
+
+def test_handoff_builder_rejects_multiple_calls_to_participant_factories():
+    """Test that multiple calls to .participant_factories() raises an error."""
+    with pytest.raises(ValueError, match=r"participant_factories\(\) has already been called"):
+        (
+            HandoffBuilder()
+            .participant_factories({"agent1": lambda: _RecordingAgent(name="agent1")})
+            .participant_factories({"agent2": lambda: _RecordingAgent(name="agent2")})
+        )
+
+
+def test_handoff_builder_rejects_multiple_calls_to_participants():
+    """Test that multiple calls to .participants() raises an error."""
+    with pytest.raises(ValueError, match="participants have already been assigned"):
+        (HandoffBuilder().participants([_RecordingAgent(name="agent1")]).participants([_RecordingAgent(name="agent2")]))
+
+
+def test_handoff_builder_rejects_duplicate_factories():
+    """Test that multiple calls to participant_factories are rejected."""
+    factories = {
+        "triage": lambda: _RecordingAgent(name="triage"),
+        "specialist": lambda: _RecordingAgent(name="specialist"),
+    }
+
+    # Multiple calls to participant_factories should fail
+    builder = HandoffBuilder(participant_factories=factories)
+    with pytest.raises(ValueError, match=r"participant_factories\(\) has already been called"):
+        builder.participant_factories({"triage": lambda: _RecordingAgent(name="triage2")})
+
+
+def test_handoff_builder_rejects_instance_coordinator_with_factories():
+    """Test that using an agent instance for set_coordinator when using factories raises an error."""
+
+    def create_triage() -> _RecordingAgent:
+        return _RecordingAgent(name="triage")
+
+    def create_specialist() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist")
+
+    # Create an agent instance
+    coordinator_instance = _RecordingAgent(name="coordinator")
+
+    with pytest.raises(ValueError, match=r"Call participants\(\.\.\.\) before coordinator\(\.\.\.\)"):
+        (
+            HandoffBuilder(
+                participant_factories={"triage": create_triage, "specialist": create_specialist}
+            ).set_coordinator(coordinator_instance)  # Instance, not factory name
+        )
+
+
+def test_handoff_builder_rejects_factory_name_coordinator_with_instances():
+    """Test that using a factory name for set_coordinator when using instances raises an error."""
+    triage = _RecordingAgent(name="triage")
+    specialist = _RecordingAgent(name="specialist")
+
+    with pytest.raises(
+        ValueError, match="coordinator factory name 'triage' is not part of the participant_factories list"
+    ):
+        (
+            HandoffBuilder(participants=[triage, specialist]).set_coordinator(
+                "triage"
+            )  # String factory name, not instance
+        )
+
+
+def test_handoff_builder_rejects_mixed_types_in_add_handoff_source():
+    """Test that add_handoff rejects factory name source with instance-based participants."""
+    triage = _RecordingAgent(name="triage")
+    specialist = _RecordingAgent(name="specialist")
+
+    with pytest.raises(TypeError, match="Cannot mix factory names \\(str\\) and AgentProtocol/Executor instances"):
+        (
+            HandoffBuilder(participants=[triage, specialist])
+            .set_coordinator(triage)
+            .add_handoff("triage", specialist)  # String source with instance participants
+        )
+
+
+def test_handoff_builder_accepts_all_factory_names_in_add_handoff():
+    """Test that add_handoff accepts all factory names when using participant_factories."""
+
+    def create_triage() -> _RecordingAgent:
+        return _RecordingAgent(name="triage")
+
+    def create_specialist_a() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist_a")
+
+    def create_specialist_b() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist_b")
+
+    # This should work - all strings with participant_factories
+    builder = (
+        HandoffBuilder(
+            participant_factories={
+                "triage": create_triage,
+                "specialist_a": create_specialist_a,
+                "specialist_b": create_specialist_b,
+            }
+        )
+        .set_coordinator("triage")
+        .add_handoff("triage", ["specialist_a", "specialist_b"])
+    )
+
+    workflow = builder.build()
+    assert "triage" in workflow.executors
+    assert "specialist_a" in workflow.executors
+    assert "specialist_b" in workflow.executors
+
+
+def test_handoff_builder_accepts_all_instances_in_add_handoff():
+    """Test that add_handoff accepts all instances when using participants."""
+    triage = _RecordingAgent(name="triage", handoff_to="specialist_a")
+    specialist_a = _RecordingAgent(name="specialist_a")
+    specialist_b = _RecordingAgent(name="specialist_b")
+
+    # This should work - all instances with participants
+    builder = (
+        HandoffBuilder(participants=[triage, specialist_a, specialist_b])
+        .set_coordinator(triage)
+        .add_handoff(triage, [specialist_a, specialist_b])
+    )
+
+    workflow = builder.build()
+    assert "triage" in workflow.executors
+    assert "specialist_a" in workflow.executors
+    assert "specialist_b" in workflow.executors
+
+
+async def test_handoff_with_participant_factories():
+    """Test workflow creation using participant_factories."""
+    call_count = 0
+
+    def create_triage() -> _RecordingAgent:
+        nonlocal call_count
+        call_count += 1
+        return _RecordingAgent(name="triage", handoff_to="specialist")
+
+    def create_specialist() -> _RecordingAgent:
+        nonlocal call_count
+        call_count += 1
+        return _RecordingAgent(name="specialist")
+
+    workflow = (
+        HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
+        .set_coordinator("triage")
+        .with_termination_condition(lambda conv: sum(1 for m in conv if m.role == Role.USER) >= 2)
+        .build()
+    )
+
+    # Factories should be called during build
+    assert call_count == 2
+
+    events = await _drain(workflow.run_stream("Need help"))
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert requests
+
+    # Follow-up message
+    events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "More details"}))
+    outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
+    assert outputs
+
+
+async def test_handoff_participant_factories_reusable_builder():
+    """Test that the builder can be reused to build multiple workflows with factories."""
+    call_count = 0
+
+    def create_triage() -> _RecordingAgent:
+        nonlocal call_count
+        call_count += 1
+        return _RecordingAgent(name="triage", handoff_to="specialist")
+
+    def create_specialist() -> _RecordingAgent:
+        nonlocal call_count
+        call_count += 1
+        return _RecordingAgent(name="specialist")
+
+    builder = HandoffBuilder(
+        participant_factories={"triage": create_triage, "specialist": create_specialist}
+    ).set_coordinator("triage")
+
+    # Build first workflow
+    wf1 = builder.build()
+    assert call_count == 2
+
+    # Build second workflow
+    wf2 = builder.build()
+    assert call_count == 4
+
+    # Verify that the two workflows have different agent instances
+    assert wf1.executors["triage"] is not wf2.executors["triage"]
+    assert wf1.executors["specialist"] is not wf2.executors["specialist"]
+
+
+async def test_handoff_with_participant_factories_and_add_handoff():
+    """Test that .add_handoff() works correctly with participant_factories."""
+
+    def create_triage() -> _RecordingAgent:
+        return _RecordingAgent(name="triage", handoff_to="specialist_a")
+
+    def create_specialist_a() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist_a", handoff_to="specialist_b")
+
+    def create_specialist_b() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist_b")
+
+    workflow = (
+        HandoffBuilder(
+            participant_factories={
+                "triage": create_triage,
+                "specialist_a": create_specialist_a,
+                "specialist_b": create_specialist_b,
+            }
+        )
+        .set_coordinator("triage")
+        .add_handoff("triage", ["specialist_a", "specialist_b"])
+        .add_handoff("specialist_a", "specialist_b")
+        .with_termination_condition(lambda conv: sum(1 for m in conv if m.role == Role.USER) >= 3)
+        .build()
+    )
+
+    # Start conversation - triage hands off to specialist_a
+    events = await _drain(workflow.run_stream("Initial request"))
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert requests
+
+    # Verify specialist_a executor exists and was called
+    assert "specialist_a" in workflow.executors
+
+    # Second user message - specialist_a hands off to specialist_b
+    events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "Need escalation"}))
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert requests
+
+    # Verify specialist_b executor exists
+    assert "specialist_b" in workflow.executors
+
+
+async def test_handoff_participant_factories_with_checkpointing():
+    """Test checkpointing with participant_factories."""
+    from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
+
+    storage = InMemoryCheckpointStorage()
+
+    def create_triage() -> _RecordingAgent:
+        return _RecordingAgent(name="triage", handoff_to="specialist")
+
+    def create_specialist() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist")
+
+    workflow = (
+        HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
+        .set_coordinator("triage")
+        .with_checkpointing(storage)
+        .with_termination_condition(lambda conv: sum(1 for m in conv if m.role == Role.USER) >= 2)
+        .build()
+    )
+
+    # Run workflow and capture output
+    events = await _drain(workflow.run_stream("checkpoint test"))
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert requests
+
+    events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "follow up"}))
+    outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
+    assert outputs, "Should have workflow output after termination condition is met"
+
+    # List checkpoints - just verify they were created
+    checkpoints = await storage.list_checkpoints()
+    assert checkpoints, "Checkpoints should be created during workflow execution"
+
+
+def test_handoff_set_coordinator_with_factory_name():
+    """Test that set_coordinator accepts factory name as string."""
+
+    def create_triage() -> _RecordingAgent:
+        return _RecordingAgent(name="triage")
+
+    def create_specialist() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist")
+
+    builder = HandoffBuilder(
+        participant_factories={"triage": create_triage, "specialist": create_specialist}
+    ).set_coordinator("triage")
+
+    workflow = builder.build()
+    assert "triage" in workflow.executors
+
+
+def test_handoff_add_handoff_with_factory_names():
+    """Test that add_handoff accepts factory names as strings."""
+
+    def create_triage() -> _RecordingAgent:
+        return _RecordingAgent(name="triage", handoff_to="specialist_a")
+
+    def create_specialist_a() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist_a")
+
+    def create_specialist_b() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist_b")
+
+    builder = (
+        HandoffBuilder(
+            participant_factories={
+                "triage": create_triage,
+                "specialist_a": create_specialist_a,
+                "specialist_b": create_specialist_b,
+            }
+        )
+        .set_coordinator("triage")
+        .add_handoff("triage", ["specialist_a", "specialist_b"])
+    )
+
+    workflow = builder.build()
+    assert "triage" in workflow.executors
+    assert "specialist_a" in workflow.executors
+    assert "specialist_b" in workflow.executors
+
+
+async def test_handoff_participant_factories_autonomous_mode():
+    """Test autonomous mode with participant_factories."""
+
+    def create_triage() -> _RecordingAgent:
+        return _RecordingAgent(name="triage", handoff_to="specialist")
+
+    def create_specialist() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist")
+
+    workflow = (
+        HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
+        .set_coordinator("triage")
+        .with_interaction_mode("autonomous", autonomous_turn_limit=2)
+        .build()
+    )
+
+    events = await _drain(workflow.run_stream("Issue"))
+    outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
+    assert outputs, "Autonomous mode should yield output"
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert not requests, "Autonomous mode should not request user input"
+
+
+async def test_handoff_participant_factories_with_request_info():
+    """Test that .with_request_info() works with participant_factories."""
+
+    def create_triage() -> _RecordingAgent:
+        return _RecordingAgent(name="triage")
+
+    def create_specialist() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist")
+
+    builder = (
+        HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
+        .set_coordinator("triage")
+        .with_request_info(agents=["triage"])
+    )
+
+    workflow = builder.build()
+    assert "triage" in workflow.executors
+
+
+def test_handoff_participant_factories_invalid_coordinator_name():
+    """Test that set_coordinator raises error for non-existent factory name."""
+
+    def create_triage() -> _RecordingAgent:
+        return _RecordingAgent(name="triage")
+
+    with pytest.raises(
+        ValueError, match="coordinator factory name 'nonexistent' is not part of the participant_factories list"
+    ):
+        (HandoffBuilder(participant_factories={"triage": create_triage}).set_coordinator("nonexistent").build())
+
+
+def test_handoff_participant_factories_invalid_handoff_target():
+    """Test that add_handoff raises error for non-existent target factory name."""
+
+    def create_triage() -> _RecordingAgent:
+        return _RecordingAgent(name="triage")
+
+    def create_specialist() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist")
+
+    with pytest.raises(ValueError, match="Target factory name 'nonexistent' is not in the participant_factories list"):
+        (
+            HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
+            .set_coordinator("triage")
+            .add_handoff("triage", "nonexistent")
+            .build()
+        )
+
+
+async def test_handoff_participant_factories_enable_return_to_previous():
+    """Test return_to_previous works with participant_factories."""
+
+    def create_triage() -> _RecordingAgent:
+        return _RecordingAgent(name="triage", handoff_to="specialist_a")
+
+    def create_specialist_a() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist_a", handoff_to="specialist_b")
+
+    def create_specialist_b() -> _RecordingAgent:
+        return _RecordingAgent(name="specialist_b")
+
+    workflow = (
+        HandoffBuilder(
+            participant_factories={
+                "triage": create_triage,
+                "specialist_a": create_specialist_a,
+                "specialist_b": create_specialist_b,
+            }
+        )
+        .set_coordinator("triage")
+        .add_handoff("triage", ["specialist_a", "specialist_b"])
+        .add_handoff("specialist_a", "specialist_b")
+        .enable_return_to_previous(True)
+        .with_termination_condition(lambda conv: sum(1 for m in conv if m.role == Role.USER) >= 3)
+        .build()
+    )
+
+    # Start conversation - triage hands off to specialist_a
+    events = await _drain(workflow.run_stream("Initial request"))
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert requests
+
+    # Second user message - specialist_a hands off to specialist_b
+    events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "Need escalation"}))
+    requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+    assert requests
+
+    # Third user message - should route back to specialist_b (return to previous)
+    events = await _drain(workflow.send_responses_streaming({requests[-1].request_id: "Follow up"}))
+    outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
+    assert outputs or [ev for ev in events if isinstance(ev, RequestInfoEvent)]
+
+
+# endregion Participant Factory Tests
+
+
 async def test_handoff_user_input_request_checkpoint_excludes_conversation():
     """Test that HandoffUserInputRequest serialization excludes conversation to prevent duplication.
 

python/samples/getting_started/workflows/README.md

@@ -86,7 +86,6 @@ Once comfortable with these, explore the rest of the samples below.
 | ConcurrentBuilder Request Info | [human-in-the-loop/concurrent_request_info.py](./human-in-the-loop/concurrent_request_info.py) | Review concurrent agent outputs before aggregation using `.with_request_info()` on ConcurrentBuilder |
 | GroupChatBuilder Request Info | [human-in-the-loop/group_chat_request_info.py](./human-in-the-loop/group_chat_request_info.py) | Steer group discussions with periodic guidance using `.with_request_info()` on GroupChatBuilder |
 
-
 ### tool-approval
 
 Tool approval samples demonstrate using `@ai_function(approval_mode="always_require")` to gate sensitive tool executions with human approval. These work with the high-level builder APIs.
@@ -120,6 +119,7 @@ For additional observability samples in Agent Framework, see the [observability
 | Handoff (Specialist-to-Specialist) | [orchestration/handoff_specialist_to_specialist.py](./orchestration/handoff_specialist_to_specialist.py) | Multi-tier routing: specialists can hand off to other specialists using `.add_handoff()` fluent API |
 | Handoff (Return-to-Previous) | [orchestration/handoff_return_to_previous.py](./orchestration/handoff_return_to_previous.py) | Return-to-previous routing: after user input, routes back to the previous specialist instead of coordinator using `.enable_return_to_previous()` |
 | Handoff (Autonomous) | [orchestration/handoff_autonomous.py](./orchestration/handoff_autonomous.py) | Autonomous mode: specialists iterate independently until invoking a handoff tool using `.with_interaction_mode("autonomous", autonomous_turn_limit=N)` |
+| Handoff (Participant Factory) | [orchestration/handoff_participant_factory.py](./orchestration/handoff_participant_factory.py) | Use participant factories for state isolation between workflow instances |
 | Magentic Workflow (Multi-Agent) | [orchestration/magentic.py](./orchestration/magentic.py) | Orchestrate multiple agents with Magentic manager and streaming |
 | Magentic + Human Plan Review | [orchestration/magentic_human_plan_update.py](./orchestration/magentic_human_plan_update.py) | Human reviews/updates the plan before execution |
 | Magentic + Human Stall Intervention | [orchestration/magentic_human_replan.py](./orchestration/magentic_human_replan.py) | Human intervenes when workflow stalls with `with_human_input_on_stall()` |

python/samples/getting_started/workflows/orchestration/handoff_autonomous.py

@@ -2,14 +2,15 @@
 
 import asyncio
 import logging
-from collections.abc import AsyncIterable
 from typing import cast
 
 from agent_framework import (
+    AgentRunResponseUpdate,
     AgentRunUpdateEvent,
     ChatAgent,
     ChatMessage,
     HandoffBuilder,
+    HostedWebSearchTool,
     WorkflowEvent,
     WorkflowOutputEvent,
 )
@@ -44,31 +45,29 @@ def create_agents(
     """Create coordinator and specialists for autonomous iteration."""
     coordinator = chat_client.create_agent(
         instructions=(
-            "You are a coordinator. Route user requests to either research_agent or summary_agent. "
-            "Always call exactly one handoff tool with a short routing acknowledgement. "
-            "If unsure, default to research_agent. Never request information yourself. "
-            "After a specialist hands off back to you, provide a concise final summary and stop."
+            "You are a coordinator. You break down a user query into a research task and a summary task. "
+            "Assign the two tasks to the appropriate specialists, one after the other."
         ),
         name="coordinator",
     )
 
     research_agent = chat_client.create_agent(
         instructions=(
-            "You are a research specialist that explores topics thoroughly. "
+            "You are a research specialist that explores topics thoroughly on the Microsoft Learn Site."
             "When given a research task, break it down into multiple aspects and explore each one. "
-            "Continue your research across multiple responses - don't try to finish everything in one response. "
-            "After each response, think about what else needs to be explored. "
-            "When you have covered the topic comprehensively (at least 3-4 different aspects), "
-            "call the handoff tool to return to coordinator with your findings. "
-            "Keep each individual response focused on one aspect."
+            "Continue your research across multiple responses - don't try to finish everything in one "
+            "response. After each response, think about what else needs to be explored. When you have "
+            "covered the topic comprehensively (at least 3-4 different aspects), return control to the "
+            "coordinator. Keep each individual response focused on one aspect."
         ),
         name="research_agent",
+        tools=[HostedWebSearchTool()],
     )
 
     summary_agent = chat_client.create_agent(
         instructions=(
-            "You summarize research findings. Provide a concise, well-organized summary. "
-            "When done, hand off to coordinator."
+            "You summarize research findings. Provide a concise, well-organized summary. When done, return "
+            "control to the coordinator."
         ),
         name="summary_agent",
     )
@@ -76,25 +75,29 @@ def create_agents(
     return coordinator, research_agent, summary_agent
 
 
-async def _drain(stream: AsyncIterable[WorkflowEvent]) -> list[WorkflowEvent]:
-    """Collect all events from an async stream."""
-    return [event async for event in stream]
+last_response_id: str | None = None
 
 
-def _print_conversation(events: list[WorkflowEvent]) -> None:
+def _display_event(event: WorkflowEvent) -> None:
     """Print the final conversation snapshot from workflow output events."""
-    for event in events:
-        if isinstance(event, AgentRunUpdateEvent):
-            print(event.data, flush=True, end="")
-        elif isinstance(event, WorkflowOutputEvent):
-            conversation = cast(list[ChatMessage], event.data)
-            print("\n=== Final Conversation (Autonomous with Iteration) ===")
-            for message in conversation:
-                speaker = message.author_name or message.role.value
-                text_preview = message.text[:200] + "..." if len(message.text) > 200 else message.text
-                print(f"- {speaker}: {text_preview}")
-            print(f"\nTotal messages: {len(conversation)}")
-            print("=====================================================")
+    if isinstance(event, AgentRunUpdateEvent) and event.data:
+        update: AgentRunResponseUpdate = event.data
+        if not update.text:
+            return
+        global last_response_id
+        if update.response_id != last_response_id:
+            last_response_id = update.response_id
+            print(f"\n- {update.author_name}: ", flush=True, end="")
+        print(event.data, flush=True, end="")
+    elif isinstance(event, WorkflowOutputEvent):
+        conversation = cast(list[ChatMessage], event.data)
+        print("\n=== Final Conversation (Autonomous with Iteration) ===")
+        for message in conversation:
+            speaker = message.author_name or message.role.value
+            text_preview = message.text[:200] + "..." if len(message.text) > 200 else message.text
+            print(f"- {speaker}: {text_preview}")
+        print(f"\nTotal messages: {len(conversation)}")
+        print("=====================================================")
 
 
 async def main() -> None:
@@ -122,12 +125,10 @@ async def main() -> None:
         .build()
     )
 
-    initial_request = "Research the key benefits and challenges of renewable energy adoption."
-    print("Initial request:", initial_request)
-    print("\nExpecting multiple iterations from the research agent...\n")
-
-    events = await _drain(workflow.run_stream(initial_request))
-    _print_conversation(events)
+    request = "Perform a comprehensive research on Microsoft Agent Framework."
+    print("Request:", request)
+    async for event in workflow.run_stream(request):
+        _display_event(event)
 
     """
     Expected behavior:

python/samples/getting_started/workflows/orchestration/handoff_participant_factory.py

@@ -0,0 +1,265 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+import logging
+from collections.abc import AsyncIterable
+from typing import cast
+
+from agent_framework import (
+    ChatAgent,
+    ChatMessage,
+    HandoffBuilder,
+    HandoffUserInputRequest,
+    RequestInfoEvent,
+    Role,
+    Workflow,
+    WorkflowEvent,
+    WorkflowOutputEvent,
+    ai_function,
+)
+from agent_framework.azure import AzureOpenAIChatClient
+from azure.identity import AzureCliCredential
+from typing import Annotated
+
+logging.basicConfig(level=logging.ERROR)
+
+"""Sample: Autonomous handoff workflow with agent factory.
+
+This sample demonstrates how to use participant factories in HandoffBuilder to create
+agents dynamically.
+
+Using participant factories allows you to set up proper state isolation between workflow
+instances created by the same builder. This is particularly useful when you need to handle
+requests or tasks in parallel with stateful participants.
+
+Routing Pattern:
+    User -> Coordinator -> Specialist (iterates N times) -> Handoff -> Final Output
+
+Prerequisites:
+    - `az login` (Azure CLI authentication)
+    - Environment variables for AzureOpenAIChatClient (AZURE_OPENAI_ENDPOINT, etc.)
+
+Key Concepts:
+    - Participant factories: create agents via factory functions for isolation
+"""
+
+
+@ai_function
+def process_refund(order_number: Annotated[str, "Order number to process refund for"]) -> str:
+    """Simulated function to process a refund for a given order number."""
+    return f"Refund processed successfully for order {order_number}."
+
+
+@ai_function
+def check_order_status(order_number: Annotated[str, "Order number to check status for"]) -> str:
+    """Simulated function to check the status of a given order number."""
+    return f"Order {order_number} is currently being processed and will ship in 2 business days."
+
+
+@ai_function
+def process_return(order_number: Annotated[str, "Order number to process return for"]) -> str:
+    """Simulated function to process a return for a given order number."""
+    return f"Return initiated successfully for order {order_number}. You will receive return instructions via email."
+
+
+def create_triage_agent() -> ChatAgent:
+    """Factory function to create a triage agent instance."""
+    return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
+        instructions=(
+            "You are frontline support triage. Route customer issues to the appropriate specialist agents "
+            "based on the problem described."
+        ),
+        name="triage_agent",
+    )
+
+
+def create_refund_agent() -> ChatAgent:
+    """Factory function to create a refund agent instance."""
+    return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
+        instructions="You process refund requests.",
+        name="refund_agent",
+        # In a real application, an agent can have multiple tools; here we keep it simple
+        tools=[process_refund],
+    )
+
+
+def create_order_status_agent() -> ChatAgent:
+    """Factory function to create an order status agent instance."""
+    return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
+        instructions="You handle order and shipping inquiries.",
+        name="order_agent",
+        # In a real application, an agent can have multiple tools; here we keep it simple
+        tools=[check_order_status],
+    )
+
+
+def create_return_agent() -> ChatAgent:
+    """Factory function to create a return agent instance."""
+    return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
+        instructions="You manage product return requests.",
+        name="return_agent",
+        # In a real application, an agent can have multiple tools; here we keep it simple
+        tools=[process_return],
+    )
+
+
+async def _drain(stream: AsyncIterable[WorkflowEvent]) -> list[WorkflowEvent]:
+    """Collect all events from an async stream into a list.
+
+    This helper drains the workflow's event stream so we can process events
+    synchronously after each workflow step completes.
+
+    Args:
+        stream: Async iterable of WorkflowEvent
+
+    Returns:
+        List of all events from the stream
+    """
+    return [event async for event in stream]
+
+
+def _handle_events(events: list[WorkflowEvent]) -> list[RequestInfoEvent]:
+    """Process workflow events and extract any pending user input requests.
+
+    This function inspects each event type and:
+    - Prints workflow status changes (IDLE, IDLE_WITH_PENDING_REQUESTS, etc.)
+    - Displays final conversation snapshots when workflow completes
+    - Prints user input request prompts
+    - Collects all RequestInfoEvent instances for response handling
+
+    Args:
+        events: List of WorkflowEvent to process
+
+    Returns:
+        List of RequestInfoEvent representing pending user input requests
+    """
+    requests: list[RequestInfoEvent] = []
+
+    for event in events:
+        # WorkflowOutputEvent: Contains the final conversation when workflow terminates
+        if isinstance(event, WorkflowOutputEvent):
+            conversation = cast(list[ChatMessage], event.data)
+            if isinstance(conversation, list):
+                print("\n=== Final Conversation Snapshot ===")
+                for message in conversation:
+                    speaker = message.author_name or message.role.value
+                    print(f"- {speaker}: {message.text}")
+                print("===================================")
+
+        # RequestInfoEvent: Workflow is requesting user input
+        elif isinstance(event, RequestInfoEvent):
+            if isinstance(event.data, HandoffUserInputRequest):
+                _print_agent_responses_since_last_user_message(event.data)
+            requests.append(event)
+
+    return requests
+
+
+def _print_agent_responses_since_last_user_message(request: HandoffUserInputRequest) -> None:
+    """Display agent responses since the last user message in a handoff request.
+
+    The HandoffUserInputRequest contains the full conversation history so far,
+    allowing the user to see what's been discussed before providing their next input.
+
+    Args:
+        request: The user input request containing conversation and prompt
+    """
+    if not request.conversation:
+        raise RuntimeError("HandoffUserInputRequest missing conversation history.")
+
+    # Reverse iterate to collect agent responses since last user message
+    agent_responses: list[ChatMessage] = []
+    for message in request.conversation[::-1]:
+        if message.role == Role.USER:
+            break
+        agent_responses.append(message)
+
+    # Print agent responses in original order
+    agent_responses.reverse()
+    for message in agent_responses:
+        speaker = message.author_name or message.role.value
+        print(f"- {speaker}: {message.text}")
+
+
+async def _run_Workflow(workflow: Workflow, user_inputs: list[str]) -> None:
+    """Run the workflow with the given user input and display events."""
+    print(f"- User: {user_inputs[0]}")
+    events = await _drain(workflow.run_stream(user_inputs[0]))
+    pending_requests = _handle_events(events)
+
+    # Process the request/response cycle
+    # The workflow will continue requesting input until:
+    # 1. The termination condition is met (4 user messages in this case), OR
+    # 2. We run out of scripted responses
+    while pending_requests and user_inputs[1:]:
+        # Get the next scripted response
+        user_response = user_inputs.pop(1)
+        print(f"\n- User: {user_response}")
+
+        # Send response(s) to all pending requests
+        # In this demo, there's typically one request per cycle, but the API supports multiple
+        responses = {req.request_id: user_response for req in pending_requests}
+
+        # Send responses and get new events
+        # We use send_responses_streaming() to get events as they occur, allowing us to
+        # display agent responses in real-time and handle new requests as they arrive
+        events = await _drain(workflow.send_responses_streaming(responses))
+        pending_requests = _handle_events(events)
+
+
+async def main() -> None:
+    """Run the autonomous handoff workflow with participant factories."""
+    # Build the handoff workflow using participant factories
+    workflow_builder = (
+        HandoffBuilder(
+            name="Autonomous Handoff with Participant Factories",
+            participant_factories={
+                "triage": create_triage_agent,
+                "refund": create_refund_agent,
+                "order_status": create_order_status_agent,
+                "return": create_return_agent,
+            },
+        )
+        .set_coordinator("triage")
+        .with_termination_condition(
+            # Custom termination: Check if the triage agent has provided a closing message.
+            # This looks for the last message being from triage_agent and containing "welcome",
+            # which indicates the conversation has concluded naturally.
+            lambda conversation: len(conversation) > 0
+            and conversation[-1].author_name == "triage_agent"
+            and "welcome" in conversation[-1].text.lower()
+        )
+    )
+
+    # Scripted user responses for reproducible demo
+    # In a console application, replace this with:
+    #   user_input = input("Your response: ")
+    # or integrate with a UI/chat interface
+    user_inputs = [
+        "Hello, I need assistance with my recent purchase.",
+        "My order 1234 arrived damaged and the packaging was destroyed. I'd like to return it.",
+        "Is my return being processed?",
+        "Thanks for resolving this.",
+    ]
+
+    workflow_a = workflow_builder.build()
+    print("=== Running workflow_a ===")
+    await _run_Workflow(workflow_a, list(user_inputs))
+
+    workflow_b = workflow_builder.build()
+    print("=== Running workflow_b ===")
+    # Only provide the last two inputs to workflow_b to demonstrate state isolation
+    # The agents in this workflow have no prior context thus should not have knowledge of
+    # order 1234 or previous interactions.
+    await _run_Workflow(workflow_b, user_inputs[2:])
+    """
+    Expected behavior:
+    - workflow_a and workflow_b maintain separate states for their participants.
+    - Each workflow processes its requests independently without interference.
+    - workflow_a will answer the follow-up request based on its own conversation history,
+      while workflow_b will provide a general answer without prior context.
+    """
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/getting_started/workflows/orchestration/handoff_simple.py

@@ -2,7 +2,7 @@
 
 import asyncio
 from collections.abc import AsyncIterable
-from typing import cast
+from typing import Annotated, cast
 
 from agent_framework import (
     ChatAgent,
@@ -10,10 +10,12 @@ from agent_framework import (
     HandoffBuilder,
     HandoffUserInputRequest,
     RequestInfoEvent,
+    Role,
     WorkflowEvent,
     WorkflowOutputEvent,
     WorkflowRunState,
     WorkflowStatusEvent,
+    ai_function,
 )
 from agent_framework.azure import AzureOpenAIChatClient
 from azure.identity import AzureCliCredential
@@ -22,10 +24,10 @@ from azure.identity import AzureCliCredential
 
 This sample demonstrates the basic handoff pattern where only the triage agent can
 route to specialists. Specialists cannot hand off to other specialists - after any
-specialist responds, control returns to the user for the next input.
+specialist responds, control returns to the user (via the triage agent) for the next input.
 
 Routing Pattern:
-    User → Triage Agent → Specialist → Back to User → Triage Agent → ...
+    User → Triage Agent → Specialist → Triage Agent → User → Triage Agent → ...
 
 This is the simplest handoff configuration, suitable for straightforward support
 scenarios where a triage agent dispatches to domain specialists, and each specialist
@@ -39,12 +41,31 @@ Prerequisites:
 
 Key Concepts:
     - Single-tier routing: Only triage agent has handoff capabilities
-    - Auto-registered handoff tools: HandoffBuilder creates tools automatically
+    - Auto-registered handoff tools: HandoffBuilder automatically creates handoff tools
+      for each participant, allowing the coordinator to transfer control to specialists
     - Termination condition: Controls when the workflow stops requesting user input
     - Request/response cycle: Workflow requests input, user responds, cycle continues
 """
 
 
+@ai_function
+def process_refund(order_number: Annotated[str, "Order number to process refund for"]) -> str:
+    """Simulated function to process a refund for a given order number."""
+    return f"Refund processed successfully for order {order_number}."
+
+
+@ai_function
+def check_order_status(order_number: Annotated[str, "Order number to check status for"]) -> str:
+    """Simulated function to check the status of a given order number."""
+    return f"Order {order_number} is currently being processed and will ship in 2 business days."
+
+
+@ai_function
+def process_return(order_number: Annotated[str, "Order number to process return for"]) -> str:
+    """Simulated function to process a return for a given order number."""
+    return f"Return initiated successfully for order {order_number}. You will receive return instructions via email."
+
+
 def create_agents(chat_client: AzureOpenAIChatClient) -> tuple[ChatAgent, ChatAgent, ChatAgent, ChatAgent]:
     """Create and configure the triage and specialist agents.
 
@@ -54,51 +75,46 @@ def create_agents(chat_client: AzureOpenAIChatClient) -> tuple[ChatAgent, ChatAg
     - Signaling handoff by calling one of the explicit handoff tools exposed to it
 
     Specialist agents are invoked only when the triage agent explicitly hands off to them.
-    After a specialist responds, control returns to the triage agent.
+    After a specialist responds, control returns to the triage agent, which then prompts
+    the user for their next message.
 
     Returns:
-        Tuple of (triage_agent, refund_agent, order_agent, support_agent)
+        Tuple of (triage_agent, refund_agent, order_agent, return_agent)
     """
     # Triage agent: Acts as the frontline dispatcher
-    # NOTE: The instructions explicitly tell it to call the correct handoff tool when routing.
-    # The HandoffBuilder intercepts these tool calls and routes to the matching specialist.
-    triage = chat_client.create_agent(
+    triage_agent = chat_client.create_agent(
         instructions=(
-            "You are frontline support triage. Read the latest user message and decide whether "
-            "to hand off to refund_agent, order_agent, or support_agent. Provide a brief natural-language "
-            "response for the user. When delegation is required, call the matching handoff tool "
-            "(`handoff_to_refund_agent`, `handoff_to_order_agent`, or `handoff_to_support_agent`)."
+            "You are frontline support triage. Route customer issues to the appropriate specialist agents "
+            "based on the problem described."
         ),
         name="triage_agent",
     )
 
     # Refund specialist: Handles refund requests
-    refund = chat_client.create_agent(
-        instructions=(
-            "You handle refund workflows. Ask for any order identifiers you require and outline the refund steps."
-        ),
+    refund_agent = chat_client.create_agent(
+        instructions="You process refund requests.",
         name="refund_agent",
+        # In a real application, an agent can have multiple tools; here we keep it simple
+        tools=[process_refund],
     )
 
     # Order/shipping specialist: Resolves delivery issues
-    order = chat_client.create_agent(
-        instructions=(
-            "You resolve shipping and fulfillment issues. Clarify the delivery problem and describe the actions "
-            "you will take to remedy it."
-        ),
+    order_agent = chat_client.create_agent(
+        instructions="You handle order and shipping inquiries.",
         name="order_agent",
+        # In a real application, an agent can have multiple tools; here we keep it simple
+        tools=[check_order_status],
     )
 
-    # General support specialist: Fallback for other issues
-    support = chat_client.create_agent(
-        instructions=(
-            "You are a general support agent. Offer empathetic troubleshooting and gather missing details if the "
-            "issue does not match other specialists."
-        ),
-        name="support_agent",
+    # Return specialist: Handles return requests
+    return_agent = chat_client.create_agent(
+        instructions="You manage product return requests.",
+        name="return_agent",
+        # In a real application, an agent can have multiple tools; here we keep it simple
+        tools=[process_return],
     )
 
-    return triage, refund, order, support
+    return triage_agent, refund_agent, order_agent, return_agent
 
 
 async def _drain(stream: AsyncIterable[WorkflowEvent]) -> list[WorkflowEvent]:
@@ -139,7 +155,7 @@ def _handle_events(events: list[WorkflowEvent]) -> list[RequestInfoEvent]:
             WorkflowRunState.IDLE,
             WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
         }:
-            print(f"[status] {event.state.name}")
+            print(f"\n[Workflow Status] {event.state.name}")
 
         # WorkflowOutputEvent: Contains the final conversation when workflow terminates
         elif isinstance(event, WorkflowOutputEvent):
@@ -154,14 +170,14 @@ def _handle_events(events: list[WorkflowEvent]) -> list[RequestInfoEvent]:
         # RequestInfoEvent: Workflow is requesting user input
         elif isinstance(event, RequestInfoEvent):
             if isinstance(event.data, HandoffUserInputRequest):
-                _print_handoff_request(event.data)
+                _print_agent_responses_since_last_user_message(event.data)
             requests.append(event)
 
     return requests
 
 
-def _print_handoff_request(request: HandoffUserInputRequest) -> None:
-    """Display a user input request prompt with conversation context.
+def _print_agent_responses_since_last_user_message(request: HandoffUserInputRequest) -> None:
+    """Display agent responses since the last user message in a handoff request.
 
     The HandoffUserInputRequest contains the full conversation history so far,
     allowing the user to see what's been discussed before providing their next input.
@@ -169,11 +185,21 @@ def _print_handoff_request(request: HandoffUserInputRequest) -> None:
     Args:
         request: The user input request containing conversation and prompt
     """
-    print("\n=== User Input Requested ===")
-    for message in request.conversation:
+    if not request.conversation:
+        raise RuntimeError("HandoffUserInputRequest missing conversation history.")
+
+    # Reverse iterate to collect agent responses since last user message
+    agent_responses: list[ChatMessage] = []
+    for message in request.conversation[::-1]:
+        if message.role == Role.USER:
+            break
+        agent_responses.append(message)
+
+    # Print agent responses in original order
+    agent_responses.reverse()
+    for message in agent_responses:
         speaker = message.author_name or message.role.value
         print(f"- {speaker}: {message.text}")
-    print("============================")
 
 
 async def main() -> None:
@@ -196,20 +222,26 @@ async def main() -> None:
     triage, refund, order, support = create_agents(chat_client)
 
     # Build the handoff workflow
-    # - participants: All agents that can participate (triage MUST be first or explicitly set as set_coordinator)
-    # - set_coordinator: The triage agent receives all user input first
-    # - with_termination_condition: Custom logic to stop the request/response loop
-    #   Default is 10 user messages; here we terminate after 4 to match our scripted demo
+    # - participants: All agents that can participate in the workflow
+    # - set_coordinator: The triage agent is designated as the coordinator, which means
+    #   it receives all user input first and orchestrates handoffs to specialists
+    # - with_termination_condition: Custom logic to stop the request/response loop.
+    #   Without this, the default behavior continues requesting user input until max_turns
+    #   is reached. Here we use a custom condition that checks if the conversation has ended
+    #   naturally (when triage agent says something like "you're welcome").
     workflow = (
         HandoffBuilder(
             name="customer_support_handoff",
             participants=[triage, refund, order, support],
         )
-        .set_coordinator("triage_agent")
+        .set_coordinator(triage)
         .with_termination_condition(
-            # Terminate after 4 user messages (initial + 3 scripted responses)
-            # Count only USER role messages to avoid counting agent responses
-            lambda conv: sum(1 for msg in conv if msg.role.value == "user") >= 4
+            # Custom termination: Check if the triage agent has provided a closing message.
+            # This looks for the last message being from triage_agent and containing "welcome",
+            # which indicates the conversation has concluded naturally.
+            lambda conversation: len(conversation) > 0
+            and conversation[-1].author_name == "triage_agent"
+            and "welcome" in conversation[-1].text.lower()
         )
         .build()
     )
@@ -219,15 +251,16 @@ async def main() -> None:
     #   user_input = input("Your response: ")
     # or integrate with a UI/chat interface
     scripted_responses = [
-        "My order 1234 arrived damaged and the packaging was destroyed.",
-        "Yes, I'd like a refund if that's possible.",
+        "My order 1234 arrived damaged and the packaging was destroyed. I'd like to return it.",
         "Thanks for resolving this.",
     ]
 
     # Start the workflow with the initial user message
     # run_stream() returns an async iterator of WorkflowEvent
-    print("\n[Starting workflow with initial user message...]")
-    events = await _drain(workflow.run_stream("Hello, I need assistance with my recent purchase."))
+    print("[Starting workflow with initial user message...]\n")
+    initial_message = "Hello, I need assistance with my recent purchase."
+    print(f"- User: {initial_message}")
+    events = await _drain(workflow.run_stream(initial_message))
     pending_requests = _handle_events(events)
 
     # Process the request/response cycle
@@ -237,13 +270,15 @@ async def main() -> None:
     while pending_requests and scripted_responses:
         # Get the next scripted response
         user_response = scripted_responses.pop(0)
-        print(f"\n[User responding: {user_response}]")
+        print(f"\n- User: {user_response}")
 
         # Send response(s) to all pending requests
         # In this demo, there's typically one request per cycle, but the API supports multiple
         responses = {req.request_id: user_response for req in pending_requests}
 
         # Send responses and get new events
+        # We use send_responses_streaming() to get events as they occur, allowing us to
+        # display agent responses in real-time and handle new requests as they arrive
         events = await _drain(workflow.send_responses_streaming(responses))
         pending_requests = _handle_events(events)
 
@@ -252,84 +287,30 @@ async def main() -> None:
 
     [Starting workflow with initial user message...]
 
-    === User Input Requested ===
+    - User: Hello, I need assistance with my recent purchase.
+    - triage_agent: Could you please provide more details about the issue you're experiencing with your recent purchase? This will help me route you to the appropriate specialist.
+
+    [Workflow Status] IDLE_WITH_PENDING_REQUESTS
+
+    - User: My order 1234 arrived damaged and the packaging was destroyed. I'd like to return it.
+    - triage_agent: I've directed your request to our return agent, who will assist you with returning the damaged order. Thank you for your patience!
+    - return_agent: The return for your order 1234 has been successfully initiated. You will receive return instructions via email shortly. If you have any other questions or need further assistance, feel free to ask!
+
+    [Workflow Status] IDLE_WITH_PENDING_REQUESTS
+
+    - User: Thanks for resolving this.
+
+    === Final Conversation Snapshot ===
     - user: Hello, I need assistance with my recent purchase.
-    - triage_agent: I'd be happy to help you with your recent purchase. Could you please provide more details about the issue you're experiencing?
-    ============================
-    [status] IDLE_WITH_PENDING_REQUESTS
-
-    [User responding: My order 1234 arrived damaged and the packaging was destroyed.]
-
-    === User Input Requested ===
-    - user: Hello, I need assistance with my recent purchase.
-    - triage_agent: I'd be happy to help you with your recent purchase. Could you please provide more details about the issue you're experiencing?
-    - user: My order 1234 arrived damaged and the packaging was destroyed.
-    - triage_agent: I'm sorry to hear that your order arrived damaged and the packaging was destroyed. I will connect you with a specialist who can assist you further with this issue.
-
-    Tool Call: handoff_to_support_agent (awaiting approval)
-    - support_agent: I'm so sorry to hear that your order arrived in such poor condition. I'll help you get this sorted out.
-
-    To assist you better, could you please let me know:
-    - Which item(s) from order 1234 arrived damaged?
-    - Could you describe the damage, or provide photos if possible?
-    - Would you prefer a replacement or a refund?
-
-    Once I have this information, I can help resolve this for you as quickly as possible.
-    ============================
-    [status] IDLE_WITH_PENDING_REQUESTS
-
-    [User responding: Yes, I'd like a refund if that's possible.]
-
-    === User Input Requested ===
-    - user: Hello, I need assistance with my recent purchase.
-    - triage_agent: I'd be happy to help you with your recent purchase. Could you please provide more details about the issue you're experiencing?
-    - user: My order 1234 arrived damaged and the packaging was destroyed.
-    - triage_agent: I'm sorry to hear that your order arrived damaged and the packaging was destroyed. I will connect you with a specialist who can assist you further with this issue.
-
-    Tool Call: handoff_to_support_agent (awaiting approval)
-    - support_agent: I'm so sorry to hear that your order arrived in such poor condition. I'll help you get this sorted out.
-
-    To assist you better, could you please let me know:
-    - Which item(s) from order 1234 arrived damaged?
-    - Could you describe the damage, or provide photos if possible?
-    - Would you prefer a replacement or a refund?
-
-    Once I have this information, I can help resolve this for you as quickly as possible.
-    - user: Yes, I'd like a refund if that's possible.
-    - triage_agent: Thank you for letting me know you'd prefer a refund. I'll connect you with a specialist who can process your refund request.
-
-    Tool Call: handoff_to_refund_agent (awaiting approval)
-    - refund_agent: Thank you for confirming that you'd like a refund for order 1234.
-
-    Here's what will happen next:
-
-    ...
-
-    Tool Call: handoff_to_refund_agent (awaiting approval)
-    - refund_agent: Thank you for confirming that you'd like a refund for order 1234.
-
-    Here's what will happen next:
-
-    **1. Verification:**
-    I will need to verify a few more details to proceed.
-    - Can you confirm the items in order 1234 that arrived damaged?
-    - Do you have any photos of the damaged items/packaging? (Photos help speed up the process.)
-
-    **2. Refund Request Submission:**
-    - Once I have the details, I will submit your refund request for review.
-
-    **3. Return Instructions (if needed):**
-    - In some cases, we may provide instructions on how to return the damaged items.
-    - You will receive a prepaid return label if necessary.
-
-    **4. Refund Processing:**
-    - After your request is approved (and any returns are received if required), your refund will be processed.
-    - Refunds usually appear on your original payment method within 5-10 business days.
-
-    Could you please reply with the specific item(s) damaged and, if possible, attach photos? This will help me get your refund started right away.
+    - triage_agent: Could you please provide more details about the issue you're experiencing with your recent purchase? This will help me route you to the appropriate specialist.
+    - user: My order 1234 arrived damaged and the packaging was destroyed. I'd like to return it.
+    - triage_agent: I've directed your request to our return agent, who will assist you with returning the damaged order. Thank you for your patience!
+    - return_agent: The return for your order 1234 has been successfully initiated. You will receive return instructions via email shortly. If you have any other questions or need further assistance, feel free to ask!
     - user: Thanks for resolving this.
+    - triage_agent: You're welcome! If you have any more questions or need assistance in the future, feel free to reach out. Have a great day!
     ===================================
-    [status] IDLE
+
+    [Workflow Status] IDLE
     """  # noqa: E501
 
 

python/samples/getting_started/workflows/orchestration/handoff_with_code_interpreter_file.py

@@ -26,9 +26,8 @@ Prerequisites:
 """
 
 import asyncio
-from collections.abc import AsyncIterable
+from collections.abc import AsyncIterable, AsyncIterator
 from contextlib import asynccontextmanager
-from collections.abc import AsyncIterator
 
 from agent_framework import (
     AgentRunUpdateEvent,