Python: Read orchestration outputs as agent responses in tool approval samples

python/samples/03-workflows/orchestrations/concurrent_custom_agent_executors.py

@@ -2,14 +2,14 @@
 
 import asyncio
 import os
-from typing import Any
+from typing import cast
 
 from agent_framework import (
     Agent,
     AgentExecutorRequest,
     AgentExecutorResponse,
+    AgentResponse,
     Executor,
-    Message,
     WorkflowContext,
     handler,
 )
@@ -124,8 +124,8 @@ async def main() -> None:
 
     if outputs:
         print("===== Final Aggregated Conversation (messages) =====")
-        messages: list[Message] | Any = outputs[0]  # Get the first (and typically only) output
-        for i, msg in enumerate(messages, start=1):
+        response = cast(AgentResponse, outputs[0])  # The aggregated conversation
+        for i, msg in enumerate(response.messages, start=1):
             name = msg.author_name if msg.author_name else "user"
             print(f"{'-' * 60}\n\n{i:02d} [{name}]:\n{msg.text}")
 

python/samples/03-workflows/tool-approval/concurrent_builder_tool_approval.py

@@ -7,8 +7,8 @@
 
 from agent_framework import (
     Agent,
+    AgentResponse,
     Content,
-    Message,
     WorkflowEvent,
     tool,
 )
@@ -108,10 +108,10 @@ def _print_output(event: WorkflowEvent) -> None:
     if not event.data:
         raise ValueError("WorkflowEvent has no data")
 
-    if not isinstance(event.data, list) and not all(isinstance(msg, Message) for msg in event.data):
-        raise ValueError("WorkflowEvent data is not a list of Message")
+    if not isinstance(event.data, AgentResponse):
+        raise ValueError("WorkflowEvent data is not an AgentResponse")
 
-    messages: list[Message] = event.data  # type: ignore
+    messages = event.data.messages
 
     print("\n" + "-" * 60)
     print("Workflow completed. Aggregated results from both agents:")

python/samples/03-workflows/tool-approval/group_chat_builder_tool_approval.py

@@ -3,12 +3,13 @@
 import asyncio
 import os
 from collections.abc import AsyncIterable
-from typing import Annotated, cast
+from typing import Annotated
 
 from agent_framework import (
     Agent,
+    AgentResponse,
+    AgentResponseUpdate,
     Content,
-    Message,
     WorkflowEvent,
     tool,
 )
@@ -105,16 +106,20 @@ def select_next_speaker(state: GroupChatState) -> str:
 async def process_event_stream(stream: AsyncIterable[WorkflowEvent]) -> dict[str, Content] | None:
     """Process events from the workflow stream to capture human feedback requests."""
     requests: dict[str, Content] = {}
+    output_updates: list[AgentResponseUpdate] = []
     async for event in stream:
         if event.type == "request_info" and isinstance(event.data, Content):
             # We are only expecting tool approval requests in this sample
             requests[event.request_id] = event.data
-        elif event.type == "output":
-            # The output of the workflow comes from the orchestrator and it's a list of messages
-            print("\n" + "=" * 60)
-            print("Workflow summary:")
-            outputs = cast(list[Message], event.data)
-            for msg in outputs:
+        elif event.type == "output" and isinstance(event.data, AgentResponseUpdate):
+            # The workflow output streams as agent response updates
+            output_updates.append(event.data)
+
+    if output_updates:
+        print("\n" + "=" * 60)
+        print("Workflow summary:")
+        for msg in AgentResponse.from_updates(output_updates).messages:
+            if msg.text:
                 speaker = msg.author_name or msg.role
                 print(f"[{speaker}]: {msg.text}")
 

python/samples/03-workflows/tool-approval/sequential_builder_tool_approval.py

@@ -3,12 +3,13 @@
 import asyncio
 import os
 from collections.abc import AsyncIterable
-from typing import Annotated, cast
+from typing import Annotated
 
 from agent_framework import (
     Agent,
+    AgentResponse,
+    AgentResponseUpdate,
     Content,
-    Message,
     WorkflowEvent,
     tool,
 )
@@ -78,16 +79,20 @@ def get_database_schema() -> str:
 async def process_event_stream(stream: AsyncIterable[WorkflowEvent]) -> dict[str, Content] | None:
     """Process events from the workflow stream to capture human feedback requests."""
     requests: dict[str, Content] = {}
+    output_updates: list[AgentResponseUpdate] = []
     async for event in stream:
         if event.type == "request_info" and isinstance(event.data, Content):
             # We are only expecting tool approval requests in this sample
             requests[event.request_id] = event.data
-        elif event.type == "output":
-            # The output of the workflow comes from the orchestrator and it's a list of messages
-            print("\n" + "=" * 60)
-            print("Workflow summary:")
-            outputs = cast(list[Message], event.data)
-            for msg in outputs:
+        elif event.type == "output" and isinstance(event.data, AgentResponseUpdate):
+            # The workflow output streams as agent response updates
+            output_updates.append(event.data)
+
+    if output_updates:
+        print("\n" + "=" * 60)
+        print("Workflow summary:")
+        for msg in AgentResponse.from_updates(output_updates).messages:
+            if msg.text:
                 speaker = msg.author_name or msg.role
                 print(f"[{speaker}]: {msg.text}")