pydantic_ai.mcp

MCPError

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.

Attributes

message

The error message.

Type: str Default: message

code

The error code returned by the server.

Type: int Default: code

data

Additional information about the error, if provided by the server.

Type: dict[str, Any] | None Default: data

Methods

from_mcp_sdk

@classmethod

def from_mcp_sdk(cls, error: mcp_exceptions.McpError) -> MCPError

Create an MCPError from an MCP SDK McpError.

Returns

MCPError

Parameters

error : mcp_exceptions.McpError

An McpError from the MCP SDK.

ResourceAnnotations

Additional properties describing MCP entities.

See the resource annotations in the MCP specification.

Attributes

audience

Intended audience for this entity.

Type: list[mcp_types.Role] | None Default: None

priority

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

Methods

from_mcp_sdk

@classmethod

def from_mcp_sdk(cls, mcp_annotations: mcp_types.Annotations) -> ResourceAnnotations

Convert from MCP SDK Annotations to ResourceAnnotations.

Returns

ResourceAnnotations

Parameters

mcp_annotations : mcp_types.Annotations

The MCP SDK annotations object.

BaseResource

Bases: ABC

Base class for MCP resources.

Attributes

name

The programmatic name of the resource.

Type: str

title

Human-readable title for UI contexts.

Type: str | None Default: None

description

A description of what this resource represents.

Type: str | None Default: None

mime_type

The MIME type of the resource, if known.

Type: str | None Default: None

annotations

Optional annotations for the resource.

Type: ResourceAnnotations | None Default: None

metadata

Optional metadata for the resource.

Type: dict[str, Any] | None Default: None

Resource

Bases: BaseResource

A resource that can be read from an MCP server.

See the resources in the MCP specification.

Attributes

uri

The URI of the resource.

Type: str

size

The size of the raw resource content in bytes (before base64 encoding), if known.

Type: int | None Default: None

Methods

from_mcp_sdk

@classmethod

def from_mcp_sdk(cls, mcp_resource: mcp_types.Resource) -> Resource

Convert from MCP SDK Resource to PydanticAI Resource.

Returns

Resource

Parameters

mcp_resource : mcp_types.Resource

The MCP SDK Resource object.

ResourceTemplate

Bases: BaseResource

A template for parameterized resources on an MCP server.

See the resource templates in the MCP specification.

Attributes

uri_template

URI template (RFC 6570) for constructing resource URIs.

Type: str

Methods

from_mcp_sdk

@classmethod

def from_mcp_sdk(cls, mcp_template: mcp_types.ResourceTemplate) -> ResourceTemplate

Convert from MCP SDK ResourceTemplate to PydanticAI ResourceTemplate.

Returns

ResourceTemplate

Parameters

mcp_template : mcp_types.ResourceTemplate

The MCP SDK ResourceTemplate object.

ServerCapabilities

Capabilities that an MCP server supports.

Attributes

experimental

Experimental, non-standard capabilities that the server supports.

Type: list[str] | None Default: None

logging

Whether the server supports sending log messages to the client.

Type: bool Default: False

prompts

Whether the server offers any prompt templates.

Type: bool Default: False

prompts_list_changed

Whether the server will emit notifications when the list of prompts changes.

Type: bool Default: False

resources

Whether the server offers any resources to read.

Type: bool Default: False

resources_list_changed

Whether the server will emit notifications when the list of resources changes.

Type: bool Default: False

tools

Whether the server offers any tools to call.

Type: bool Default: False

tools_list_changed

Whether the server will emit notifications when the list of tools changes.

Type: bool Default: False

completions

Whether the server offers autocompletion suggestions for prompts and resources.

Type: bool Default: False

Methods

from_mcp_sdk

@classmethod

def from_mcp_sdk(
    cls,
    mcp_capabilities: mcp_types.ServerCapabilities,
) -> ServerCapabilities

Convert from MCP SDK ServerCapabilities to PydanticAI ServerCapabilities.

Returns

ServerCapabilities

Parameters

mcp_capabilities : mcp_types.ServerCapabilities

The MCP SDK ServerCapabilities object.

MCPServer

Bases: AbstractToolset[Any], ABC

