pydantic_ai.mcp
Bases: RuntimeError
Raised when an MCP server returns an error response.
This exception wraps error responses from MCP servers, following the ErrorData schema from the MCP specification.
The error message.
Type: str Default: message
The error code returned by the server.
Type: int Default: code
Additional information about the error, if provided by the server.
Type: dict[str, Any] | None Default: data
@classmethod
def from_mcp_sdk(cls, error: mcp_exceptions.McpError) -> MCPError
Create an MCPError from an MCP SDK McpError.
MCPError
An McpError from the MCP SDK.
Additional properties describing MCP entities.
See the resource annotations in the MCP specification.
Intended audience for this entity.
Type: list[mcp_types.Role] | None Default: None
Priority level for this entity, ranging from 0.0 to 1.0.
Type: Annotated[float, Field(ge=0.0, le=1.0)] | None Default: None
ISO 8601 timestamp of the last modification.
Type: str | None Default: None
@classmethod
def from_mcp_sdk(cls, mcp_annotations: mcp_types.Annotations) -> ResourceAnnotations
Convert from MCP SDK Annotations to ResourceAnnotations.
ResourceAnnotations
The MCP SDK annotations object.
An icon for display in user interfaces.
URL or data URI for the icon.
Type: str
Optional MIME type for the icon.
Type: str | None Default: None
Optional list of strings specifying icon dimensions (e.g., [“48x48”, “96x96”]).
Type: list[str] | None Default: None
Bases: ABC
Base class for MCP resources.
The programmatic name of the resource.
Type: str
Human-readable title for UI contexts.
Type: str | None Default: None
A description of what this resource represents.
Type: str | None Default: None
The MIME type of the resource, if known.
Type: str | None Default: None
Optional annotations for the resource.
Type: ResourceAnnotations | None Default: None
Optional icons for the resource.
Type: list[Icon] | None Default: None
Optional metadata for the resource.
Type: dict[str, Any] | None Default: None
Bases: BaseResource
A resource that can be read from an MCP server.
See the resources in the MCP specification.
The URI of the resource.
Type: str
The size of the raw resource content in bytes (before base64 encoding), if known.
Type: int | None Default: None
@classmethod
def from_mcp_sdk(cls, mcp_resource: mcp_types.Resource) -> Resource
Convert from MCP SDK Resource to PydanticAI Resource.
Resource
The MCP SDK Resource object.
Bases: BaseResource
A template for parameterized resources on an MCP server.
See the resource templates in the MCP specification.
URI template (RFC 6570) for constructing resource URIs.
Type: str
@classmethod
def from_mcp_sdk(cls, mcp_template: mcp_types.ResourceTemplate) -> ResourceTemplate
Convert from MCP SDK ResourceTemplate to PydanticAI ResourceTemplate.
ResourceTemplate
The MCP SDK ResourceTemplate object.
A resource link referenced in a prompt or tool call result.
Unlike EmbeddedResource, this does not include the resource
content directly — it is a reference to a resource that the server can read.
Note: resource links returned by tools are not guaranteed to appear in the results of
resources/list requests.
See the MCP specification.
The URI of the linked resource.
Type: str
The programmatic name of the linked resource.
Type: str
Human-readable title for UI contexts.
Type: str | None Default: None
A description of what this linked resource represents.
Type: str | None Default: None
The MIME type of the linked resource, if known.
Type: str | None Default: None
The size of the raw resource content in bytes (before base64 encoding), if known.
Type: int | None Default: None
Optional annotations for the linked resource.
Type: ResourceAnnotations | None Default: None
Optional icons for the linked resource.
Type: list[Icon] | None Default: None
Optional metadata for the linked resource.
Type: dict[str, Any] | None Default: None
Discriminator for resource link content.
Type: Literal[‘resource_link’] Default: 'resource_link'
@classmethod
def from_mcp_sdk(cls, mcp_resource_link: mcp_types.ResourceLink) -> ResourceLink
Convert from MCP SDK ResourceLink to PydanticAI ResourceLink.
ResourceLink
An argument for a prompt template.
The name of the argument.
Type: str
Human-readable title for the argument.
Type: str | None Default: None
A human-readable description of the argument.
Type: str | None Default: None
Whether the argument is required or optional. If not specified, the server may determine this based on context.
Type: bool | None Default: None
A prompt or prompt template that the server offers.
The programmatic name of the prompt.
Type: str
Human-readable title for prompt.
Type: str | None Default: None
An optional description of what this prompt provides.
Type: str | None Default: None
A list of arguments to use for templating the prompt.
Type: list[PromptArgument] | None Default: None
An optional list of icons for this prompt.
Type: list[Icon] | None Default: None
See MCP specification for notes on _meta usage.
Type: dict[str, Any] | None Default: None
@classmethod
def from_mcp_sdk(cls, mcp_prompt: mcp_types.Prompt) -> Prompt
Convert from MCP SDK Prompt to PydanticAI Prompt.
Prompt
The MCP SDK Prompt object.
A resource embedded into a prompt or tool call result.
Contains the actual resource content alongside its metadata, unlike
ResourceLink which is only a reference.
See the MCP specification.
The URI of the embedded resource.
Type: str
The content of the embedded resource.
Type: str | messages.BinaryContent
Discriminator for embedded resource content.
Type: Literal[‘resource’] Default: 'resource'
The MIME type of the resource, if known.
Type: str | None Default: None
Optional annotations for the resource.
Type: ResourceAnnotations | None Default: None
See MCP specification for notes on _meta usage.
Type: dict[str, Any] | None Default: None
_meta carried on the nested resource contents (separate from the embedding’s own _meta).
Type: dict[str, Any] | None Default: None
@classmethod
def from_mcp_sdk(
cls,
part: mcp_types.EmbeddedResource,
content: str | messages.BinaryContent,
) -> EmbeddedResource
Convert from MCP SDK EmbeddedResource to PydanticAI EmbeddedResource.
EmbeddedResource
A message returned as part of a prompt result.
The role of the message sender.
Type: PromptRole
The content of the message.
Type: ContentBlock
The result of a get_prompt request.
The prompt messages.
Type: list[PromptMessage]
An optional description for the prompt.
Type: str | None Default: None
See MCP specification for notes on _meta usage.
Type: dict[str, Any] | None Default: None
Capabilities that an MCP server supports.
Experimental, non-standard capabilities that the server supports.
Type: list[str] | None Default: None
Whether the server supports sending log messages to the client.
Type: bool Default: False
Whether the server offers any prompt templates.
Type: bool Default: False
Whether the server will emit notifications when the list of prompts changes.
Type: bool Default: False
Whether the server offers any resources to read.
Type: bool Default: False
Whether the server will emit notifications when the list of resources changes.
Type: bool Default: False
Whether the server offers any tools to call.
Type: bool Default: False
Whether the server will emit notifications when the list of tools changes.
Type: bool Default: False
Whether the server offers autocompletion suggestions for prompts and resources.
Type: bool Default: False
@classmethod
def from_mcp_sdk(
cls,
mcp_capabilities: mcp_types.ServerCapabilities,
) -> ServerCapabilities
Convert from MCP SDK ServerCapabilities to PydanticAI ServerCapabilities.
ServerCapabilities
The MCP SDK ServerCapabilities object.
Bases: Protocol
A callable that invokes an MCP tool — typically MCPToolset.direct_call_tool or its legacy equivalent.
Passed to user-defined ProcessToolCallback functions as
the underlying call hook. metadata is keyword-only — pass it as
await call_tool(name, args, metadata=...).
Bases: AbstractToolset[AgentDepsT]
A toolset for connecting to an MCP server.
MCPToolset is the recommended way to use Model Context Protocol
servers in Pydantic AI. It is built on the FastMCP Client, which
supports the full MCP protocol — tools, resources, sampling, elicitation, OAuth — and a wide
range of transports (HTTP, SSE, stdio, in-process FastMCP servers, multi-server configs).
Pass any input that FastMCP can build a transport from — a URL, a script path, a FastMCP
server instance for in-process testing — or a pre-built fastmcp.Client for full control over
its configuration. For multi-server JSON config files, use
load_mcp_toolsets instead.
Example — connect to a streamable-HTTP MCP server:
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
toolset = MCPToolset('http://localhost:8000/mcp')
agent = Agent('openai:gpt-5', toolsets=[toolset])
Example — connect to a local stdio MCP server:
from pydantic_ai.mcp import MCPToolset
toolset = MCPToolset('my_mcp_server.py')
Example — pass a pre-built FastMCP Client for full configuration control:
from fastmcp.client import Client
from fastmcp.client.transports import StreamableHttpTransport
from pydantic_ai.mcp import MCPToolset
client = Client(StreamableHttpTransport('http://localhost:8000/mcp'), auth='oauth')
toolset = MCPToolset(client)
The underlying FastMCP Client. Always normalized to a fastmcp.Client regardless of how
the toolset was constructed.
Type: FastMCPClient[Any]
Maximum number of times a tool call may be retried after a ModelRetry.
None (default) inherits the agent’s retry count at runtime. Set explicitly to override.
Type: int | None Default: max_retries
How to handle tool errors raised by the server.
'retry' (default) raises ModelRetry so the model can
self-correct; 'error' propagates the underlying fastmcp.exceptions.ToolError to the caller.
Type: Literal[‘retry’, ‘error’] Default: tool_error_behavior
Hook to wrap tool calls — useful for adding request-level metadata, custom retry policies,
or telemetry. See ProcessToolCallback.
Type: ProcessToolCallback | None Default: process_tool_call
Whether to cache the list of tools across get_tools() calls.
When enabled (default), tools are fetched once and cached until either:
- The server sends a
notifications/tools/list_changednotification - The toolset is fully exited (last
__aexit__matches the first__aenter__)
Set to False for servers that change tools dynamically without sending notifications, or when
passing a pre-built FastMCP Client (the cache-invalidation message handler isn’t installed in
that case, so caches are only invalidated by session close).
Type: bool Default: cache_tools
Whether to cache the list of resources across list_resources() calls.
Same semantics as cache_tools but for
notifications/resources/list_changed notifications.
Type: bool Default: cache_resources
Whether to cache the list of prompts across list_prompts() calls.
Same semantics as cache_tools but for
notifications/prompts/list_changed notifications.
Type: bool Default: cache_prompts
Whether to include the server’s initialize instructions string in the agent’s instruction set.
Defaults to False for backward compatibility. When True, the instructions returned by the
server during initialization are added to the agent’s instructions.
Type: bool Default: include_instructions
Whether to include each tool’s outputSchema in the schema sent to the model.
When None (the default), defaults to False unless the
IncludeToolReturnSchemas capability is
used.
Type: bool | None Default: include_return_schema
A Pydantic AI model that the server may sample from via the MCP sampling/createMessage flow.
When set (and no explicit sampling_handler is passed), Pydantic AI builds a sampling handler
that delegates to this model with the request’s maxTokens/temperature/stopSequences
settings applied. If both sampling_model and sampling_handler are passed, an error is raised.
Type: models.Model | None Default: sampling_model
Log level requested from the server via logging/setLevel after initialization.
None (default) leaves the server’s default log level alone. Combine with log_handler to
receive log messages.
Type: mcp_types.LoggingLevel | None Default: log_level
The server-implementation info sent during initialization.
Raises AttributeError when accessed before the toolset has been entered.
Type: mcp_types.Implementation
The capabilities advertised by the server during initialization.
Raises AttributeError when accessed before the toolset has been entered.
Type: ServerCapabilities
The instructions sent by the server during initialization.
Raises AttributeError when accessed before the toolset has been entered.
Whether the toolset is currently entered (the FastMCP session is open).
Type: bool
def __init__(
client: MCPToolsetClient,
id: str | None = None,
max_retries: int | None = None,
tool_error_behavior: Literal['retry', 'error'] = 'retry',
process_tool_call: ProcessToolCallback | None = None,
cache_tools: bool = True,
cache_resources: bool = True,
cache_prompts: bool = True,
include_instructions: bool = False,
include_return_schema: bool | None = None,
sampling_model: models.Model | None = None,
sampling_handler: SamplingHandler[Any, Any] | None = None,
elicitation_handler: ElicitationHandler[Any, Any] | None = None,
log_handler: LogHandler | None = None,
log_level: mcp_types.LoggingLevel | None = None,
progress_handler: ProgressHandler | None = None,
message_handler: MessageHandlerT | None = None,
client_info: mcp_types.Implementation | None = None,
init_timeout: float | None = _UNSET,
read_timeout: float | None = _UNSET,
roots: RootsList | RootsHandler[Any] | None = None,
auth: httpx.Auth | Literal['oauth'] | str | None = None,
verify: ssl.SSLContext | bool | str | None = None,
headers: dict[str, str] | None = None,
http_client: httpx.AsyncClient | None = None,
)
Build a new MCPToolset.
How to connect to the MCP server. See the class docstring for accepted shapes.
An optional unique identifier for this toolset. Required for use in durable execution environments like Temporal or DBOS, where it identifies the toolset’s activities/steps within a workflow.
Maximum number of times a tool call may be retried after a ModelRetry.
None inherits the agent’s retry count at runtime.
tool_error_behavior : Literal[‘retry’, ‘error’] Default: 'retry'
'retry' (default) raises
ModelRetry on tool errors so the model can
self-correct; 'error' propagates the underlying exception.
process_tool_call : ProcessToolCallback | None Default: None
Hook to wrap tool calls. See
ProcessToolCallback.
cache_tools : bool Default: True
Whether to cache the list of tools. See
MCPToolset.cache_tools.
cache_resources : bool Default: True
Whether to cache the list of resources. See
MCPToolset.cache_resources.
cache_prompts : bool Default: True
Whether to cache the list of prompts. See
MCPToolset.cache_prompts.
include_instructions : bool Default: False
Whether to include the server’s instructions in the agent’s
instructions. See
MCPToolset.include_instructions.
Whether to include return schemas in tool definitions. See
MCPToolset.include_return_schema.
sampling_model : models.Model | None Default: None
A Pydantic AI model the server may sample from. Mutually exclusive with
sampling_handler.
A FastMCP-shaped sampling handler. Use for full control over the sampling response.
A FastMCP-shaped elicitation handler that receives MCP
elicitation/create requests from the server.
log_handler : LogHandler | None Default: None
A FastMCP-shaped log handler that receives log messages from the server.
log_level : mcp_types.LoggingLevel | None Default: None
Log level requested from the server via logging/setLevel after
initialization.
progress_handler : ProgressHandler | None Default: None
A FastMCP-shaped progress handler.
message_handler : MessageHandlerT | None Default: None
A FastMCP-shaped message handler called for every server-sent message.
Pydantic AI installs its own message handler internally to invalidate caches on
list_changed notifications; if you provide one, both run (yours after ours).
client_info : mcp_types.Implementation | None Default: None
Information describing the MCP client implementation, sent to the server during initialization.
Timeout in seconds for the initial connection and initialize handshake.
Maximum time in seconds to wait for new messages on the long-lived connection. Defaults to 5 minutes.
Filesystem roots advertised to the server.
HTTP authentication for HTTP transports — an httpx.Auth, the literal string
'oauth' to enable FastMCP’s OAuth flow, or a bearer-token string.
verify : ssl.SSLContext | bool | str | None Default: None
SSL verification mode for HTTP transports — an ssl.SSLContext, a CA bundle
path string, or a bool.
Extra HTTP headers for HTTP transports. Mutually exclusive with http_client.
http_client : httpx.AsyncClient | None Default: None
A pre-configured httpx.AsyncClient to use for HTTP transports — useful
for self-signed certificates or custom connection pooling. Mutually exclusive with
headers.
ValueError— If a pre-builtfastmcp.Clientis passed alongside any of the kwargs that would otherwise build a default Client (sampling, elicitation, headers, etc.), or ifsampling_modelandsampling_handlerare both passed, or ifheadersandhttp_clientare both passed.
def set_sampling_model(model: models.Model) -> None
Set the sampling_model on an already-constructed toolset.
Swaps both the public attribute and the underlying FastMCP client’s sampling callback. Takes effect on the next session opened by the client; calls already in flight on an existing session continue using the previously configured handler.
@async
def get_instructions(ctx: RunContext[AgentDepsT]) -> messages.InstructionPart | None
Return the server’s instructions if include_instructions is enabled.
messages.InstructionPart | None
@async
def list_tools() -> list[mcp_types.Tool]
Retrieve the tools currently exposed by the server.
When cache_tools is enabled (default), results
are cached and invalidated by notifications/tools/list_changed or the toolset’s last
__aexit__.
list[mcp_types.Tool]
@async
def direct_call_tool(
name: str,
args: dict[str, Any],
metadata: dict[str, Any] | None = None,
use_task: bool = False,
) -> Any
Call a tool on the server directly.
name : str
The name of the tool to call.
The arguments to pass to the tool.
Optional request-level _meta payload sent alongside the call.
use_task : bool Default: False
When True, send the call with task=True per MCP
SEP-1686 so
the server wraps execution in a durable, cancelable, pollable task; the result is awaited via
tasks/result. Only valid for tools whose execution.taskSupport is 'required' or 'optional'.
ModelRetry— If the tool errors andtool_error_behavior='retry'(the default).fastmcp.exceptions.ToolError— If the tool errors andtool_error_behavior='error'.
@async
def list_prompts() -> list[Prompt]
Retrieve the prompts currently exposed by the server.
When cache_prompts is enabled (default),
results are cached and invalidated by notifications/prompts/list_changed or the
toolset’s last __aexit__.
Returns an empty list if the server does not advertise the prompts capability.
list[Prompt]
MCPError— If the server returns an error.
@async
def get_prompt(name: str, arguments: dict[str, str] | None = None) -> PromptResult
Retrieve a specific prompt from the server, optionally parameterized.
PromptResult
name : str
The name of the prompt to retrieve.
Arguments to parameterize the prompt, if applicable.
MCPError— If the server doesn’t advertise thepromptscapability, or if it returns an error response.
@async
def list_resources() -> list[Resource]
Retrieve the resources currently exposed by the server.
When cache_resources is enabled (default),
results are cached and invalidated by notifications/resources/list_changed or the
toolset’s last __aexit__.
Returns an empty list if the server does not advertise the resources capability.
list[Resource]
MCPError— If the server returns an error.
@async
def list_resource_templates() -> list[ResourceTemplate]
Retrieve the resource templates currently exposed by the server.
Returns an empty list if the server does not advertise the resources capability.
list[ResourceTemplate]
MCPError— If the server returns an error.
@async
def read_resource(
uri: str,
) -> str | messages.BinaryContent | list[str | messages.BinaryContent]
def read_resource(
uri: Resource,
) -> str | messages.BinaryContent | list[str | messages.BinaryContent]
Read the contents of a specific resource by URI.
str | messages.BinaryContent | list[str | messages.BinaryContent] — The resource contents — a single value if the resource has one content item, or a list
str | messages.BinaryContent | list[str | messages.BinaryContent] — otherwise. Text content is returned as str, binary content as
str | messages.BinaryContent | list[str | messages.BinaryContent] — BinaryContent.
uri : str | Resource
The URI of the resource to read, or a Resource object.
MCPError— If the server returns an error.
def load_mcp_toolsets(config_path: str | Path) -> list[AbstractToolset[Any]]
Load MCPToolsets from a configuration file.
The configuration file uses the same mcpServers JSON shape as Claude Desktop, Cursor, and the
MCP specification. Each server entry produces one MCPToolset,
wrapped in a PrefixedToolset using the server’s name
as prefix to disambiguate tools across multiple servers.
Environment variables can be referenced in the configuration file using:
${VAR_NAME}syntax — expands to the value ofVAR_NAME, raises if not defined${VAR_NAME:-default}syntax — expands toVAR_NAMEif set, otherwise the default
list[AbstractToolset[Any]] — A list of toolsets, one per server in the config file, each prefixed with the server name.
config_path : str | Path
Path to the JSON configuration file.
FileNotFoundError— If the configuration file does not exist.ValidationError— If the configuration file does not match the schema.ValueError— If an environment variable referenced in the configuration is not defined and no default is provided.
A content block that can be used in prompts and tool results.
Default: messages.TextContent | messages.BinaryContent | ResourceLink | EmbeddedResource
The result type of an MCP tool call.
Default: str | messages.BinaryContent | dict[str, Any] | list[Any] | Sequence[str | messages.BinaryContent | dict[str, Any] | list[Any]]
A process tool callback.
It accepts a run context, the original tool call function, a tool name, and arguments.
Allows wrapping an MCP server tool call to customize it, including adding extra request metadata.
Default: Callable[[RunContext[Any], CallToolFunc, str, dict[str, Any]], Awaitable[ToolResult]]
Anything MCPToolset accepts as its client argument — a pre-built fastmcp.Client, a FastMCP
ClientTransport, an in-process FastMCP server, an AnyUrl/URL string, a script Path, or a
URL/path/script string.
For multi-server JSON config files, use load_mcp_toolsets
instead — it expands env vars and constructs one MCPToolset per server entry.
Type: TypeAlias Default: FastMCPClient[Any] | ClientTransport | FastMCP | FastMCP1Server | AnyUrl | Path | str