pydantic_ai.run
Bases: Generic[AgentDepsT, OutputDataT]
A stateful, async-iterable run of an Agent.
You generally obtain an AgentRun instance by calling async with my_agent.iter(...) as agent_run:.
Once you have an instance, you can use it to iterate through the run’s nodes as they execute. When an
End is reached, the run finishes and result
becomes available.
Example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
async def main():
nodes = []
# Iterate through the run, recording each node along the way:
async with agent.iter('What is the capital of France?') as agent_run:
async for node in agent_run:
nodes.append(node)
print(nodes)
'''
[
UserPromptNode(
user_prompt='What is the capital of France?',
instructions_functions=[],
system_prompts=(),
system_prompt_functions=[],
system_prompt_dynamic_functions={},
),
ModelRequestNode(
request=ModelRequest(
parts=[
UserPromptPart(
content='What is the capital of France?',
timestamp=datetime.datetime(...),
)
],
timestamp=datetime.datetime(...),
run_id='...',
conversation_id='...',
)
),
CallToolsNode(
model_response=ModelResponse(
parts=[TextPart(content='The capital of France is Paris.')],
usage=RequestUsage(input_tokens=56, output_tokens=7),
model_name='gpt-5.2',
timestamp=datetime.datetime(...),
run_id='...',
conversation_id='...',
)
),
End(data=FinalResult(output='The capital of France is Paris.')),
]
'''
print(agent_run.result.output)
#> The capital of France is Paris.
You can also manually drive the iteration using the next method for
more granular control.
The current context of the agent run.
Type: GraphRunContext[_agent_graph.GraphAgentState, _agent_graph.GraphAgentDeps[AgentDepsT, Any]]
The next node that will be run in the agent graph.
This is the next node that will be used during async iteration, or if a node is not passed to self.next(...).
Type: _agent_graph.AgentNode[AgentDepsT, OutputDataT] | End[FinalResult[OutputDataT]]
The final result of the run if it has ended, otherwise None.
Once the run returns an End node, result is populated
with an AgentRunResult.
Type: AgentRunResult[OutputDataT] | None
Get usage statistics for the run so far, including token usage, model requests, and so on.
Type: _usage.RunUsage
Metadata associated with this agent run, if configured.
The unique identifier for the agent run.
Type: str
The unique identifier for the conversation this run belongs to.
Type: str
Internal: live view of the queue mutated by enqueue and drained by the internal PendingMessageDrainCapability.
Exposed for inspection / debugging; use enqueue to add messages.
Type: list[PendingMessage]
def all_messages() -> list[_messages.ModelMessage]
Return all messages for the run so far.
Messages from older runs are included.
list[_messages.ModelMessage]
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.
def new_messages() -> list[_messages.ModelMessage]
Return the messages produced during this run so far.
Messages provided via message_history and messages from older runs are excluded.
list[_messages.ModelMessage]
def new_messages_json() -> bytes
Return new messages from new_messages as JSON bytes.
bytes — JSON bytes representing the new messages.
def __aiter__(
) -> AsyncIterator[_agent_graph.AgentNode[AgentDepsT, OutputDataT] | End[FinalResult[OutputDataT]]]
Provide async-iteration over the nodes in the agent run.
AsyncIterator[_agent_graph.AgentNode[AgentDepsT, OutputDataT] | End[FinalResult[OutputDataT]]]
@async
def __anext__(
) -> _agent_graph.AgentNode[AgentDepsT, OutputDataT] | End[FinalResult[OutputDataT]]
Advance to the next node automatically based on the last returned node.
Note: this uses the graph run’s internal iteration which does NOT call
node hooks (before_node_run, wrap_node_run, after_node_run,
on_node_run_error). Use next() for capability-hooked iteration, or
use agent.run() which drives via next() automatically.
_agent_graph.AgentNode[AgentDepsT, OutputDataT] | End[FinalResult[OutputDataT]]
@async
def next(
node: _agent_graph.AgentNode[AgentDepsT, OutputDataT],
) -> _agent_graph.AgentNode[AgentDepsT, OutputDataT] | End[FinalResult[OutputDataT]]
Manually drive the agent run by passing in the node you want to run next.
This lets you inspect or mutate the node before continuing execution, or skip certain nodes
under dynamic conditions. The agent run should be stopped when you return an End
node.
Example:
from pydantic_ai import Agent
from pydantic_graph import End
agent = Agent('openai:gpt-5.2')
async def main():
async with agent.iter('What is the capital of France?') as agent_run:
next_node = agent_run.next_node # start with the first node
nodes = [next_node]
while not isinstance(next_node, End):
next_node = await agent_run.next(next_node)
nodes.append(next_node)
# Once `next_node` is an End, we've finished:
print(nodes)
'''
[
UserPromptNode(
user_prompt='What is the capital of France?',
instructions_functions=[],
system_prompts=(),
system_prompt_functions=[],
system_prompt_dynamic_functions={},
),
ModelRequestNode(
request=ModelRequest(
parts=[
UserPromptPart(
content='What is the capital of France?',
timestamp=datetime.datetime(...),
)
],
timestamp=datetime.datetime(...),
run_id='...',
conversation_id='...',
)
),
CallToolsNode(
model_response=ModelResponse(
parts=[TextPart(content='The capital of France is Paris.')],
usage=RequestUsage(input_tokens=56, output_tokens=7),
model_name='gpt-5.2',
timestamp=datetime.datetime(...),
run_id='...',
conversation_id='...',
)
),
End(data=FinalResult(output='The capital of France is Paris.')),
]
'''
print('Final result:', agent_run.result.output)
#> Final result: The capital of France is Paris.
_agent_graph.AgentNode[AgentDepsT, OutputDataT] | End[FinalResult[OutputDataT]] — The next node returned by the graph logic, or an End node if
_agent_graph.AgentNode[AgentDepsT, OutputDataT] | End[FinalResult[OutputDataT]] — the run has completed.
The node to run next in the graph.
def enqueue(
*content: EnqueueContent,
priority: PendingMessagePriority = 'asap',
) -> str | None
Enqueue content to be injected into the conversation.
Designed to be called from the same event loop driving agent.iter(). If
you’re forwarding events from a different thread (e.g. a webhook handler
running on its own loop or thread), marshal the call back onto the agent’s
loop first (e.g. loop.call_soon_threadsafe(agent_run.enqueue, msg)).
The drain’s queue[:] = remaining pattern in _drain_by_priority isn’t
atomic against concurrent appends from a different thread.
str | None — The enqueue_id of the queued message, echoed on the
str | None — EnqueuedMessagesEvent emitted when it’s
str | None — delivered, or None when there was nothing to enqueue (an empty call).
One or more EnqueueContent items.
Adjacent UserContent (a str or multi-modal
content like an ImageUrl) is gathered into one
UserPromptPart, and each
ModelRequestPart (e.g. a
SystemPromptPart) is coalesced with adjacent
part-style items into one ModelRequest; a complete
ModelRequest or
ModelResponse is kept as its own message. The
assembled sequence must end in a request. Calling with no positional args is a no-op.
When to deliver:
'asap' (default) — at the earliest opportunity (next model request,
or a redirect if the agent would otherwise end).
'when_idle' — only when the agent would otherwise end, after 'asap' messages.
Bases: Generic[OutputDataT]
The final result of an agent run.
The output data from the agent run.
Type: OutputDataT
Return the last response from the message history.
Type: _messages.ModelResponse
Return the usage of the whole run.
Type: _usage.RunUsage
Return the timestamp of last response.
Type: datetime
Metadata associated with this agent run, if configured.
The unique identifier for the agent run.
Type: str
The unique identifier for the conversation this run belongs to.
Type: str
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.
Bases: Generic[OutputDataT]
An event indicating the agent run ended and containing the final result of the agent run.
The result of the run.
Type: AgentRunResult[OutputDataT]
Event type identifier, used as a discriminator.
Type: Literal[‘agent_run_result’] Default: 'agent_run_result'
A single item accepted by RunContext.enqueue
and AgentRun.enqueue.
enqueue is variadic, so each item is one positional argument:
UserContent(astror a piece of multi-modal content like anImageUrl): adjacent user content is gathered into a singleUserPromptPart, soenqueue('caption', image)forms one user turn. To pass an existing list, spread it:enqueue(*items).ModelRequestPart(e.g. aSystemPromptPart): included verbatim.ModelMessage(a completeModelRequestorModelResponse): emitted as its own message.
Consecutive part-style items (user content and ModelRequestParts) are coalesced into a single
ModelRequest; complete ModelMessages stay separate. This lets one enqueue call inject an
interleaved exchange (e.g. a synthetic tool call + result — a ModelResponse followed by a
ModelRequest). The assembled sequence must end in a ModelRequest so the agent has something to
respond to.
Type: TypeAlias Default: 'UserContent | ModelRequestPart | ModelMessage'
One or more ModelMessages queued for injection into the agent conversation.
Enqueued via RunContext.enqueue or
AgentRun.enqueue and automatically drained
at the appropriate time during the agent run by the internal PendingMessageDrainCapability.
The message(s) to inject, in order. Always ends in a
ModelRequest.
Type: list[ModelMessage]
When to deliver these messages:
'asap': at the earliest opportunity (next model request, or redirect if the agent would otherwise terminate).'when_idle': only when the agent would otherwise terminate, after'asap'messages.
Type: PendingMessagePriority Default: 'asap'
Unique identifier for this enqueue call, surfaced on the
EnqueuedMessagesEvent emitted when the messages
are delivered, and returned by enqueue.
Type: str Default: field(default_factory=(lambda: str(uuid7())))
@classmethod
def from_content(
cls,
*content: EnqueueContent,
priority: PendingMessagePriority = 'asap',
) -> PendingMessage | None
Build a PendingMessage from enqueue arguments, or None when there’s nothing to send.
Returns None for an empty call (enqueueing nothing is a no-op rather than an error).
PendingMessage | None
UserError— If the assembled messages don’t end in aModelRequest— e.g. a loneModelResponse— since the agent needs a request to respond to.
When to deliver a pending message.
'asap': Delivered at the earliest opportunity — either prepended to the nextModelRequest, or, if the agent would otherwise terminate before another request, used to redirect the run into one more request.'when_idle': Delivered only when the agent would otherwise terminate, after any'asap'messages. Doesn’t interrupt in-flight work.
Type: TypeAlias Default: Literal['asap', 'when_idle']