Base class for attaching agents to MCP servers.

See https://modelcontextprotocol.io for more information.

Deprecated

This class hierarchy (MCPServer, MCPServerStdio, MCPServerSSE, MCPServerStreamableHTTP, MCPServerHTTP) is deprecated in favor of MCPToolset, which is built on the more capable FastMCP client and supports the full MCP protocol. The concrete subclasses will be removed in v2.

Attributes

tool_prefix

A prefix to add to all tools that are registered with the server.

If not empty, will include a trailing underscore(_).

e.g. if tool_prefix='foo', then a tool named bar will be registered as foo_bar

Type: str | None Default: tool_prefix

log_level

The log level to set when connecting to the server, if any.

See https://modelcontextprotocol.io/specification/2025-03-26/server/utilities/logging#logging for more details.

If None, no log level will be set.

Type: mcp_types.LoggingLevel | None Default: log_level

log_handler

A handler for logging messages from the server.

Type: LoggingFnT | None Default: log_handler

timeout

The timeout in seconds to wait for the client to initialize.

Type: float Default: timeout

read_timeout

Maximum time in seconds to wait for new messages before timing out.

This timeout applies to the long-lived connection after it’s established. If no new messages are received within this time, the connection will be considered stale and may be closed. Defaults to 5 minutes (300 seconds).

Type: float Default: read_timeout

process_tool_call

Hook to customize tool calling and optionally pass extra metadata.

Type: ProcessToolCallback | None Default: process_tool_call

allow_sampling

Whether to allow MCP sampling through this client.

Type: bool Default: allow_sampling

sampling_model

The model to use for sampling.

Type: models.Model | None Default: sampling_model

max_retries

The maximum number of times to retry a tool call.

Type: int Default: max_retries

elicitation_callback

Callback function to handle elicitation requests from the server.

Type: ElicitationFnT | None Default: elicitation_callback

cache_tools

Whether to cache the list of tools.

When enabled (default), tools are fetched once and cached until either:

• The server sends a notifications/tools/list_changed notification
• MCPServer.__aexit__ is called (when the last context exits)

Set to False for servers that change tools dynamically without sending notifications.

Note: When using durable execution (Temporal, DBOS), tool definitions are additionally cached at the wrapper level across activities/steps, to avoid redundant MCP connections. This wrapper-level cache is not invalidated by tools/list_changed notifications. Set to False to disable all caching if tools may change during a workflow.

Type: bool Default: cache_tools

cache_resources

Whether to cache the list of resources.

When enabled (default), resources are fetched once and cached until either:

• The server sends a notifications/resources/list_changed notification
• MCPServer.__aexit__ is called (when the last context exits)

Set to False for servers that change resources dynamically without sending notifications.

Type: bool Default: cache_resources

include_instructions

Whether to include the server’s instructions in the agent’s instructions.

Defaults to False for backward compatibility.

Type: bool Default: include_instructions

include_return_schema

Whether to include return schemas in tool definitions sent to the model.

When None (default), defaults to False unless the IncludeToolReturnSchemas capability is used.

Type: bool | None Default: include_return_schema

server_info

Access the information send by the MCP server during initialization.

Type: mcp_types.Implementation

capabilities

Access the capabilities advertised by the MCP server during initialization.

Type: ServerCapabilities

instructions

Access the instructions sent by the MCP server during initialization.

Type: str | None

is_running

Check if the MCP server is running.

Type: bool

Methods

client_streams

@abstractmethod

@async

def client_streams(

) -> AsyncIterator[tuple[MemoryObjectReceiveStream[SessionMessage | Exception], MemoryObjectSendStream[SessionMessage]]]

Create the streams for the MCP server.

Returns

AsyncIterator[tuple[MemoryObjectReceiveStream[SessionMessage | Exception], MemoryObjectSendStream[SessionMessage]]]

get_instructions

@async

def get_instructions(ctx: RunContext[Any]) -> messages.InstructionPart | None

Return the MCP server’s instructions for how to use its tools.

If include_instructions is True, returns the instructions sent by the MCP server during initialization. Otherwise, returns None.

Instructions from external servers are marked as dynamic since they may change between connections.

Returns

