Python: Introduce add_agent functionality and added output_response to AgentExecutor; agent streaming behavior to follow workflow invocation (#1184)

* refactor AgentExecutor, add output_response flag for switching on or off workflow output for each agent.

* introduce add_agent

* make default agent's streaming to false

* address comments

* fix test

* add is_streaming to RunnerContext and WorkflowContext

* fix add_agent return

* fix tests

* address comments

* resolve conflict

* update to address comments

* fix

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

@@ -8,22 +8,22 @@ from typing_extensions import Never
 
 from agent_framework import (
     AgentExecutor,
+    AgentExecutorResponse,
     AgentRunResponse,
     AgentRunResponseUpdate,
     AgentThread,
     BaseAgent,
     ChatMessage,
+    Executor,
     Role,
     SequentialBuilder,
     TextContent,
     WorkflowBuilder,
-    WorkflowOutputEvent,
+    WorkflowContext,
     WorkflowRunState,
     WorkflowStatusEvent,
     handler,
 )
-from agent_framework._workflows._executor import AgentExecutorResponse, Executor
-from agent_framework._workflows._workflow_context import WorkflowContext
 
 
 class _SimpleAgent(BaseAgent):
@@ -71,28 +71,22 @@ class _CaptureFullConversation(Executor):
 
 
 async def test_agent_executor_populates_full_conversation_non_streaming() -> None:
-    # Arrange: non-streaming AgentExecutor for deterministic response composition
+    # Arrange: AgentExecutor will be non-streaming when using workflow.run()
     agent = _SimpleAgent(id="agent1", name="A", reply_text="agent-reply")
-    agent_exec = AgentExecutor(agent, streaming=False, id="agent1-exec")
+    agent_exec = AgentExecutor(agent, id="agent1-exec")
     capturer = _CaptureFullConversation(id="capture")
 
     wf = WorkflowBuilder().set_start_executor(agent_exec).add_edge(agent_exec, capturer).build()
 
-    # Act: run with a simple user prompt
-    completed = False
-    output: dict | None = None
-    async for ev in wf.run_stream("hello world"):
-        if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
-            completed = True
-        elif isinstance(ev, WorkflowOutputEvent):
-            output = ev.data  # type: ignore[assignment]
-        if completed and output is not None:
-            break
+    # Act: use run() instead of run_stream() to test non-streaming mode
+    result = await wf.run("hello world")
+
+    # Extract output from run result
+    outputs = result.get_outputs()
+    assert len(outputs) == 1
+    payload = outputs[0]
 
     # Assert: full_conversation contains [user("hello world"), assistant("agent-reply")]
-    assert completed
-    assert output is not None
-    payload = output
     assert isinstance(payload, dict)
     assert payload["length"] == 2
     assert payload["roles"][0] == Role.USER and "hello world" in (payload["texts"][0] or "")

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

@@ -2,12 +2,21 @@
 
 import asyncio
 import tempfile
+from collections.abc import AsyncIterable
 from dataclasses import dataclass
 from typing import Any
 
 import pytest
 
 from agent_framework import (
+    AgentExecutor,
+    AgentRunEvent,
+    AgentRunResponse,
+    AgentRunResponseUpdate,
+    AgentRunUpdateEvent,
+    AgentThread,
+    BaseAgent,
+    ChatMessage,
     Executor,
     FileCheckpointStorage,
     Message,
@@ -15,6 +24,8 @@ from agent_framework import (
     RequestInfoExecutor,
     RequestInfoMessage,
     RequestResponse,
+    Role,
+    TextContent,
     WorkflowBuilder,
     WorkflowContext,
     WorkflowEvent,
@@ -789,3 +800,76 @@ async def test_workflow_concurrent_execution_prevention_mixed_methods():
     # Now all methods should work again
     result = await workflow.run(NumberMessage(data=0))
     assert result.get_final_state() == WorkflowRunState.IDLE
+
+
+class _StreamingTestAgent(BaseAgent):
+    """Test agent that supports both streaming and non-streaming modes."""
+
+    def __init__(self, *, reply_text: str, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self._reply_text = reply_text
+
+    async def run(
+        self,
+        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        *,
+        thread: AgentThread | None = None,
+        **kwargs: Any,
+    ) -> AgentRunResponse:
+        """Non-streaming run - returns complete response."""
+        return AgentRunResponse(messages=[ChatMessage(role=Role.ASSISTANT, text=self._reply_text)])
+
+    async def run_stream(
+        self,
+        messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
+        *,
+        thread: AgentThread | None = None,
+        **kwargs: Any,
+    ) -> AsyncIterable[AgentRunResponseUpdate]:
+        """Streaming run - yields incremental updates."""
+        # Simulate streaming by yielding character by character
+        for char in self._reply_text:
+            yield AgentRunResponseUpdate(contents=[TextContent(text=char)])
+
+
+async def test_agent_streaming_vs_non_streaming() -> None:
+    """Test that run() emits AgentRunEvent while run_stream() emits AgentRunUpdateEvent."""
+    agent = _StreamingTestAgent(id="test_agent", name="TestAgent", reply_text="Hello World")
+    agent_exec = AgentExecutor(agent, id="agent_exec")
+
+    workflow = WorkflowBuilder().set_start_executor(agent_exec).build()
+
+    # Test non-streaming mode with run()
+    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)]
+
+    # 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.messages[0].text == "Hello World"
+
+    # Test streaming mode with run_stream()
+    stream_events: list[WorkflowEvent] = []
+    async for event in workflow.run_stream("test message"):
+        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)]
+
+    # 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"
+
+    # 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"
+
+    # 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.contents and e.data.contents[0].text
+    )
+    assert accumulated_text == "Hello World", f"Expected 'Hello World', got '{accumulated_text}'"

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

@@ -47,14 +47,6 @@ def test_builder_accepts_agents_directly():
     assert any(isinstance(e, AgentExecutor) and e.id in {"writer", "reviewer"} for e in wf.executors.values())
 
 
-def test_builder_agents_always_stream():
-    agent = DummyAgent(id="agentX", name="streamer")
-    wf = WorkflowBuilder().set_start_executor(agent).build()
-    exec_obj = wf.get_start_executor()
-    assert isinstance(exec_obj, AgentExecutor)
-    assert getattr(exec_obj, "_streaming", False) is True
-
-
 @dataclass
 class MockMessage:
     """A mock message for testing purposes."""
@@ -111,3 +103,108 @@ def test_workflow_builder_fluent_api():
     assert len(workflow.edge_groups) == 4
     assert workflow.start_executor_id == executor_a.id
     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
+    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")
+    builder = WorkflowBuilder()
+
+    # Add agent with specific parameters
+    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)
+
+    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
+    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()
+
+    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"
+
+
+def test_add_agent_duplicate_id_raises_error():
+    """Test that adding agents with duplicate IDs raises an error."""
+    agent1 = DummyAgent(id="agent1", name="first")
+    agent2 = DummyAgent(id="agent2", name="first")  # Same name as agent1
+    builder = WorkflowBuilder()
+
+    # Add first agent
+    builder.add_agent(agent1)
+
+    # Adding second agent with same name should raise ValueError
+    with pytest.raises(ValueError, match="Duplicate executor ID"):
+        builder.add_agent(agent2)