Added structured output to agent streaming (#943)

python/packages/main/agent_framework/_types.py

@@ -2096,29 +2096,47 @@ class AgentRunResponse(AFBaseModel):
 
     @classmethod
     def from_agent_run_response_updates(
-        cls: type[TAgentRunResponse], updates: Sequence["AgentRunResponseUpdate"]
+        cls: type[TAgentRunResponse],
+        updates: Sequence["AgentRunResponseUpdate"],
+        *,
+        output_format_type: type[BaseModel] | None = None,
     ) -> TAgentRunResponse:
         """Joins multiple updates into a single AgentRunResponse."""
         msg = cls(messages=[])
         for update in updates:
             _process_update(msg, update)
         _finalize_response(msg)
+        if output_format_type:
+            msg.try_parse_value(output_format_type)
         return msg
 
     @classmethod
     async def from_agent_response_generator(
-        cls: type[TAgentRunResponse], updates: AsyncIterable["AgentRunResponseUpdate"]
+        cls: type[TAgentRunResponse],
+        updates: AsyncIterable["AgentRunResponseUpdate"],
+        *,
+        output_format_type: type[BaseModel] | None = None,
     ) -> TAgentRunResponse:
         """Joins multiple updates into a single AgentRunResponse."""
         msg = cls(messages=[])
         async for update in updates:
             _process_update(msg, update)
         _finalize_response(msg)
+        if output_format_type:
+            msg.try_parse_value(output_format_type)
         return msg
 
     def __str__(self) -> str:
         return self.text
 
+    def try_parse_value(self, output_format_type: type[BaseModel]) -> None:
+        """If there is a value, does nothing, otherwise tries to parse the text into the value."""
+        if self.value is None:
+            try:
+                self.value = output_format_type.model_validate_json(self.text)  # type: ignore[reportUnknownMemberType]
+            except ValidationError as ex:
+                logger.debug("Failed to parse value from agent run response text: %s", ex)
+
 
 # region AgentRunResponseUpdate
 

python/samples/getting_started/agents/openai/openai_responses_client_with_structured_output.py

@@ -2,6 +2,7 @@
 
 import asyncio
 
+from agent_framework import AgentRunResponse
 from agent_framework.openai import OpenAIResponsesClient
 from pydantic import BaseModel
 
@@ -13,24 +14,23 @@ class OutputStruct(BaseModel):
     description: str
 
 
-async def main():
-    print("=== OpenAI Responses Agent with Structured Output ===")
+async def non_streaming_example() -> None:
+    print("=== Non-streaming example ===")
 
-    # 1. Create an OpenAI Responses agent
+    # Create an OpenAI Responses agent
     agent = OpenAIResponsesClient().create_agent(
         name="CityAgent",
         instructions="You are a helpful agent that describes cities in a structured format.",
     )
 
-    # 2. Ask the agent about a city
+    # Ask the agent about a city
     query = "Tell me about Paris, France"
-
     print(f"User: {query}")
 
-    # 3. Get structured response from the agent using response_format parameter
+    # Get structured response from the agent using response_format parameter
     result = await agent.run(query, response_format=OutputStruct)
 
-    # 4. Access the structured output directly from the response value
+    # Access the structured output directly from the response value
     if result.value:
         structured_data = result.value
         print("Structured Output Agent (from result.value):")
@@ -40,5 +40,42 @@ async def main():
         print("Error: No structured data found in result.value")
 
 
+async def streaming_example() -> None:
+    print("=== Streaming example ===")
+
+    # Create an OpenAI Responses agent
+    agent = OpenAIResponsesClient().create_agent(
+        name="CityAgent",
+        instructions="You are a helpful agent that describes cities in a structured format.",
+    )
+
+    # Ask the agent about a city
+    query = "Tell me about Tokyo, Japan"
+    print(f"User: {query}")
+
+    # Get structured response from streaming agent using AgentRunResponse.from_agent_response_generator
+    # This method collects all streaming updates and combines them into a single AgentRunResponse
+    result = await AgentRunResponse.from_agent_response_generator(
+        agent.run_stream(query, response_format=OutputStruct),
+        output_format_type=OutputStruct,
+    )
+
+    # Access the structured output directly from the response value
+    if result.value:
+        structured_data = result.value
+        print("Structured Output (from streaming with AgentRunResponse.from_agent_response_generator):")
+        print(f"City: {structured_data.city}")
+        print(f"Description: {structured_data.description}")
+    else:
+        print("Error: No structured data found in result.value")
+
+
+async def main() -> None:
+    print("=== OpenAI Responses Agent with Structured Output ===")
+
+    await non_streaming_example()
+    await streaming_example()
+
+
 if __name__ == "__main__":
     asyncio.run(main())