messages.InstructionPart | None — An InstructionPart with the server’s instructions if include_instructions is enabled, otherwise None.

Parameters

ctx : RunContext[Any]

The run context for this agent run.

list_tools

@async

def list_tools() -> list[mcp_types.Tool]

Retrieve tools that are currently active on the server.

Tools are cached by default, with cache invalidation on:

• notifications/tools/list_changed notifications from the server
• __aexit__ when the last context exits

Set cache_tools=False for servers that change tools without sending notifications.

Returns

list[mcp_types.Tool]

direct_call_tool

@async

def direct_call_tool(
    name: str,
    args: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> ToolResult

Call a tool on the server.

Returns

ToolResult — The result of the tool call.

Parameters

name : str

The name of the tool to call.

args : dict[str, Any]

The arguments to pass to the tool.

metadata : dict[str, Any] | None Default: None

Request-level metadata (optional)

Raises

• ModelRetry — If the tool call fails.

list_resources

@async

def list_resources() -> list[Resource]

Retrieve resources that are currently present on the server.

Resources are cached by default, with cache invalidation on:

• notifications/resources/list_changed notifications from the server
• __aexit__ when the last context exits

Set cache_resources=False for servers that change resources without sending notifications.

Returns

list[Resource]

Raises

• MCPError — If the server returns an error.

list_resource_templates

@async

def list_resource_templates() -> list[ResourceTemplate]

Retrieve resource templates that are currently present on the server.

Returns

list[ResourceTemplate]

Raises

• MCPError — If the server returns an error.

read_resource

@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.

Returns

str | messages.BinaryContent | list[str | messages.BinaryContent] — The resource contents. If the resource has a single content item, returns that item directly. str | messages.BinaryContent | list[str | messages.BinaryContent] — If the resource has multiple content items, returns a list of items.

Parameters

uri : str | Resource

The URI of the resource to read, or a Resource object.

Raises

• MCPError — If the server returns an error.

__aenter__

@async

def __aenter__() -> Self

Enter the MCP server context.

The first call starts the connection (spawning a subprocess for stdio servers, opening an HTTP connection for HTTP servers). Subsequent calls — from any task — share the same connection via reference counting. The connection is torn down when the last async with scope exits.

Because the session runs in a dedicated background task, entering and exiting from different tasks (e.g. asyncio.gather children, fasta2a workers, or graph node tasks) is safe: the underlying transport’s cancel scopes never cross task boundaries.

Returns

Self

MCPServerStdio

Bases: MCPServer

Runs an MCP server in a subprocess and communicates with it over stdin/stdout.

This class implements the stdio transport from the MCP specification. See https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#stdio for more information.

Note

Using this class as an async context manager will start the server as a subprocess when entering the context, and stop it when exiting the context.

Example:

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

server = MCPServerStdio(  # (1)
  'uv', args=['run', 'mcp-run-python', 'stdio'], timeout=10
)
agent = Agent('openai:gpt-5.2', toolsets=[server])

See MCP Run Python for more information.

Attributes

command

The command to run.

Type: str Default: command

args

The arguments to pass to the command.

Type: Sequence[str] Default: args

env

The environment variables the CLI server will have access to.

By default the subprocess will not inherit any environment variables from the parent process. If you want to inherit the environment variables from the parent process, use env=os.environ.

Type: dict[str, str] | None Default: env

cwd

The working directory to use when spawning the process.

Type: str | Path | None Default: cwd

Methods

__init__

def __init__(
    command: str,
    args: Sequence[str],
    env: dict[str, str] | None = None,
    cwd: str | Path | None = None,
    tool_prefix: str | None = None,
    log_level: mcp_types.LoggingLevel | None = None,
    log_handler: LoggingFnT | None = None,
    timeout: float = 5,
    read_timeout: float = 5 * 60,
    process_tool_call: ProcessToolCallback | None = None,
    allow_sampling: bool = True,
    sampling_model: models.Model | None = None,
    max_retries: int = 1,
    elicitation_callback: ElicitationFnT | None = None,
    cache_tools: bool = True,
    cache_resources: bool = True,
    include_instructions: bool = False,
    include_return_schema: bool | None = None,
    id: str | None = None,
    client_info: mcp_types.Implementation | None = None,
)

Build a new MCP server.

Parameters

command : str

The command to run.

args : Sequence[str]

The arguments to pass to the command.

env : dict[str, str] | None Default: None

The environment variables to set in the subprocess.

cwd : str | Path | None Default: None

The working directory to use when spawning the process.

tool_prefix : str | None Default: None

A prefix to add to all tools that are registered with the server.

log_level : mcp_types.LoggingLevel | None Default: None

The log level to set when connecting to the server, if any.

log_handler : LoggingFnT | None Default: None

A handler for logging messages from the server.

timeout : float Default: 5

The timeout in seconds to wait for the client to initialize.

read_timeout : float Default: 5 * 60

Maximum time in seconds to wait for new messages before timing out.

process_tool_call : ProcessToolCallback | None Default: None

Hook to customize tool calling and optionally pass extra metadata.

allow_sampling : bool Default: True

Whether to allow MCP sampling through this client.

sampling_model : models.Model | None Default: None

The model to use for sampling.

max_retries : int Default: 1

The maximum number of times to retry a tool call.

elicitation_callback : ElicitationFnT | None Default: None

Callback function to handle elicitation requests from the server.

cache_tools : bool Default: True

Whether to cache the list of tools. See MCPServer.cache_tools.

cache_resources : bool Default: True

Whether to cache the list of resources. See MCPServer.cache_resources.

include_instructions : bool Default: False

Whether to include the server’s instructions in the agent’s instructions. See MCPServer.include_instructions.

include_return_schema : bool | None Default: None

Whether to include return schemas in tool definitions. See MCPServer.include_return_schema.

id : str | None Default: None

An optional unique ID for the MCP server. An MCP server needs to have an ID in order to be used in a durable execution environment like Temporal, in which case the ID will be used to identify the server’s activities within the workflow.

client_info : mcp_types.Implementation | None Default: None

Information describing the MCP client implementation.

MCPServerSSE

Bases: _MCPServerHTTP

An MCP server that connects over streamable HTTP connections.

This class implements the SSE transport from the MCP specification. See https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#http-with-sse for more information.

Note

Using this class as an async context manager will create a new pool of HTTP connections to connect to a server which should already be running.

Example:

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerSSE

server = MCPServerSSE('http://localhost:3001/sse')
agent = Agent('openai:gpt-5.2', toolsets=[server])

MCPServerHTTP

Bases: MCPServerSSE

An MCP server that connects over HTTP using the old SSE transport.

This class implements the SSE transport from the MCP specification. See https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#http-with-sse for more information.

Note

Using this class as an async context manager will create a new pool of HTTP connections to connect to a server which should already be running.

Example:

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP

server = MCPServerHTTP('http://localhost:3001/sse')
agent = Agent('openai:gpt-5.2', toolsets=[server])

MCPServerStreamableHTTP

Bases: _MCPServerHTTP

An MCP server that connects over HTTP using the Streamable HTTP transport.

This class implements the Streamable HTTP transport from the MCP specification. See https://modelcontextprotocol.io/introduction#streamable-http for more information.

Note

Using this class as an async context manager will create a new pool of HTTP connections to connect to a server which should already be running.

Example:

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

server = MCPServerStreamableHTTP('http://localhost:8000/mcp')
agent = Agent('openai:gpt-5.2', toolsets=[server])

CallToolFunc

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=...).

