[BREAKING] Python: Fix workflow as agent streaming output (#3649)

* WIP: with_output_from

* Add with_output_from to other modules; next: workflow as agent

* WIP: remove agent run events

* orchestrations

* WIP: update samples; next start at guessing_game_With_human_input.py

* Update all samples

* WIP: consolidate workflow as agent streaming vs non-streaming

* Consolidate workflow as agent streaming vs non-streaming

* Move request info event processing to a share method

* Final pass on the samples

* Fix mypy

* Fix mypy

* Comments

---------

Co-authored-by: Evan Mattson <35585003+moonbox3@users.noreply.github.com>

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

@@ -2,7 +2,7 @@
 
 """Tests for AgentExecutor handling of tool calls and results in streaming mode."""
 
-from collections.abc import AsyncIterable
+from collections.abc import AsyncIterable, Sequence
 from typing import Any
 
 from typing_extensions import Never
@@ -12,7 +12,6 @@ from agent_framework import (
     AgentExecutorResponse,
     AgentResponse,
     AgentResponseUpdate,
-    AgentRunUpdateEvent,
     AgentThread,
     BaseAgent,
     ChatAgent,
@@ -38,7 +37,7 @@ class _ToolCallingAgent(BaseAgent):
 
     async def run(
         self,
-        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
         *,
         thread: AgentThread | None = None,
         **kwargs: Any,
@@ -48,7 +47,7 @@ class _ToolCallingAgent(BaseAgent):
 
     async def run_stream(
         self,
-        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
         *,
         thread: AgentThread | None = None,
         **kwargs: Any,
@@ -99,9 +98,9 @@ async def test_agent_executor_emits_tool_calls_in_streaming_mode() -> None:
     workflow = WorkflowBuilder().set_start_executor(agent_exec).build()
 
     # Act: run in streaming mode
-    events: list[AgentRunUpdateEvent] = []
+    events: list[WorkflowOutputEvent] = []
     async for event in workflow.run_stream("What's the weather?"):
-        if isinstance(event, AgentRunUpdateEvent):
+        if isinstance(event, WorkflowOutputEvent):
             events.append(event)
 
     # Assert: we should receive 4 events (text, function call, function result, text)
@@ -148,7 +147,7 @@ class MockChatClient:
 
     async def get_response(
         self,
-        messages: str | ChatMessage | list[str] | list[ChatMessage],
+        messages: str | ChatMessage | Sequence[str | ChatMessage],
         **kwargs: Any,
     ) -> ChatResponse:
         if self._iteration == 0:
@@ -185,7 +184,7 @@ class MockChatClient:
 
     async def get_streaming_response(
         self,
-        messages: str | ChatMessage | list[str] | list[ChatMessage],
+        messages: str | ChatMessage | Sequence[str | ChatMessage],
         **kwargs: Any,
     ) -> AsyncIterable[ChatResponseUpdate]:
         if self._iteration == 0:
@@ -231,7 +230,13 @@ async def test_agent_executor_tool_call_with_approval() -> None:
         tools=[mock_tool_requiring_approval],
     )
 
-    workflow = WorkflowBuilder().set_start_executor(agent).add_edge(agent, test_executor).build()
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(agent)
+        .add_edge(agent, test_executor)
+        .with_output_from([test_executor])
+        .build()
+    )
 
     # Act
     events = await workflow.run("Invoke tool requiring approval")
@@ -300,7 +305,13 @@ async def test_agent_executor_parallel_tool_call_with_approval() -> None:
         tools=[mock_tool_requiring_approval],
     )
 
-    workflow = WorkflowBuilder().set_start_executor(agent).add_edge(agent, test_executor).build()
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(agent)
+        .add_edge(agent, test_executor)
+        .with_output_from([test_executor])
+        .build()
+    )
 
     # Act
     events = await workflow.run("Invoke tool requiring approval")

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

@@ -1,15 +1,15 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-"""Tests for AgentRunEvent and AgentRunUpdateEvent type annotations."""
+"""Tests for agent run event typing."""
 
 from agent_framework import AgentResponse, AgentResponseUpdate, ChatMessage
-from agent_framework._workflows._events import AgentRunEvent, AgentRunUpdateEvent
+from agent_framework._workflows._events import WorkflowOutputEvent
 
 
 def test_agent_run_event_data_type() -> None:
-    """Verify AgentRunEvent.data is typed as AgentResponse | None."""
-    response = AgentResponse(messages=[ChatMessage("assistant", ["Hello"])])
-    event = AgentRunEvent(executor_id="test", data=response)
+    """Verify WorkflowOutputEvent.data is typed as AgentResponse | None."""
+    response = AgentResponse(messages=[ChatMessage(role="assistant", text="Hello")])
+    event = WorkflowOutputEvent(data=response, executor_id="test")
 
     # This assignment should pass type checking without a cast
     data: AgentResponse | None = event.data
@@ -18,9 +18,9 @@ def test_agent_run_event_data_type() -> None:
 
 
 def test_agent_run_update_event_data_type() -> None:
-    """Verify AgentRunUpdateEvent.data is typed as AgentResponseUpdate | None."""
+    """Verify WorkflowOutputEvent.data is typed as AgentResponseUpdate | None."""
     update = AgentResponseUpdate()
-    event = AgentRunUpdateEvent(executor_id="test", data=update)
+    event = WorkflowOutputEvent(data=update, executor_id="test")
 
     # This assignment should pass type checking without a cast
     data: AgentResponseUpdate | None = event.data

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

@@ -1,6 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from collections.abc import AsyncIterable
+from collections.abc import AsyncIterable, Sequence
 from typing import Any
 
 from pydantic import PrivateAttr
