pydantic_ai.agent
Bases: AbstractAgent[AgentDepsT, OutputDataT]
Class for defining “agents” - a way to have a specific type of “conversation” with an LLM.
Agents are generic in the dependency type they take AgentDepsT
and the output type they return, OutputDataT.
By default, if neither generic parameter is customised, agents have type Agent[None, str].
Minimal usage example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
result = agent.run_sync('What is the capital of France?')
print(result.output)
#> The capital of France is Paris.
The strategy for handling multiple tool calls when a final result is found.
'early'(default): Output tools are executed first. Once a valid final result is found, remaining function and output tool calls are skipped'exhaustive': Output tools are executed first, then all function tools are executed. The first valid output tool result becomes the final output
Type: EndStrategy Default: end_strategy
Optional model request settings to use for this agent’s runs, by default.
Can be a static ModelSettings dict or a callable that takes a
RunContext and returns ModelSettings.
Callables are called before each model request, allowing dynamic per-step settings.
Note, if model_settings is also provided at run time, those settings will be merged
on top of the agent-level settings, with the run-level argument taking priority.
Type: AgentModelSettings[AgentDepsT] | None Default: model_settings
Options to automatically instrument with OpenTelemetry.
Type: InstrumentationSettings | bool | None Default: instrument
The default model configured for this agent.
Type: models.Model | models.KnownModelName | str | None
The name of the agent, used for logging.
If None, we try to infer the agent name from the call frame when the agent is first run.
A human-readable description of the agent.
If the description is a TemplateStr, returns the raw template source. The rendered description is available at runtime via OTel span attributes.
The type of dependencies used by the agent.
Type: type
The type of data output by agent runs, used to validate the data returned by the model, defaults to str.
Type: OutputSpec[OutputDataT]
Optional handler for events from the model’s streaming response and the agent’s execution of tools.
Type: EventStreamHandler[AgentDepsT] | None
All toolsets registered on the agent, including a function toolset holding tools that were registered on the agent directly.
Output tools are not included.
Type: Sequence[AbstractToolset[AgentDepsT]]
def __init__(
model: models.Model | models.KnownModelName | str | None = None,
output_type: OutputSpec[OutputDataT] = str,
instructions: AgentInstructions[AgentDepsT] = None,
system_prompt: str | Sequence[str] = (),
deps_type: type[AgentDepsT] = NoneType,
name: str | None = None,
description: TemplateStr[AgentDepsT] | str | None = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
retries: int = 1,
validation_context: Any | Callable[[RunContext[AgentDepsT]], Any] = None,
output_retries: int | None = None,
tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] = (),
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] = (),
prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
prepare_output_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
toolsets: Sequence[AgentToolset[AgentDepsT]] | None = None,
defer_model_check: bool = False,
end_strategy: EndStrategy = 'early',
instrument: InstrumentationSettings | bool | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
history_processors: Sequence[HistoryProcessor[AgentDepsT]] | None = None,
event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
tool_timeout: float | None = None,
max_concurrency: _concurrency.AnyConcurrencyLimit = None,
capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
) -> None
def __init__(
model: models.Model | models.KnownModelName | str | None = None,
output_type: OutputSpec[OutputDataT] = str,
instructions: AgentInstructions[AgentDepsT] = None,
system_prompt: str | Sequence[str] = (),
deps_type: type[AgentDepsT] = NoneType,
name: str | None = None,
description: TemplateStr[AgentDepsT] | str | None = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
retries: int = 1,
validation_context: Any | Callable[[RunContext[AgentDepsT]], Any] = None,
output_retries: int | None = None,
tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] = (),
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] = (),
prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
prepare_output_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
mcp_servers: Sequence[MCPServer] = (),
defer_model_check: bool = False,
end_strategy: EndStrategy = 'early',
instrument: InstrumentationSettings | bool | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
history_processors: Sequence[HistoryProcessor[AgentDepsT]] | None = None,
event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
tool_timeout: float | None = None,
max_concurrency: _concurrency.AnyConcurrencyLimit = None,
capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
) -> None
Create an agent.
model : models.Model | models.KnownModelName | str | None Default: None
The default model to use for this agent, if not provided,
you must provide the model when calling it. We allow str here since the actual list of allowed models changes frequently.
The type of the output data, used to validate the data returned by the model,
defaults to str.
Instructions to use for this agent, you can also register instructions via a function with
instructions or pass additional, temporary, instructions when executing a run.
Static system prompts to use for this agent, you can also register system
prompts via a function with system_prompt.
deps_type : type[AgentDepsT] Default: NoneType
The type used for dependency injection, this parameter exists solely to allow you to fully
parameterize the agent, and therefore get the best out of static type checking.
If you’re not using deps, but want type checking to pass, you can set deps=None to satisfy Pyright
or add a type hint : Agent[None, <return type>].
The name of the agent, used for logging. If None, we try to infer the agent name from the call frame
when the agent is first run.
A human-readable description of the agent, attached to the agent run span as
gen_ai.agent.description when instrumentation is enabled.
model_settings : AgentModelSettings[AgentDepsT] | None Default: None
Optional model request settings to use for this agent’s runs, by default.
Can be a static ModelSettings dict or a callable that takes a
RunContext and returns ModelSettings.
Callables are called before each model request, allowing dynamic per-step settings.
retries : int Default: 1
The default number of retries to allow for tool calls and output validation, before raising an error. For model request retries, see the HTTP Request Retries documentation.
validation_context : Any | Callable[[RunContext[AgentDepsT]], Any] Default: None
Pydantic validation context used to validate tool arguments and outputs.
The maximum number of retries to allow for output validation, defaults to retries.
Tools to register with the agent, you can also register tools via the decorators
@agent.tool and @agent.tool_plain.
builtin_tools : Sequence[AgentBuiltinTool[AgentDepsT]] Default: ()
The builtin tools that the agent will use. This depends on the model, as some models may not support certain tools. If the model doesn’t support the builtin tools, an error will be raised.
prepare_tools : ToolsPrepareFunc[AgentDepsT] | None Default: None
Custom function to prepare the tool definition of all tools for each step, except output tools.
This is useful if you want to customize the definition of multiple tools or you want to register
a subset of tools for a given step. See ToolsPrepareFunc
prepare_output_tools : ToolsPrepareFunc[AgentDepsT] | None Default: None
Custom function to prepare the tool definition of all output tools for each step.
This is useful if you want to customize the definition of multiple output tools or you want to register
a subset of output tools for a given step. See ToolsPrepareFunc
Toolsets to register with the agent, including MCP servers and functions which take a run context
and return a toolset. See ToolsetFunc for more information.
defer_model_check : bool Default: False
by default, if you provide a named model,
it’s evaluated to create a Model instance immediately,
which checks for the necessary environment variables. Set this to false
to defer the evaluation until the first run. Useful if you want to
override the model for testing.
end_strategy : EndStrategy Default: 'early'
Strategy for handling tool calls that are requested alongside a final result.
See EndStrategy for more information.
instrument : InstrumentationSettings | bool | None Default: None
Set to True to automatically instrument with OpenTelemetry,
which will use Logfire if it’s configured.
Set to an instance of InstrumentationSettings to customize.
If this isn’t set, then the last value set by
Agent.instrument_all()
will be used, which defaults to False.
See the Debugging and Monitoring guide for more info.
metadata : AgentMetadata[AgentDepsT] | None Default: None
Optional metadata to store with each run.
Provide a dictionary of primitives, or a callable returning one
computed from the RunContext on each run.
Metadata is resolved when a run starts and recomputed after a successful run finishes so it
can reflect the final state.
Resolved metadata can be read after the run completes via
AgentRun.metadata,
AgentRunResult.metadata, and
StreamedRunResult.metadata,
and is attached to the agent run span when instrumentation is enabled.
Optional list of callables to process the message history before sending it to the model. Each processor takes a list of messages and returns a modified list of messages. Processors can be sync or async and are applied in sequence.
event_stream_handler : EventStreamHandler[AgentDepsT] | None Default: None
Optional handler for events from the model’s streaming response and the agent’s execution of tools.
Default timeout in seconds for tool execution. If a tool takes longer than this, the tool is considered to have failed and a retry prompt is returned to the model (counting towards the retry limit). Individual tools can override this with their own timeout. Defaults to None (no timeout).
Optional limit on concurrent agent runs. Can be an integer for simple limiting,
a ConcurrencyLimit for advanced configuration with backpressure,
a ConcurrencyLimiter for sharing limits across
multiple agents, or None (default) for no limiting. When the limit is reached, additional calls
to run() or iter() will wait until a slot becomes available.
Optional list of capabilities to configure the agent with.
Custom capabilities can be created by subclassing
AbstractCapability.
@classmethod
def from_spec(
cls,
spec: dict[str, Any] | AgentSpec,
custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
model: models.Model | models.KnownModelName | str | None = None,
output_type: OutputSpec[Any] = str,
instructions: AgentInstructions[Any] = None,
system_prompt: str | Sequence[str] = (),
name: str | None = None,
description: TemplateStr[Any] | str | None = None,
model_settings: ModelSettings | None = None,
retries: int | None = None,
validation_context: Any = None,
output_retries: int | None = None,
tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
builtin_tools: Sequence[AgentBuiltinTool[Any]] = (),
prepare_tools: ToolsPrepareFunc[Any] | None = None,
prepare_output_tools: ToolsPrepareFunc[Any] | None = None,
toolsets: Sequence[AgentToolset[Any]] | None = None,
defer_model_check: bool = False,
end_strategy: EndStrategy | None = None,
instrument: InstrumentationSettings | bool | None = None,
metadata: AgentMetadata[Any] | None = None,
history_processors: Sequence[HistoryProcessor[Any]] | None = None,
event_stream_handler: EventStreamHandler[Any] | None = None,
tool_timeout: float | None = None,
max_concurrency: _concurrency.AnyConcurrencyLimit = None,
capabilities: Sequence[AbstractCapability[Any]] | None = None,
) -> Agent[None, str]
def from_spec(
cls,
spec: dict[str, Any] | AgentSpec,
deps_type: type[T],
custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
model: models.Model | models.KnownModelName | str | None = None,
output_type: OutputSpec[Any] = str,
instructions: AgentInstructions[Any] = None,
system_prompt: str | Sequence[str] = (),
name: str | None = None,
description: TemplateStr[Any] | str | None = None,
model_settings: ModelSettings | None = None,
retries: int | None = None,
validation_context: Any = None,
output_retries: int | None = None,
tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
builtin_tools: Sequence[AgentBuiltinTool[Any]] = (),
prepare_tools: ToolsPrepareFunc[Any] | None = None,
prepare_output_tools: ToolsPrepareFunc[Any] | None = None,
toolsets: Sequence[AgentToolset[Any]] | None = None,
defer_model_check: bool = False,
end_strategy: EndStrategy | None = None,
instrument: InstrumentationSettings | bool | None = None,
metadata: AgentMetadata[Any] | None = None,
history_processors: Sequence[HistoryProcessor[Any]] | None = None,
event_stream_handler: EventStreamHandler[Any] | None = None,
tool_timeout: float | None = None,
max_concurrency: _concurrency.AnyConcurrencyLimit = None,
capabilities: Sequence[AbstractCapability[Any]] | None = None,
) -> Agent[T, str]
Construct an Agent from a spec dict or AgentSpec.
This allows defining agents declaratively in YAML/JSON/dict form.
Keyword arguments supplement the spec: scalar spec fields (like name,
retries) are used as defaults that explicit arguments override, while
capabilities from both sources are merged.
Agent[Any, Any] — A new Agent instance.
The agent specification, either a dict or an AgentSpec instance.
The type of the dependencies for the agent. When provided,
template strings in capabilities (e.g. "Hello \{\{name\}\}") are
compiled and validated against this type.
Additional capability classes to make available beyond the built-in defaults.
model : models.Model | models.KnownModelName | str | None Default: None
Override the model from the spec.
output_type : OutputSpec[Any] Default: str
The type of the output data, defaults to str.
instructions : AgentInstructions[Any] Default: None
Instructions for the agent.
Static system prompts.
The agent name, overrides spec name if provided.
The agent description, overrides spec description if provided.
model_settings : ModelSettings | None Default: None
Model request settings.
Default retries for tool calls and output validation, overrides spec retries if provided.
validation_context : Any Default: None
Pydantic validation context for tool arguments and outputs.
Max retries for output validation, overrides spec output_retries if provided.
Tools to register with the agent.
builtin_tools : Sequence[AgentBuiltinTool[Any]] Default: ()
Builtin tools for the agent.
Custom function to prepare tool definitions.
Custom function to prepare output tool definitions.
Toolsets to register with the agent.
defer_model_check : bool Default: False
Defer model evaluation until first run.
end_strategy : EndStrategy | None Default: None
Strategy for tool calls alongside a final result, overrides spec end_strategy if provided.
instrument : InstrumentationSettings | bool | None Default: None
Instrumentation settings, overrides spec instrument if provided.
Metadata to store with each run, overrides spec metadata if provided.
Processors for message history.
Handler for streaming events.
Default timeout for tool execution, overrides spec tool_timeout if provided.
Limit on concurrent agent runs.
Additional capabilities merged with those from the spec.
@classmethod
def from_file(
cls,
path: Path | str,
fmt: Literal['yaml', 'json'] | None = None,
custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
model: models.Model | models.KnownModelName | str | None = None,
output_type: OutputSpec[Any] = str,
instructions: AgentInstructions[Any] = None,
system_prompt: str | Sequence[str] = (),
name: str | None = None,
description: TemplateStr[Any] | str | None = None,
model_settings: ModelSettings | None = None,
retries: int | None = None,
validation_context: Any = None,
output_retries: int | None = None,
tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
builtin_tools: Sequence[AgentBuiltinTool[Any]] = (),
prepare_tools: ToolsPrepareFunc[Any] | None = None,
prepare_output_tools: ToolsPrepareFunc[Any] | None = None,
toolsets: Sequence[AgentToolset[Any]] | None = None,
defer_model_check: bool = False,
end_strategy: EndStrategy | None = None,
instrument: InstrumentationSettings | bool | None = None,
metadata: AgentMetadata[Any] | None = None,
history_processors: Sequence[HistoryProcessor[Any]] | None = None,
event_stream_handler: EventStreamHandler[Any] | None = None,
tool_timeout: float | None = None,
max_concurrency: _concurrency.AnyConcurrencyLimit = None,
capabilities: Sequence[AbstractCapability[Any]] | None = None,
) -> Agent[None, str]
def from_file(
cls,
path: Path | str,
fmt: Literal['yaml', 'json'] | None = None,
deps_type: type[T],
custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
model: models.Model | models.KnownModelName | str | None = None,
output_type: OutputSpec[Any] = str,
instructions: AgentInstructions[Any] = None,
system_prompt: str | Sequence[str] = (),
name: str | None = None,
description: TemplateStr[Any] | str | None = None,
model_settings: ModelSettings | None = None,
retries: int | None = None,
validation_context: Any = None,
output_retries: int | None = None,
tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
builtin_tools: Sequence[AgentBuiltinTool[Any]] = (),
prepare_tools: ToolsPrepareFunc[Any] | None = None,
prepare_output_tools: ToolsPrepareFunc[Any] | None = None,
toolsets: Sequence[AgentToolset[Any]] | None = None,
defer_model_check: bool = False,
end_strategy: EndStrategy | None = None,
instrument: InstrumentationSettings | bool | None = None,
metadata: AgentMetadata[Any] | None = None,
history_processors: Sequence[HistoryProcessor[Any]] | None = None,
event_stream_handler: EventStreamHandler[Any] | None = None,
tool_timeout: float | None = None,
max_concurrency: _concurrency.AnyConcurrencyLimit = None,
capabilities: Sequence[AbstractCapability[Any]] | None = None,
) -> Agent[T, str]
Construct an Agent from a YAML or JSON spec file.
This is a convenience method equivalent to
Agent.from_spec(AgentSpec.from_file(path), ...).
The file format is inferred from the extension (.yaml/.yml or .json)
unless overridden with the fmt argument.
All other arguments are forwarded to from_spec.
@staticmethod
def instrument_all(instrument: InstrumentationSettings | bool = True) -> None
Set the instrumentation options for all agents where instrument is not set.
@async
def iter(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: None = None,
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AbstractAsyncContextManager[AgentRun[AgentDepsT, OutputDataT]]
def iter(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: OutputSpec[RunOutputDataT],
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AbstractAsyncContextManager[AgentRun[AgentDepsT, RunOutputDataT]]
A contextmanager which can be used to iterate over the agent graph’s nodes as they are executed.
This method builds an internal agent graph (using system prompts, tools and output schemas) and then returns an
AgentRun object. The AgentRun can be used to async-iterate over the nodes of the graph as they are
executed. This is the API to use if you want to consume the outputs coming from each LLM model response, or the
stream of events coming from the execution of tools.
The AgentRun also provides methods to access the full message history, new messages, and usage statistics,
and the final result of the run once it has completed.
For more details, see the documentation of AgentRun.
Example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
async def main():
nodes = []
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='...',
)
),
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='...',
)
),
End(data=FinalResult(output='The capital of France is Paris.')),
]
'''
print(agent_run.result.output)
#> The capital of France is Paris.
AsyncIterator[AgentRun[AgentDepsT, Any]] — The result of the run.
User input to start/continue the conversation.
Custom output type to use for this run, output_type may only be used if the agent has no
output validators since output validators would expect an argument that matches the agent’s output type.
History of the conversation so far.
deferred_tool_results : DeferredToolResults | None Default: None
Optional results for deferred tool calls in the message history.
model : models.Model | models.KnownModelName | str | None Default: None
Optional model to use for this run, required if model was not set when creating the agent.
Optional additional instructions to use for this run.
Optional dependencies to use for this run.
model_settings : AgentModelSettings[AgentDepsT] | None Default: None
Optional settings to use for this model’s request, or a callable
that receives RunContext and returns settings.
Callables are called before each model request, allowing dynamic per-step settings.
usage_limits : _usage.UsageLimits | None Default: None
Optional limits on model request count or token usage.
usage : _usage.RunUsage | None Default: None
Optional usage to start with, useful for resuming a conversation or agents used in tools.
metadata : AgentMetadata[AgentDepsT] | None Default: None
Optional metadata to attach to this run. Accepts a dictionary or a callable taking
RunContext; merged with the agent’s configured metadata.
infer_name : bool Default: True
Whether to try to infer the agent name from the call frame if it’s not set.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | None Default: None
Optional additional toolsets for this run.
builtin_tools : Sequence[AgentBuiltinTool[AgentDepsT]] | None Default: None
Optional additional builtin tools for this run.
Optional agent spec to apply for this run. At run time, spec values are additive.
def override(
name: str | _utils.Unset = _utils.UNSET,
deps: AgentDepsT | _utils.Unset = _utils.UNSET,
model: models.Model | models.KnownModelName | str | _utils.Unset = _utils.UNSET,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | _utils.Unset = _utils.UNSET,
tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] | _utils.Unset = _utils.UNSET,
instructions: AgentInstructions[AgentDepsT] | _utils.Unset = _utils.UNSET,
metadata: AgentMetadata[AgentDepsT] | _utils.Unset = _utils.UNSET,
model_settings: AgentModelSettings[AgentDepsT] | _utils.Unset = _utils.UNSET,
spec: dict[str, Any] | AgentSpec | None = None,
) -> Iterator[None]
Context manager to temporarily override agent name, dependencies, model, toolsets, tools, or instructions.
This is particularly useful when testing. You can find an example of this here.
name : str | _utils.Unset Default: _utils.UNSET
The name to use instead of the name passed to the agent constructor and agent run.
The dependencies to use instead of the dependencies passed to the agent run.
model : models.Model | models.KnownModelName | str | _utils.Unset Default: _utils.UNSET
The model to use instead of the model passed to the agent run.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | _utils.Unset Default: _utils.UNSET
The toolsets to use instead of the toolsets passed to the agent constructor and agent run.
tools : Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, …]] | _utils.Unset Default: _utils.UNSET
The tools to use instead of the tools registered with the agent.
The instructions to use instead of the instructions registered with the agent.
Note: this also replaces capability-contributed instructions (e.g. from
get_instructions).
The metadata to use instead of the metadata passed to the agent constructor. When set, any
per-run metadata argument is ignored.
The model settings to use instead of the model settings passed to the agent constructor.
When set, any per-run model_settings argument is ignored.
Optional agent spec providing defaults for override. Explicit params take precedence
over spec values. When the spec includes capabilities, they replace (not merge with)
the agent’s existing capabilities. To add capabilities without replacing, pass spec
to run() or iter() instead.
def instructions(
func: Callable[[RunContext[AgentDepsT]], str | None],
) -> Callable[[RunContext[AgentDepsT]], str | None]
def instructions(
func: Callable[[RunContext[AgentDepsT]], Awaitable[str | None]],
) -> Callable[[RunContext[AgentDepsT]], Awaitable[str | None]]
def instructions(func: Callable[[], str | None]) -> Callable[[], str | None]
def instructions(
func: Callable[[], Awaitable[str | None]],
) -> Callable[[], Awaitable[str | None]]
def instructions(
) -> Callable[[_system_prompt.SystemPromptFunc[AgentDepsT]], _system_prompt.SystemPromptFunc[AgentDepsT]]
Decorator to register an instructions function.
Optionally takes RunContext as its only argument.
Can decorate a sync or async functions.
The decorator can be used bare (agent.instructions).
Overloads for every possible signature of instructions are included so the decorator doesn’t obscure
the type of the function.
Example:
from pydantic_ai import Agent, RunContext
agent = Agent('test', deps_type=str)
@agent.instructions
def simple_instructions() -> str:
return 'foobar'
@agent.instructions
async def async_instructions(ctx: RunContext[str]) -> str:
return f'{ctx.deps} is the best'
Callable[[_system_prompt.SystemPromptFunc[AgentDepsT]], _system_prompt.SystemPromptFunc[AgentDepsT]] | _system_prompt.SystemPromptFunc[AgentDepsT]
def system_prompt(
func: Callable[[RunContext[AgentDepsT]], str | None],
) -> Callable[[RunContext[AgentDepsT]], str | None]
def system_prompt(
func: Callable[[RunContext[AgentDepsT]], Awaitable[str | None]],
) -> Callable[[RunContext[AgentDepsT]], Awaitable[str | None]]
def system_prompt(func: Callable[[], str | None]) -> Callable[[], str | None]
def system_prompt(
func: Callable[[], Awaitable[str | None]],
) -> Callable[[], Awaitable[str | None]]
def system_prompt(
dynamic: bool = False,
) -> Callable[[_system_prompt.SystemPromptFunc[AgentDepsT]], _system_prompt.SystemPromptFunc[AgentDepsT]]
Decorator to register a system prompt function.
Optionally takes RunContext as its only argument.
Can decorate a sync or async functions.
The decorator can be used either bare (agent.system_prompt) or as a function call
(agent.system_prompt(...)), see the examples below.
Overloads for every possible signature of system_prompt are included so the decorator doesn’t obscure
the type of the function, see tests/typed_agent.py for tests.
Example:
from pydantic_ai import Agent, RunContext
agent = Agent('test', deps_type=str)
@agent.system_prompt
def simple_system_prompt() -> str:
return 'foobar'
@agent.system_prompt(dynamic=True)
async def async_system_prompt(ctx: RunContext[str]) -> str:
return f'{ctx.deps} is the best'
Callable[[_system_prompt.SystemPromptFunc[AgentDepsT]], _system_prompt.SystemPromptFunc[AgentDepsT]] | _system_prompt.SystemPromptFunc[AgentDepsT]
func : _system_prompt.SystemPromptFunc[AgentDepsT] | None Default: None
The function to decorate
dynamic : bool Default: False
If True, the system prompt will be reevaluated even when messages_history is provided,
see SystemPromptPart.dynamic_ref
def output_validator(
func: Callable[[RunContext[AgentDepsT], OutputDataT], OutputDataT],
) -> Callable[[RunContext[AgentDepsT], OutputDataT], OutputDataT]
def output_validator(
func: Callable[[RunContext[AgentDepsT], OutputDataT], Awaitable[OutputDataT]],
) -> Callable[[RunContext[AgentDepsT], OutputDataT], Awaitable[OutputDataT]]
def output_validator(
func: Callable[[OutputDataT], OutputDataT],
) -> Callable[[OutputDataT], OutputDataT]
def output_validator(
func: Callable[[OutputDataT], Awaitable[OutputDataT]],
) -> Callable[[OutputDataT], Awaitable[OutputDataT]]
Decorator to register an output validator function.
Optionally takes RunContext as its first argument.
Can decorate a sync or async functions.
Overloads for every possible signature of output_validator are included so the decorator doesn’t obscure
the type of the function, see tests/typed_agent.py for tests.
Example:
from pydantic_ai import Agent, ModelRetry, RunContext
agent = Agent('test', deps_type=str)
@agent.output_validator
def output_validator_simple(data: str) -> str:
if 'wrong' in data:
raise ModelRetry('wrong response')
return data
@agent.output_validator
async def output_validator_deps(ctx: RunContext[str], data: str) -> str:
if ctx.deps in data:
raise ModelRetry('wrong response')
return data
result = agent.run_sync('foobar', deps='spam')
print(result.output)
#> success (no tool calls)
_output.OutputValidatorFunc[AgentDepsT, OutputDataT]
def tool(
func: ToolFuncContext[AgentDepsT, ToolParams],
) -> ToolFuncContext[AgentDepsT, ToolParams]
def tool(
name: str | None = None,
description: str | None = None,
retries: int | None = None,
prepare: ToolPrepareFunc[AgentDepsT] | None = None,
args_validator: ArgsValidatorFunc[AgentDepsT, ToolParams] | None = None,
docstring_format: DocstringFormat = 'auto',
require_parameter_descriptions: bool = False,
schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
strict: bool | None = None,
sequential: bool = False,
requires_approval: bool = False,
metadata: dict[str, Any] | None = None,
timeout: float | None = None,
defer_loading: bool = False,
) -> Callable[[ToolFuncContext[AgentDepsT, ToolParams]], ToolFuncContext[AgentDepsT, ToolParams]]
Decorator to register a tool function which takes RunContext as its first argument.
Can decorate a sync or async functions.
The docstring is inspected to extract both the tool description and description of each parameter, learn more.
We can’t add overloads for every possible signature of tool, since the return type is a recursive union
so the signature of functions decorated with @agent.tool is obscured.
Example:
from pydantic_ai import Agent, RunContext
agent = Agent('test', deps_type=int)
@agent.tool
def foobar(ctx: RunContext[int], x: int) -> int:
return ctx.deps + x
@agent.tool(retries=2)
async def spam(ctx: RunContext[str], y: float) -> float:
return ctx.deps + y
result = agent.run_sync('foobar', deps=1)
print(result.output)
#> {"foobar":1,"spam":1.0}
func : ToolFuncContext[AgentDepsT, ToolParams] | None Default: None
The tool function to register.
The name of the tool, defaults to the function name.
The description of the tool, defaults to the function docstring.
The number of retries to allow for this tool, defaults to the agent’s default retries, which defaults to 1.
prepare : ToolPrepareFunc[AgentDepsT] | None Default: None
custom method to prepare the tool definition for each step, return None to omit this
tool from a given step. This is useful if you want to customise a tool at call time,
or omit it completely from a step. See ToolPrepareFunc.
args_validator : ArgsValidatorFunc[AgentDepsT, ToolParams] | None Default: None
custom method to validate tool arguments after schema validation has passed,
before execution. The validator receives the already-validated and type-converted parameters,
with RunContext as the first argument.
Should raise ModelRetry on validation failure,
return None on success.
See ArgsValidatorFunc.
The format of the docstring, see DocstringFormat.
Defaults to 'auto', such that the format is inferred from the structure of the docstring.
require_parameter_descriptions : bool Default: False
If True, raise an error if a parameter description is missing. Defaults to False.
schema_generator : type[GenerateJsonSchema] Default: GenerateToolJsonSchema
The JSON schema generator class to use for this tool. Defaults to GenerateToolJsonSchema.
Whether to enforce JSON schema compliance (only affects OpenAI).
See ToolDefinition for more info.
sequential : bool Default: False
Whether the function requires a sequential/serial execution environment. Defaults to False.
requires_approval : bool Default: False
Whether this tool requires human-in-the-loop approval. Defaults to False. See the tools documentation for more info.
Optional metadata for the tool. This is not sent to the model but can be used for filtering and tool behavior customization.
Timeout in seconds for tool execution. If the tool takes longer, a retry prompt is returned to the model.
Overrides the agent-level tool_timeout if set. Defaults to None (no timeout).
defer_loading : bool Default: False
Whether to hide this tool until it’s discovered via tool search. Defaults to False. See Tool Search for more info.
def tool_plain(func: ToolFuncPlain[ToolParams]) -> ToolFuncPlain[ToolParams]
def tool_plain(
name: str | None = None,
description: str | None = None,
retries: int | None = None,
prepare: ToolPrepareFunc[AgentDepsT] | None = None,
args_validator: ArgsValidatorFunc[AgentDepsT, ToolParams] | None = None,
docstring_format: DocstringFormat = 'auto',
require_parameter_descriptions: bool = False,
schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
strict: bool | None = None,
sequential: bool = False,
requires_approval: bool = False,
metadata: dict[str, Any] | None = None,
timeout: float | None = None,
defer_loading: bool = False,
) -> Callable[[ToolFuncPlain[ToolParams]], ToolFuncPlain[ToolParams]]
Decorator to register a tool function which DOES NOT take RunContext as an argument.
Can decorate a sync or async functions.
The docstring is inspected to extract both the tool description and description of each parameter, learn more.
We can’t add overloads for every possible signature of tool, since the return type is a recursive union
so the signature of functions decorated with @agent.tool is obscured.
Example:
from pydantic_ai import Agent, RunContext
agent = Agent('test')
@agent.tool
def foobar(ctx: RunContext[int]) -> int:
return 123
@agent.tool(retries=2)
async def spam(ctx: RunContext[str]) -> float:
return 3.14
result = agent.run_sync('foobar', deps=1)
print(result.output)
#> {"foobar":123,"spam":3.14}
func : ToolFuncPlain[ToolParams] | None Default: None
The tool function to register.
The name of the tool, defaults to the function name.
The description of the tool, defaults to the function docstring.
The number of retries to allow for this tool, defaults to the agent’s default retries, which defaults to 1.
prepare : ToolPrepareFunc[AgentDepsT] | None Default: None
custom method to prepare the tool definition for each step, return None to omit this
tool from a given step. This is useful if you want to customise a tool at call time,
or omit it completely from a step. See ToolPrepareFunc.
args_validator : ArgsValidatorFunc[AgentDepsT, ToolParams] | None Default: None
custom method to validate tool arguments after schema validation has passed,
before execution. The validator receives the already-validated and type-converted parameters,
with RunContext as the first argument — even though the
tool function itself does not take RunContext when using tool_plain.
Should raise ModelRetry on validation failure,
return None on success.
See ArgsValidatorFunc.
The format of the docstring, see DocstringFormat.
Defaults to 'auto', such that the format is inferred from the structure of the docstring.
require_parameter_descriptions : bool Default: False
If True, raise an error if a parameter description is missing. Defaults to False.
schema_generator : type[GenerateJsonSchema] Default: GenerateToolJsonSchema
The JSON schema generator class to use for this tool. Defaults to GenerateToolJsonSchema.
Whether to enforce JSON schema compliance (only affects OpenAI).
See ToolDefinition for more info.
sequential : bool Default: False
Whether the function requires a sequential/serial execution environment. Defaults to False.
requires_approval : bool Default: False
Whether this tool requires human-in-the-loop approval. Defaults to False. See the tools documentation for more info.
Optional metadata for the tool. This is not sent to the model but can be used for filtering and tool behavior customization.
Timeout in seconds for tool execution. If the tool takes longer, a retry prompt is returned to the model.
Overrides the agent-level tool_timeout if set. Defaults to None (no timeout).
defer_loading : bool Default: False
Whether to hide this tool until it’s discovered via tool search. Defaults to False. See Tool Search for more info.
def toolset(func: ToolsetFunc[AgentDepsT]) -> ToolsetFunc[AgentDepsT]
def toolset(
per_run_step: bool = True,
id: str | None = None,
) -> Callable[[ToolsetFunc[AgentDepsT]], ToolsetFunc[AgentDepsT]]
Decorator to register a toolset function which takes RunContext as its only argument.
Can decorate a sync or async functions.
The decorator can be used bare (agent.toolset).
Example:
from pydantic_ai import AbstractToolset, Agent, FunctionToolset, RunContext
agent = Agent('test', deps_type=str)
@agent.toolset
async def simple_toolset(ctx: RunContext[str]) -> AbstractToolset[str]:
return FunctionToolset()
func : ToolsetFunc[AgentDepsT] | None Default: None
The toolset function to register.
per_run_step : bool Default: True
Whether to re-evaluate the toolset for each run step. Defaults to True.
An optional unique ID for the dynamic toolset. Required for use with durable execution environments like Temporal, where the ID identifies the toolset’s activities within the workflow.
@async
def __aenter__() -> Self
Enter the agent context.
This will start all MCPServerStdios registered as toolsets so they are ready to be used.
This is a no-op if the agent has already been entered.
def set_mcp_sampling_model(
model: models.Model | models.KnownModelName | str | None = None,
) -> None
Set the sampling model on all MCP servers registered with the agent.
If no sampling model is provided, the agent’s model will be used.
def to_web(
models: ModelsParam = None,
builtin_tools: list[AbstractBuiltinTool] | None = None,
deps: AgentDepsT = None,
model_settings: ModelSettings | None = None,
instructions: str | None = None,
html_source: str | Path | None = None,
) -> Starlette
Create a Starlette app that serves a web chat UI for this agent.
This method returns a pre-configured Starlette application that provides a web-based chat interface for interacting with the agent. By default, the UI is fetched from a CDN and cached on first use.
The returned Starlette application can be mounted into a FastAPI app or run directly with any ASGI server (uvicorn, hypercorn, etc.).
Note that the deps and model_settings will be the same for each request.
To provide different deps for each request use the lower-level adapters directly.
Starlette — A configured Starlette application ready to be served (e.g., with uvicorn)
Additional models to make available in the UI. Can be:
- A sequence of model names/instances (e.g.,
['openai:gpt-5', 'anthropic:claude-sonnet-4-6']) - A dict mapping display labels to model names/instances
(e.g.,
\{'GPT 5': 'openai:gpt-5', 'Claude': 'anthropic:claude-sonnet-4-6'\}) The agent’s model is always included. Builtin tool support is automatically determined from each model’s profile.
Additional builtin tools to make available in the UI.
The agent’s configured builtin tools are always included. Tool labels
in the UI are derived from the tool’s label property.
Optional dependencies to use for all requests.
model_settings : ModelSettings | None Default: None
Optional settings to use for all model requests.
Optional extra instructions to pass to each agent run.
Path or URL for the chat UI HTML. Can be:
- None (default): Fetches from CDN and caches locally
- A Path instance: Reads from the local file
- A URL string (http:// or https://): Fetches from the URL
- A file path string: Reads from the local file
@async
@deprecated
def run_mcp_servers(
model: models.Model | models.KnownModelName | str | None = None,
) -> AsyncIterator[None]
Run MCPServerStdios so they can be used by the agent.
Deprecated: use async with agent instead.
If you need to set a sampling model on all MCP servers, use agent.set_mcp_sampling_model().
Returns: a context manager to start and shutdown the servers.
Bases: Generic[AgentDepsT, OutputDataT], ABC
Abstract superclass for Agent, WrapperAgent, and your own custom agent implementations.
The default model configured for this agent.
Type: models.Model | models.KnownModelName | str | None
The name of the agent, used for logging.
If None, we try to infer the agent name from the call frame when the agent is first run.
A human-readable description of the agent.
The type of dependencies used by the agent.
Type: type
The type of data output by agent runs, used to validate the data returned by the model, defaults to str.
Type: OutputSpec[OutputDataT]
Optional handler for events from the model’s streaming response and the agent’s execution of tools.
Type: EventStreamHandler[AgentDepsT] | None
All toolsets registered on the agent.
Output tools are not included.
Type: Sequence[AbstractToolset[AgentDepsT]]
def output_json_schema(
output_type: OutputSpec[OutputDataT | RunOutputDataT] | None = None,
) -> JsonSchema
The output return JSON schema.
JsonSchema
@async
def run(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: None = None,
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AgentRunResult[OutputDataT]
def run(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: OutputSpec[RunOutputDataT],
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AgentRunResult[RunOutputDataT]
Run the agent with a user prompt in async mode.
This method builds an internal agent graph (using system prompts, tools and output schemas) and then runs the graph to completion. The result of the run is returned.
Example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
async def main():
agent_run = await agent.run('What is the capital of France?')
print(agent_run.output)
#> The capital of France is Paris.
AgentRunResult[Any] — The result of the run.
User input to start/continue the conversation.
output_type : OutputSpec[RunOutputDataT] | None Default: None
Custom output type to use for this run, output_type may only be used if the agent has no
output validators since output validators would expect an argument that matches the agent’s output type.
History of the conversation so far.
deferred_tool_results : DeferredToolResults | None Default: None
Optional results for deferred tool calls in the message history.
model : models.Model | models.KnownModelName | str | None Default: None
Optional model to use for this run, required if model was not set when creating the agent.
Optional additional instructions to use for this run.
Optional dependencies to use for this run.
model_settings : AgentModelSettings[AgentDepsT] | None Default: None
Optional settings to use for this model’s request, or a callable
that receives RunContext and returns settings.
Callables are called before each model request, allowing dynamic per-step settings.
usage_limits : _usage.UsageLimits | None Default: None
Optional limits on model request count or token usage.
usage : _usage.RunUsage | None Default: None
Optional usage to start with, useful for resuming a conversation or agents used in tools.
metadata : AgentMetadata[AgentDepsT] | None Default: None
Optional metadata to attach to this run. Accepts a dictionary or a callable taking
RunContext; merged with the agent’s configured metadata.
infer_name : bool Default: True
Whether to try to infer the agent name from the call frame if it’s not set.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | None Default: None
Optional additional toolsets for this run.
event_stream_handler : EventStreamHandler[AgentDepsT] | None Default: None
Optional handler for events from the model’s streaming response and the agent’s execution of tools to use for this run.
builtin_tools : Sequence[AgentBuiltinTool[AgentDepsT]] | None Default: None
Optional additional builtin tools for this run.
Optional agent spec to apply for this run. At run time, spec values are additive.
def run_sync(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: None = None,
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AgentRunResult[OutputDataT]
def run_sync(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: OutputSpec[RunOutputDataT],
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AgentRunResult[RunOutputDataT]
Synchronously run the agent with a user prompt.
This is a convenience method that wraps self.run with loop.run_until_complete(...).
You therefore can’t use this method inside async code or if there’s an active event loop.
Example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
result_sync = agent.run_sync('What is the capital of Italy?')
print(result_sync.output)
#> The capital of Italy is Rome.
AgentRunResult[Any] — The result of the run.
User input to start/continue the conversation.
output_type : OutputSpec[RunOutputDataT] | None Default: None
Custom output type to use for this run, output_type may only be used if the agent has no
output validators since output validators would expect an argument that matches the agent’s output type.
History of the conversation so far.
deferred_tool_results : DeferredToolResults | None Default: None
Optional results for deferred tool calls in the message history.
model : models.Model | models.KnownModelName | str | None Default: None
Optional model to use for this run, required if model was not set when creating the agent.
Optional additional instructions to use for this run.
Optional dependencies to use for this run.
model_settings : AgentModelSettings[AgentDepsT] | None Default: None
Optional settings to use for this model’s request, or a callable
that receives RunContext and returns settings.
Callables are called before each model request, allowing dynamic per-step settings.
usage_limits : _usage.UsageLimits | None Default: None
Optional limits on model request count or token usage.
usage : _usage.RunUsage | None Default: None
Optional usage to start with, useful for resuming a conversation or agents used in tools.
metadata : AgentMetadata[AgentDepsT] | None Default: None
Optional metadata to attach to this run. Accepts a dictionary or a callable taking
RunContext; merged with the agent’s configured metadata.
infer_name : bool Default: True
Whether to try to infer the agent name from the call frame if it’s not set.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | None Default: None
Optional additional toolsets for this run.
event_stream_handler : EventStreamHandler[AgentDepsT] | None Default: None
Optional handler for events from the model’s streaming response and the agent’s execution of tools to use for this run.
builtin_tools : Sequence[AgentBuiltinTool[AgentDepsT]] | None Default: None
Optional additional builtin tools for this run.
Optional agent spec to apply for this run. At run time, spec values are additive.
@async
def run_stream(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: None = None,
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AbstractAsyncContextManager[result.StreamedRunResult[AgentDepsT, OutputDataT]]
def run_stream(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: OutputSpec[RunOutputDataT],
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AbstractAsyncContextManager[result.StreamedRunResult[AgentDepsT, RunOutputDataT]]
Run the agent with a user prompt in async streaming mode.
This method builds an internal agent graph (using system prompts, tools and output schemas) and then
runs the graph until the model produces output matching the output_type, for example text or structured data.
At this point, a streaming run result object is yielded from which you can stream the output as it comes in,
and — once this output has completed streaming — get the complete output, message history, and usage.
As this method will consider the first output matching the output_type to be the final output,
it will stop running the agent graph and will not execute any tool calls made by the model after this “final” output.
If you want to always run the agent graph to completion and stream events and output at the same time,
use agent.run() with an event_stream_handler or agent.iter() instead.
Example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
async def main():
async with agent.run_stream('What is the capital of the UK?') as response:
print(await response.get_output())
#> The capital of the UK is London.
AsyncIterator[result.StreamedRunResult[AgentDepsT, Any]] — The result of the run.
User input to start/continue the conversation.
output_type : OutputSpec[RunOutputDataT] | None Default: None
Custom output type to use for this run, output_type may only be used if the agent has no
output validators since output validators would expect an argument that matches the agent’s output type.
History of the conversation so far.
deferred_tool_results : DeferredToolResults | None Default: None
Optional results for deferred tool calls in the message history.
model : models.Model | models.KnownModelName | str | None Default: None
Optional model to use for this run, required if model was not set when creating the agent.
Optional additional instructions to use for this run.
Optional dependencies to use for this run.
model_settings : AgentModelSettings[AgentDepsT] | None Default: None
Optional settings to use for this model’s request, or a callable
that receives RunContext and returns settings.
Callables are called before each model request, allowing dynamic per-step settings.
usage_limits : _usage.UsageLimits | None Default: None
Optional limits on model request count or token usage.
usage : _usage.RunUsage | None Default: None
Optional usage to start with, useful for resuming a conversation or agents used in tools.
metadata : AgentMetadata[AgentDepsT] | None Default: None
Optional metadata to attach to this run. Accepts a dictionary or a callable taking
RunContext; merged with the agent’s configured metadata.
infer_name : bool Default: True
Whether to try to infer the agent name from the call frame if it’s not set.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | None Default: None
Optional additional toolsets for this run.
builtin_tools : Sequence[AgentBuiltinTool[AgentDepsT]] | None Default: None
Optional additional builtin tools for this run.
event_stream_handler : EventStreamHandler[AgentDepsT] | None Default: None
Optional handler for events from the model’s streaming response and the agent’s execution of tools to use for this run. It will receive all the events up until the final result is found, which you can then read or stream from inside the context manager. Note that it does not receive any events after the final result is found.
Optional agent spec to apply for this run. At run time, spec values are additive.
def run_stream_sync(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: None = None,
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> result.StreamedRunResultSync[AgentDepsT, OutputDataT]
def run_stream_sync(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: OutputSpec[RunOutputDataT],
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AbstractBuiltinTool] | None = None,
event_stream_handler: EventStreamHandler[AgentDepsT] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> result.StreamedRunResultSync[AgentDepsT, RunOutputDataT]
Run the agent with a user prompt in sync streaming mode.
This is a convenience method that wraps run_stream() with loop.run_until_complete(...).
You therefore can’t use this method inside async code or if there’s an active event loop.
This method builds an internal agent graph (using system prompts, tools and output schemas) and then
runs the graph until the model produces output matching the output_type, for example text or structured data.
At this point, a streaming run result object is yielded from which you can stream the output as it comes in,
and — once this output has completed streaming — get the complete output, message history, and usage.
As this method will consider the first output matching the output_type to be the final output,
it will stop running the agent graph and will not execute any tool calls made by the model after this “final” output.
If you want to always run the agent graph to completion and stream events and output at the same time,
use agent.run() with an event_stream_handler or agent.iter() instead.
Example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
def main():
response = agent.run_stream_sync('What is the capital of the UK?')
print(response.get_output())
#> The capital of the UK is London.
result.StreamedRunResultSync[AgentDepsT, Any] — The result of the run.
User input to start/continue the conversation.
output_type : OutputSpec[RunOutputDataT] | None Default: None
Custom output type to use for this run, output_type may only be used if the agent has no
output validators since output validators would expect an argument that matches the agent’s output type.
History of the conversation so far.
deferred_tool_results : DeferredToolResults | None Default: None
Optional results for deferred tool calls in the message history.
model : models.Model | models.KnownModelName | str | None Default: None
Optional model to use for this run, required if model was not set when creating the agent.
Optional dependencies to use for this run.
model_settings : AgentModelSettings[AgentDepsT] | None Default: None
Optional settings to use for this model’s request, or a callable
that receives RunContext and returns settings.
Callables are called before each model request, allowing dynamic per-step settings.
usage_limits : _usage.UsageLimits | None Default: None
Optional limits on model request count or token usage.
usage : _usage.RunUsage | None Default: None
Optional usage to start with, useful for resuming a conversation or agents used in tools.
metadata : AgentMetadata[AgentDepsT] | None Default: None
Optional metadata to attach to this run. Accepts a dictionary or a callable taking
RunContext; merged with the agent’s configured metadata.
infer_name : bool Default: True
Whether to try to infer the agent name from the call frame if it’s not set.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | None Default: None
Optional additional toolsets for this run.
builtin_tools : Sequence[AgentBuiltinTool[AgentDepsT]] | None Default: None
Optional additional builtin tools for this run.
event_stream_handler : EventStreamHandler[AgentDepsT] | None Default: None
Optional handler for events from the model’s streaming response and the agent’s execution of tools to use for this run. It will receive all the events up until the final result is found, which you can then read or stream from inside the context manager. Note that it does not receive any events after the final result is found.
Optional agent spec to apply for this run. At run time, spec values are additive.
def run_stream_events(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: None = None,
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AsyncIterator[_messages.AgentStreamEvent | AgentRunResultEvent[OutputDataT]]
def run_stream_events(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: OutputSpec[RunOutputDataT],
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AsyncIterator[_messages.AgentStreamEvent | AgentRunResultEvent[RunOutputDataT]]
Run the agent with a user prompt in async mode and stream events from the run.
This is a convenience method that wraps self.run and
uses the event_stream_handler kwarg to get a stream of events from the run.
Example:
from pydantic_ai import Agent, AgentRunResultEvent, AgentStreamEvent
agent = Agent('openai:gpt-5.2')
async def main():
events: list[AgentStreamEvent | AgentRunResultEvent] = []
async for event in agent.run_stream_events('What is the capital of France?'):
events.append(event)
print(events)
'''
[
PartStartEvent(index=0, part=TextPart(content='The capital of ')),
FinalResultEvent(tool_name=None, tool_call_id=None),
PartDeltaEvent(index=0, delta=TextPartDelta(content_delta='France is Paris. ')),
PartEndEvent(
index=0, part=TextPart(content='The capital of France is Paris. ')
),
AgentRunResultEvent(
result=AgentRunResult(output='The capital of France is Paris. ')
),
]
'''
Arguments are the same as for self.run,
except that event_stream_handler is now allowed.
AsyncIterator[_messages.AgentStreamEvent | AgentRunResultEvent[Any]] — An async iterable of stream events AgentStreamEvent and finally a AgentRunResultEvent with the final
AsyncIterator[_messages.AgentStreamEvent | AgentRunResultEvent[Any]] — run result.
User input to start/continue the conversation.
output_type : OutputSpec[RunOutputDataT] | None Default: None
Custom output type to use for this run, output_type may only be used if the agent has no
output validators since output validators would expect an argument that matches the agent’s output type.
History of the conversation so far.
deferred_tool_results : DeferredToolResults | None Default: None
Optional results for deferred tool calls in the message history.
model : models.Model | models.KnownModelName | str | None Default: None
Optional model to use for this run, required if model was not set when creating the agent.
Optional additional instructions to use for this run.
Optional dependencies to use for this run.
model_settings : AgentModelSettings[AgentDepsT] | None Default: None
Optional settings to use for this model’s request, or a callable
that receives RunContext and returns settings.
Callables are called before each model request, allowing dynamic per-step settings.
usage_limits : _usage.UsageLimits | None Default: None
Optional limits on model request count or token usage.
usage : _usage.RunUsage | None Default: None
Optional usage to start with, useful for resuming a conversation or agents used in tools.
metadata : AgentMetadata[AgentDepsT] | None Default: None
Optional metadata to attach to this run. Accepts a dictionary or a callable taking
RunContext; merged with the agent’s configured metadata.
infer_name : bool Default: True
Whether to try to infer the agent name from the call frame if it’s not set.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | None Default: None
Optional additional toolsets for this run.
builtin_tools : Sequence[AgentBuiltinTool[AgentDepsT]] | None Default: None
Optional additional builtin tools for this run.
Optional agent spec to apply for this run. At run time, spec values are additive.
@abstractmethod
@async
def iter(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: None = None,
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AbstractAsyncContextManager[AgentRun[AgentDepsT, OutputDataT]]
def iter(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: OutputSpec[RunOutputDataT],
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AbstractAsyncContextManager[AgentRun[AgentDepsT, RunOutputDataT]]
A contextmanager which can be used to iterate over the agent graph’s nodes as they are executed.
This method builds an internal agent graph (using system prompts, tools and output schemas) and then returns an
AgentRun object. The AgentRun can be used to async-iterate over the nodes of the graph as they are
executed. This is the API to use if you want to consume the outputs coming from each LLM model response, or the
stream of events coming from the execution of tools.
The AgentRun also provides methods to access the full message history, new messages, and usage statistics,
and the final result of the run once it has completed.
For more details, see the documentation of AgentRun.
Example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
async def main():
nodes = []
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='...',
)
),
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='...',
)
),
End(data=FinalResult(output='The capital of France is Paris.')),
]
'''
print(agent_run.result.output)
#> The capital of France is Paris.
AsyncIterator[AgentRun[AgentDepsT, Any]] — The result of the run.
User input to start/continue the conversation.
output_type : OutputSpec[RunOutputDataT] | None Default: None
Custom output type to use for this run, output_type may only be used if the agent has no
output validators since output validators would expect an argument that matches the agent’s output type.
History of the conversation so far.
deferred_tool_results : DeferredToolResults | None Default: None
Optional results for deferred tool calls in the message history.
model : models.Model | models.KnownModelName | str | None Default: None
Optional model to use for this run, required if model was not set when creating the agent.
Optional additional instructions to use for this run.
Optional dependencies to use for this run.
model_settings : AgentModelSettings[AgentDepsT] | None Default: None
Optional settings to use for this model’s request, or a callable
that receives RunContext and returns settings.
Callables are called before each model request, allowing dynamic per-step settings.
usage_limits : _usage.UsageLimits | None Default: None
Optional limits on model request count or token usage.
usage : _usage.RunUsage | None Default: None
Optional usage to start with, useful for resuming a conversation or agents used in tools.
metadata : AgentMetadata[AgentDepsT] | None Default: None
Optional metadata to attach to this run. Accepts a dictionary or a callable taking
RunContext; merged with the agent’s configured metadata.
infer_name : bool Default: True
Whether to try to infer the agent name from the call frame if it’s not set.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | None Default: None
Optional additional toolsets for this run.
builtin_tools : Sequence[AgentBuiltinTool[AgentDepsT]] | None Default: None
Optional additional builtin tools for this run.
Optional agent spec to apply for this run. At run time, spec values are additive.
@abstractmethod
def override(
name: str | _utils.Unset = _utils.UNSET,
deps: AgentDepsT | _utils.Unset = _utils.UNSET,
model: models.Model | models.KnownModelName | str | _utils.Unset = _utils.UNSET,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | _utils.Unset = _utils.UNSET,
tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] | _utils.Unset = _utils.UNSET,
instructions: _instructions.AgentInstructions[AgentDepsT] | _utils.Unset = _utils.UNSET,
model_settings: AgentModelSettings[AgentDepsT] | _utils.Unset = _utils.UNSET,
spec: dict[str, Any] | AgentSpec | None = None,
) -> Iterator[None]
Context manager to temporarily override agent name, dependencies, model, toolsets, tools, or instructions.
This is particularly useful when testing. You can find an example of this here.
name : str | _utils.Unset Default: _utils.UNSET
The name to use instead of the name passed to the agent constructor and agent run.
The dependencies to use instead of the dependencies passed to the agent run.
model : models.Model | models.KnownModelName | str | _utils.Unset Default: _utils.UNSET
The model to use instead of the model passed to the agent run.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | _utils.Unset Default: _utils.UNSET
The toolsets to use instead of the toolsets passed to the agent constructor and agent run.
tools : Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, …]] | _utils.Unset Default: _utils.UNSET
The tools to use instead of the tools registered with the agent.
The instructions to use instead of the instructions registered with the agent.
The model settings to use instead of the model settings passed to the agent constructor.
When set, any per-run model_settings argument is ignored.
Optional agent spec providing defaults for override.
@staticmethod
def parallel_tool_call_execution_mode(
mode: _tool_manager.ParallelExecutionMode = 'parallel',
) -> Iterator[None]
Set the parallel execution mode during the context.
The execution mode for tool calls:
- ‘parallel’: Run tool calls in parallel, yielding events as they complete (default).
- ‘sequential’: Run tool calls one at a time in order.
- ‘parallel_ordered_events’: Run tool calls in parallel, but events are emitted in order, after all calls complete.
@deprecated
@staticmethod
def sequential_tool_calls() -> Iterator[None]
Run tool calls sequentially during the context.
@staticmethod
def using_thread_executor(executor: Executor) -> Iterator[None]
Use a custom executor for running sync functions in threads during the context.
By default, sync tool functions and other sync callbacks are run in threads using
anyio.to_thread.run_sync, which creates ephemeral threads.
In long-running servers (e.g. FastAPI), this can lead to thread accumulation under sustained load.
This context manager lets you provide a bounded
ThreadPoolExecutor (or any
Executor) to control thread lifecycle:
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from pydantic_ai import Agent
@asynccontextmanager
async def lifespan(app):
executor = ThreadPoolExecutor(max_workers=16)
with Agent.using_thread_executor(executor):
yield
executor.shutdown(wait=True)
For per-agent configuration, use the
ThreadExecutor capability instead.
The executor to use for running sync functions.
@staticmethod
def is_model_request_node(
node: _agent_graph.AgentNode[T, S] | End[result.FinalResult[S]],
) -> TypeIs[_agent_graph.ModelRequestNode[T, S]]
Check if the node is a ModelRequestNode, narrowing the type if it is.
This method preserves the generic parameters while narrowing the type, unlike a direct call to isinstance.
TypeIs[_agent_graph.ModelRequestNode[T, S]]
@staticmethod
def is_call_tools_node(
node: _agent_graph.AgentNode[T, S] | End[result.FinalResult[S]],
) -> TypeIs[_agent_graph.CallToolsNode[T, S]]
Check if the node is a CallToolsNode, narrowing the type if it is.
This method preserves the generic parameters while narrowing the type, unlike a direct call to isinstance.
TypeIs[_agent_graph.CallToolsNode[T, S]]
@staticmethod
def is_user_prompt_node(
node: _agent_graph.AgentNode[T, S] | End[result.FinalResult[S]],
) -> TypeIs[_agent_graph.UserPromptNode[T, S]]
Check if the node is a UserPromptNode, narrowing the type if it is.
This method preserves the generic parameters while narrowing the type, unlike a direct call to isinstance.
TypeIs[_agent_graph.UserPromptNode[T, S]]
@staticmethod
def is_end_node(
node: _agent_graph.AgentNode[T, S] | End[result.FinalResult[S]],
) -> TypeIs[End[result.FinalResult[S]]]
Check if the node is a End, narrowing the type if it is.
This method preserves the generic parameters while narrowing the type, unlike a direct call to isinstance.
TypeIs[End[result.FinalResult[S]]]
def to_ag_ui(
output_type: OutputSpec[OutputDataT] | None = None,
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
deps: AgentDepsT = None,
model_settings: ModelSettings | None = None,
usage_limits: UsageLimits | None = None,
usage: RunUsage | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
debug: bool = False,
routes: Sequence[BaseRoute] | None = None,
middleware: Sequence[Middleware] | None = None,
exception_handlers: Mapping[Any, ExceptionHandler] | None = None,
on_startup: Sequence[Callable[[], Any]] | None = None,
on_shutdown: Sequence[Callable[[], Any]] | None = None,
lifespan: Lifespan[AGUIApp[AgentDepsT, OutputDataT]] | None = None,
) -> AGUIApp[AgentDepsT, OutputDataT]
Returns an ASGI application that handles every AG-UI request by running the agent.
Note that the deps will be the same for each request, with the exception of the AG-UI state that’s
injected into the state field of a deps object that implements the StateHandler protocol.
To provide different deps for each request (e.g. based on the authenticated user),
use pydantic_ai.ag_ui.run_ag_ui or
pydantic_ai.ag_ui.handle_ag_ui_request instead.
Example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
app = agent.to_ag_ui()
The app is an ASGI application that can be used with any ASGI server.
To run the application, you can use the following command:
uvicorn app:app --host 0.0.0.0 --port 8000
See AG-UI docs for more information.
AGUIApp[AgentDepsT, OutputDataT] — An ASGI application for running Pydantic AI agents with AG-UI protocol support.
output_type : OutputSpec[OutputDataT] | None Default: None
Custom output type to use for this run, output_type may only be used if the agent has
no output validators since output validators would expect an argument that matches the agent’s
output type.
History of the conversation so far.
deferred_tool_results : DeferredToolResults | None Default: None
Optional results for deferred tool calls in the message history.
model : models.Model | models.KnownModelName | str | None Default: None
Optional model to use for this run, required if model was not set when creating the agent.
Optional dependencies to use for this run.
model_settings : ModelSettings | None Default: None
Optional settings to use for this model’s request.
usage_limits : UsageLimits | None Default: None
Optional limits on model request count or token usage.
Optional usage to start with, useful for resuming a conversation or agents used in tools.
infer_name : bool Default: True
Whether to try to infer the agent name from the call frame if it’s not set.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | None Default: None
Optional additional toolsets for this run.
debug : bool Default: False
Boolean indicating if debug tracebacks should be returned on errors.
A list of routes to serve incoming HTTP and WebSocket requests.
A list of middleware to run for every request. A starlette application will always
automatically include two middleware classes. ServerErrorMiddleware is added as the very
outermost middleware, to handle any uncaught errors occurring anywhere in the entire stack.
ExceptionMiddleware is added as the very innermost middleware, to deal with handled
exception cases occurring in the routing or endpoints.
A mapping of either integer status codes, or exception class types onto
callables which handle the exceptions. Exception handler callables should be of the form
handler(request, exc) -> response and may be either standard functions, or async functions.
A list of callables to run on application startup. Startup handler callables do not take any arguments, and may be either standard functions, or async functions.
A list of callables to run on application shutdown. Shutdown handler callables do not take any arguments, and may be either standard functions, or async functions.
lifespan : Lifespan[AGUIApp[AgentDepsT, OutputDataT]] | None Default: None
A lifespan context function, which can be used to perform startup and shutdown tasks.
This is a newer style that replaces the on_startup and on_shutdown handlers. Use one or
the other, not both.
def to_a2a(
storage: Storage | None = None,
broker: Broker | None = None,
name: str | None = None,
url: str = 'http://localhost:8000',
version: str = '1.0.0',
description: str | None = None,
provider: AgentProvider | None = None,
skills: list[Skill] | None = None,
debug: bool = False,
routes: Sequence[Route] | None = None,
middleware: Sequence[Middleware] | None = None,
exception_handlers: dict[Any, ExceptionHandler] | None = None,
lifespan: Lifespan[FastA2A] | None = None,
) -> FastA2A
Convert the agent to a FastA2A application.
Example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
app = agent.to_a2a()
The app is an ASGI application that can be used with any ASGI server.
To run the application, you can use the following command:
uvicorn app:app --host 0.0.0.0 --port 8000
FastA2A
@async
def to_cli(
deps: AgentDepsT = None,
prog_name: str = 'pydantic-ai',
message_history: Sequence[_messages.ModelMessage] | None = None,
model_settings: ModelSettings | None = None,
usage_limits: _usage.UsageLimits | None = None,
) -> None
Run the agent in a CLI chat interface.
Example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2', instructions='You always respond in Italian.')
async def main():
await agent.to_cli()
The dependencies to pass to the agent.
prog_name : str Default: 'pydantic-ai'
The name of the program to use for the CLI. Defaults to ‘pydantic-ai’.
History of the conversation so far.
model_settings : ModelSettings | None Default: None
Optional settings to use for this model’s request.
usage_limits : _usage.UsageLimits | None Default: None
Optional limits on model request count or token usage.
def to_cli_sync(
deps: AgentDepsT = None,
prog_name: str = 'pydantic-ai',
message_history: Sequence[_messages.ModelMessage] | None = None,
model_settings: ModelSettings | None = None,
usage_limits: _usage.UsageLimits | None = None,
) -> None
Run the agent in a CLI chat interface with the non-async interface.
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2', instructions='You always respond in Italian.')
agent.to_cli_sync()
agent.to_cli_sync(prog_name='assistant')
The dependencies to pass to the agent.
prog_name : str Default: 'pydantic-ai'
The name of the program to use for the CLI. Defaults to ‘pydantic-ai’.
History of the conversation so far.
model_settings : ModelSettings | None Default: None
Optional settings to use for this model’s request.
usage_limits : _usage.UsageLimits | None Default: None
Optional limits on model request count or token usage.
Bases: AbstractAgent[AgentDepsT, OutputDataT]
Agent which wraps another agent.
Does nothing on its own, used as a base class.
@async
def iter(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: None = None,
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AbstractAsyncContextManager[AgentRun[AgentDepsT, OutputDataT]]
def iter(
user_prompt: str | Sequence[_messages.UserContent] | None = None,
output_type: OutputSpec[RunOutputDataT],
message_history: Sequence[_messages.ModelMessage] | None = None,
deferred_tool_results: DeferredToolResults | None = None,
model: models.Model | models.KnownModelName | str | None = None,
instructions: _instructions.AgentInstructions[AgentDepsT] = None,
deps: AgentDepsT = None,
model_settings: AgentModelSettings[AgentDepsT] | None = None,
usage_limits: _usage.UsageLimits | None = None,
usage: _usage.RunUsage | None = None,
metadata: AgentMetadata[AgentDepsT] | None = None,
infer_name: bool = True,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
spec: dict[str, Any] | AgentSpec | None = None,
) -> AbstractAsyncContextManager[AgentRun[AgentDepsT, RunOutputDataT]]
A contextmanager which can be used to iterate over the agent graph’s nodes as they are executed.
This method builds an internal agent graph (using system prompts, tools and output schemas) and then returns an
AgentRun object. The AgentRun can be used to async-iterate over the nodes of the graph as they are
executed. This is the API to use if you want to consume the outputs coming from each LLM model response, or the
stream of events coming from the execution of tools.
The AgentRun also provides methods to access the full message history, new messages, and usage statistics,
and the final result of the run once it has completed.
For more details, see the documentation of AgentRun.
Example:
from pydantic_ai import Agent
agent = Agent('openai:gpt-5.2')
async def main():
nodes = []
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='...',
)
),
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='...',
)
),
End(data=FinalResult(output='The capital of France is Paris.')),
]
'''
print(agent_run.result.output)
#> The capital of France is Paris.
AsyncIterator[AgentRun[AgentDepsT, Any]] — The result of the run.
User input to start/continue the conversation.
output_type : OutputSpec[RunOutputDataT] | None Default: None
Custom output type to use for this run, output_type may only be used if the agent has no
output validators since output validators would expect an argument that matches the agent’s output type.
History of the conversation so far.
deferred_tool_results : DeferredToolResults | None Default: None
Optional results for deferred tool calls in the message history.
model : models.Model | models.KnownModelName | str | None Default: None
Optional model to use for this run, required if model was not set when creating the agent.
Optional additional instructions to use for this run.
Optional dependencies to use for this run.
model_settings : AgentModelSettings[AgentDepsT] | None Default: None
Optional settings to use for this model’s request.
usage_limits : _usage.UsageLimits | None Default: None
Optional limits on model request count or token usage.
usage : _usage.RunUsage | None Default: None
Optional usage to start with, useful for resuming a conversation or agents used in tools.
metadata : AgentMetadata[AgentDepsT] | None Default: None
Optional metadata to attach to this run.
infer_name : bool Default: True
Whether to try to infer the agent name from the call frame if it’s not set.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | None Default: None
Optional additional toolsets for this run.
builtin_tools : Sequence[AgentBuiltinTool[AgentDepsT]] | None Default: None
Optional additional builtin tools for this run.
Optional agent spec to apply for this run.
def override(
name: str | _utils.Unset = _utils.UNSET,
deps: AgentDepsT | _utils.Unset = _utils.UNSET,
model: models.Model | models.KnownModelName | str | _utils.Unset = _utils.UNSET,
toolsets: Sequence[AbstractToolset[AgentDepsT]] | _utils.Unset = _utils.UNSET,
tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] | _utils.Unset = _utils.UNSET,
instructions: _instructions.AgentInstructions[AgentDepsT] | _utils.Unset = _utils.UNSET,
model_settings: AgentModelSettings[AgentDepsT] | _utils.Unset = _utils.UNSET,
spec: dict[str, Any] | AgentSpec | None = None,
) -> Iterator[None]
Context manager to temporarily override agent name, dependencies, model, toolsets, tools, or instructions.
This is particularly useful when testing. You can find an example of this here.
name : str | _utils.Unset Default: _utils.UNSET
The name to use instead of the name passed to the agent constructor and agent run.
The dependencies to use instead of the dependencies passed to the agent run.
model : models.Model | models.KnownModelName | str | _utils.Unset Default: _utils.UNSET
The model to use instead of the model passed to the agent run.
toolsets : Sequence[AbstractToolset[AgentDepsT]] | _utils.Unset Default: _utils.UNSET
The toolsets to use instead of the toolsets passed to the agent constructor and agent run.
tools : Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, …]] | _utils.Unset Default: _utils.UNSET
The tools to use instead of the tools registered with the agent.
The instructions to use instead of the instructions registered with the agent.
The model settings to use instead of the model settings passed to the agent constructor.
When set, any per-run model_settings argument is ignored.
Optional agent spec to apply as overrides.
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='...',
)
),
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='...',
)
),
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
Metadata associated with this agent run, if configured.
The unique identifier for the agent run.
Type: str
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 new messages for the run so far.
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='...',
)
),
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='...',
)
),
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 usage() -> _usage.RunUsage
Get usage statistics for the run so far, including token usage, model requests, and so on.
_usage.RunUsage
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
Metadata associated with this agent run, if configured.
The unique identifier for the agent run.
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 new messages associated with this run.
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 usage() -> _usage.RunUsage
Return the usage of the whole run.
_usage.RunUsage
def timestamp() -> datetime
Return the timestamp of last response.
Options for instrumenting models and agents with OpenTelemetry.
Used in:
Agent(instrument=...)Agent.instrument_all()InstrumentedModel
See the Debugging and Monitoring guide for more info.
def __init__(
tracer_provider: TracerProvider | None = None,
meter_provider: MeterProvider | None = None,
include_binary_content: bool = True,
include_content: bool = True,
version: Literal[1, 2, 3, 4, 5] = DEFAULT_INSTRUMENTATION_VERSION,
event_mode: Literal['attributes', 'logs'] = 'attributes',
logger_provider: LoggerProvider | None = None,
use_aggregated_usage_attribute_names: bool = False,
)
Create instrumentation options.
tracer_provider : TracerProvider | None Default: None
The OpenTelemetry tracer provider to use.
If not provided, the global tracer provider is used.
Calling logfire.configure() sets the global tracer provider, so most users don’t need this.
meter_provider : MeterProvider | None Default: None
The OpenTelemetry meter provider to use.
If not provided, the global meter provider is used.
Calling logfire.configure() sets the global meter provider, so most users don’t need this.
include_binary_content : bool Default: True
Whether to include binary content in the instrumentation events.
include_content : bool Default: True
Whether to include prompts, completions, and tool call arguments and responses in the instrumentation events.
version : Literal[1, 2, 3, 4, 5] Default: DEFAULT_INSTRUMENTATION_VERSION
Version of the data format. This is unrelated to the Pydantic AI package version.
Version 1 is based on the legacy event-based OpenTelemetry GenAI spec
and will be removed in a future release.
The parameters event_mode and logger_provider are only relevant for version 1.
Version 2 uses the newer OpenTelemetry GenAI spec and stores messages in the following attributes:
gen_ai.system_instructionsfor instructions passed to the agent.gen_ai.input.messagesandgen_ai.output.messageson model request spans.pydantic_ai.all_messageson agent run spans. Version 3 is the same as version 2, with additional support for thinking tokens. Version 4 is the same as version 3, with GenAI semantic conventions for multimodal content: URL-based media uses type=‘uri’ with uri and mime_type fields (and modality for image/audio/video). Inline binary content uses type=‘blob’ with mime_type and content fields (and modality for image/audio/video). https://opentelemetry.io/docs/specs/semconv/gen-ai/non-normative/examples-llm-calls/#multimodal-inputs-example Version 5 is the same as version 4, but CallDeferred and ApprovalRequired exceptions no longer record an exception event or set the span status to ERROR — the span is left as UNSET, since deferrals are control flow, not errors.
event_mode : Literal[‘attributes’, ‘logs’] Default: 'attributes'
The mode for emitting events in version 1.
If 'attributes', events are attached to the span as attributes.
If 'logs', events are emitted as OpenTelemetry log-based events.
logger_provider : LoggerProvider | None Default: None
The OpenTelemetry logger provider to use.
If not provided, the global logger provider is used.
Calling logfire.configure() sets the global logger provider, so most users don’t need this.
This is only used if event_mode='logs' and version=1.
use_aggregated_usage_attribute_names : bool Default: False
Whether to use gen_ai.aggregated_usage.* attribute names
for token usage on agent run spans instead of the standard gen_ai.usage.* names.
Enable this to prevent double-counting in observability backends that aggregate span
attributes across parent and child spans. Defaults to False.
Note: gen_ai.aggregated_usage.* is a custom namespace, not part of the OpenTelemetry
Semantic Conventions. It may be updated if OTel introduces an official convention.
def messages_to_otel_events(
messages: list[ModelMessage],
parameters: ModelRequestParameters | None = None,
) -> list[LogRecord]
Convert a list of model messages to OpenTelemetry events.
list[LogRecord] — A list of OpenTelemetry events.
messages : list[ModelMessage]
The messages to convert.
parameters : ModelRequestParameters | None Default: None
The model request parameters.
def capture_run_messages() -> Iterator[list[_messages.ModelMessage]]
Context manager to access the messages used in a run, run_sync, or run_stream call.
Useful when a run may raise an exception, see model errors for more information.
Examples:
from pydantic_ai import Agent, capture_run_messages
agent = Agent('test')
with capture_run_messages() as messages:
try:
result = agent.run_sync('foobar')
except Exception:
print(messages)
raise
Iterator[list[_messages.ModelMessage]]
Default: Literal['early', 'exhaustive']
Type variable for the result data of a run where output_type was customized on the run call.
Default: TypeVar('RunOutputDataT')
A function that receives agent RunContext and an async iterable of events from the model’s streaming response and the agent’s execution of tools.
Type: TypeAlias Default: Callable[[RunContext[AgentDepsT], AsyncIterable[_messages.AgentStreamEvent]], Awaitable[None]]