MCPToolset

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)

Attributes

client

The underlying FastMCP Client. Always normalized to a fastmcp.Client regardless of how the toolset was constructed.

Type: FastMCPClient[Any]

max_retries

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

tool_error_behavior

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

process_tool_call

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

cache_tools

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_changed notification
• 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

cache_resources

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

include_instructions

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

include_return_schema

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

sampling_model

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

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

server_info

The server-implementation info sent during initialization.

Raises AttributeError when accessed before the toolset has been entered.

Type: mcp_types.Implementation

capabilities

The capabilities advertised by the server during initialization.

Raises AttributeError when accessed before the toolset has been entered.

Type: ServerCapabilities

instructions

The instructions sent by the server during initialization.

Raises AttributeError when accessed before the toolset has been entered.

Type: str | None

is_running

Whether the toolset is currently entered (the FastMCP session is open).

Type: bool

Methods

__init__

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,
    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.

Parameters

client : MCPToolsetClient

How to connect to the MCP server. See the class docstring for accepted shapes.

id : str | None Default: None

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.

max_retries : int | None Default: None

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.

include_instructions : bool Default: False

Whether to include the server’s instructions in the agent’s instructions. See MCPToolset.include_instructions.

include_return_schema : bool | None Default: None

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.

sampling_handler : SamplingHandler[Any, Any] | None Default: None

