agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: [BREAKING] Standardize orchestration terminal outputs as AgentResponse (#5301)

Browse Source 
* Fix orchestration outputs so as_agent() returns the final answer only. Align other orchestration outputs

* Fix orchestration output issues from review comments

1. Sample cleanup: Remove commented-out FoundryChatClient block and update
   prerequisites to reference OPENAI_CHAT_MODEL_ID instead of FOUNDRY_* vars.

2. Sequential approval output: Change _EndWithConversation.end_with_agent_executor_response
   from a no-op sink to yield response.agent_response. When the last participant is
   AgentApprovalExecutor (via with_request_info), _EndWithConversation is the output
   executor so the yield produces the terminal answer. When the last participant is a
   regular AgentExecutor, _EndWithConversation is not in output_executors so the yield
   is silently filtered out.

3. Forward data events through WorkflowExecutor: _process_workflow_result now also
   forwards 'data' events from sub-workflows so that emit_intermediate_data=True on
   AgentExecutor works correctly when wrapped in AgentApprovalExecutor.

4. Concurrent docstring: Update _AggregateAgentConversations docstring to say
   'deterministic participant order' instead of 'completion order'.

5. Add test_concurrent_intermediate_outputs_emits_data_events verifying that
   ConcurrentBuilder(intermediate_outputs=True) emits per-participant data events
   alongside the single aggregated output event.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Add tests for sequential workflow with_request_info and intermediate_outputs (#5301)

Address PR review comments 2, 3, and 5:

- Add test_sequential_request_info_last_participant_emits_output:
  Verifies that when the last participant is wrapped via with_request_info()
  (AgentApprovalExecutor), the workflow still emits a terminal output after
  approval, exercising the _EndWithConversation.end_with_agent_executor_response
  fallback path.

- Add test_sequential_request_info_with_intermediate_outputs_emits_data_events:
  Verifies that emit_intermediate_data=True works correctly through
  AgentApprovalExecutor wrapping—WorkflowExecutor._process_result already
  forwards data events from sub-workflows, so intermediate agent responses
  surface as data events in the parent workflow.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Fix pyright type errors from AgentResponse output refactor (#5301)

Update cast() calls in _group_chat.py and _magentic.py to use
WorkflowContext[Never, AgentResponse] instead of the old
WorkflowContext[Never, list[Message]], matching the updated method
signatures in _base_group_chat_orchestrator.py.

Fix _sequential.py _EndWithConversation.end_with_agent_executor_response
to declare WorkflowContext[Any, AgentResponse] so yield_output accepts
AgentResponse[None].

Fix _workflow_executor.py data event forwarding to handle nullable
executor_id.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Fix pyright reportUnknownVariableType in _agent.py (#5301)

Extract event.data into a typed local variable before the isinstance
check to avoid pyright narrowing it to AgentResponse[Unknown].

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Fix pyright reportMissingImports for orjson in file history samples (#5301)

Add pyright: ignore[reportMissingImports] to orjson imports that are
already guarded by try/except ImportError, matching the existing pattern
used elsewhere in the samples.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Address review feedback for #5301: review comment fixes

* Address review feedback for #5301: review comment fixes

* Revert sequential_workflow_as_agent sample to FoundryChatClient

Reverts the mistaken switch from FoundryChatClient to OpenAIChatClient
in the sequential workflow as agent sample.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Address ultrareview feedback: emit_data_events rename + WorkflowAgent reasoning conversion

Layered on top of the prior review-feedback work in this branch.

Renames:
- AgentExecutor.emit_intermediate_data -> emit_data_events (mechanical
  rename; orchestration semantics live at the orchestration layer, not
  the general-purpose executor). Forwarded through MagenticAgentExecutor,
  AgentApprovalExecutor, and all orchestration call sites.
- HandoffAgentExecutor._check_terminate_and_yield -> _should_terminate
  (pure predicate; no longer yields anything). HandoffBuilder docstring
  rewritten to describe the new per-agent AgentResponse output contract.

WorkflowAgent reasoning-content conversion:
- Add _rewrite_text_to_reasoning(contents) and _msg_as_reasoning(msg)
  helpers; the as_agent() path now reframes text content from data events
  as text_reasoning Content blocks before merging into the AgentResponse.
- Consumers iterate msg.contents and branch on content.type — same path
  they already use for Claude thinking and OpenAI reasoning. No new
  field on Message/AgentResponse/WorkflowEvent.
- Streaming branch constructs fresh AgentResponseUpdate instances instead
  of mutating shared payloads (regression test added).
- Helper _msg_maybe_reasoning consolidates the conditional rewrite at
  three call sites in the non-streaming conversion.

Tests:
- TestWorkflowAgentReasoningHelpers + TestWorkflowAgentDataEventReasoningConversion
  add 9 new tests covering helpers, non-streaming, streaming, mixed content,
  already-reasoning passthrough, and mutation-safety regression.
- Updated test_sequential_as_agent_with_intermediate_outputs_includes_chain
  to assert text_reasoning content for intermediate agents.

* Fix pyright: widen event.data to Any to avoid partial-unknown narrowing

The streaming conversion path narrowed event.data via isinstance against
generic AgentResponse, producing AgentResponse[Unknown] and tripping
reportUnknownVariableType/reportUnknownMemberType. Binding data: Any
before the check keeps runtime behavior identical while restoring a fully
known type for downstream access.

* Clean up design

* Scope to agent output semantics only

* yield AgentResponseUpdate streaming, AgentResponse non-streaming

* Fix mypy/pyright: widen cast types at GroupChat callsites

Eight callsites in _group_chat.py still cast to WorkflowContext[Never,
AgentResponse] but the base orchestrator methods now accept the wider
WorkflowContext[Never, AgentResponse | AgentResponseUpdate] (mode-aware
yields). W_OutT is invariant, so the narrower cast is not assignable.
Magentic was widened in the same commit; this catches the GroupChat
callsites that were missed.

* Python: skip flaky Foundry / Foundry Hosting integration tests (#5553)

These two integration tests have been failing in the merge queue across
multiple unrelated PRs (5301, 5531). Both are marked `@pytest.mark.flaky`
with 3 retries, but all attempts fail back-to-back. Skipping both with a
reason pointing to #5553 so they can be fixed properly without continuing
to block unrelated merges.

- packages/foundry_hosting/tests/test_responses_int.py::TestOptions::test_temperature_and_max_tokens
- packages/foundry/tests/foundry/test_foundry_embedding_client.py::TestFoundryEmbeddingIntegration::test_text_embedding_live

Also includes a one-line uv.lock specifier-ordering normalization
auto-applied by the poe-check pre-commit hook.

---------

Co-authored-by: Copilot <copilot@github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Evan Mattson
2026-04-29 09:35:36 +09:00
committed by GitHub
Unverified
parent 40e90c96c3
commit 866a325b48
22 changed files with 785 additions and 490 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/orchestrations/tests/test_concurrent.py
+71 -31
View File
@@ -1,14 +1,21 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import Any, cast
from collections.abc import AsyncIterable, Awaitable
from typing import Any, Literal, cast, overload
import pytest
from agent_framework import (
AgentExecutorRequest,
AgentExecutorResponse,
AgentResponse,
AgentResponseUpdate,
AgentRunInputs,
AgentSession,
BaseAgent,
Content,
Executor,
Message,
ResponseStream,
WorkflowContext,
WorkflowRunState,
handler,
@@ -49,36 +56,26 @@ def test_concurrent_builder_rejects_duplicate_executors() -> None:
ConcurrentBuilder(participants=[a, b])
async def test_concurrent_default_aggregator_emits_single_user_and_assistants() -> None:
# Three synthetic agent executors
async def test_concurrent_default_aggregator_emits_assistants_only() -> None:
"""Default aggregator yields a single AgentResponse with one assistant message per participant.
The user prompt is intentionally not included — that belongs in the input, not the answer.
"""
e1 = _FakeAgentExec("agentA", "Alpha")
e2 = _FakeAgentExec("agentB", "Beta")
e3 = _FakeAgentExec("agentC", "Gamma")
wf = ConcurrentBuilder(participants=[e1, e2, e3]).build()
completed = False
output: list[Message] | None = None
async for ev in wf.run("prompt: hello world", stream=True):
if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
completed = True
elif ev.type == "output":
output = cast(list[Message], ev.data)
if completed and output is not None:
break
output_events = [ev for ev in await wf.run("prompt: hello world") if ev.type == "output"]
assert len(output_events) == 1
response = output_events[0].data
assert isinstance(response, AgentResponse)
assert completed
assert output is not None
messages: list[Message] = output
# Expect one user message + one assistant message per participant
assert len(messages) == 1 + 3
assert messages[0].role == "user"
assert "hello world" in messages[0].text
assistant_texts = {m.text for m in messages[1:]}
assert assistant_texts == {"Alpha", "Beta", "Gamma"}
assert all(m.role == "assistant" for m in messages[1:])
# Exactly one assistant message per participant; no user prompt.
assert len(response.messages) == 3
assert all(m.role == "assistant" for m in response.messages)
assert {m.text for m in response.messages} == {"Alpha", "Beta", "Gamma"}
async def test_concurrent_custom_aggregator_callback_is_used() -> None:
@@ -215,7 +212,7 @@ async def test_concurrent_checkpoint_resume_round_trip() -> None:
wf = ConcurrentBuilder(participants=list(participants), checkpoint_storage=storage).build()
baseline_output: list[Message] | None = None
baseline_output: AgentResponse | None = None
async for ev in wf.run("checkpoint concurrent", stream=True):
if ev.type == "output":
baseline_output = ev.data # type: ignore[assignment]
@@ -236,7 +233,7 @@ async def test_concurrent_checkpoint_resume_round_trip() -> None:
)
wf_resume = ConcurrentBuilder(participants=list(resumed_participants), checkpoint_storage=storage).build()
resumed_output: list[Message] | None = None
resumed_output: AgentResponse | None = None
async for ev in wf_resume.run(checkpoint_id=resume_checkpoint.checkpoint_id, stream=True):
if ev.type == "output":
resumed_output = ev.data # type: ignore[assignment]
@@ -247,8 +244,8 @@ async def test_concurrent_checkpoint_resume_round_trip() -> None:
break
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
assert [m.role for m in resumed_output.messages] == [m.role for m in baseline_output.messages]
assert [m.text for m in resumed_output.messages] == [m.text for m in baseline_output.messages]
async def test_concurrent_checkpoint_runtime_only() -> None:
@@ -258,7 +255,7 @@ async def test_concurrent_checkpoint_runtime_only() -> None:
agents = [_FakeAgentExec(id="agent1", reply_text="A1"), _FakeAgentExec(id="agent2", reply_text="A2")]
wf = ConcurrentBuilder(participants=agents).build()
baseline_output: list[Message] | None = None
baseline_output: AgentResponse | None = None
async for ev in wf.run("runtime checkpoint test", checkpoint_storage=storage, stream=True):
if ev.type == "output":
baseline_output = ev.data # type: ignore[assignment]
@@ -278,7 +275,7 @@ async def test_concurrent_checkpoint_runtime_only() -> None:
resumed_agents = [_FakeAgentExec(id="agent1", reply_text="A1"), _FakeAgentExec(id="agent2", reply_text="A2")]
wf_resume = ConcurrentBuilder(participants=resumed_agents).build()
resumed_output: list[Message] | None = None
resumed_output: AgentResponse | None = None
async for ev in wf_resume.run(
checkpoint_id=resume_checkpoint.checkpoint_id, checkpoint_storage=storage, stream=True
):
@@ -291,7 +288,7 @@ async def test_concurrent_checkpoint_runtime_only() -> None:
break
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
assert [m.role for m in resumed_output.messages] == [m.role for m in baseline_output.messages]
async def test_concurrent_checkpoint_runtime_overrides_buildtime() -> None:
@@ -334,3 +331,46 @@ async def test_concurrent_builder_reusable_after_build_with_participants() -> No
assert builder._participants[0] is e1 # type: ignore
assert builder._participants[1] is e2 # type: ignore
class _EchoAgent(BaseAgent):
"""Simple agent that appends a single assistant message with its name."""
@overload
def run(
self,
messages: AgentRunInputs | None = ...,
*,
stream: Literal[False] = ...,
session: AgentSession | None = ...,
**kwargs: Any,
) -> Awaitable[AgentResponse[Any]]: ...
@overload
def run(
self,
messages: AgentRunInputs | None = ...,
*,
stream: Literal[True],
session: AgentSession | None = ...,
**kwargs: Any,
) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...
def run(
self,
messages: AgentRunInputs | None = None,
*,
stream: bool = False,
session: AgentSession | None = None,
**kwargs: Any,
) -> Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
if stream:
async def _stream() -> AsyncIterable[AgentResponseUpdate]:
yield AgentResponseUpdate(contents=[Content.from_text(text=f"{self.name} reply")])
return ResponseStream(_stream(), finalizer=AgentResponse.from_updates)
async def _run() -> AgentResponse:
return AgentResponse(messages=[Message("assistant", [f"{self.name} reply"])])
return _run()
python/packages/orchestrations/tests/test_group_chat.py
+150 -116
View File
@@ -238,18 +238,16 @@ async def test_group_chat_builder_basic_flow() -> None:
orchestrator_name="manager",
).build()
outputs: list[list[Message]] = []
updates: list[AgentResponseUpdate] = []
async for event in workflow.run("coordinate task", stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list):
outputs.append(cast(list[Message], data))
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
assert len(outputs) == 1
assert len(outputs[0]) >= 1
# Check that both agents contributed
authors = {msg.author_name for msg in outputs[0] if msg.author_name in ["alpha", "beta"]}
assert len(authors) == 2
# Exactly one terminal `output` event = the orchestrator's completion AgentResponseUpdate
# (mode-aware: streaming yields a single update chunk for the synthesized message).
assert len(updates) == 1
# The completion message is authored by the orchestrator.
assert updates[0].author_name == "manager"
async def test_group_chat_as_agent_accepts_conversation() -> None:
@@ -283,18 +281,16 @@ async def test_agent_manager_handles_concatenated_json_output() -> None:
orchestrator_agent=manager,
).build()
outputs: list[list[Message]] = []
updates: list[AgentResponseUpdate] = []
async for event in workflow.run("coordinate task", stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list):
outputs.append(cast(list[Message], data))
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
assert outputs
conversation = outputs[-1]
assert any(msg.author_name == "agent" and msg.text == "worker response" for msg in conversation)
assert conversation[-1].author_name == manager.name
assert conversation[-1].text == "concatenated manager final"
assert updates
final_update = updates[-1]
# Terminal update is the orchestrator's completion message.
assert final_update.author_name == manager.name
assert final_update.text == "concatenated manager final"
# Comprehensive tests for group chat functionality
@@ -400,20 +396,14 @@ class TestGroupChatWorkflow:
selection_func=selector,
).build()
outputs: list[list[Message]] = []
updates: list[AgentResponseUpdate] = []
async for event in workflow.run("test task", stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list):
outputs.append(cast(list[Message], data))
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
# Should have terminated due to max_rounds, expect at least one output
assert len(outputs) >= 1
# The final message in the conversation should be about round limit
conversation = outputs[-1]
assert len(conversation) >= 1
final_output = conversation[-1]
assert "maximum number of rounds" in final_output.text.lower()
# Exactly one terminal output event = orchestrator's max-rounds completion update.
assert len(updates) == 1
assert "maximum number of rounds" in (updates[0].text or "").lower()
async def test_termination_condition_halts_conversation(self) -> None:
"""Test that a custom termination condition stops the workflow."""
@@ -433,20 +423,89 @@ class TestGroupChatWorkflow:
selection_func=selector,
).build()
outputs: list[list[Message]] = []
updates: list[AgentResponseUpdate] = []
async for event in workflow.run("test task", stream=True):
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
assert updates, "Expected termination to yield output"
# Terminal update is the orchestrator's completion message only.
assert "termination condition" in (updates[-1].text or "").lower()
async def test_termination_yields_update_in_streaming(self) -> None:
"""In streaming mode, the orchestrator's terminal completion surfaces as `AgentResponseUpdate`.
Mirrors AgentExecutor's mode-aware behavior: streaming workflows produce per-chunk
`AgentResponseUpdate` events; the synthesized termination message is logically a
single chunk, so it should be a single `AgentResponseUpdate`.
"""
def selector(state: GroupChatState) -> str:
return "agent"
def termination_condition(conversation: list[Message]) -> bool:
replies = [msg for msg in conversation if msg.role == "assistant" and msg.author_name == "agent"]
return len(replies) >= 2
workflow = GroupChatBuilder(
participants=[StubAgent("agent", "response")],
termination_condition=termination_condition,
selection_func=selector,
).build()
terminal: AgentResponseUpdate | None = None
async for event in workflow.run("test task", stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list):
outputs.append(cast(list[Message], data))
terminal = event.data # last output event wins
assert outputs, "Expected termination to yield output"
conversation = outputs[-1]
agent_replies = [msg for msg in conversation if msg.author_name == "agent" and msg.role == "assistant"]
assert len(agent_replies) == 2
final_output = conversation[-1]
# The orchestrator uses its ID as author_name by default
assert "termination condition" in final_output.text.lower()
assert isinstance(terminal, AgentResponseUpdate), (
f"Expected AgentResponseUpdate in streaming mode, got {type(terminal).__name__}"
)
assert "termination condition" in (terminal.text or "").lower()
async def test_termination_yields_response_in_non_streaming(self) -> None:
"""In non-streaming mode, the orchestrator's terminal completion surfaces as `AgentResponse`."""
def selector(state: GroupChatState) -> str:
return "agent"
def termination_condition(conversation: list[Message]) -> bool:
replies = [msg for msg in conversation if msg.role == "assistant" and msg.author_name == "agent"]
return len(replies) >= 2
workflow = GroupChatBuilder(
participants=[StubAgent("agent", "response")],
termination_condition=termination_condition,
selection_func=selector,
).build()
events = await workflow.run("test task")
outputs = [ev for ev in events if ev.type == "output"]
assert len(outputs) == 1
assert isinstance(outputs[0].data, AgentResponse)
assert "termination condition" in outputs[0].data.messages[-1].text.lower()
async def test_max_rounds_yields_update_in_streaming(self) -> None:
"""Max-rounds completion in streaming mode surfaces as `AgentResponseUpdate`."""
def selector(state: GroupChatState) -> str:
return "agent"
workflow = GroupChatBuilder(
participants=[StubAgent("agent", "response")],
max_rounds=2,
selection_func=selector,
).build()
terminal: AgentResponseUpdate | None = None
async for event in workflow.run("test task", stream=True):
if event.type == "output":
terminal = event.data
assert isinstance(terminal, AgentResponseUpdate), (
f"Expected AgentResponseUpdate in streaming mode, got {type(terminal).__name__}"
)
assert "maximum number of rounds" in (terminal.text or "").lower()
async def test_termination_condition_agent_manager_finalizes(self) -> None:
"""Test that termination condition with agent orchestrator produces default termination message."""
@@ -459,17 +518,15 @@ class TestGroupChatWorkflow:
orchestrator_agent=manager,
).build()
outputs: list[list[Message]] = []
updates: list[AgentResponseUpdate] = []
async for event in workflow.run("test task", stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list):
outputs.append(cast(list[Message], data))
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
assert outputs, "Expected termination to yield output"
conversation = outputs[-1]
assert conversation[-1].text == BaseGroupChatOrchestrator.TERMINATION_CONDITION_MET_MESSAGE
assert conversation[-1].author_name == manager.name
assert updates, "Expected termination to yield output"
final_update = updates[-1]
assert final_update.text == BaseGroupChatOrchestrator.TERMINATION_CONDITION_MET_MESSAGE
assert final_update.author_name == manager.name
async def test_unknown_participant_error(self) -> None:
"""Test that unknown participant selection raises error."""
@@ -505,14 +562,12 @@ class TestCheckpointing:
selection_func=selector,
).build()
outputs: list[list[Message]] = []
updates: list[AgentResponseUpdate] = []
async for event in workflow.run("test task", stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list):
outputs.append(cast(list[Message], data))
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
assert len(outputs) == 1 # Should complete normally
assert len(updates) == 1 # Should complete normally
class TestConversationHandling:
@@ -546,14 +601,12 @@ class TestConversationHandling:
workflow = GroupChatBuilder(participants=[agent], max_rounds=1, selection_func=selector).build()
outputs: list[list[Message]] = []
updates: list[AgentResponseUpdate] = []
async for event in workflow.run("test string", stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list):
outputs.append(cast(list[Message], data))
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
assert len(outputs) == 1
assert len(updates) == 1
async def test_handle_chat_message_input(self) -> None:
"""Test handling Message input directly."""
@@ -569,14 +622,12 @@ class TestConversationHandling:
workflow = GroupChatBuilder(participants=[agent], max_rounds=1, selection_func=selector).build()
outputs: list[list[Message]] = []
updates: list[AgentResponseUpdate] = []
async for event in workflow.run(task_message, stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list):
outputs.append(cast(list[Message], data))
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
assert len(outputs) == 1
assert len(updates) == 1
async def test_handle_conversation_list_input(self) -> None:
"""Test handling conversation list preserves context."""
@@ -595,14 +646,12 @@ class TestConversationHandling:
workflow = GroupChatBuilder(participants=[agent], max_rounds=1, selection_func=selector).build()
outputs: list[list[Message]] = []
updates: list[AgentResponseUpdate] = []
async for event in workflow.run(conversation, stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list):
outputs.append(cast(list[Message], data))
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
assert len(outputs) == 1
assert len(updates) == 1
class TestRoundLimitEnforcement:
@@ -625,20 +674,14 @@ class TestRoundLimitEnforcement:
selection_func=selector,
).build()
outputs: list[list[Message]] = []
updates: list[AgentResponseUpdate] = []
async for event in workflow.run("test", stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list):
outputs.append(cast(list[Message], data))
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
# Should have at least one output (the round limit message)
assert len(outputs) >= 1
# The last message in the conversation should be about round limit
conversation = outputs[-1]
assert len(conversation) >= 1
final_output = conversation[-1]
assert "maximum number of rounds" in final_output.text.lower()
# Exactly one terminal output event = orchestrator's max-rounds completion update.
assert len(updates) == 1
assert "maximum number of rounds" in (updates[0].text or "").lower()
async def test_round_limit_in_ingest_participant_message(self) -> None:
"""Test round limit enforcement after participant response."""
@@ -658,20 +701,14 @@ class TestRoundLimitEnforcement:
selection_func=selector,
).build()
outputs: list[list[Message]] = []
updates: list[AgentResponseUpdate] = []
async for event in workflow.run("test", stream=True):
if event.type == "output":
data = event.data
if isinstance(data, list):
outputs.append(cast(list[Message], data))
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
# Should have at least one output (the round limit message)
assert len(outputs) >= 1
# The last message in the conversation should be about round limit
conversation = outputs[-1]
assert len(conversation) >= 1
final_output = conversation[-1]
assert "maximum number of rounds" in final_output.text.lower()
# Exactly one terminal output event = orchestrator's max-rounds completion update.
assert len(updates) == 1
assert "maximum number of rounds" in (updates[0].text or "").lower()
async def test_group_chat_checkpoint_runtime_only() -> None:
@@ -684,17 +721,17 @@ async def test_group_chat_checkpoint_runtime_only() -> None:
wf = GroupChatBuilder(participants=[agent_a, agent_b], max_rounds=2, selection_func=selector).build()
baseline_output: list[Message] | None = None
baseline_update: AgentResponseUpdate | None = None
async for ev in wf.run("runtime checkpoint test", checkpoint_storage=storage, stream=True):
if ev.type == "output":
baseline_output = cast(list[Message], ev.data) if isinstance(ev.data, list) else None # type: ignore
if ev.type == "output" and isinstance(ev.data, AgentResponseUpdate):
baseline_update = ev.data
if ev.type == "status" and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert baseline_output is not None
assert baseline_update is not None
checkpoints = await storage.list_checkpoints(workflow_name=wf.name)
assert len(checkpoints) > 0, "Runtime-only checkpointing should have created checkpoints"
@@ -720,17 +757,17 @@ async def test_group_chat_checkpoint_runtime_overrides_buildtime() -> None:
checkpoint_storage=buildtime_storage,
selection_func=selector,
).build()
baseline_output: list[Message] | None = None
baseline_update: AgentResponseUpdate | None = None
async for ev in wf.run("override test", checkpoint_storage=runtime_storage, stream=True):
if ev.type == "output":
baseline_output = cast(list[Message], ev.data) if isinstance(ev.data, list) else None # type: ignore
if ev.type == "output" and isinstance(ev.data, AgentResponseUpdate):
baseline_update = ev.data
if ev.type == "status" and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert baseline_output is not None
assert baseline_update is not None
buildtime_checkpoints = await buildtime_storage.list_checkpoints(workflow_name=wf.name)
runtime_checkpoints = await runtime_storage.list_checkpoints(workflow_name=wf.name)
@@ -974,14 +1011,11 @@ async def test_group_chat_with_orchestrator_factory_returning_chat_agent():
outputs.append(event)
assert len(outputs) == 1
# The DynamicManagerAgent terminates after second call with final_message
final_messages = outputs[0].data
assert isinstance(final_messages, list)
assert any(
msg.text == "dynamic manager final"
for msg in cast(list[Message], final_messages)
if msg.author_name == "dynamic_manager"
)
# Streaming mode: terminal yield is AgentResponseUpdate. The DynamicManagerAgent
# terminates after second call with final_message.
final_update = outputs[0].data
assert isinstance(final_update, AgentResponseUpdate)
assert final_update.text == "dynamic manager final"
def test_group_chat_with_orchestrator_factory_returning_base_orchestrator():
python/packages/orchestrations/tests/test_handoff.py
+18 -14
View File
@@ -9,6 +9,8 @@ from unittest.mock import AsyncMock, MagicMock
import pytest
from agent_framework import (
Agent,
AgentResponse,
AgentResponseUpdate,
ChatResponse,
ChatResponseUpdate,
Content,
@@ -856,10 +858,15 @@ async def test_autonomous_mode_yields_output_without_user_request():
outputs = [ev for ev in events if ev.type == "output"]
assert outputs, "Autonomous mode should yield a workflow output"
final_conversation = outputs[-1].data
assert isinstance(final_conversation, list)
conversation_list = cast(list[Message], final_conversation)
assert any(msg.role == "assistant" and (msg.text or "").startswith("specialist reply") for msg in conversation_list)
# Per-agent activity surfaces as `output` events from each HandoffAgentExecutor as they
# speak. Handoff has no orchestrator that produces a separate "answer" — the conversation
# IS the result. In streaming mode payloads are AgentResponseUpdate; combined text should
# contain the specialist's reply.
payloads = [ev.data for ev in outputs if isinstance(ev.data, (AgentResponse, AgentResponseUpdate))]
combined = " ".join(
getattr(p, "text", None) or " ".join(m.text for m in getattr(p, "messages", [])) for p in payloads
)
assert "specialist reply" in combined
async def test_autonomous_mode_resumes_user_input_on_turn_limit():
@@ -923,14 +930,10 @@ async def test_handoff_async_termination_condition() -> None:
stream=True, responses={requests[-1].request_id: [Message(role="user", contents=["Second user message"])]}
)
)
outputs = [ev for ev in events if ev.type == "output"]
assert len(outputs) == 1
final_conversation = outputs[0].data
assert isinstance(final_conversation, list)
final_conv_list = cast(list[Message], final_conversation)
user_messages = [msg for msg in final_conv_list if msg.role == "user"]
assert len(user_messages) == 2
# Resume run terminates without further agent activity once the second user message
# satisfies the termination condition. The workflow returns to idle cleanly.
idle_states = [ev for ev in events if ev.type == "status" and ev.state == WorkflowRunState.IDLE]
assert idle_states, "Workflow should become idle after termination"
assert termination_call_count > 0
@@ -990,8 +993,9 @@ async def test_handoff_terminates_without_request_info_when_latest_response_meet
outputs = [event for event in events if event.type == "output"]
assert outputs
conversation_outputs = [event for event in outputs if isinstance(event.data, list)]
assert len(conversation_outputs) == 1
# Per-agent activity surfaces as output events (AgentResponseUpdate in streaming mode).
agent_payloads = [event for event in outputs if isinstance(event.data, (AgentResponse, AgentResponseUpdate))]
assert len(agent_payloads) >= 1
async def test_tool_choice_preserved_from_agent_config():
python/packages/orchestrations/tests/test_magentic.py
+76 -22
View File
@@ -190,24 +190,82 @@ async def test_magentic_builder_returns_workflow_and_runs() -> None:
assert isinstance(workflow, Workflow)
outputs: list[Message] = []
updates: list[AgentResponseUpdate] = []
orchestrator_event_count = 0
async for event in workflow.run("compose summary", stream=True):
if event.type == "output":
msg = event.data
if isinstance(msg, list):
outputs.extend(cast(list[Message], msg))
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
updates.append(event.data)
elif event.type == "magentic_orchestrator":
orchestrator_event_count += 1
assert outputs, "Expected a final output message"
assert len(outputs) >= 1
final = outputs[-1]
assert updates, "Expected a final output update"
final = updates[-1]
assert final.text == manager.FINAL_ANSWER
assert final.author_name == manager.name
assert orchestrator_event_count > 0, "Expected orchestrator events to be emitted"
async def test_magentic_final_answer_yields_update_in_streaming() -> None:
"""In streaming mode, Magentic's manager final-answer surfaces as `AgentResponseUpdate`.
Mirrors AgentExecutor's mode-aware behavior: streaming workflows produce per-chunk
`AgentResponseUpdate` events; the synthesized final answer is logically a single chunk,
so it surfaces as a single `AgentResponseUpdate`.
"""
manager = FakeManager()
workflow = MagenticBuilder(
participants=[StubAgent(manager.next_speaker_name, "first draft")],
manager=manager,
).build()
terminal: AgentResponseUpdate | None = None
async for event in workflow.run("compose summary", stream=True):
if event.type == "output":
terminal = event.data
assert isinstance(terminal, AgentResponseUpdate), (
f"Expected AgentResponseUpdate in streaming mode, got {type(terminal).__name__}"
)
assert terminal.text == manager.FINAL_ANSWER
assert terminal.author_name == manager.name
async def test_magentic_final_answer_yields_response_in_non_streaming() -> None:
"""In non-streaming mode, Magentic's manager final-answer surfaces as `AgentResponse`."""
manager = FakeManager()
workflow = MagenticBuilder(
participants=[StubAgent(manager.next_speaker_name, "first draft")],
manager=manager,
).build()
events = await workflow.run("compose summary")
outputs = [ev for ev in events if ev.type == "output"]
assert len(outputs) == 1
assert isinstance(outputs[0].data, AgentResponse)
assert outputs[0].data.messages[-1].text == manager.FINAL_ANSWER
async def test_magentic_limit_termination_yields_update_in_streaming() -> None:
"""In streaming mode, Magentic's round-limit termination surfaces as `AgentResponseUpdate`."""
manager = FakeManager(max_round_count=1)
workflow = MagenticBuilder(
participants=[DummyExec(name=manager.next_speaker_name)],
manager=manager,
).build()
terminal: AgentResponseUpdate | None = None
async for event in workflow.run("round limit test", stream=True):
if event.type == "output":
terminal = event.data
assert isinstance(terminal, AgentResponseUpdate), (
f"Expected AgentResponseUpdate in streaming mode, got {type(terminal).__name__}"
)
# Either the final answer OR the round-limit termination message — both are valid terminal states
# for max_round_count=1; the precise one depends on FakeManager's progression.
assert terminal.text
async def test_magentic_as_agent_does_not_accept_conversation() -> None:
manager = FakeManager()
writer = StubAgent(manager.next_speaker_name, "summary response")
@@ -250,7 +308,7 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
assert isinstance(req_event.data, MagenticPlanReviewRequest)
completed = False
output: list[Message] | None = None
output: AgentResponseUpdate | None = None
async for ev in wf.run(stream=True, responses={req_event.request_id: req_event.data.approve()}):
if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
completed = True
@@ -261,8 +319,8 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
assert completed
assert output is not None
assert isinstance(output, list)
assert all(isinstance(msg, Message) for msg in output)
# Streaming mode: terminal output is AgentResponseUpdate.
assert isinstance(output, AgentResponseUpdate)
async def test_magentic_plan_review_with_revise():
@@ -333,14 +391,12 @@ async def test_magentic_orchestrator_round_limit_produces_partial_result():
None,
)
assert idle_status is not None
# Check that we got workflow output via WorkflowEvent with type "output"
# Streaming mode: terminal output is AgentResponseUpdate.
output_event = next((e for e in events if e.type == "output"), None)
assert output_event is not None
data = output_event.data
assert isinstance(data, list)
assert len(data) > 0 # type: ignore
assert data[-1].role == "assistant" # type: ignore
assert all(isinstance(msg, Message) for msg in data) # type: ignore
assert isinstance(data, AgentResponseUpdate)
assert data.role == "assistant"
async def test_magentic_checkpoint_resume_round_trip():
@@ -578,7 +634,7 @@ async def _collect_agent_responses_setup(participant: SupportsAgentRun) -> list[
# Run a bounded stream to allow one invoke and then completion
events: list[WorkflowEvent] = []
async for ev in wf.run("task", stream=True): # plan review disabled
async for ev in wf.run("task", stream=True):
events.append(ev)
# Capture streaming updates (type="output" with AgentResponseUpdate data)
if ev.type == "output" and isinstance(ev.data, AgentResponseUpdate):
@@ -753,11 +809,9 @@ async def test_magentic_stall_and_reset_reach_limits():
assert idle_status is not None
output_event = next((e for e in events if e.type == "output"), None)
assert output_event is not None
assert isinstance(output_event.data, list)
assert all(isinstance(msg, Message) for msg in output_event.data) # type: ignore
assert len(output_event.data) > 0 # type: ignore
assert output_event.data[-1].text is not None # type: ignore
assert output_event.data[-1].text == "Workflow terminated due to reaching maximum reset count." # type: ignore
# Streaming mode: terminal output is AgentResponseUpdate.
assert isinstance(output_event.data, AgentResponseUpdate)
assert output_event.data.text == "Workflow terminated due to reaching maximum reset count."
async def test_magentic_checkpoint_runtime_only() -> None:
python/packages/orchestrations/tests/test_sequential.py
+136 -52
View File
@@ -22,6 +22,7 @@ from agent_framework import (
)
from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
from agent_framework.orchestrations import SequentialBuilder
from typing_extensions import Never
class _EchoAgent(BaseAgent):
@@ -67,16 +68,20 @@ class _EchoAgent(BaseAgent):
return _run()
class _SummarizerExec(Executor):
"""Custom executor that summarizes by appending a short assistant message."""
class _SummarizerTerminator(Executor):
"""Custom-executor terminator that yields a synthesized summary as the workflow's final answer."""
@handler
async def summarize(self, agent_response: AgentExecutorResponse, ctx: WorkflowContext[list[Message]]) -> None:
async def summarize(
self,
agent_response: AgentExecutorResponse,
ctx: WorkflowContext[Never, AgentResponse],
) -> None:
conversation = agent_response.full_conversation or []
user_texts = [m.text for m in conversation if m.role == "user"]
agents = [m.author_name or m.role for m in conversation if m.role == "assistant"]
summary = Message("assistant", [f"Summary of users:{len(user_texts)} agents:{len(agents)}"])
await ctx.send_message(list(conversation) + [summary])
await ctx.yield_output(AgentResponse(messages=[summary]))
class _InvalidExecutor(Executor):
@@ -98,58 +103,91 @@ def test_sequential_builder_validation_rejects_invalid_executor() -> None:
SequentialBuilder(participants=[_EchoAgent(id="agent1", name="A1"), _InvalidExecutor(id="invalid")]).build()
async def test_sequential_agents_append_to_context() -> None:
async def test_sequential_streaming_yields_only_last_agent_updates() -> None:
"""Streaming mode surfaces only the last agent's AgentResponseUpdate chunks as outputs.
Intermediate agents do NOT emit `output` events; only the last agent (the workflow's
output_executor) emits chunks of the final answer.
"""
a1 = _EchoAgent(id="agent1", name="A1")
a2 = _EchoAgent(id="agent2", name="A2")
wf = SequentialBuilder(participants=[a1, a2]).build()
completed = False
output: list[Message] | None = None
update_events: list[AgentResponseUpdate] = []
async for ev in wf.run("hello sequential", stream=True):
if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
completed = True
elif ev.type == "output":
output = ev.data # type: ignore[assignment]
if completed and output is not None:
update_events.append(ev.data) # type: ignore[arg-type]
if completed:
break
assert completed
assert output is not None
assert isinstance(output, list)
msgs: list[Message] = output
assert len(msgs) == 3
assert msgs[0].role == "user" and "hello sequential" in msgs[0].text
assert msgs[1].role == "assistant" and (msgs[1].author_name == "A1" or True)
assert msgs[2].role == "assistant" and (msgs[2].author_name == "A2" or True)
assert "A1 reply" in msgs[1].text
assert "A2 reply" in msgs[2].text
# Only the last agent's streaming chunks surface as `output` events.
assert update_events, "Expected at least one streaming update from the last agent"
for upd in update_events:
assert isinstance(upd, AgentResponseUpdate)
combined_text = "".join(u.text for u in update_events if hasattr(u, "text"))
assert "A2 reply" in combined_text
assert "A1 reply" not in combined_text
async def test_sequential_non_streaming_yields_only_last_agent_response() -> None:
"""Non-streaming mode emits a single `output` event with the last agent's AgentResponse."""
a1 = _EchoAgent(id="agent1", name="A1")
a2 = _EchoAgent(id="agent2", name="A2")
wf = SequentialBuilder(participants=[a1, a2]).build()
output_events = [ev for ev in await wf.run("hello sequential") if ev.type == "output"]
assert len(output_events) == 1
response = output_events[0].data
assert isinstance(response, AgentResponse)
assert all(m.role == "assistant" for m in response.messages)
combined = " ".join(m.text for m in response.messages)
assert "A2 reply" in combined
assert "A1 reply" not in combined
async def test_sequential_as_agent_returns_only_last_agent_response() -> None:
"""`workflow.as_agent().run(prompt)` returns ONLY the last agent's messages — not the user
input or earlier agents' replies. This is the core fix for the orchestration-as-agent
output contract."""
a1 = _EchoAgent(id="agent1", name="A1")
a2 = _EchoAgent(id="agent2", name="A2")
agent = SequentialBuilder(participants=[a1, a2]).build().as_agent()
response = await agent.run("hello as_agent")
assert isinstance(response, AgentResponse)
# Only the last agent's reply — no user prompt, no agent1 messages.
combined = " ".join(m.text for m in response.messages)
assert "A2 reply" in combined
assert "A1 reply" not in combined
assert "hello as_agent" not in combined
async def test_sequential_with_custom_executor_summary() -> None:
"""A custom-executor terminator yields its own AgentResponse — that becomes the workflow output.
Custom executors used as the terminator must call `ctx.yield_output(AgentResponse(...))`
directly (rather than `ctx.send_message(list[Message])` like an intermediate executor would),
because the terminator IS the workflow's output executor.
"""
a1 = _EchoAgent(id="agent1", name="A1")
summarizer = _SummarizerExec(id="summarizer")
summarizer = _SummarizerTerminator(id="summarizer")
wf = SequentialBuilder(participants=[a1, summarizer]).build()
completed = False
output: list[Message] | None = None
async for ev in wf.run("topic X", stream=True):
if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
completed = True
elif ev.type == "output":
output = ev.data
if completed and output is not None:
break
assert completed
assert output is not None
msgs: list[Message] = output
# Expect: [user, A1 reply, summary]
assert len(msgs) == 3
assert msgs[0].role == "user"
assert msgs[1].role == "assistant" and "A1 reply" in msgs[1].text
assert msgs[2].role == "assistant" and msgs[2].text.startswith("Summary of users:")
output_events = [ev for ev in await wf.run("topic X") if ev.type == "output"]
assert len(output_events) == 1
response = output_events[0].data
assert isinstance(response, AgentResponse)
assert len(response.messages) == 1
assert response.messages[0].role == "assistant"
assert response.messages[0].text.startswith("Summary of users:")
async def test_sequential_checkpoint_resume_round_trip() -> None:
@@ -158,14 +196,14 @@ async def test_sequential_checkpoint_resume_round_trip() -> None:
initial_agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf = SequentialBuilder(participants=list(initial_agents), checkpoint_storage=storage).build()
baseline_output: list[Message] | None = None
baseline_updates: list[AgentResponseUpdate] = []
async for ev in wf.run("checkpoint sequential", stream=True):
if ev.type == "output":
baseline_output = ev.data # type: ignore[assignment]
baseline_updates.append(ev.data) # type: ignore[arg-type]
if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
assert baseline_updates
checkpoints = await storage.list_checkpoints(workflow_name=wf.name)
assert checkpoints
@@ -175,19 +213,20 @@ async def test_sequential_checkpoint_resume_round_trip() -> None:
resumed_agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf_resume = SequentialBuilder(participants=list(resumed_agents), checkpoint_storage=storage).build()
resumed_output: list[Message] | None = None
resumed_updates: list[AgentResponseUpdate] = []
async for ev in wf_resume.run(checkpoint_id=resume_checkpoint.checkpoint_id, stream=True):
if ev.type == "output":
resumed_output = ev.data # type: ignore[assignment]
resumed_updates.append(ev.data) # type: ignore[arg-type]
if ev.type == "status" and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
assert resumed_updates
baseline_text = "".join(u.text for u in baseline_updates if hasattr(u, "text"))
resumed_text = "".join(u.text for u in resumed_updates if hasattr(u, "text"))
assert baseline_text == resumed_text
async def test_sequential_checkpoint_runtime_only() -> None:
@@ -197,14 +236,14 @@ async def test_sequential_checkpoint_runtime_only() -> None:
agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf = SequentialBuilder(participants=list(agents)).build()
baseline_output: list[Message] | None = None
baseline_updates: list[AgentResponseUpdate] = []
async for ev in wf.run("runtime checkpoint test", checkpoint_storage=storage, stream=True):
if ev.type == "output":
baseline_output = ev.data # type: ignore[assignment]
baseline_updates.append(ev.data) # type: ignore[arg-type]
if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
assert baseline_updates
checkpoints = await storage.list_checkpoints(workflow_name=wf.name)
assert checkpoints
@@ -214,21 +253,22 @@ async def test_sequential_checkpoint_runtime_only() -> None:
resumed_agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf_resume = SequentialBuilder(participants=list(resumed_agents)).build()
resumed_output: list[Message] | None = None
resumed_updates: list[AgentResponseUpdate] = []
async for ev in wf_resume.run(
checkpoint_id=resume_checkpoint.checkpoint_id, checkpoint_storage=storage, stream=True
):
if ev.type == "output":
resumed_output = ev.data # type: ignore[assignment]
resumed_updates.append(ev.data) # type: ignore[arg-type]
if ev.type == "status" and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
assert resumed_updates
baseline_text = "".join(u.text for u in baseline_updates if hasattr(u, "text"))
resumed_text = "".join(u.text for u in resumed_updates if hasattr(u, "text"))
assert baseline_text == resumed_text
async def test_sequential_checkpoint_runtime_overrides_buildtime() -> None:
@@ -390,3 +430,47 @@ async def test_chain_only_agent_responses_three_agents() -> None:
# a3 should see only A2's reply
assert len(a3.last_messages) == 1
assert a3.last_messages[0].role == "assistant" and "A2 reply" in (a3.last_messages[0].text or "")
# ---------------------------------------------------------------------------
# with_request_info tests
# ---------------------------------------------------------------------------
async def test_sequential_request_info_last_participant_emits_output() -> None:
"""When the last participant is wrapped via with_request_info(), the workflow
still emits a terminal output event after approval.
This exercises the _EndWithConversation.end_with_agent_executor_response path
that converts the AgentApprovalExecutor's forwarded AgentExecutorResponse into
the workflow's final AgentResponse output.
"""
from agent_framework_orchestrations._orchestration_request_info import AgentRequestInfoResponse
a1 = _EchoAgent(id="agent1", name="A1")
a2 = _EchoAgent(id="agent2", name="A2")
wf = SequentialBuilder(participants=[a1, a2]).with_request_info().build()
# First run: collect request_info events for both agents
request_events: list[Any] = []
async for ev in wf.run("hello with approval", stream=True):
if ev.type == "request_info" and isinstance(ev.data, AgentExecutorResponse):
request_events.append(ev)
# Approve each agent in sequence until the workflow completes
while request_events:
responses = {req.request_id: AgentRequestInfoResponse.approve() for req in request_events}
request_events = []
output_events: list[Any] = []
async for ev in wf.run(stream=True, responses=responses):
if ev.type == "request_info" and isinstance(ev.data, AgentExecutorResponse):
request_events.append(ev)
elif ev.type == "output":
output_events.append(ev)
# The workflow must produce a terminal output with the last agent's response.
assert len(output_events) == 1
response = output_events[0].data
assert isinstance(response, AgentResponse)
assert any("A2 reply" in m.text for m in response.messages)