StreamedRunResultSync

Author: DouweM

pydantic_ai_slim/pydantic_ai/agent/abstract.py

@@ -597,7 +597,7 @@ def run_stream_sync(
         toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
         builtin_tools: Sequence[AbstractBuiltinTool] | None = None,
         event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
-    ) -> result.StreamedRunResult[AgentDepsT, OutputDataT]: ...
+    ) -> result.StreamedRunResultSync[AgentDepsT, OutputDataT]: ...
 
     @overload
     def run_stream_sync(
@@ -616,7 +616,7 @@ def run_stream_sync(
         toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
         builtin_tools: Sequence[AbstractBuiltinTool] | None = None,
         event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
-    ) -> result.StreamedRunResult[AgentDepsT, RunOutputDataT]: ...
+    ) -> result.StreamedRunResultSync[AgentDepsT, RunOutputDataT]: ...
 
     def run_stream_sync(
         self,
@@ -634,7 +634,7 @@ def run_stream_sync(
         toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
         builtin_tools: Sequence[AbstractBuiltinTool] | None = None,
         event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
-    ) -> result.StreamedRunResult[AgentDepsT, Any]:
+    ) -> result.StreamedRunResultSync[AgentDepsT, Any]:
         """Run the agent with a user prompt in sync streaming mode.
 
         This is a convenience method that wraps [`run_stream()`][pydantic_ai.agent.AbstractAgent.run_stream] with `loop.run_until_complete(...)`.
@@ -658,7 +658,7 @@ def run_stream_sync(
 
         def main():
             response = agent.run_stream_sync('What is the capital of the UK?')
-            print(response.get_output_sync())
+            print(response.get_output())
             #> The capital of the UK is London.
         ```
 
@@ -704,7 +704,8 @@ async def _consume_stream():
             ) as stream_result:
                 yield stream_result
 
-        return _utils.get_event_loop().run_until_complete(anext(_consume_stream()))
+        async_result = _utils.get_event_loop().run_until_complete(anext(_consume_stream()))
+        return result.StreamedRunResultSync(async_result)
 
     @overload
     def run_stream_events(

pydantic_ai_slim/pydantic_ai/result.py

@@ -35,6 +35,7 @@
     'OutputDataT_inv',
     'ToolOutput',
     'OutputValidatorFunc',
+    'StreamedRunResultSync',
 )
 
 
@@ -420,26 +421,6 @@ async def stream_output(self, *, debounce_by: float | None = 0.1) -> AsyncIterat
         else:
             raise ValueError('No stream response or run result provided')  # pragma: no cover
 
-    def stream_output_sync(self, *, debounce_by: float | None = 0.1) -> Iterator[OutputDataT]:
-        """Stream the output as an iterable.
-
-        This is a convenience method that wraps [`stream_output()`][pydantic_ai.result.StreamedRunResult.stream_output] with `loop.run_until_complete(...)`.
-        You therefore can't use this method inside async code or if there's an active event loop.
-
-        The pydantic validator for structured data will be called in
-        [partial mode](https://docs.pydantic.dev/dev/concepts/experimental/#partial-validation)
-        on each iteration.
-
-        Args:
-            debounce_by: by how much (if at all) to debounce/group the output chunks by. `None` means no debouncing.
-                Debouncing is particularly important for long structured outputs to reduce the overhead of
-                performing validation as each token is received.
-
-        Returns:
-            An iterable of the response data.
-        """
-        return _utils.sync_async_iterator(self.stream_output(debounce_by=debounce_by))
-
     async def stream_text(self, *, delta: bool = False, debounce_by: float | None = 0.1) -> AsyncIterator[str]:
         """Stream the text result as an async iterable.
 
@@ -468,24 +449,6 @@ async def stream_text(self, *, delta: bool = False, debounce_by: float | None =
         else:
             raise ValueError('No stream response or run result provided')  # pragma: no cover
 
-    def stream_text_sync(self, *, delta: bool = False, debounce_by: float | None = 0.1) -> Iterator[str]:
-        """Stream the text result as a sync iterable.
-
-        This is a convenience method that wraps [`stream_text()`][pydantic_ai.result.StreamedRunResult.stream_text] with `loop.run_until_complete(...)`.
-        You therefore can't use this method inside async code or if there's an active event loop.
-
-        !!! note
-            Result validators will NOT be called on the text result if `delta=True`.
-
-        Args:
-            delta: if `True`, yield each chunk of text as it is received, if `False` (default), yield the full text
-                up to the current point.
-            debounce_by: by how much (if at all) to debounce/group the response chunks by. `None` means no debouncing.
-                Debouncing is particularly important for long structured responses to reduce the overhead of
-                performing validation as each token is received.
-        """
-        return _utils.sync_async_iterator(self.stream_text(delta=delta, debounce_by=debounce_by))
-
     @deprecated('`StreamedRunResult.stream_structured` is deprecated, use `stream_responses` instead.')
     async def stream_structured(
         self, *, debounce_by: float | None = 0.1
@@ -521,24 +484,6 @@ async def stream_responses(
         else:
             raise ValueError('No stream response or run result provided')  # pragma: no cover
 
-    def stream_responses_sync(
-        self, *, debounce_by: float | None = 0.1
-    ) -> Iterator[tuple[_messages.ModelResponse, bool]]:
-        """Stream the response as an iterable of Structured LLM Messages.
-
-        This is a convenience method that wraps [`stream_responses()`][pydantic_ai.result.StreamedRunResult.stream_responses] with `loop.run_until_complete(...)`.
-        You therefore can't use this method inside async code or if there's an active event loop.
-
-        Args:
-            debounce_by: by how much (if at all) to debounce/group the response chunks by. `None` means no debouncing.
-                Debouncing is particularly important for long structured responses to reduce the overhead of
-                performing validation as each token is received.
-
-        Returns:
-            An iterable of the structured response message and whether that is the last message.
-        """
-        return _utils.sync_async_iterator(self.stream_responses(debounce_by=debounce_by))
-
     async def get_output(self) -> OutputDataT:
         """Stream the whole response, validate and return it."""
         if self._run_result is not None:
@@ -552,14 +497,6 @@ async def get_output(self) -> OutputDataT:
         else:
             raise ValueError('No stream response or run result provided')  # pragma: no cover
 
-    def get_output_sync(self) -> OutputDataT:
-        """Stream the whole response, validate and return it.
-
-        This is a convenience method that wraps [`get_output()`][pydantic_ai.result.StreamedRunResult.get_output] with `loop.run_until_complete(...)`.
-        You therefore can't use this method inside async code or if there's an active event loop.
-        """
-        return _utils.get_event_loop().run_until_complete(self.get_output())
-
     @property
     def response(self) -> _messages.ModelResponse:
         """Return the current state of the response."""
@@ -611,18 +548,6 @@ async def validate_response_output(
         else:
             raise ValueError('No stream response or run result provided')  # pragma: no cover
 
-    def validate_response_output_sync(
-        self, message: _messages.ModelResponse, *, allow_partial: bool = False
-    ) -> OutputDataT:
-        """Validate a structured result message.
-
-        This is a convenience method that wraps [`validate_response_output()`][pydantic_ai.result.StreamedRunResult.validate_response_output] with `loop.run_until_complete(...)`.
-        You therefore can't use this method inside async code or if there's an active event loop.
-        """
-        return _utils.get_event_loop().run_until_complete(
-            self.validate_response_output(message, allow_partial=allow_partial)
-        )
-
     async def _marked_completed(self, message: _messages.ModelResponse | None = None) -> None:
         self.is_complete = True
         if message is not None:
@@ -631,6 +556,158 @@ async def _marked_completed(self, message: _messages.ModelResponse | None = None
             await self._on_complete()
 
 
+@dataclass(init=False)
+class StreamedRunResultSync(Generic[AgentDepsT, OutputDataT]):
+    """Synchronous wrapper for [`StreamedRunResult`][pydantic_ai.result.StreamedRunResult] that only exposes sync methods."""
+
+    _streamed_run_result: StreamedRunResult[AgentDepsT, OutputDataT]
+
+    def __init__(self, streamed_run_result: StreamedRunResult[AgentDepsT, OutputDataT]) -> None:
+        self._streamed_run_result = streamed_run_result
+
+    def all_messages(self, *, output_tool_return_content: str | None = None) -> list[_messages.ModelMessage]:
+        """Return the history of messages.
+
+        Args:
+            output_tool_return_content: The return content of the tool call to set in the last message.
+                This provides a convenient way to modify the content of the output tool call if you want to continue
+                the conversation and want to set the response to the output tool call. If `None`, the last message will
+                not be modified.
+
+        Returns:
+            List of messages.
+        """
+        return self._streamed_run_result.all_messages(output_tool_return_content=output_tool_return_content)
+
+    def all_messages_json(self, *, output_tool_return_content: str | None = None) -> bytes:  # pragma: no cover
+        """Return all messages from [`all_messages`][pydantic_ai.result.StreamedRunResultSync.all_messages] as JSON bytes.
+
+        Args:
+            output_tool_return_content: The return content of the tool call to set in the last message.
+                This provides a convenient way to modify the content of the output tool call if you want to continue
+                the conversation and want to set the response to the output tool call. If `None`, the last message will
+                not be modified.
+
+        Returns:
+            JSON bytes representing the messages.
+        """
+        return self._streamed_run_result.all_messages_json(output_tool_return_content=output_tool_return_content)
+
+    def new_messages(self, *, output_tool_return_content: str | None = None) -> list[_messages.ModelMessage]:
+        """Return new messages associated with this run.
+
+        Messages from older runs are excluded.
+
+        Args:
+            output_tool_return_content: The return content of the tool call to set in the last message.
+                This provides a convenient way to modify the content of the output tool call if you want to continue
+                the conversation and want to set the response to the output tool call. If `None`, the last message will
+                not be modified.
+
+        Returns:
+            List of new messages.
+        """
+        return self._streamed_run_result.new_messages(output_tool_return_content=output_tool_return_content)
+
+    def new_messages_json(self, *, output_tool_return_content: str | None = None) -> bytes:  # pragma: no cover
+        """Return new messages from [`new_messages`][pydantic_ai.result.StreamedRunResultSync.new_messages] as JSON bytes.
+
+        Args:
+            output_tool_return_content: The return content of the tool call to set in the last message.
+                This provides a convenient way to modify the content of the output tool call if you want to continue
+                the conversation and want to set the response to the output tool call. If `None`, the last message will
+                not be modified.
+
+        Returns:
+            JSON bytes representing the new messages.
+        """
+        return self._streamed_run_result.new_messages_json(output_tool_return_content=output_tool_return_content)
+
+    def stream_output(self, *, debounce_by: float | None = 0.1) -> Iterator[OutputDataT]:
+        """Stream the output as an iterable.
+
+        The pydantic validator for structured data will be called in
+        [partial mode](https://docs.pydantic.dev/dev/concepts/experimental/#partial-validation)
+        on each iteration.
+
+        Args:
+            debounce_by: by how much (if at all) to debounce/group the output chunks by. `None` means no debouncing.
+                Debouncing is particularly important for long structured outputs to reduce the overhead of
+                performing validation as each token is received.
+
+        Returns:
+            An iterable of the response data.
+        """
+        return _utils.sync_async_iterator(self._streamed_run_result.stream_output(debounce_by=debounce_by))
+
+    def stream_text(self, *, delta: bool = False, debounce_by: float | None = 0.1) -> Iterator[str]:
+        """Stream the text result as an iterable.
+
+        !!! note
+            Result validators will NOT be called on the text result if `delta=True`.
+
+        Args:
+            delta: if `True`, yield each chunk of text as it is received, if `False` (default), yield the full text
+                up to the current point.
+            debounce_by: by how much (if at all) to debounce/group the response chunks by. `None` means no debouncing.
+                Debouncing is particularly important for long structured responses to reduce the overhead of
+                performing validation as each token is received.
+        """
+        return _utils.sync_async_iterator(self._streamed_run_result.stream_text(delta=delta, debounce_by=debounce_by))
+
+    def stream_responses(self, *, debounce_by: float | None = 0.1) -> Iterator[tuple[_messages.ModelResponse, bool]]:
+        """Stream the response as an iterable of Structured LLM Messages.
+
+        Args:
+            debounce_by: by how much (if at all) to debounce/group the response chunks by. `None` means no debouncing.
+                Debouncing is particularly important for long structured responses to reduce the overhead of
+                performing validation as each token is received.
+
+        Returns:
+            An iterable of the structured response message and whether that is the last message.
+        """
+        return _utils.sync_async_iterator(self._streamed_run_result.stream_responses(debounce_by=debounce_by))
+
+    def get_output(self) -> OutputDataT:
+        """Stream the whole response, validate and return it."""
+        return _utils.get_event_loop().run_until_complete(self._streamed_run_result.get_output())
+
+    @property
+    def response(self) -> _messages.ModelResponse:
+        """Return the current state of the response."""
+        return self._streamed_run_result.response
+
+    def usage(self) -> RunUsage:
+        """Return the usage of the whole run.
+
+        !!! note
+            This won't return the full usage until the stream is finished.
+        """
+        return self._streamed_run_result.usage()
+
+    def timestamp(self) -> datetime:
+        """Get the timestamp of the response."""
+        return self._streamed_run_result.timestamp()
+
+    def validate_response_output(self, message: _messages.ModelResponse, *, allow_partial: bool = False) -> OutputDataT:
+        """Validate a structured result message."""
+        return _utils.get_event_loop().run_until_complete(
+            self._streamed_run_result.validate_response_output(message, allow_partial=allow_partial)
+        )
+
+    @property
+    def is_complete(self) -> bool:
+        """Whether the stream has all been received.
+
+        This is set to `True` when one of
+        [`stream_output`][pydantic_ai.result.StreamedRunResultSync.stream_output],
+        [`stream_text`][pydantic_ai.result.StreamedRunResultSync.stream_text],
+        [`stream_responses`][pydantic_ai.result.StreamedRunResultSync.stream_responses] or
+        [`get_output`][pydantic_ai.result.StreamedRunResultSync.get_output] completes.
+        """
+        return self._streamed_run_result.is_complete
+
+
 @dataclass(repr=False)
 class FinalResult(Generic[OutputDataT]):
     """Marker class storing the final output of an agent run and associated metadata."""