A FastMCP-shaped sampling handler. Use for full control over the sampling response.

elicitation_handler : ElicitationHandler[Any, Any] | None Default: None

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.

init_timeout : float | None Default: _UNSET

Timeout in seconds for the initial connection and initialize handshake.

read_timeout : float | None Default: _UNSET

Maximum time in seconds to wait for new messages on the long-lived connection. Defaults to 5 minutes.

roots : RootsList | RootsHandler[Any] | None Default: None

Filesystem roots advertised to the server.

auth : httpx.Auth | Literal[‘oauth’] | str | None Default: None

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.

headers : dict[str, str] | None Default: None

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.

Raises

• ValueError — If a pre-built fastmcp.Client is passed alongside any of the kwargs that would otherwise build a default Client (sampling, elicitation, headers, etc.), or if sampling_model and sampling_handler are both passed, or if headers and http_client are both passed.
• ImportError — If the fastmcp client isn’t installed. Install the mcp extra (which pulls fastmcp-slim[client]): pip install "pydantic-ai-slim[mcp]".

get_instructions

@async

def get_instructions(ctx: RunContext[AgentDepsT]) -> messages.InstructionPart | None

Return the server’s instructions if include_instructions is enabled.

Returns

messages.InstructionPart | None

list_tools

@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__.

Returns

list[mcp_types.Tool]

direct_call_tool

@async

def direct_call_tool(
    name: str,
    args: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> Any

Call a tool on the server directly.

Returns

Any

Parameters

name : str

The name of the tool to call.

args : dict[str, Any]

The arguments to pass to the tool.

metadata : dict[str, Any] | None Default: None

Optional request-level _meta payload sent alongside the call.

Raises

• ModelRetry — If the tool errors and tool_error_behavior='retry' (the default).
• fastmcp.exceptions.ToolError — If the tool errors and tool_error_behavior='error'.

list_resources

@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.

Returns

list[Resource]

Raises

• MCPError — If the server returns an error.

list_resource_templates

@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.

Returns

list[ResourceTemplate]

Raises

• MCPError — If the server returns an error.

read_resource

@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.

Returns

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.

Parameters

uri : str | Resource

The URI of the resource to read, or a Resource object.

Raises

• MCPError — If the server returns an error.

load_mcp_servers

@deprecated

def load_mcp_servers(
    config_path: str | Path,
) -> list[MCPServerStdio | MCPServerStreamableHTTP | MCPServerSSE]

Load MCP servers from a configuration file.

Environment variables can be referenced in the configuration file using:

• ${VAR_NAME} syntax - expands to the value of VAR_NAME, raises error if not defined
• ${VAR_NAME:-default} syntax - expands to VAR_NAME if set, otherwise uses the default value

Returns

list[MCPServerStdio | MCPServerStreamableHTTP | MCPServerSSE] — A list of MCP servers.

Parameters

config_path : str | Path

The path to the configuration file.

Raises

• 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 value is provided.

load_mcp_toolsets

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 of VAR_NAME, raises if not defined
• ${VAR_NAME:-default} syntax — expands to VAR_NAME if set, otherwise the default

Returns

list[AbstractToolset[Any]] — A list of toolsets, one per server in the config file, each prefixed with the server name.

Parameters

config_path : str | Path

Path to the JSON configuration file.

Raises

• 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.
• ImportError — If the fastmcp client isn’t installed. Install the mcp extra (which pulls fastmcp-slim[client]): pip install "pydantic-ai-slim[mcp]".

ToolResult

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]]

ProcessToolCallback

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]]

MCPToolsetClient

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'