@@ -34,7 +34,7 @@ class _SimpleAgent(BaseAgent):
 
     async def run(  # type: ignore[override]
         self,
-        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
         *,
         thread: AgentThread | None = None,
         **kwargs: Any,
@@ -43,7 +43,7 @@ class _SimpleAgent(BaseAgent):
 
     async def run_stream(  # type: ignore[override]
         self,
-        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
         *,
         thread: AgentThread | None = None,
         **kwargs: Any,
@@ -56,7 +56,7 @@ class _CaptureFullConversation(Executor):
     """Captures AgentExecutorResponse.full_conversation and completes the workflow."""
 
     @handler
-    async def capture(self, response: AgentExecutorResponse, ctx: WorkflowContext[Never, dict]) -> None:
+    async def capture(self, response: AgentExecutorResponse, ctx: WorkflowContext[Never, dict[str, Any]]) -> None:
         full = response.full_conversation
         # The AgentExecutor contract guarantees full_conversation is populated.
         assert full is not None
@@ -75,7 +75,13 @@ async def test_agent_executor_populates_full_conversation_non_streaming() -> Non
     agent_exec = AgentExecutor(agent, id="agent1-exec")
     capturer = _CaptureFullConversation(id="capture")
 
-    wf = WorkflowBuilder().set_start_executor(agent_exec).add_edge(agent_exec, capturer).build()
+    wf = (
+        WorkflowBuilder()
+        .set_start_executor(agent_exec)
+        .add_edge(agent_exec, capturer)
+        .with_output_from([capturer])
+        .build()
+    )
 
     # Act: use run() instead of run_stream() to test non-streaming mode
     result = await wf.run("hello world")
@@ -103,7 +109,7 @@ class _CaptureAgent(BaseAgent):
 
     async def run(  # type: ignore[override]
         self,
-        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
         *,
         thread: AgentThread | None = None,
         **kwargs: Any,
@@ -121,7 +127,7 @@ class _CaptureAgent(BaseAgent):
 
     async def run_stream(  # type: ignore[override]
         self,
-        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
         *,
         thread: AgentThread | None = None,
         **kwargs: Any,

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

@@ -11,7 +11,6 @@ from agent_framework import (
     AgentProtocol,
     AgentResponse,
     AgentResponseUpdate,
-    AgentRunUpdateEvent,
     AgentThread,
     BaseAgent,
     ChatMessage,
@@ -574,15 +573,19 @@ class StubAssistantsAgent(BaseAgent):
 async def _collect_agent_responses_setup(participant: AgentProtocol) -> list[ChatMessage]:
     captured: list[ChatMessage] = []
 
-    wf = MagenticBuilder().participants([participant]).with_manager(manager=InvokeOnceManager()).build()
+    wf = (
+        MagenticBuilder()
+        .participants([participant])
+        .with_manager(manager=InvokeOnceManager())
+        .with_intermediate_outputs()
+        .build()
+    )
 
     # Run a bounded stream to allow one invoke and then completion
     events: list[WorkflowEvent] = []
     async for ev in wf.run_stream("task"):  # plan review disabled
         events.append(ev)
-        if isinstance(ev, WorkflowOutputEvent):
-            break
-        if isinstance(ev, AgentRunUpdateEvent):
+        if isinstance(ev, WorkflowOutputEvent) and isinstance(ev.data, AgentResponseUpdate):
             captured.append(
                 ChatMessage(
                     role=ev.data.role or "assistant",
@@ -597,7 +600,6 @@ async def _collect_agent_responses_setup(participant: AgentProtocol) -> list[Cha
 async def test_agent_executor_invoke_with_thread_chat_client():
     agent = StubThreadAgent()
     captured = await _collect_agent_responses_setup(agent)
-    # Should have at least one response from agentA via _MagenticAgentExecutor path
     assert any((m.author_name == agent.name and "ok" in (m.text or "")) for m in captured)
 
 

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

@@ -177,7 +177,7 @@ def test_graph_connectivity_isolated_executors():
     executors: dict[str, Executor] = {executor1.id: executor1, executor2.id: executor2, executor3.id: executor3}
 
     with pytest.raises(GraphConnectivityError) as exc_info:
-        validate_workflow_graph(edge_groups, executors, executor1)
+        validate_workflow_graph(edge_groups, executors, executor1, [])
 
     assert "unreachable" in str(exc_info.value).lower()
     assert "executor3" in str(exc_info.value)
@@ -258,12 +258,12 @@ def test_direct_validation_function():
     executors: dict[str, Executor] = {executor1.id: executor1, executor2.id: executor2}
 
     # This should not raise any exceptions
-    validate_workflow_graph(edge_groups, executors, executor1)
+    validate_workflow_graph(edge_groups, executors, executor1, [])
 
     # Test with invalid start executor
     executor3 = StringExecutor(id="executor3")
     with pytest.raises(GraphConnectivityError):
-        validate_workflow_graph(edge_groups, executors, executor3)
+        validate_workflow_graph(edge_groups, executors, executor3, [])
 
 
 def test_fan_out_validation():
@@ -557,3 +557,155 @@ def test_handler_ctx_any_is_allowed_but_skips_type_checks(caplog: Any) -> None:
     # Builds; later edges from this executor will skip type compatibility when outputs are unspecified
     wf = WorkflowBuilder().add_edge(start, any_out).set_start_executor(start).build()
     assert wf is not None
+
+
+# region Output Validation Tests
+
+
+class OutputExecutor(Executor):
+    @handler
+    async def handle_string(self, message: str, ctx: WorkflowContext[str, str]) -> None:
+        pass
+
+
+def test_output_validation_with_valid_output_executors():
+    """Test that output validation passes when output executors exist and have output types."""
+    executor1 = OutputExecutor(id="executor1")
+    executor2 = OutputExecutor(id="executor2")
+
+    # Build workflow with valid output executors
+    workflow = (
+        WorkflowBuilder()
+        .add_edge(executor1, executor2)
+        .set_start_executor(executor1)
+        .with_output_from([executor2])
+        .build()
+    )
+
+    assert workflow is not None
+    assert workflow._output_executors == ["executor2"]
+
+
+def test_output_validation_with_multiple_valid_output_executors():
+    """Test that output validation passes with multiple valid output executors."""
+    executor1 = OutputExecutor(id="executor1")
+    executor2 = OutputExecutor(id="executor2")
+    executor3 = OutputExecutor(id="executor3")
+
+    workflow = (
+        WorkflowBuilder()
+        .add_edge(executor1, executor2)
+        .add_edge(executor2, executor3)
+        .set_start_executor(executor1)
+        .with_output_from([executor1, executor3])
+        .build()
+    )
+
+    assert workflow is not None
+    assert set(workflow._output_executors) == {"executor1", "executor3"}
+
+
+def test_output_validation_fails_for_nonexistent_executor():
+    """Test that output validation fails when an output executor doesn't exist in the graph."""
+    executor1 = OutputExecutor(id="executor1")
+    executor2 = OutputExecutor(id="executor2")
+    edge_groups = [SingleEdgeGroup(executor1.id, executor2.id)]
+    executors: dict[str, Executor] = {executor1.id: executor1, executor2.id: executor2}
+
+    # Directly test validation with a nonexistent output executor
+    with pytest.raises(WorkflowValidationError) as exc_info:
+        validate_workflow_graph(edge_groups, executors, executor1, ["nonexistent_executor"])
+
+    assert "not present in the workflow graph" in str(exc_info.value)
+    assert "nonexistent_executor" in str(exc_info.value)
+    assert exc_info.value.validation_type == ValidationTypeEnum.OUTPUT_VALIDATION
+
+
+def test_output_validation_fails_for_executor_without_output_types():
+    """Test that output validation fails when an output executor has no output type annotations."""
+    executor1 = OutputExecutor(id="executor1")
+    no_output_executor = NoOutputTypesExecutor(id="no_output")
+
+    with pytest.raises(WorkflowValidationError) as exc_info:
+        (
+            WorkflowBuilder()
+            .add_edge(executor1, no_output_executor)
+            .set_start_executor(executor1)
+            .with_output_from([no_output_executor])
+            .build()
+        )
+
+    assert "must have output type annotations defined" in str(exc_info.value)
+    assert "no_output" in str(exc_info.value)
+    assert exc_info.value.validation_type == ValidationTypeEnum.OUTPUT_VALIDATION
+
+
+def test_output_validation_empty_list_passes():
+    """Test that output validation passes with an empty output executors list."""
+    executor1 = OutputExecutor(id="executor1")
+    executor2 = OutputExecutor(id="executor2")
+
+    workflow = (
+        WorkflowBuilder().add_edge(executor1, executor2).set_start_executor(executor1).with_output_from([]).build()
+    )
+
+    assert workflow is not None
+    # All executors are outputs
+    assert workflow._output_executors == ["executor1", "executor2"]  # type: ignore
+
+
+def test_output_validation_with_direct_validate_workflow_graph():
+    """Test _output_validation directly via validate_workflow_graph function."""
+    executor1 = OutputExecutor(id="executor1")
+    executor2 = OutputExecutor(id="executor2")
+    edge_groups = [SingleEdgeGroup(executor1.id, executor2.id)]
+    executors: dict[str, Executor] = {executor1.id: executor1, executor2.id: executor2}
+
+    # Valid output executors
+    validate_workflow_graph(edge_groups, executors, executor1, ["executor2"])
+
+    # Invalid output executor (doesn't exist)
+    with pytest.raises(WorkflowValidationError) as exc_info:
+        validate_workflow_graph(edge_groups, executors, executor1, ["nonexistent"])
+
+    assert "not present in the workflow graph" in str(exc_info.value)
+    assert exc_info.value.validation_type == ValidationTypeEnum.OUTPUT_VALIDATION
+
+
+def test_output_validation_with_no_output_types_via_direct_validation():
+    """Test _output_validation fails for executors without output types via direct validation."""
+    executor1 = OutputExecutor(id="executor1")
+    no_output_executor = NoOutputTypesExecutor(id="no_output")
+    edge_groups = [SingleEdgeGroup(executor1.id, no_output_executor.id)]
+    executors: dict[str, Executor] = {executor1.id: executor1, no_output_executor.id: no_output_executor}
+
+    # Should fail because no_output_executor has no output types
+    with pytest.raises(WorkflowValidationError) as exc_info:
+        validate_workflow_graph(edge_groups, executors, executor1, ["no_output"])
+
+    assert "must have output type annotations defined" in str(exc_info.value)
+    assert exc_info.value.validation_type == ValidationTypeEnum.OUTPUT_VALIDATION
+
+
+def test_output_validation_partial_invalid_list():
+    """Test that output validation fails if any executor in the list is invalid."""
+    executor1 = OutputExecutor(id="executor1")
+    executor2 = OutputExecutor(id="executor2")
+    edge_groups = [SingleEdgeGroup(executor1.id, executor2.id)]
+    executors: dict[str, Executor] = {executor1.id: executor1, executor2.id: executor2}
+
+    # First executor is valid, second doesn't exist - validation should fail
+    with pytest.raises(WorkflowValidationError) as exc_info:
+        validate_workflow_graph(edge_groups, executors, executor1, ["executor2", "nonexistent"])
+
+    assert "not present in the workflow graph" in str(exc_info.value)
+    assert "nonexistent" in str(exc_info.value)
+
+
+def test_output_validation_type_enum_value():
+    """Test that OUTPUT_VALIDATION is properly defined in ValidationTypeEnum."""
+    assert hasattr(ValidationTypeEnum, "OUTPUT_VALIDATION")
+    assert ValidationTypeEnum.OUTPUT_VALIDATION.value == "OUTPUT_VALIDATION"
+
+
+# endregion

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

@@ -2,9 +2,9 @@
 
 import asyncio
 import tempfile
-from collections.abc import AsyncIterable
+from collections.abc import AsyncIterable, Sequence
 from dataclasses import dataclass, field
-from typing import Any
+from typing import Any, cast
 from uuid import uuid4
 
 import pytest
@@ -13,8 +13,6 @@ from agent_framework import (
     AgentExecutor,
     AgentResponse,
     AgentResponseUpdate,
-    AgentRunEvent,
-    AgentRunUpdateEvent,
     AgentThread,
     BaseAgent,
     ChatMessage,
@@ -862,7 +860,7 @@ class _StreamingTestAgent(BaseAgent):
 
     async def run(
         self,
-        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
         *,
         thread: AgentThread | None = None,
         **kwargs: Any,
@@ -872,7 +870,7 @@ class _StreamingTestAgent(BaseAgent):
 
     async def run_stream(
         self,
-        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        messages: str | ChatMessage | Sequence[str | ChatMessage] | None = None,
         *,
         thread: AgentThread | None = None,
         **kwargs: Any,
@@ -884,7 +882,7 @@ class _StreamingTestAgent(BaseAgent):
 
 
 async def test_agent_streaming_vs_non_streaming() -> None:
-    """Test that run() emits AgentRunEvent while run_stream() emits AgentRunUpdateEvent."""
+    """Test that run() and run_stream() both emits WorkflowOutputEvents correctly with the right data types."""
     agent = _StreamingTestAgent(id="test_agent", name="TestAgent", reply_text="Hello World")
     agent_exec = AgentExecutor(agent, id="agent_exec")
 
@@ -894,15 +892,17 @@ async def test_agent_streaming_vs_non_streaming() -> None:
     result = await workflow.run("test message")
 
     # Filter for agent events (result is a list of events)
-    agent_run_events = [e for e in result if isinstance(e, AgentRunEvent)]
-    agent_update_events = [e for e in result if isinstance(e, AgentRunUpdateEvent)]
+    agent_response = [e for e in result if isinstance(e, WorkflowOutputEvent) and isinstance(e.data, AgentResponse)]
+    agent_response_updates = [
+        e for e in result if isinstance(e, WorkflowOutputEvent) and isinstance(e.data, AgentResponseUpdate)
+    ]
 
-    # In non-streaming mode, should have AgentRunEvent, no AgentRunUpdateEvent
-    assert len(agent_run_events) == 1, "Expected exactly one AgentRunEvent in non-streaming mode"
-    assert len(agent_update_events) == 0, "Expected no AgentRunUpdateEvent in non-streaming mode"
-    assert agent_run_events[0].executor_id == "agent_exec"
-    assert agent_run_events[0].data is not None
-    assert agent_run_events[0].data.messages[0].text == "Hello World"
+    # In non-streaming mode, should have AgentResponse, no AgentResponseUpdate
+    assert len(agent_response) == 1, "Expected exactly one AgentResponse in non-streaming mode"
+    assert len(agent_response_updates) == 0, "Expected no AgentResponseUpdate in non-streaming mode"
+    assert agent_response[0].executor_id == "agent_exec"
+    assert agent_response[0].data is not None
+    assert agent_response[0].data.messages[0].text == "Hello World"
 
     # Test streaming mode with run_stream()
     stream_events: list[WorkflowEvent] = []
@@ -910,22 +910,31 @@ async def test_agent_streaming_vs_non_streaming() -> None:
         stream_events.append(event)
 
     # Filter for agent events
-    stream_agent_run_events = [e for e in stream_events if isinstance(e, AgentRunEvent)]
-    stream_agent_update_events = [e for e in stream_events if isinstance(e, AgentRunUpdateEvent)]
+    agent_response = [
+        cast(AgentResponse, e.data)  # type: ignore
+        for e in stream_events
+        if isinstance(e, WorkflowOutputEvent) and isinstance(e.data, AgentResponse)
+    ]
+    agent_response_updates = [
+        e.data for e in stream_events if isinstance(e, WorkflowOutputEvent) and isinstance(e.data, AgentResponseUpdate)
+    ]
 
-    # In streaming mode, should have AgentRunUpdateEvent, no AgentRunEvent
-    assert len(stream_agent_run_events) == 0, "Expected no AgentRunEvent in streaming mode"
-    assert len(stream_agent_update_events) > 0, "Expected AgentRunUpdateEvent events in streaming mode"
+    # In streaming mode, should have AgentResponseUpdate, no AgentResponse
+    assert len(agent_response) == 0, "Expected no AgentResponse in streaming mode"
+    assert len(agent_response_updates) > 0, "Expected AgentResponseUpdate events in streaming mode"
 
     # Verify we got incremental updates (one per character in "Hello World")
-    assert len(stream_agent_update_events) == len("Hello World"), "Expected one update per character"
+    assert len(agent_response_updates) == len("Hello World"), "Expected one update per character"
 
     # Verify the updates build up to the full message
-    accumulated_text = "".join(
-        e.data.contents[0].text
-        for e in stream_agent_update_events
-        if e.data and e.data.contents and e.data.contents[0].text
-    )
+    accumulated_text = "".join([
+        e.contents[0].text
+        for e in agent_response_updates
+        if e.contents
+        and isinstance(e.contents[0], Content)
+        and e.contents[0].type == "text"
+        and e.contents[0].text is not None
+    ])
     assert accumulated_text == "Hello World", f"Expected 'Hello World', got '{accumulated_text}'"
 
 
@@ -974,3 +983,253 @@ async def test_workflow_run_stream_parameter_validation(
 
     # Invalid combinations already tested in test_workflow_run_parameter_validation
     # This test ensures streaming works correctly for valid parameters
+
+
+# region Output executor filtering tests
+
+
+class OutputProducerExecutor(Executor):
+    """An executor that produces a unique output value for testing output filtering."""
+
+    def __init__(self, id: str, output_value: int) -> None:
+        super().__init__(id=id)
+        self.output_value = output_value
+
+    @handler
+    async def handle_message(self, message: NumberMessage, ctx: WorkflowContext[NumberMessage, int]) -> None:
+        await ctx.yield_output(self.output_value)
+
+
+class PassthroughExecutor(Executor):
+    """An executor that passes through messages and produces an output."""
+
+    def __init__(self, id: str, output_value: int) -> None:
+        super().__init__(id=id)
+        self.output_value = output_value
+
+    @handler
+    async def handle_message(self, message: NumberMessage, ctx: WorkflowContext[NumberMessage, int]) -> None:
+        await ctx.yield_output(self.output_value)
+        await ctx.send_message(message)
+
+
+async def test_output_executors_empty_yields_all_outputs() -> None:
+    """Test that when _output_executors is empty (default), all outputs are yielded."""
+    # Create executors that each produce different outputs
+    executor_a = PassthroughExecutor(id="executor_a", output_value=10)
+    executor_b = OutputProducerExecutor(id="executor_b", output_value=20)
+
+    # Build workflow with a -> b
+    workflow = WorkflowBuilder().set_start_executor(executor_a).add_edge(executor_a, executor_b).build()
+
+    result = await workflow.run(NumberMessage(data=0))
+    outputs = result.get_outputs()
+
+    # Both executors' outputs should be present
+    assert len(outputs) == 2
+    assert outputs == [10, 20]
+
+    output_events = [event for event in result if isinstance(event, WorkflowOutputEvent)]
+    assert len(output_events) == 2
+    assert output_events[0].executor_id == "executor_a"
+    assert output_events[1].executor_id == "executor_b"
+
+
+async def test_output_executors_filters_outputs_non_streaming() -> None:
+    """Test that only outputs from specified executors are yielded in non-streaming mode."""
+    # Create executors that each produce different outputs
+    executor_a = PassthroughExecutor(id="executor_a", output_value=10)
+    executor_b = OutputProducerExecutor(id="executor_b", output_value=20)
+
+    # Build workflow with a -> b
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .with_output_from([executor_b])
+        .build()
+    )
+
+    result = await workflow.run(NumberMessage(data=0))
+    outputs = result.get_outputs()
+
+    # Only executor_b's output should be present
+    assert len(outputs) == 1
+    assert outputs[0] == 20
+
+    output_events = [event for event in result if isinstance(event, WorkflowOutputEvent)]
+    assert len(output_events) == 1
+    assert output_events[0].executor_id == "executor_b"
+
+
+async def test_output_executors_filters_outputs_streaming() -> None:
+    """Test that only outputs from specified executors are yielded in streaming mode."""
+    # Create executors that each produce different outputs
+    executor_a = PassthroughExecutor(id="executor_a", output_value=100)
+    executor_b = OutputProducerExecutor(id="executor_b", output_value=200)
+
+    # Build workflow with a -> b
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .with_output_from([executor_a])
+        .build()
+    )
+
+    # Collect outputs from streaming
+    output_events: list[WorkflowOutputEvent] = []
+    async for event in workflow.run_stream(NumberMessage(data=0)):
+        if isinstance(event, WorkflowOutputEvent):
+            output_events.append(event)
+
+    # Only executor_a's output should be present
+    assert len(output_events) == 1
+    assert output_events[0].data == 100
+    assert output_events[0].executor_id == "executor_a"
+
+
+async def test_output_executors_with_multiple_specified_executors() -> None:
+    """Test filtering with multiple executors in the output list."""
+    # Create three executors with pass-through to reach all of them
+    executor_a = PassthroughExecutor(id="executor_a", output_value=1)
+    executor_b = PassthroughExecutor(id="executor_b", output_value=2)
+    executor_c = OutputProducerExecutor(id="executor_c", output_value=3)
+
+    # Build workflow with a -> b -> c
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .add_edge(executor_b, executor_c)
+        .with_output_from([executor_a, executor_c])
+        .build()
+    )
+
+    result = await workflow.run(NumberMessage(data=0))
+    outputs = result.get_outputs()
+
+    # Only executor_a and executor_c outputs should be present
+    assert len(outputs) == 2
+    assert 1 in outputs  # executor_a
+    assert 3 in outputs  # executor_c
+    assert 2 not in outputs  # executor_b should be filtered out
+
+
+async def test_output_executors_with_nonexistent_executor_id() -> None:
+    """Test that specifying a non-existent executor ID doesn't break the workflow."""
+    executor_a = OutputProducerExecutor(id="executor_a", output_value=42)
+
+    workflow = WorkflowBuilder().set_start_executor(executor_a).build()
+
+    # Set output_executors to an ID that doesn't exist
+    workflow._output_executors = ["nonexistent_executor"]  # type: ignore
+
+    result = await workflow.run(NumberMessage(data=0))
+    outputs = result.get_outputs()
+
+    # No outputs should be yielded since the executor ID doesn't match
+    assert len(outputs) == 0
+
+
+async def test_output_executors_filtering_with_fan_in() -> None:
+    """Test output filtering in a fan-in workflow."""
+
+    class FanOutStartExecutor(Executor):
+        """Executor that sends messages to fan-out targets."""
+
+        @handler
+        async def handle(self, message: NumberMessage, ctx: WorkflowContext[NumberMessage, int]) -> None:
+            await ctx.yield_output(999)  # This should be filtered out
+            await ctx.send_message(NumberMessage(data=5))
+
+    class FanOutTargetExecutor(Executor):
+        """Executor that processes fan-out messages."""
+
+        def __init__(self, id: str, increment: int) -> None:
+            super().__init__(id=id)
+            self.increment = increment
+
+        @handler
+        async def handle(self, message: NumberMessage, ctx: WorkflowContext[NumberMessage, int]) -> None:
+            await ctx.yield_output(888)  # This should be filtered out
+            await ctx.send_message(NumberMessage(data=message.data + self.increment))
+
+    # Create executors for fan-in pattern
+    executor_start = FanOutStartExecutor(id="executor_start")
+    executor_a = FanOutTargetExecutor(id="executor_a", increment=10)
+    executor_b = FanOutTargetExecutor(id="executor_b", increment=20)
+    aggregator = AggregatorExecutor(id="aggregator")
+
+    # Build fan-in workflow: start -> [a, b] -> aggregator
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_start)
+        .add_fan_out_edges(executor_start, [executor_a, executor_b])
+        .add_fan_in_edges([executor_a, executor_b], aggregator)
+        .with_output_from([aggregator])
+        .build()
+    )
+
+    result = await workflow.run(NumberMessage(data=0))
+    outputs = result.get_outputs()
+
+    # Only aggregator output should be present
+    # executor_a sends 5+10=15, executor_b sends 5+20=25, aggregator sums: 15+25=40
+    assert len(outputs) == 1
+    assert outputs[0] == 40
+
+
+async def test_output_executors_filtering_with_send_responses() -> None:
+    """Test output filtering works correctly with send_responses method."""
+    executor = MockExecutorRequestApproval(id="approval_executor")
+
+    workflow = WorkflowBuilder().set_start_executor(executor).with_output_from([executor]).build()
+
+    # Run workflow which will request approval
+    result = await workflow.run(NumberMessage(data=42))
+
+    # Get request info events
+    request_events = result.get_request_info_events()
+    assert len(request_events) == 1
+
+    # Send approval response
+    responses = {request_events[0].request_id: ApprovalMessage(approved=True)}
+    response_result = await workflow.send_responses(responses)
+    outputs = response_result.get_outputs()
+
+    # Output should be yielded since approval_executor is in output_executors
+    assert len(outputs) == 1
+    assert outputs[0] == 42
+
+
+async def test_output_executors_filtering_with_send_responses_streaming() -> None:
+    """Test output filtering works correctly with send_responses_streaming method."""
+    executor = MockExecutorRequestApproval(id="approval_executor")
+
+    workflow = WorkflowBuilder().set_start_executor(executor).build()
+
+    # Run workflow which will request approval
+    events_list: list[WorkflowEvent] = []
+    async for event in workflow.run_stream(NumberMessage(data=99)):
+        events_list.append(event)
+
+    # Get request info events
+    request_events = [e for e in events_list if isinstance(e, RequestInfoEvent)]
+    assert len(request_events) == 1
+
+    # Set output_executors to exclude the approval executor
+    workflow._output_executors = ["other_executor"]  # type: ignore
+
+    # Send approval response via streaming
+    responses = {request_events[0].request_id: ApprovalMessage(approved=True)}
+    output_events: list[WorkflowOutputEvent] = []
+    async for event in workflow.send_responses_streaming(responses):
+        if isinstance(event, WorkflowOutputEvent):
+            output_events.append(event)
+
+    # No outputs should be yielded since approval_executor is not in output_executors
+    assert len(output_events) == 0
+
+
+# endregion

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

@@ -1,16 +1,17 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import uuid
-from collections.abc import AsyncIterable
+from collections.abc import AsyncIterable, Sequence
 from typing import Any
 
 import pytest
+from typing_extensions import Never
 
 from agent_framework import (
+    AgentExecutorRequest,
     AgentProtocol,
     AgentResponse,
     AgentResponseUpdate,
-    AgentRunUpdateEvent,
     AgentThread,
     ChatMessage,
     ChatMessageStore,
@@ -27,26 +28,34 @@ from agent_framework import (
 
 
 class SimpleExecutor(Executor):
-    """Simple executor that emits AgentRunEvent or AgentRunStreamingEvent."""
+    """Simple executor that emits a response based on input."""
 
-    def __init__(self, id: str, response_text: str, emit_streaming: bool = False):
+    def __init__(self, id: str, response_text: str, streaming: bool = False):
         super().__init__(id=id)
         self.response_text = response_text
-        self.emit_streaming = emit_streaming
+        self.streaming = streaming
 
     @handler
-    async def handle_message(self, message: list[ChatMessage], ctx: WorkflowContext[list[ChatMessage]]) -> None:
+    async def handle_message(
+        self,
+        message: list[ChatMessage],
+        ctx: WorkflowContext[list[ChatMessage], AgentResponseUpdate | AgentResponse],
+    ) -> None:
         input_text = message[0].contents[0].text if message and message[0].contents[0].type == "text" else "no input"
         response_text = f"{self.response_text}: {input_text}"
 
         # Create response message for both streaming and non-streaming cases
         response_message = ChatMessage("assistant", [Content.from_text(text=response_text)])
 
-        # Emit update event.
-        streaming_update = AgentResponseUpdate(
-            contents=[Content.from_text(text=response_text)], role="assistant", message_id=str(uuid.uuid4())
-        )
-        await ctx.add_event(AgentRunUpdateEvent(executor_id=self.id, data=streaming_update))
+        if self.streaming:
+            # Emit update event.
+            streaming_update = AgentResponseUpdate(
+                contents=[Content.from_text(text=response_text)], role="assistant", message_id=str(uuid.uuid4())
+            )
+            await ctx.yield_output(streaming_update)
+        else:
+            response = AgentResponse(messages=[response_message])
+            await ctx.yield_output(response)
 
         # Pass message to next executor if any (for both streaming and non-streaming)
         await ctx.send_message([response_message])
@@ -55,6 +64,10 @@ class SimpleExecutor(Executor):
 class RequestingExecutor(Executor):
     """Executor that requests info."""
 
+    def __init__(self, id: str, streaming: bool = False):
+        super().__init__(id=id)
+        self.streaming = streaming
+
     @handler
     async def handle_message(self, _: list[ChatMessage], ctx: WorkflowContext) -> None:
         # Send a RequestInfoMessage to trigger the request info process
@@ -62,26 +75,49 @@ class RequestingExecutor(Executor):
 
     @response_handler
     async def handle_request_response(
-        self, original_request: str, response: str, ctx: WorkflowContext[ChatMessage]
+        self,
+        original_request: str,
+        response: str,
+        ctx: WorkflowContext[ChatMessage, AgentResponseUpdate | AgentResponse],
     ) -> None:
         # Handle the response and emit completion response
-        update = AgentResponseUpdate(
-            contents=[Content.from_text(text="Request completed successfully")],
-            role="assistant",
-            message_id=str(uuid.uuid4()),
+        content = Content.from_text(text=f"Request completed with response: {response}")
+        if self.streaming:
+            await ctx.yield_output(
+                AgentResponseUpdate(
+                    contents=[content],
+                    role="assistant",
+                    message_id=str(uuid.uuid4()),
+                )
+            )
+            return
+
+        await ctx.yield_output(
+            AgentResponse(
+                messages=[
+                    ChatMessage(
+                        role="assistant",
+                        contents=[content],
+                    )
+                ],
+            )
         )
-        await ctx.add_event(AgentRunUpdateEvent(executor_id=self.id, data=update))
 
 
 class ConversationHistoryCapturingExecutor(Executor):
     """Executor that captures the received conversation history for verification."""
 
-    def __init__(self, id: str):
+    def __init__(self, id: str, streaming: bool = False):
         super().__init__(id=id)
         self.received_messages: list[ChatMessage] = []
+        self.streaming = streaming
 
     @handler
-    async def handle_message(self, messages: list[ChatMessage], ctx: WorkflowContext[list[ChatMessage]]) -> None:
+    async def handle_message(
+        self,
+        messages: list[ChatMessage],
+        ctx: WorkflowContext[list[ChatMessage], AgentResponseUpdate | AgentResponse],
+    ) -> None:
         # Capture all received messages
         self.received_messages = list(messages)
 
@@ -91,10 +127,16 @@ class ConversationHistoryCapturingExecutor(Executor):
 
         response_message = ChatMessage("assistant", [Content.from_text(text=response_text)])
 
-        streaming_update = AgentResponseUpdate(
-            contents=[Content.from_text(text=response_text)], role="assistant", message_id=str(uuid.uuid4())
-        )
-        await ctx.add_event(AgentRunUpdateEvent(executor_id=self.id, data=streaming_update))
+        if self.streaming:
+            # Emit streaming update
+            streaming_update = AgentResponseUpdate(
+                contents=[Content.from_text(text=response_text)], role="assistant", message_id=str(uuid.uuid4())
+            )
+            await ctx.yield_output(streaming_update)
+        else:
+            response = AgentResponse(messages=[response_message])
+            await ctx.yield_output(response)
+
         await ctx.send_message([response_message])
 
 
@@ -102,10 +144,10 @@ class TestWorkflowAgent:
     """Test cases for WorkflowAgent end-to-end functionality."""
 
     async def test_end_to_end_basic_workflow(self):
-        """Test basic end-to-end workflow execution with 2 executors emitting AgentRunEvent."""
+        """Test basic end-to-end workflow execution with 2 executors emitting AgentResponse."""
         # Create workflow with two executors
-        executor1 = SimpleExecutor(id="executor1", response_text="Step1", emit_streaming=False)
-        executor2 = SimpleExecutor(id="executor2", response_text="Step2", emit_streaming=False)
+        executor1 = SimpleExecutor(id="executor1", response_text="Step1", streaming=False)
+        executor2 = SimpleExecutor(id="executor2", response_text="Step2", streaming=False)
 
         workflow = WorkflowBuilder().set_start_executor(executor1).add_edge(executor1, executor2).build()
 
@@ -126,6 +168,7 @@ class TestWorkflowAgent:
             first_content = message.contents[0]
             if first_content.type == "text":
                 text = first_content.text
+                assert text is not None
                 if text.startswith("Step1:"):
                     step1_messages.append(message)
                 elif text.startswith("Step2:"):
@@ -136,16 +179,18 @@ class TestWorkflowAgent:
         assert len(step2_messages) >= 1, "Should have received message from Step2 executor"
 
         # Verify the processing worked for both
-        step1_text: str = step1_messages[0].contents[0].text  # type: ignore[attr-defined]
-        step2_text: str = step2_messages[0].contents[0].text  # type: ignore[attr-defined]
+        step1_text = step1_messages[0].contents[0].text
+        step2_text = step2_messages[0].contents[0].text
+        assert step1_text is not None
+        assert step2_text is not None
         assert "Step1: Hello World" in step1_text
         assert "Step2: Step1: Hello World" in step2_text
 
     async def test_end_to_end_basic_workflow_streaming(self):
         """Test end-to-end workflow with streaming executor that emits AgentRunStreamingEvent."""
         # Create a single streaming executor
-        executor1 = SimpleExecutor(id="stream1", response_text="Streaming1", emit_streaming=True)
-        executor2 = SimpleExecutor(id="stream2", response_text="Streaming2", emit_streaming=True)
+        executor1 = SimpleExecutor(id="stream1", response_text="Streaming1")
+        executor2 = SimpleExecutor(id="stream2", response_text="Streaming2")
 
         # Create workflow with just one executor
         workflow = WorkflowBuilder().set_start_executor(executor1).add_edge(executor1, executor2).build()
@@ -165,15 +210,17 @@ class TestWorkflowAgent:
         first_content: Content = updates[0].contents[0]  # type: ignore[assignment]
         second_content: Content = updates[1].contents[0]  # type: ignore[assignment]
         assert first_content.type == "text"
+        assert first_content.text is not None
         assert "Streaming1: Test input" in first_content.text
         assert second_content.type == "text"
+        assert second_content.text is not None
         assert "Streaming2: Streaming1: Test input" in second_content.text
 
     async def test_end_to_end_request_info_handling(self):
         """Test end-to-end workflow with RequestInfoEvent handling."""
         # Create workflow with requesting executor -> request info executor (no cycle)
-        simple_executor = SimpleExecutor(id="simple", response_text="SimpleResponse", emit_streaming=False)
-        requesting_executor = RequestingExecutor(id="requester")
+        simple_executor = SimpleExecutor(id="simple", response_text="SimpleResponse", streaming=False)
+        requesting_executor = RequestingExecutor(id="requester", streaming=False)
 
         workflow = (
             WorkflowBuilder().set_start_executor(simple_executor).add_edge(simple_executor, requesting_executor).build()
@@ -208,6 +255,8 @@ class TestWorkflowAgent:
         assert function_call.arguments.get("request_id") == approval_request.id
 
         # Approval request should reference the same function call
+        assert approval_request.id is not None
+        assert approval_request.function_call is not None
         assert approval_request.function_call.call_id == function_call.call_id
         assert approval_request.function_call.name == function_call.name
 
@@ -245,7 +294,7 @@ class TestWorkflowAgent:
     def test_workflow_as_agent_method(self) -> None:
         """Test that Workflow.as_agent() creates a properly configured WorkflowAgent."""
         # Create a simple workflow
-        executor = SimpleExecutor(id="executor1", response_text="Response", emit_streaming=False)
+        executor = SimpleExecutor(id="executor1", response_text="Response")
         workflow = WorkflowBuilder().set_start_executor(executor).build()
 
         # Test as_agent with a name
@@ -286,7 +335,7 @@ class TestWorkflowAgent:
         """
 
         @executor
-        async def yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
+        async def yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext[Never, str]) -> None:
             # Extract text from input for demonstration
             input_text = messages[0].text if messages else "no input"
             await ctx.yield_output(f"processed: {input_text}")
@@ -311,7 +360,7 @@ class TestWorkflowAgent:
         """Test that ctx.yield_output() surfaces as AgentResponseUpdate when streaming."""
 
         @executor
-        async def yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
+        async def yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext[Never, str]) -> None:
             await ctx.yield_output("first output")
             await ctx.yield_output("second output")
 
@@ -331,7 +380,7 @@ class TestWorkflowAgent:
         """Test that yield_output preserves different content types (Content, Content, etc.)."""
 
         @executor
-        async def content_yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
+        async def content_yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext[Never, Content]) -> None:
             # Yield different content types
             await ctx.yield_output(Content.from_text(text="text content"))
             await ctx.yield_output(Content.from_data(data=b"binary data", media_type="application/octet-stream"))
@@ -359,7 +408,7 @@ class TestWorkflowAgent:
         """Test that yield_output with ChatMessage preserves the message structure."""
 
         @executor
-        async def chat_message_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
+        async def chat_message_executor(messages: list[ChatMessage], ctx: WorkflowContext[Never, ChatMessage]) -> None:
             msg = ChatMessage(
                 role="assistant",
                 contents=[Content.from_text(text="response text")],
@@ -389,7 +438,9 @@ class TestWorkflowAgent:
                 return f"CustomData({self.value})"
 
         @executor
-        async def raw_yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
+        async def raw_yielding_executor(
+            messages: list[ChatMessage], ctx: WorkflowContext[Never, Content | CustomData | str]
+        ) -> None:
             # Yield different types of data
             await ctx.yield_output("simple string")
             await ctx.yield_output(Content.from_text(text="text content"))
@@ -408,8 +459,11 @@ class TestWorkflowAgent:
 
         # Verify raw_representation is set for each update
         assert updates[0].raw_representation == "simple string"
+
+        assert isinstance(updates[1].raw_representation, Content)
         assert updates[1].raw_representation.type == "text"
         assert updates[1].raw_representation.text == "text content"
+
         assert isinstance(updates[2].raw_representation, CustomData)
         assert updates[2].raw_representation.value == 42
 
@@ -421,7 +475,9 @@ class TestWorkflowAgent:
         """
 
         @executor
-        async def list_yielding_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
+        async def list_yielding_executor(
+            messages: list[ChatMessage], ctx: WorkflowContext[Never, list[ChatMessage]]
+        ) -> None:
             # Yield a list of ChatMessages (as SequentialBuilder does)
             msg_list = [
                 ChatMessage("user", [Content.from_text(text="first message")]),
@@ -441,19 +497,20 @@ class TestWorkflowAgent:
         async for update in agent.run_stream("test"):
             updates.append(update)
 
-        assert len(updates) == 1
-        assert len(updates[0].contents) == 4
-        texts = [c.text for c in updates[0].contents if c.type == "text"]
-        assert texts == ["first message", "second message", "third", "fourth"]
+        assert len(updates) == 3
+        full_response = AgentResponse.from_updates(updates)
+        assert len(full_response.messages) == 3
+        texts = [message.text for message in full_response.messages]
+        # Note: `from_agent_run_response_updates` coalesces multiple text contents into one content
+        assert texts == ["first message", "second message", "thirdfourth"]
 
-        # Verify run() coalesces text contents (expected behavior)
+        # Verify run()
         result = await agent.run("test")
 
         assert isinstance(result, AgentResponse)
-        assert len(result.messages) == 1
-        # Content items are coalesced into one
-        assert len(result.messages[0].contents) == 1
-        assert result.messages[0].text == "first messagesecond messagethirdfourth"
+        assert len(result.messages) == 3
+        texts = [message.text for message in result.messages]
+        assert texts == ["first message", "second message", "third fourth"]
 
     async def test_thread_conversation_history_included_in_workflow_run(self) -> None:
         """Test that conversation history from thread is included when running WorkflowAgent.
@@ -462,7 +519,7 @@ class TestWorkflowAgent:
         the workflow receives the complete conversation history (thread history + new messages).
         """
         # Create an executor that captures all received messages
-        capturing_executor = ConversationHistoryCapturingExecutor(id="capturing")
+        capturing_executor = ConversationHistoryCapturingExecutor(id="capturing", streaming=False)
         workflow = WorkflowBuilder().set_start_executor(capturing_executor).build()
         agent = WorkflowAgent(workflow=workflow, name="Thread History Test Agent")
 
@@ -561,41 +618,41 @@ class TestWorkflowAgent:
             """Mock agent for testing."""
 
             def __init__(self, name: str, response_text: str) -> None:
-                self._name = name
+                self.id = str(uuid.uuid4())
+                self.name = name
+                self.description: str | None = None
                 self._response_text = response_text
-                self._description: str | None = None
 
-            @property
-            def name(self) -> str | None:
-                return self._name
-
-            @property
-            def description(self) -> str | None:
-                return self._description
-
-            def get_new_thread(self) -> AgentThread:
+            def get_new_thread(self, **kwargs: Any) -> AgentThread:
                 return AgentThread()
 
-            async def run(self, messages: Any, *, thread: AgentThread | None = None, **kwargs: Any) -> AgentResponse:
+            async def run(
+                self,
+                messages: str | Content | ChatMessage | Sequence[str | Content | ChatMessage] | None = None,
+                *,
+                thread: AgentThread | None = None,
+                **kwargs: Any,
+            ) -> AgentResponse:
                 return AgentResponse(
                     messages=[ChatMessage("assistant", [self._response_text])],
-                    text=self._response_text,
                 )
 
             async def run_stream(
-                self, messages: Any, *, thread: AgentThread | None = None, **kwargs: Any
+                self,
+                messages: str | Content | ChatMessage | Sequence[str | Content | ChatMessage] | None = None,
+                *,
+                thread: AgentThread | None = None,
+                **kwargs: Any,
             ) -> AsyncIterable[AgentResponseUpdate]:
                 for word in self._response_text.split():
                     yield AgentResponseUpdate(
                         contents=[Content.from_text(text=word + " ")],
                         role="assistant",
-                        author_name=self._name,
+                        author_name=self.name,
                     )
 
         @executor
-        async def start_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
-            from agent_framework import AgentExecutorRequest
-
+        async def start_executor(messages: list[ChatMessage], ctx: WorkflowContext[AgentExecutorRequest, str]) -> None:
             await ctx.yield_output("Start output")
             await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True))
 
@@ -604,12 +661,11 @@ class TestWorkflowAgent:
             WorkflowBuilder()
             .register_executor(lambda: start_executor, "start")
             .register_agent(lambda: MockAgent("agent1", "Agent1 output - should NOT appear"), "agent1")
-            .register_agent(
-                lambda: MockAgent("agent2", "Agent2 output - SHOULD appear"), "agent2", output_response=True
-            )
+            .register_agent(lambda: MockAgent("agent2", "Agent2 output - SHOULD appear"), "agent2")
             .set_start_executor("start")
             .add_edge("start", "agent1")
             .add_edge("agent1", "agent2")
+            .with_output_from(["start", "agent2"])
             .build()
         )
 
@@ -635,47 +691,45 @@ class TestWorkflowAgent:
             """Mock agent for testing."""
 
             def __init__(self, name: str, response_text: str) -> None:
-                self._name = name
+                self.id = str(uuid.uuid4())
+                self.name = name
+                self.description: str | None = None
                 self._response_text = response_text
-                self._description: str | None = None
 
-            @property
-            def name(self) -> str | None:
-                return self._name
-
-            @property
-            def description(self) -> str | None:
-                return self._description
-
-            def get_new_thread(self) -> AgentThread:
+            def get_new_thread(self, **kwargs: Any) -> AgentThread:
                 return AgentThread()
 
-            async def run(self, messages: Any, *, thread: AgentThread | None = None, **kwargs: Any) -> AgentResponse:
-                return AgentResponse(
-                    messages=[ChatMessage("assistant", [self._response_text])],
-                    text=self._response_text,
-                )
+            async def run(
+                self,
+                messages: str | Content | ChatMessage | Sequence[str | Content | ChatMessage] | None = None,
+                *,
+                thread: AgentThread | None = None,
+                **kwargs: Any,
+            ) -> AgentResponse:
+                return AgentResponse(messages=[ChatMessage("assistant", [self._response_text])])
 
             async def run_stream(
-                self, messages: Any, *, thread: AgentThread | None = None, **kwargs: Any
+                self,
+                messages: str | Content | ChatMessage | Sequence[str | Content | ChatMessage] | None = None,
+                *,
+                thread: AgentThread | None = None,
+                **kwargs: Any,
             ) -> AsyncIterable[AgentResponseUpdate]:
                 yield AgentResponseUpdate(
                     contents=[Content.from_text(text=self._response_text)],
                     role="assistant",
-                    author_name=self._name,
+                    author_name=self.name,
                 )
 
         @executor
-        async def start_executor(messages: list[ChatMessage], ctx: WorkflowContext) -> None:
-            from agent_framework import AgentExecutorRequest
-
+        async def start_executor(messages: list[ChatMessage], ctx: WorkflowContext[AgentExecutorRequest]) -> None:
             await ctx.send_message(AgentExecutorRequest(messages=messages, should_respond=True))
 
-        # Build workflow with single agent that has output_response=True
+        # Build workflow with single agent
         workflow = (
             WorkflowBuilder()
             .register_executor(lambda: start_executor, "start")
-            .register_agent(lambda: MockAgent("agent", "Unique response text"), "agent", output_response=True)
+            .register_agent(lambda: MockAgent("agent", "Unique response text"), "agent")
             .set_start_executor("start")
             .add_edge("start", "agent")
             .build()
@@ -694,14 +748,14 @@ class TestWorkflowAgent:
 class TestWorkflowAgentAuthorName:
     """Test cases for author_name enrichment in WorkflowAgent (GitHub issue #1331)."""
 
-    async def test_agent_run_update_event_gets_executor_id_as_author_name(self):
-        """Test that AgentRunUpdateEvent gets executor_id as author_name when not already set.
+    async def test_agent_response_update_gets_executor_id_as_author_name(self):
+        """Test that AgentResponseUpdate gets executor_id as author_name when not already set.
 
         This validates the fix for GitHub issue #1331: agent responses should include
         identification of which agent produced them in multi-agent workflows.
         """
-        # Create workflow with executor that emits AgentRunUpdateEvent without author_name
-        executor1 = SimpleExecutor(id="my_executor_id", response_text="Response", emit_streaming=False)
+        # Create workflow with executor that emits AgentResponseUpdate without author_name
+        executor1 = SimpleExecutor(id="my_executor_id", response_text="Response")
         workflow = WorkflowBuilder().set_start_executor(executor1).build()
         agent = WorkflowAgent(workflow=workflow, name="Test Agent")
 
@@ -716,14 +770,18 @@ class TestWorkflowAgentAuthorName:
         # Verify author_name is set to executor_id
         assert updates[0].author_name == "my_executor_id"
 
-    async def test_agent_run_update_event_preserves_existing_author_name(self):
+    async def test_agent_response_update_preserves_existing_author_name(self):
         """Test that existing author_name is preserved and not overwritten."""
 
         class AuthorNameExecutor(Executor):
             """Executor that sets author_name explicitly."""
 
             @handler
-            async def handle_message(self, message: list[ChatMessage], ctx: WorkflowContext[list[ChatMessage]]) -> None:
+            async def handle_message(
+                self,
+                message: list[ChatMessage],
+                ctx: WorkflowContext[list[ChatMessage], AgentResponseUpdate],
+            ) -> None:
                 # Emit update with explicit author_name
                 update = AgentResponseUpdate(
                     contents=[Content.from_text(text="Response with author")],
@@ -731,7 +789,7 @@ class TestWorkflowAgentAuthorName:
                     author_name="custom_author_name",  # Explicitly set
                     message_id=str(uuid.uuid4()),
                 )
-                await ctx.add_event(AgentRunUpdateEvent(executor_id=self.id, data=update))
+                await ctx.yield_output(update)
 
         executor = AuthorNameExecutor(id="executor_id")
         workflow = WorkflowBuilder().set_start_executor(executor).build()
@@ -749,8 +807,8 @@ class TestWorkflowAgentAuthorName:
     async def test_multiple_executors_have_distinct_author_names(self):
         """Test that multiple executors in a workflow have their own author_name."""
         # Create workflow with two executors
-        executor1 = SimpleExecutor(id="first_executor", response_text="First", emit_streaming=False)
-        executor2 = SimpleExecutor(id="second_executor", response_text="Second", emit_streaming=False)
+        executor1 = SimpleExecutor(id="first_executor", response_text="First")
+        executor2 = SimpleExecutor(id="second_executor", response_text="Second")
 
         workflow = WorkflowBuilder().set_start_executor(executor1).add_edge(executor1, executor2).build()
         agent = WorkflowAgent(workflow=workflow, name="Multi-Executor Agent")
@@ -834,11 +892,11 @@ class TestWorkflowAgentMergeUpdates:
         # The exact order depends on dict iteration order for response_ids,
         # but within each response group, chronological order should be maintained
         # and global dangling should be last
-        assert "Global-Dangling" in message_texts[-1]  # Global dangling at end
+        assert "Global-Dangling" in message_texts[-1]  # type: ignore # Global dangling at end
 
         # Find positions of resp-a and resp-b messages
-        resp_a_positions = [i for i, text in enumerate(message_texts) if "RespA" in text]
-        resp_b_positions = [i for i, text in enumerate(message_texts) if "RespB" in text]
+        resp_a_positions = [i for i, text in enumerate(message_texts) if "RespA" in text]  # type: ignore
+        resp_b_positions = [i for i, text in enumerate(message_texts) if "RespB" in text]  # type: ignore
 
         # Within resp-a group: Msg1 (earlier) should come before Msg2 (later)
         resp_a_texts = [message_texts[i] for i in resp_a_positions]
@@ -1013,7 +1071,7 @@ class TestWorkflowAgentMergeUpdates:
         assert len(result.messages) == 4
 
         # Extract content types for verification
-        content_sequence = []
+        content_sequence: list[tuple[str, str]] = []
         for msg in result.messages:
             for content in msg.contents:
                 if content.type == "text":
@@ -1128,7 +1186,7 @@ class TestWorkflowAgentMergeUpdates:
         assert len(result.messages) == 6
 
         # Build a sequence of (content_type, call_id_if_applicable)
-        content_sequence = []
+        content_sequence: list[tuple[str, str | None]] = []
         for msg in result.messages:
             for content in msg.contents:
                 if content.type == "text":
@@ -1194,7 +1252,7 @@ class TestWorkflowAgentMergeUpdates:
         assert len(result.messages) == 3
 
         # Orphan function result should be at the end since it can't be matched
-        content_types = []
+        content_types: list[str] = []
         for msg in result.messages:
             for content in msg.contents:
                 if content.type == "text":

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

@@ -15,6 +15,7 @@ from agent_framework import (
     Executor,
     WorkflowBuilder,
     WorkflowContext,
+    WorkflowValidationError,
     handler,
 )
 
@@ -57,7 +58,7 @@ class MockExecutor(Executor):
     """A mock executor for testing purposes."""
 
     @handler
-    async def mock_handler(self, message: MockMessage, ctx: WorkflowContext[MockMessage]) -> None:
+    async def mock_handler(self, message: MockMessage, ctx: WorkflowContext[MockMessage, MockMessage]) -> None:
         """A mock handler that does nothing."""
         pass
 
@@ -104,99 +105,23 @@ def test_workflow_builder_fluent_api():
     assert len(workflow.executors) == 6
 
 
-def test_add_agent_with_custom_parameters():
-    """Test adding an agent with custom parameters."""
-    agent = DummyAgent(id="agent_custom", name="custom_agent")
-    builder = WorkflowBuilder()
-
-    # Add agent with custom parameters
-    with pytest.deprecated_call():
-        result = builder.add_agent(agent, output_response=True, id="my_custom_id")
-
-    # Verify that add_agent returns the builder for chaining
-    assert result is builder
-
-    # Build workflow and verify executor is present
-    workflow = builder.set_start_executor(agent).build()
-    assert "my_custom_id" in workflow.executors
-
-    # Verify the executor was created with correct parameters
-    executor = workflow.executors["my_custom_id"]
-    assert isinstance(executor, AgentExecutor)
-    assert executor.id == "my_custom_id"
-    assert getattr(executor, "_output_response", False) is True
-
-
 def test_add_agent_reuses_same_wrapper():
     """Test that using the same agent instance multiple times reuses the same wrapper."""
-    agent = DummyAgent(id="agent_reuse", name="reuse_agent")
+    reuse_agent = DummyAgent(id="agent_reuse", name="reuse_agent")
+    agent_a = DummyAgent(id="agent_a", name="agent_a")
+
     builder = WorkflowBuilder()
-
-    # Add agent with specific parameters
-    with pytest.deprecated_call():
-        builder.add_agent(agent, output_response=True, id="agent_exec")
-
     # Use the same agent instance in add_edge - should reuse the same wrapper
-    builder.set_start_executor(agent)
+    builder.set_start_executor(reuse_agent)
+    builder.add_edge(reuse_agent, agent_a)
+    builder.add_edge(agent_a, reuse_agent)
 
     workflow = builder.build()
 
     # Verify only one executor exists for this agent
-    assert workflow.start_executor_id == "agent_exec"
-    assert "agent_exec" in workflow.executors
-    assert len([e for e in workflow.executors.values() if isinstance(e, AgentExecutor)]) == 1
-
-    # Verify the executor has the parameters from add_agent
-    start_executor = workflow.get_start_executor()
-    assert isinstance(start_executor, AgentExecutor)
-    assert getattr(start_executor, "_output_response", False) is True
-
-
-def test_add_agent_then_use_in_edges():
-    """Test that an agent added via add_agent can be used in edge definitions."""
-    agent1 = DummyAgent(id="agent1", name="first")
-    agent2 = DummyAgent(id="agent2", name="second")
-    builder = WorkflowBuilder()
-
-    # Add agents with specific settings
-    with pytest.deprecated_call():
-        builder.add_agent(agent1, output_response=False, id="exec1")
-        builder.add_agent(agent2, output_response=True, id="exec2")
-
-    # Use the same agent instances to create edges
-    workflow = builder.set_start_executor(agent1).add_edge(agent1, agent2).build()
-
-    # Verify the executors maintain their settings
-    assert workflow.start_executor_id == "exec1"
-    assert "exec1" in workflow.executors
-    assert "exec2" in workflow.executors
-
-    e1 = workflow.executors["exec1"]
-    e2 = workflow.executors["exec2"]
-
-    assert isinstance(e1, AgentExecutor)
-    assert isinstance(e2, AgentExecutor)
-    assert getattr(e1, "_output_response", True) is False
-    assert getattr(e2, "_output_response", False) is True
-
-
-def test_add_agent_without_explicit_id_uses_agent_name():
-    """Test that add_agent uses agent name as id when no explicit id is provided."""
-    agent = DummyAgent(id="agent_x", name="named_agent")
-    builder = WorkflowBuilder()
-
-    with pytest.deprecated_call():
-        result = builder.add_agent(agent)
-
-    # Verify that add_agent returns the builder for chaining
-    assert result is builder
-
-    workflow = builder.set_start_executor(agent).build()
-    assert "named_agent" in workflow.executors
-
-    # Verify the executor id matches the agent name
-    executor = workflow.executors["named_agent"]
-    assert executor.id == "named_agent"
+    assert workflow.start_executor_id == "reuse_agent"
+    assert "reuse_agent" in workflow.executors
+    assert len([e for e in workflow.executors.values() if isinstance(e, AgentExecutor)]) == 2
 
 
 def test_add_agent_duplicate_id_raises_error():
@@ -205,13 +130,8 @@ def test_add_agent_duplicate_id_raises_error():
     agent2 = DummyAgent(id="agent2", name="first")  # Same name as agent1
     builder = WorkflowBuilder()
 
-    # Add first agent
-    with pytest.deprecated_call():
-        builder.add_agent(agent1)
-
-    # Adding second agent with same name should raise ValueError
-    with pytest.deprecated_call(), pytest.raises(ValueError, match="Duplicate executor ID"):
-        builder.add_agent(agent2)
+    with pytest.raises(ValueError, match="Duplicate executor ID"):
+        builder.set_start_executor(agent1).add_edge(agent1, agent2).build()
 
 
 # Tests for new executor registration patterns
@@ -303,7 +223,7 @@ def test_register_duplicate_id_raises_error():
     builder.set_start_executor("MyExecutor1")
 
     # Registering second executor with same ID should raise ValueError
-    with pytest.raises(ValueError, match="Executor with ID 'executor' has already been created."):
+    with pytest.raises(ValueError, match="Executor with ID 'executor' has already been registered."):
         builder.build()
 
 
@@ -312,9 +232,7 @@ def test_register_agent_basic():
     builder = WorkflowBuilder()
 
     # Register an agent factory
-    result = builder.register_agent(
-        lambda: DummyAgent(id="agent_test", name="test_agent"), name="TestAgent", output_response=True
-    )
+    result = builder.register_agent(lambda: DummyAgent(id="agent_test", name="test_agent"), name="TestAgent")
 
     # Verify that register_agent returns the builder for chaining
     assert result is builder
@@ -323,7 +241,6 @@ def test_register_agent_basic():
     workflow = builder.set_start_executor("TestAgent").build()
     assert "test_agent" in workflow.executors
     assert isinstance(workflow.executors["test_agent"], AgentExecutor)
-    assert workflow.executors["test_agent"]._output_response is True  # type: ignore
 
 
 def test_register_agent_with_thread():
@@ -336,7 +253,6 @@ def test_register_agent_with_thread():
         lambda: DummyAgent(id="agent_with_thread", name="threaded_agent"),
         name="ThreadedAgent",
         agent_thread=custom_thread,
-        output_response=False,
     )
 
     # Build workflow and verify agent executor configuration
@@ -345,7 +261,6 @@ def test_register_agent_with_thread():
 
     assert isinstance(executor, AgentExecutor)
     assert executor.id == "threaded_agent"
-    assert executor._output_response is False  # type: ignore
     assert executor._agent_thread is custom_thread  # type: ignore
 
 
@@ -549,3 +464,151 @@ def test_register_agent_creates_unique_instances():
     # Verify that two different agent instances were created
     assert len(instance_ids) == 2
     assert instance_ids[0] != instance_ids[1]
+
+
+# region with_output_from tests
+
+
+def test_with_output_from_returns_builder():
+    """Test that with_output_from returns the builder for method chaining."""
+    executor_a = MockExecutor(id="executor_a")
+    builder = WorkflowBuilder()
+
+    result = builder.with_output_from([executor_a])
+
+    assert result is builder
+
+
+def test_with_output_from_with_executor_instances():
+    """Test with_output_from with direct executor instances."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .with_output_from([executor_b])
+        .build()
+    )
+
+    # Verify that the workflow was built with the correct output executors
+    assert workflow._output_executors == ["executor_b"]  # type: ignore
+
+
+def test_with_output_from_with_agent_instances():
+    """Test with_output_from with agent instances."""
+    agent_a = DummyAgent(id="agent_a", name="writer")
+    agent_b = DummyAgent(id="agent_b", name="reviewer")
+
+    workflow = (
+        WorkflowBuilder().set_start_executor(agent_a).add_edge(agent_a, agent_b).with_output_from([agent_b]).build()
+    )
+
+    # Verify that the workflow was built with the agent's name as output executor
+    assert workflow._output_executors == ["reviewer"]  # type: ignore
+
+
+def test_with_output_from_with_registered_names():
+    """Test with_output_from with registered factory names (strings)."""
+    workflow = (
+        WorkflowBuilder()
+        .register_executor(lambda: MockExecutor(id="ExecutorA"), name="ExecutorAFactory")
+        .register_executor(lambda: MockExecutor(id="ExecutorB"), name="ExecutorBFactory")
+        .set_start_executor("ExecutorAFactory")
+        .add_edge("ExecutorAFactory", "ExecutorBFactory")
+        .with_output_from(["ExecutorBFactory"])
+        .build()
+    )
+
+    # Verify that the workflow was built with the correct output executors
+    assert workflow._output_executors == ["ExecutorB"]  # type: ignore
+
+
+def test_with_output_from_with_multiple_executors():
+    """Test with_output_from with multiple executors."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+    executor_c = MockExecutor(id="executor_c")
+
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .add_edge(executor_b, executor_c)
+        .with_output_from([executor_a, executor_c])
+        .build()
+    )
+
+    # Verify that the workflow was built with both output executors
+    assert set(workflow._output_executors) == {"executor_a", "executor_c"}  # type: ignore
+
+
+def test_with_output_from_can_be_called_multiple_times():
+    """Test that calling with_output_from multiple times overwrites the previous setting."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .add_edge(executor_a, executor_b)
+        .with_output_from([executor_a])
+        .with_output_from([executor_b])  # This should overwrite the previous setting
+        .build()
+    )
+
+    # Verify that only the last setting is applied
+    assert workflow._output_executors == ["executor_b"]  # type: ignore
+
+
+def test_with_output_from_with_registered_agents():
+    """Test with_output_from with registered agent factory names."""
+    workflow = (
+        WorkflowBuilder()
+        .register_agent(lambda: DummyAgent(id="agent1", name="writer"), name="WriterAgent")
+        .register_agent(lambda: DummyAgent(id="agent2", name="reviewer"), name="ReviewerAgent")
+        .set_start_executor("WriterAgent")
+        .add_edge("WriterAgent", "ReviewerAgent")
+        .with_output_from(["ReviewerAgent"])
+        .build()
+    )
+
+    # Verify that the workflow was built with the agent's resolved name
+    assert workflow._output_executors == ["reviewer"]  # type: ignore
+
+
+def test_with_output_from_in_fluent_chain():
+    """Test that with_output_from works correctly in a fluent builder chain."""
+    executor_a = MockExecutor(id="executor_a")
+    executor_b = MockExecutor(id="executor_b")
+    executor_c = MockExecutor(id="executor_c")
+
+    # Build workflow with with_output_from in the middle of the chain
+    workflow = (
+        WorkflowBuilder()
+        .set_start_executor(executor_a)
+        .with_output_from([executor_c])  # Set early in the chain
+        .add_edge(executor_a, executor_b)
+        .add_edge(executor_b, executor_c)
+        .build()
+    )
+
+    # Verify that the setting persists through the chain
+    assert workflow._output_executors == ["executor_c"]  # type: ignore
+
+
+def test_with_output_from_with_invalid_executor_raises_validation_error():
+    """Test that with_output_from with an invalid executor raises an error."""
+    executor_a = MockExecutor(id="executor_a")
+
+    builder = WorkflowBuilder().set_start_executor(executor_a)
+
+    # Attempting to set output from an executor not in the workflow should raise an error
+    with pytest.raises(
+        WorkflowValidationError, match="Output executor 'executor_b' is not present in the workflow graph"
+    ):
+        builder.with_output_from([MockExecutor(id="executor_b")]).build()
+
+
+# endregion