pydantic_ai.result
Bases: Generic[AgentDepsT, OutputDataT]
Whether the stream has been cancelled via cancel().
Type: bool
The unique identifier for the agent run.
Type: str
The unique identifier for the conversation this run belongs to.
Type: str
Metadata associated with this agent run, if configured.
Get the current state of the response.
Type: _messages.ModelResponse
Return the usage of the whole run.
Type: RunUsage
Get the timestamp of the response.
Type: datetime
@async
def stream_output(*, debounce_by: float | None = 0.1) -> AsyncIterator[OutputDataT]
Asynchronously stream the (validated) agent outputs.
AsyncIterator[OutputDataT]
@async
def stream_response(
*,
debounce_by: float | None = 0.1,
) -> AsyncIterator[_messages.ModelResponse]
Asynchronously stream the (unvalidated) model responses for the agent.
Yields ModelResponse snapshots — state='incomplete' while streaming is in flight,
followed by one final state='complete' snapshot (or 'interrupted' if cancel() was
called). If the underlying response already has accumulated content when this is called,
a pre-stream yield surfaces it before iteration begins.
AsyncIterator[_messages.ModelResponse]
@async
def stream_text(
*,
delta: bool = False,
debounce_by: float | None = 0.1,
) -> AsyncIterator[str]
Stream the text result as an async iterable.
delta : bool Default: False
if True, yield each chunk of text as it is received, if False (default), yield the full text
up to the current point.
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.
@async
def cancel() -> None
Cancel the stream, stopping token generation and closing the underlying connection.
@async
def drain() -> None
Consume all remaining events from the stream, discarding them.
@async
def get_output() -> OutputDataT
Stream the whole response, validate the output and return it.
OutputDataT
@async
def validate_response_output(
message: _messages.ModelResponse,
*,
allow_partial: bool = False,
) -> OutputDataT
Validate a structured result message.
OutputDataT
def __aiter__() -> AsyncIterator[AgentStreamEvent]
Stream AgentStreamEvents, interleaving events emitted into the run’s event buffer.
AsyncIterator[AgentStreamEvent]
Bases: Generic[AgentDepsT, OutputDataT]
Result of a streamed run that returns structured data via a tool call.
Whether the stream has all been received.
This is set to True when one of
stream_output,
stream_text,
stream_response or
get_output completes.
Type: bool Default: field(default=False, init=False)
Return the current state of the response.
Type: _messages.ModelResponse
Metadata associated with this agent run, if configured.
Return the usage of the whole run.
Type: RunUsage
Get the timestamp of the response.
Type: datetime
The unique identifier for the agent run.
Type: str
The unique identifier for the conversation this run belongs to.
Type: str
Whether the stream has been cancelled via cancel().
Type: bool
def all_messages(
*,
output_tool_return_content: str | None = None,
) -> list[_messages.ModelMessage]
Return the history of _messages.
list[_messages.ModelMessage] — List of messages.
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.
def all_messages_json(*, output_tool_return_content: str | None = None) -> bytes
Return all messages from all_messages as JSON bytes.
bytes — JSON bytes representing the messages.
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.
def new_messages(
*,
output_tool_return_content: str | None = None,
) -> list[_messages.ModelMessage]
Return the messages produced during this run.
Messages provided via message_history and messages from older runs are excluded.
list[_messages.ModelMessage] — List of new messages.
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.
def new_messages_json(*, output_tool_return_content: str | None = None) -> bytes
Return new messages from new_messages as JSON bytes.
bytes — JSON bytes representing the new messages.
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.
@async
def stream_output(*, debounce_by: float | None = 0.1) -> AsyncIterator[OutputDataT]
Stream the output as an async iterable.
The pydantic validator for structured data will be called in partial mode on each iteration.
AsyncIterator[OutputDataT] — An async iterable of the response data.
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.
@async
def stream_text(
*,
delta: bool = False,
debounce_by: float | None = 0.1,
) -> AsyncIterator[str]
Stream the text result as an async iterable.
delta : bool Default: False
if True, yield each chunk of text as it is received, if False (default), yield the full text
up to the current point.
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.
@async
def stream_response(
*,
debounce_by: float | None = 0.1,
) -> AsyncIterator[_messages.ModelResponse]
Stream the response as an async iterable of ModelResponse snapshots.
Each yielded ModelResponse is the current state of the response: response.state is
'incomplete' while streaming is in flight and 'complete' (or 'interrupted' if
cancel() was called) on the final yield.
AsyncIterator[_messages.ModelResponse] — An async iterable of ModelResponse snapshots.
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.
@async
def get_output() -> OutputDataT
Stream the whole response, validate and return it.
OutputDataT
@async
def validate_response_output(
message: _messages.ModelResponse,
*,
allow_partial: bool = False,
) -> OutputDataT
Validate a structured result message.
OutputDataT
@async
def cancel() -> None
Cancel the stream, stopping token generation and closing the underlying connection.
The interrupted response state is recorded in the message history so that
all_messages() includes it.
Bases: Generic[AgentDepsT, OutputDataT]
Synchronous wrapper for StreamedRunResult that only exposes sync methods.
All of the run’s async work happens on the caller’s event loop. Context-manager and iterator lifecycles remain in stable tasks, so cancel scopes entered and exited by the agent graph never straddle tasks and OpenTelemetry spans stay correctly nested. The wrapper must be used and closed on the thread where it was created.
This is a synchronous context manager; the underlying stream is cleaned up on exit:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
def main():
with agent.run_stream_sync('What is the capital of the UK?') as response:
print(response.get_output())
#> The capital of the UK is London.
Using it without a with block also works for backwards compatibility. Garbage collection requests
best-effort cleanup on the owner loop, but it cannot drive a stopped owner loop from another thread
or while another loop is running. A with block should be used whenever deterministic cleanup matters.
Return the current state of the response.
Type: _messages.ModelResponse
Return the usage of the whole run.
Type: RunUsage
Get the timestamp of the response.
Type: datetime
The unique identifier for the agent run.
Type: str
The unique identifier for the conversation this run belongs to.
Type: str
Metadata associated with this agent run, if configured.
Whether the stream has all been received.
This is set to True when one of
stream_output,
stream_text,
stream_response or
get_output completes.
Type: bool
def all_messages(
*,
output_tool_return_content: str | None = None,
) -> list[_messages.ModelMessage]
Return the history of messages.
list[_messages.ModelMessage] — List of messages.
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.
def all_messages_json(*, output_tool_return_content: str | None = None) -> bytes
Return all messages from all_messages as JSON bytes.
bytes — JSON bytes representing the messages.
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.
def new_messages(
*,
output_tool_return_content: str | None = None,
) -> list[_messages.ModelMessage]
Return the messages produced during this run.
Messages provided via message_history and messages from older runs are excluded.
list[_messages.ModelMessage] — List of new messages.
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.
def new_messages_json(*, output_tool_return_content: str | None = None) -> bytes
Return new messages from new_messages as JSON bytes.
bytes — JSON bytes representing the new messages.
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.
def stream_output(*, 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 on each iteration.
Iterator[OutputDataT] — An iterable of the response data.
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.
def stream_text(
*,
delta: bool = False,
debounce_by: float | None = 0.1,
) -> Iterator[str]
Stream the text result as an iterable.
delta : bool Default: False
if True, yield each chunk of text as it is received, if False (default), yield the full text
up to the current point.
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.
def stream_response(
*,
debounce_by: float | None = 0.1,
) -> Iterator[_messages.ModelResponse]
Stream the response as an iterable of ModelResponse snapshots.
Each yielded ModelResponse is the current state of the response: response.state is
'incomplete' while streaming is in flight and 'complete' on the final yield.
Iterator[_messages.ModelResponse] — An iterable of ModelResponse snapshots.
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.
def get_output() -> OutputDataT
Stream the whole response, validate and return it.
OutputDataT
def validate_response_output(
message: _messages.ModelResponse,
*,
allow_partial: bool = False,
) -> OutputDataT
Validate a structured result message.
OutputDataT