refactoring

Author: eavanvalkenburg

python/packages/anthropic/agent_framework_anthropic/_chat_client.py

@@ -345,12 +345,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]:
                     if parsed_chunk:
                         yield parsed_chunk
 
-            def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
-                response_format = options.get("response_format")
-                output_format_type = response_format if isinstance(response_format, type) else None
-                return ChatResponse.from_chat_response_updates(updates, output_format_type=output_format_type)
-
-            return ResponseStream(_stream(), finalizer=_finalize)
+            return self._build_response_stream(_stream(), response_format=options.get("response_format"))
 
         # Non-streaming mode
         async def _get_response() -> ChatResponse:

python/packages/azure-ai/agent_framework_azure_ai/_chat_client.py

@@ -362,12 +362,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]:
                 ):
                     yield update
 
-            def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
-                response_format = options.get("response_format")
-                output_format_type = response_format if isinstance(response_format, type) else None
-                return ChatResponse.from_chat_response_updates(updates, output_format_type=output_format_type)
-
-            return ResponseStream(_stream(), finalizer=_finalize)
+            return self._build_response_stream(_stream(), response_format=options.get("response_format"))
 
         # Non-streaming mode - collect updates and convert to response
         async def _get_response() -> ChatResponse:

python/packages/bedrock/agent_framework_bedrock/_chat_client.py

@@ -323,7 +323,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]:
                     raw_representation=parsed_response.raw_representation,
                 )
 
-            return ResponseStream(_stream(), finalizer=ChatResponse.from_chat_response_updates)
+            return self._build_response_stream(_stream())
 
         # Non-streaming mode
         async def _get_response() -> ChatResponse:

python/packages/core/agent_framework/_clients.py

@@ -3,6 +3,7 @@
 import sys
 from abc import ABC, abstractmethod
 from collections.abc import (
+    AsyncIterable,
     Awaitable,
     Callable,
     Mapping,
@@ -289,6 +290,28 @@ async def _validate_options(self, options: Mapping[str, Any]) -> dict[str, Any]:
         """
         return await validate_chat_options(options)
 
+    def _finalize_response_updates(
+        self,
+        updates: Sequence[ChatResponseUpdate],
+        *,
+        response_format: Any | None = None,
+    ) -> ChatResponse:
+        """Finalize response updates into a single ChatResponse."""
+        output_format_type = response_format if isinstance(response_format, type) else None
+        return ChatResponse.from_chat_response_updates(updates, output_format_type=output_format_type)
+
+    def _build_response_stream(
+        self,
+        stream: AsyncIterable[ChatResponseUpdate] | Awaitable[AsyncIterable[ChatResponseUpdate]],
+        *,
+        response_format: Any | None = None,
+    ) -> ResponseStream[ChatResponseUpdate, ChatResponse]:
+        """Create a ResponseStream with the standard finalizer."""
+        return ResponseStream(
+            stream,
+            finalizer=lambda updates: self._finalize_response_updates(updates, response_format=response_format),
+        )
+
     # region Internal method to be implemented by derived classes
 
     @abstractmethod

python/packages/core/agent_framework/openai/_assistants_client.py

@@ -366,12 +366,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]:
                 async for update in self._process_stream_events(stream_obj, thread_id):
                     yield update
 
-            def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
-                response_format = options.get("response_format")
-                output_format_type = response_format if isinstance(response_format, type) else None
-                return ChatResponse.from_chat_response_updates(updates, output_format_type=output_format_type)
-
-            return ResponseStream(_stream(), finalizer=_finalize)
+            return self._build_response_stream(_stream(), response_format=options.get("response_format"))
 
         # Non-streaming mode - collect updates and convert to response
         async def _get_response() -> ChatResponse:

python/packages/core/agent_framework/openai/_chat_client.py

@@ -166,12 +166,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]:
                         inner_exception=ex,
                     ) from ex
 
-            def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
-                response_format = options.get("response_format")
-                output_format_type = response_format if isinstance(response_format, type) else None
-                return ChatResponse.from_chat_response_updates(updates, output_format_type=output_format_type)
-
-            return ResponseStream(_stream(), finalizer=_finalize)
+            return self._build_response_stream(_stream(), response_format=options.get("response_format"))
 
         # Non-streaming mode
         async def _get_response() -> ChatResponse:

python/packages/core/agent_framework/openai/_responses_client.py

@@ -262,12 +262,8 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]:
                 except Exception as ex:
                     self._handle_request_error(ex)
 
-            def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
-                response_format = validated_options.get("response_format") if validated_options else None
-                output_format_type = response_format if isinstance(response_format, type) else None
-                return ChatResponse.from_chat_response_updates(updates, output_format_type=output_format_type)
-
-            return ResponseStream(_stream(), finalizer=_finalize)
+            response_format = validated_options.get("response_format") if validated_options else None
+            return self._build_response_stream(_stream(), response_format=response_format)
 
         # Non-streaming
         async def _get_response() -> ChatResponse:

python/packages/ollama/agent_framework_ollama/_chat_client.py

@@ -352,12 +352,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]:
                 async for part in response_object:
                     yield self._parse_streaming_response_from_ollama(part)
 
-            def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse:
-                response_format = options.get("response_format")
-                output_format_type = response_format if isinstance(response_format, type) else None
-                return ChatResponse.from_chat_response_updates(updates, output_format_type=output_format_type)
-
-            return ResponseStream(_stream(), finalizer=_finalize)
+            return self._build_response_stream(_stream(), response_format=options.get("response_format"))
 
         # Non-streaming mode
         async def _get_response() -> ChatResponse: