simplification

Author: DouweM

docs/mcp/fastmcp-client.md

@@ -19,7 +19,7 @@ pip/uv-add "pydantic-ai-slim[fastmcp]"
 A `FastMCPToolset` can then be created from:
 
 - A FastMCP Server: `#!python FastMCPToolset(fastmcp.FastMCP('my_server'))`
-- A FastMCP Client: `#!python FastMCPToolset(client=fastmcp.Client(...))`
+- 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')`
@@ -85,4 +85,4 @@ 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`)_
+_(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_

pydantic_ai_slim/pydantic_ai/toolsets/fastmcp.py

@@ -3,17 +3,12 @@
 import base64
 from asyncio import Lock
 from contextlib import AsyncExitStack
-from dataclasses import KW_ONLY, dataclass, field
+from dataclasses import KW_ONLY, dataclass
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Literal, overload
+from typing import TYPE_CHECKING, Any, Literal
 
-from fastmcp.client.transports import ClientTransport
-from fastmcp.mcp_config import MCPConfig
-from fastmcp.server import FastMCP
-from mcp.server.fastmcp import FastMCP as FastMCP1Server
-from mcp.types import BlobResourceContents, EmbeddedResource, ResourceLink
 from pydantic import AnyUrl
-from typing_extensions import Self
+from typing_extensions import Self, assert_never
 
 from pydantic_ai import messages
 from pydantic_ai.exceptions import ModelRetry
@@ -23,12 +18,20 @@
 
 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,
     )
 
@@ -52,7 +55,7 @@
 UNKNOWN_BINARY_MEDIA_TYPE = 'application/octet-stream'
 
 
-@dataclass
+@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.
 
@@ -62,73 +65,38 @@ class FastMCPToolset(AbstractToolset[AgentDepsT]):
     """
 
     client: Client[Any]
-    """The FastMCP transport to use. This can be a local or remote MCP Server configuration, a transport string, or a FastMCP Client."""
+    """The FastMCP client to use."""
 
     _: KW_ONLY
 
-    tool_error_behavior: Literal['model_retry', 'error'] = field(default='error')
+    tool_error_behavior: Literal['model_retry', 'error']
     """The behavior to take when a tool error occurs."""
 
-    max_retries: int = field(default=2)
+    max_retries: int
     """The maximum number of retries to attempt if a tool call fails."""
 
-    _id: str | None = field(default=None)
+    _id: str | None
 
-    @overload
     def __init__(
         self,
-        *,
-        client: Client[Any],
-        max_retries: int = 2,
-        tool_error_behavior: Literal['model_retry', 'error'] = 'error',
-        id: str | None = None,
-    ) -> None: ...
-
-    @overload
-    def __init__(
-        self,
-        transport: ClientTransport
-        | FastMCP
-        | FastMCP1Server
-        | AnyUrl
-        | Path
-        | MCPConfig
-        | dict[str, Any]
-        | str
-        | None = None,
-        *,
-        max_retries: int = 2,
-        tool_error_behavior: Literal['model_retry', 'error'] = 'error',
-        id: str | None = None,
-    ) -> None: ...
-
-    def __init__(
-        self,
-        transport: ClientTransport
+        client: Client[Any]
+        | ClientTransport
         | FastMCP
         | FastMCP1Server
         | AnyUrl
         | Path
         | MCPConfig
         | dict[str, Any]
-        | str
-        | None = None,
+        | str,
         *,
-        client: Client[Any] | None = None,
-        max_retries: int = 2,
+        max_retries: int = 1,
         tool_error_behavior: Literal['model_retry', 'error'] = 'model_retry',
         id: str | None = None,
     ) -> None:
-        if not client and not transport:
-            raise ValueError('Either client or transport must be provided')
-
-        if client and transport:
-            raise ValueError('Either client or transport must be provided, not both')
-
-        if client:
+        if isinstance(client, Client):
             self.client = client
         else:
-            self.client = Client[Any](transport=transport)
+            self.client = Client[Any](transport=client)
 
         self._id = id
         self.max_retries = max_retries
@@ -144,7 +112,7 @@ def id(self) -> str | None:
 
     async def __aenter__(self) -> Self:
         async with self._enter_lock:
-            if self._running_count == 0 and self.client:
+            if self._running_count == 0:
                 self._exit_stack = AsyncExitStack()
                 await self._exit_stack.enter_async_context(self.client)
 
@@ -226,25 +194,22 @@ def _map_fastmcp_tool_results(parts: list[ContentBlock]) -> list[FastMCPToolResu
 def _map_fastmcp_tool_result(part: ContentBlock) -> FastMCPToolResult:
     if isinstance(part, TextContent):
         return part.text
-
-    if isinstance(part, ImageContent | AudioContent):
+    elif isinstance(part, ImageContent | AudioContent):
         return messages.BinaryContent(data=base64.b64decode(part.data), media_type=part.mimeType)
-
-    if isinstance(part, EmbeddedResource):
+    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,
             )
-
-        # If not a BlobResourceContents, it's a TextResourceContents
-        return part.resource.text
-
-    if isinstance(part, ResourceLink):
+        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.'
         )
-
-    msg = f'Unsupported/Unknown content block type: {type(part)}'  # pragma: no cover
-    raise ValueError(msg)  # pragma: no cover)
+    else:
+        assert_never(part)

tests/test_fastmcp.py

@@ -158,42 +158,30 @@ class TestFastMCPToolsetInitialization:
 
     async def test_init_with_client(self, fastmcp_client: Client[FastMCPTransport]):
         """Test initialization with a FastMCP client."""
-        toolset = FastMCPToolset(client=fastmcp_client)
+        toolset = FastMCPToolset(fastmcp_client)
 
         # Test that the client is accessible via the property
         assert toolset.id is None
 
     async def test_init_with_id(self, fastmcp_client: Client[FastMCPTransport]):
         """Test initialization with an id."""
-        toolset = FastMCPToolset(client=fastmcp_client, id='test_id')
+        toolset = FastMCPToolset(fastmcp_client, id='test_id')
 
         # Test that the client is accessible via the property
         assert toolset.id == 'test_id'
 
     async def test_init_with_custom_retries_and_error_behavior(self, fastmcp_client: Client[FastMCPTransport]):
         """Test initialization with custom retries and error behavior."""
-        toolset = FastMCPToolset(client=fastmcp_client, max_retries=5, tool_error_behavior='model_retry')
+        toolset = FastMCPToolset(fastmcp_client, max_retries=5, tool_error_behavior='model_retry')
 
         # Test that the toolset was created successfully
         assert toolset.client is fastmcp_client
 
     async def test_id_property(self, fastmcp_client: Client[FastMCPTransport]):
         """Test that the id property returns None."""
-        toolset = FastMCPToolset(client=fastmcp_client)
+        toolset = FastMCPToolset(fastmcp_client)
         assert toolset.id is None
 
-    async def test_init_without_client_or_transport(self):
-        """Test initialization without a client or transport."""
-        with pytest.raises(ValueError, match='Either client or transport must be provided'):
-            FastMCPToolset()
-
-    async def test_init_with_client_and_transport(self):
-        """Test initialization with a client and transport."""
-        with pytest.raises(ValueError, match='Either client or transport must be provided, not both'):
-            tmp_server = FastMCP('tmp_server')
-            client = Client(transport=tmp_server)
-            FastMCPToolset(client=client, transport=tmp_server)  # pyright: ignore[reportCallIssue]
-
 
 class TestFastMCPToolsetContextManagement:
     """Test FastMCP Toolset context management."""
@@ -202,7 +190,7 @@ async def test_context_manager_single_enter_exit(
         self, fastmcp_client: Client[FastMCPTransport], run_context: RunContext[None]
     ):
         """Test single enter/exit cycle."""
-        toolset = FastMCPToolset(client=fastmcp_client)
+        toolset = FastMCPToolset(fastmcp_client)
 
         async with toolset:
             # Test that we can get tools when the context is active
@@ -216,7 +204,7 @@ async def test_context_manager_no_enter(
         self, fastmcp_client: Client[FastMCPTransport], run_context: RunContext[None]
     ):
         """Test no enter/exit cycle."""
-        toolset = FastMCPToolset(client=fastmcp_client)
+        toolset = FastMCPToolset(fastmcp_client)
 
         # Test that we can get tools when the context is not active
         tools = await toolset.get_tools(run_context)
@@ -227,7 +215,7 @@ async def test_context_manager_nested_enter_exit(
         self, fastmcp_client: Client[FastMCPTransport], run_context: RunContext[None]
     ):
         """Test nested enter/exit cycles."""
-        toolset = FastMCPToolset(client=fastmcp_client)
+        toolset = FastMCPToolset(fastmcp_client)
 
         async with toolset:
             tools1 = await toolset.get_tools(run_context)
@@ -248,7 +236,7 @@ async def test_get_tools(
         run_context: RunContext[None],
     ):
         """Test getting tools from the FastMCP client."""
-        toolset = FastMCPToolset(client=fastmcp_client)
+        toolset = FastMCPToolset(fastmcp_client)
 
         async with toolset:
             tools = await toolset.get_tools(run_context)
@@ -275,7 +263,7 @@ async def test_get_tools(
             assert test_tool.tool_def.name == 'test_tool'
             assert test_tool.tool_def.description is not None
             assert 'test tool that returns a formatted string' in test_tool.tool_def.description
-            assert test_tool.max_retries == 2
+            assert test_tool.max_retries == 1
             assert test_tool.toolset is toolset
 
             # Check that the tool has proper schema
@@ -288,7 +276,7 @@ async def test_get_tools_with_empty_server(self, run_context: RunContext[None]):
         """Test getting tools from an empty FastMCP server."""
         empty_server = FastMCP('empty_server')
         empty_client = Client(transport=empty_server)
-        toolset = FastMCPToolset(client=empty_client)
+        toolset = FastMCPToolset(empty_client)
 
         async with toolset:
             tools = await toolset.get_tools(run_context)
@@ -301,7 +289,7 @@ class TestFastMCPToolsetToolCalling:
     @pytest.fixture
     async def fastmcp_toolset(self, fastmcp_client: Client[FastMCPTransport]) -> FastMCPToolset[None]:
         """Create a FastMCP Toolset."""
-        return FastMCPToolset(client=fastmcp_client)
+        return FastMCPToolset(fastmcp_client)
 
     async def test_call_tool_success(
         self,
@@ -491,7 +479,7 @@ async def test_call_tool_with_error_behavior_raise(
         run_context: RunContext[None],
     ):
         """Test tool call with error behavior set to raise."""
-        toolset = FastMCPToolset(client=fastmcp_client, tool_error_behavior='error')
+        toolset = FastMCPToolset(fastmcp_client, tool_error_behavior='error')
 
         async with toolset:
             tools = await toolset.get_tools(run_context)
@@ -506,7 +494,7 @@ async def test_call_tool_with_error_behavior_model_retry(
         run_context: RunContext[None],
     ):
         """Test tool call with error behavior set to model retry."""
-        toolset = FastMCPToolset(client=fastmcp_client, tool_error_behavior='model_retry')
+        toolset = FastMCPToolset(fastmcp_client, tool_error_behavior='model_retry')
 
         async with toolset:
             tools = await toolset.get_tools(run_context)
@@ -577,7 +565,7 @@ async def test_transports(self):
     @pytest.mark.parametrize(
         'invalid_transport', ['tomato_is_not_a_valid_transport', '/path/to/server.ini', 'ftp://localhost']
     )
-    async def test_invalid_transports_uninferrable(self, invalid_transport: str | None):
+    async def test_invalid_transports_uninferrable(self, invalid_transport: str):
         """Test creating toolset from invalid transports."""
         with pytest.raises(ValueError, match='Could not infer a valid transport from:'):
             FastMCPToolset(invalid_transport)