Add FastMCPToolset

docs/api/toolsets.md

@@ -14,3 +14,5 @@
         - PreparedToolset
         - WrapperToolset
         - ToolsetFunc
+
+::: pydantic_ai.toolsets.fastmcp

docs/install.md

@@ -55,6 +55,7 @@ pip/uv-add "pydantic-ai-slim[openai]"
 * `tavily` - installs `tavily-python` [PyPI ↗](https://pypi.org/project/tavily-python){:target="_blank"}
 * `cli` - installs `rich` [PyPI ↗](https://pypi.org/project/rich){:target="_blank"}, `prompt-toolkit` [PyPI ↗](https://pypi.org/project/prompt-toolkit){:target="_blank"}, and `argcomplete` [PyPI ↗](https://pypi.org/project/argcomplete){:target="_blank"}
 * `mcp` - installs `mcp` [PyPI ↗](https://pypi.org/project/mcp){:target="_blank"}
+* `fastmcp` - installs `fastmcp` [PyPI ↗](https://pypi.org/project/fastmcp){:target="_blank"}
 * `a2a` - installs `fasta2a` [PyPI ↗](https://pypi.org/project/fasta2a){:target="_blank"}
 * `ag-ui` - installs `ag-ui-protocol` [PyPI ↗](https://pypi.org/project/ag-ui-protocol){:target="_blank"} and `starlette` [PyPI ↗](https://pypi.org/project/starlette){:target="_blank"}
 * `dbos` - installs [`dbos`](durable_execution/dbos.md) [PyPI ↗](https://pypi.org/project/dbos){:target="_blank"}

docs/mcp/client.md

@@ -5,7 +5,7 @@ to use their tools.
 
 ## Install
 
-You need to either install [`pydantic-ai`](../install.md), or[`pydantic-ai-slim`](../install.md#slim-install) with the `mcp` optional group:
+You need to either install [`pydantic-ai`](../install.md), or [`pydantic-ai-slim`](../install.md#slim-install) with the `mcp` optional group:
 
 ```bash
 pip/uv-add "pydantic-ai-slim[mcp]"

docs/mcp/fastmcp-client.md

@@ -0,0 +1,88 @@
+# FastMCP Client
+
+[FastMCP](https://gofastmcp.com/) is a higher-level MCP framework that bills itself as "The fast, Pythonic way to build MCP servers and clients." It supports additional capabilities on top of the MCP specification like [Tool Transformation](https://gofastmcp.com/patterns/tool-transformation), [OAuth](https://gofastmcp.com/clients/auth/oauth), and more.
+
+As an alternative to Pydantic AI's standard [`MCPServer` MCP client](client.md) built on the [MCP SDK](https://github.com/modelcontextprotocol/python-sdk), you can use the [`FastMCPToolset`][pydantic_ai.toolsets.fastmcp.FastMCPToolset] [toolset](../toolsets.md) that leverages the [FastMCP Client](https://gofastmcp.com/clients/) to connect to local and remote MCP servers, whether or not they're built using [FastMCP Server](https://gofastmcp.com/servers/).
+
+Note that it does not yet support integration elicitation or sampling, which are supported by the [standard `MCPServer` client](client.md).
+
+## Install
+
+To use the `FastMCPToolset`, you will need to install [`pydantic-ai-slim`](../install.md#slim-install) with the `fastmcp` optional group:
+
+```bash
+pip/uv-add "pydantic-ai-slim[fastmcp]"
+```
+
+## Usage
+
+A `FastMCPToolset` can then be created from:
+
+- A FastMCP Server: `#!python FastMCPToolset(fastmcp.FastMCP('my_server'))`
+- A FastMCP Client: `#!python FastMCPToolset(fastmcp.Client(...))`
+- A FastMCP Transport: `#!python FastMCPToolset(fastmcp.StdioTransport(command='uvx', args=['mcp-run-python', 'stdio']))`
+- A Streamable HTTP URL: `#!python FastMCPToolset('http://localhost:8000/mcp')`
+- An HTTP SSE URL: `#!python FastMCPToolset('http://localhost:8000/sse')`
+- A Python Script: `#!python FastMCPToolset('my_server.py')`
+- A Node.js Script: `#!python FastMCPToolset('my_server.js')`
+- A JSON MCP Configuration: `#!python FastMCPToolset({'mcpServers': {'my_server': {'command': 'uvx', 'args': ['mcp-run-python', 'stdio']}}})`
+
+If you already have a [FastMCP Server](https://gofastmcp.com/servers) in the same codebase as your Pydantic AI agent, you can create a `FastMCPToolset` directly from it and save agent a network round trip:
+
+```python
+from fastmcp import FastMCP
+
+from pydantic_ai import Agent
+from pydantic_ai.toolsets.fastmcp import FastMCPToolset
+
+fastmcp_server = FastMCP('my_server')
+@fastmcp_server.tool()
+async def add(a: int, b: int) -> int:
+    return a + b
+
+toolset = FastMCPToolset(fastmcp_server)
+
+agent = Agent('openai:gpt-5', toolsets=[toolset])
+
+async def main():
+    result = await agent.run('What is 7 plus 5?')
+    print(result.output)
+    #> The answer is 12.
+```
+
+_(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_
+
+Connecting your agent to a Streamable HTTP MCP Server is as simple as:
+
+```python
+from pydantic_ai import Agent
+from pydantic_ai.toolsets.fastmcp import FastMCPToolset
+
+toolset = FastMCPToolset('http://localhost:8000/mcp')
+
+agent = Agent('openai:gpt-5', toolsets=[toolset])
+```
+
+_(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_
+
+You can also create a `FastMCPToolset` from a JSON MCP Configuration:
+
+```python
+from pydantic_ai import Agent
+from pydantic_ai.toolsets.fastmcp import FastMCPToolset
+
+mcp_config = {
+    'mcpServers': {
+        'time_mcp_server': {
+            'command': 'uvx',
+            'args': ['mcp-run-python', 'stdio']
+        }
+    }
+}
+
+toolset = FastMCPToolset(mcp_config)
+
+agent = Agent('openai:gpt-5', toolsets=[toolset])
+```
+
+_(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_

docs/mcp/overview.md

@@ -1,11 +1,12 @@
 # Model Context Protocol (MCP)
 
-Pydantic AI supports [Model Context Protocol (MCP)](https://modelcontextprotocol.io) in two ways:
+Pydantic AI supports [Model Context Protocol (MCP)](https://modelcontextprotocol.io) in multiple ways:
 
-1. [Agents](../agents.md) can connect to MCP servers and user their tools
-    1. Pydantic AI can act as an MCP client and connect directly to local and remote MCP servers, [learn more …](client.md)
-    2. Some model providers can themselves connect to remote MCP servers, [learn more …](../builtin-tools.md#mcp-server-tool)
-2. Agents can be used within MCP servers, [learn more …](server.md)
+1. [Agents](../agents.md) can connect to MCP servers and use their tools using three different methods:
+    1. Pydantic AI can act as an MCP client and connect directly to local and remote MCP servers. [Learn more](client.md) about [`MCPServer`][pydantic_ai.mcp.MCPServer].
+    2. Pydantic AI can use the [FastMCP Client](https://gofastmcp.com/clients/client/) to connect to local and remote MCP servers, whether or not they're built using [FastMCP Server](https://gofastmcp.com/servers). [Learn more](fastmcp-client.md) about [`FastMCPToolset`][pydantic_ai.toolsets.fastmcp.FastMCPToolset].
+    3. Some model providers can themselves connect to remote MCP servers using a "built-in tool". [Learn more](../builtin-tools.md#mcp-server-tool) about [`MCPServerTool`][pydantic_ai.builtin_tools.MCPServerTool].
+2. Agents can be used within MCP servers. [Learn more](server.md)
 
 ## What is MCP?
 

docs/toolsets.md

@@ -661,7 +661,10 @@ If you want to reuse a network connection or session across tool listings and ca
 
 ### MCP Servers
 
-See the [MCP Client](./mcp/client.md) documentation for how to use MCP servers with Pydantic AI.
+Pydantic AI provides two toolsets that allow an agent to connect to and call tools on local and remote MCP Servers:
+
+1. `MCPServer`: the [MCP SDK-based Client](./mcp/client.md) which offers more direct control by leveraging the MCP SDK directly
+2. `FastMCPToolset`: the [FastMCP-based Client](./mcp/fastmcp-client.md) which offers additional capabilities like Tool Transformation, simpler OAuth configuration, and more.
 
 ### LangChain Tools {#langchain-tools}
 

mkdocs.yml

@@ -49,6 +49,7 @@ nav:
       - MCP:
           - Overview: mcp/overview.md
           - mcp/client.md
+          - mcp/fastmcp-client.md
           - mcp/server.md
       - Multi-Agent Patterns: multi-agent-applications.md
       - Testing: testing.md

pydantic_ai_slim/pydantic_ai/toolsets/fastmcp.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import base64
+from asyncio import Lock
+from contextlib import AsyncExitStack
+from dataclasses import KW_ONLY, dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Literal
+
+from pydantic import AnyUrl
+from typing_extensions import Self, assert_never
+
+from pydantic_ai import messages
+from pydantic_ai.exceptions import ModelRetry
+from pydantic_ai.tools import AgentDepsT, RunContext, ToolDefinition
+from pydantic_ai.toolsets import AbstractToolset
+from pydantic_ai.toolsets.abstract import ToolsetTool
+
+try:
+    from fastmcp.client import Client
+    from fastmcp.client.transports import ClientTransport
+    from fastmcp.exceptions import ToolError
+    from fastmcp.mcp_config import MCPConfig
+    from fastmcp.server import FastMCP
+    from mcp.server.fastmcp import FastMCP as FastMCP1Server
+    from mcp.types import (
+        AudioContent,
+        BlobResourceContents,
+        ContentBlock,
+        EmbeddedResource,
+        ImageContent,
+        ResourceLink,
+        TextContent,
+        TextResourceContents,
+        Tool as MCPTool,
+    )
+
+    from pydantic_ai.mcp import TOOL_SCHEMA_VALIDATOR
+
+except ImportError as _import_error:
+    raise ImportError(
+        'Please install the `fastmcp` package to use the FastMCP server, '
+        'you can use the `fastmcp` optional group — `pip install "pydantic-ai-slim[fastmcp]"`'
+    ) from _import_error
+
+
+if TYPE_CHECKING:
+    from fastmcp.client.client import CallToolResult
+
+
+FastMCPToolResult = messages.BinaryContent | dict[str, Any] | str | None
+
+ToolErrorBehavior = Literal['model_retry', 'error']
+
+UNKNOWN_BINARY_MEDIA_TYPE = 'application/octet-stream'
+
+
+@dataclass(init=False)
+class FastMCPToolset(AbstractToolset[AgentDepsT]):
+    """A FastMCP Toolset that uses the FastMCP Client to call tools from a local or remote MCP Server.
+
+    The Toolset can accept a FastMCP Client, a FastMCP Transport, or any other object which a FastMCP Transport can be created from.
+
+    See https://gofastmcp.com/clients/transports for a full list of transports available.
+    """
+
+    client: Client[Any]
+    """The FastMCP client to use."""
+
+    _: KW_ONLY
+
+    tool_error_behavior: Literal['model_retry', 'error']
+    """The behavior to take when a tool error occurs."""
+
+    max_retries: int
+    """The maximum number of retries to attempt if a tool call fails."""
+
+    _id: str | None
+
+    def __init__(
+        self,
+        client: Client[Any]
+        | ClientTransport
+        | FastMCP
+        | FastMCP1Server
+        | AnyUrl
+        | Path
+        | MCPConfig
+        | dict[str, Any]
+        | str,
+        *,
+        max_retries: int = 1,
+        tool_error_behavior: Literal['model_retry', 'error'] = 'model_retry',
+        id: str | None = None,
+    ) -> None:
+        if isinstance(client, Client):
+            self.client = client
+        else:
+            self.client = Client[Any](transport=client)
+
+        self._id = id
+        self.max_retries = max_retries
+        self.tool_error_behavior = tool_error_behavior
+
+        self._enter_lock: Lock = Lock()
+        self._running_count: int = 0
+        self._exit_stack: AsyncExitStack | None = None
+
+    @property
+    def id(self) -> str | None:
+        return self._id
+
+    async def __aenter__(self) -> Self:
+        async with self._enter_lock:
+            if self._running_count == 0:
+                self._exit_stack = AsyncExitStack()
+                await self._exit_stack.enter_async_context(self.client)
+
+            self._running_count += 1
+
+        return self
+
+    async def __aexit__(self, *args: Any) -> bool | None:
+        async with self._enter_lock:
+            self._running_count -= 1
+            if self._running_count == 0 and self._exit_stack:
+                await self._exit_stack.aclose()
+                self._exit_stack = None
+
+        return None
+
+    async def get_tools(self, ctx: RunContext[AgentDepsT]) -> dict[str, ToolsetTool[AgentDepsT]]:
+        async with self:
+            mcp_tools: list[MCPTool] = await self.client.list_tools()
+
+            return {
+                tool.name: _convert_mcp_tool_to_toolset_tool(toolset=self, mcp_tool=tool, retries=self.max_retries)
+                for tool in mcp_tools
+            }
+
+    async def call_tool(
+        self, name: str, tool_args: dict[str, Any], ctx: RunContext[AgentDepsT], tool: ToolsetTool[AgentDepsT]
+    ) -> Any:
+        async with self:
+            try:
+                call_tool_result: CallToolResult = await self.client.call_tool(name=name, arguments=tool_args)
+            except ToolError as e:
+                if self.tool_error_behavior == 'model_retry':
+                    raise ModelRetry(message=str(e)) from e
+                else:
+                    raise e
+
+        # If we have structured content, return that
+        if call_tool_result.structured_content:
+            return call_tool_result.structured_content
+
+        # Otherwise, return the content
+        return _map_fastmcp_tool_results(parts=call_tool_result.content)
+
+
+def _convert_mcp_tool_to_toolset_tool(
+    toolset: FastMCPToolset[AgentDepsT],
+    mcp_tool: MCPTool,
+    retries: int,
+) -> ToolsetTool[AgentDepsT]:
+    """Convert an MCP tool to a toolset tool."""
+    return ToolsetTool[AgentDepsT](
+        tool_def=ToolDefinition(
+            name=mcp_tool.name,
+            description=mcp_tool.description,
+            parameters_json_schema=mcp_tool.inputSchema,
+            metadata={
+                'meta': mcp_tool.meta,
+                'annotations': mcp_tool.annotations.model_dump() if mcp_tool.annotations else None,
+                'output_schema': mcp_tool.outputSchema or None,
+            },
+        ),
+        toolset=toolset,
+        max_retries=retries,
+        args_validator=TOOL_SCHEMA_VALIDATOR,
+    )
+
+
+def _map_fastmcp_tool_results(parts: list[ContentBlock]) -> list[FastMCPToolResult] | FastMCPToolResult:
+    """Map FastMCP tool results to toolset tool results."""
+    mapped_results = [_map_fastmcp_tool_result(part) for part in parts]
+
+    if len(mapped_results) == 1:
+        return mapped_results[0]
+
+    return mapped_results
+
+
+def _map_fastmcp_tool_result(part: ContentBlock) -> FastMCPToolResult:
+    if isinstance(part, TextContent):
+        return part.text
+    elif isinstance(part, ImageContent | AudioContent):
+        return messages.BinaryContent(data=base64.b64decode(part.data), media_type=part.mimeType)
+    elif isinstance(part, EmbeddedResource):
+        if isinstance(part.resource, BlobResourceContents):
+            return messages.BinaryContent(
+                data=base64.b64decode(part.resource.blob),
+                media_type=part.resource.mimeType or UNKNOWN_BINARY_MEDIA_TYPE,
+            )
+        elif isinstance(part.resource, TextResourceContents):
+            return part.resource.text
+        else:
+            assert_never(part.resource)
+    elif isinstance(part, ResourceLink):
+        # ResourceLink is not yet supported by the FastMCP toolset as reading resources is not yet supported.
+        raise NotImplementedError(
+            'ResourceLink is not supported by the FastMCP toolset as reading resources is not yet supported.'
+        )
+    else:
+        assert_never(part)

pydantic_ai_slim/pyproject.toml

@@ -88,6 +88,8 @@ cli = [
 ]
 # MCP
 mcp = ["mcp>=1.12.3"]
+# FastMCP
+fastmcp = ["fastmcp>=2.12.0"]
 # Evals
 evals = ["pydantic-evals=={{ version }}"]
 # A2A

pyproject.toml

@@ -46,7 +46,7 @@ requires-python = ">=3.10"
 
 [tool.hatch.metadata.hooks.uv-dynamic-versioning]
 dependencies = [
-    "pydantic-ai-slim[openai,vertexai,google,groq,anthropic,mistral,cohere,bedrock,huggingface,cli,mcp,evals,ag-ui,retries,temporal,logfire]=={{ version }}",
+    "pydantic-ai-slim[openai,vertexai,google,groq,anthropic,mistral,cohere,bedrock,huggingface,cli,mcp,fastmcp,evals,ag-ui,retries,temporal,logfire]=={{ version }}",
 ]
 
 [tool.hatch.metadata.hooks.uv-dynamic-versioning.optional-dependencies]
@@ -232,6 +232,7 @@ filterwarnings = [
     "ignore:unclosed <socket:ResourceWarning",
     "ignore:unclosed event loop:ResourceWarning",
 ]
+# addopts = ["--inline-snapshot=create,fix"]
 
 # https://coverage.readthedocs.io/en/latest/config.html#run
 [tool.coverage.run]