Allow custom clientInfo when connecting to MCP servers (#3572)

Author: atinylittleshell

docs/mcp/client.md

@@ -473,6 +473,31 @@ async def main():
    request. Anything supported by **httpx** (`verify`, `cert`, custom
    proxies, timeouts, etc.) therefore applies to all MCP traffic.
 
+## Client Identification
+
+When connecting to an MCP server, you can optionally specify an [Implementation](https://modelcontextprotocol.io/specification/2025-11-25/schema#implementation) object as client information that will be sent to the server during initialization. This is useful for:
+
+- Identifying your application in server logs
+- Allowing servers to provide custom behavior based on the client
+- Debugging and monitoring MCP connections
+- Version-specific feature negotiation
+
+All MCP client classes ([`MCPServerStdio`][pydantic_ai.mcp.MCPServerStdio], [`MCPServerStreamableHTTP`][pydantic_ai.mcp.MCPServerStreamableHTTP], and [`MCPServerSSE`][pydantic_ai.mcp.MCPServerSSE]) support the `client_info` parameter:
+
+```python {title="mcp_client_with_name.py"}
+from mcp import types as mcp_types
+
+from pydantic_ai.mcp import MCPServerSSE
+
+server = MCPServerSSE(
+    'http://localhost:3001/sse',
+    client_info=mcp_types.Implementation(
+        name='MyApplication',
+        version='2.1.0',
+    ),
+)
+```
+
 ## MCP Sampling
 
 !!! info "What is MCP Sampling?"

pydantic_ai_slim/pydantic_ai/mcp.py

@@ -387,6 +387,7 @@ def __init__(
         cache_resources: bool = True,
         *,
         id: str | None = None,
+        client_info: mcp_types.Implementation | None = None,
     ):
         self.tool_prefix = tool_prefix
         self.log_level = log_level
@@ -400,6 +401,7 @@ def __init__(
         self.elicitation_callback = elicitation_callback
         self.cache_tools = cache_tools
         self.cache_resources = cache_resources
+        self.client_info = client_info
 
         self._id = id or tool_prefix
 
@@ -685,6 +687,7 @@ async def __aenter__(self) -> Self:
             if self._running_count == 0:
                 async with AsyncExitStack() as exit_stack:
                     self._read_stream, self._write_stream = await exit_stack.enter_async_context(self.client_streams())
+
                     client = ClientSession(
                         read_stream=self._read_stream,
                         write_stream=self._write_stream,
@@ -693,6 +696,7 @@ async def __aenter__(self) -> Self:
                         logging_callback=self.log_handler,
                         read_timeout_seconds=timedelta(seconds=self.read_timeout),
                         message_handler=self._handle_notification,
+                        client_info=self.client_info,
                     )
                     self._client = await exit_stack.enter_async_context(client)
 
@@ -879,6 +883,7 @@ def __init__(
         cache_tools: bool = True,
         cache_resources: bool = True,
         id: str | None = None,
+        client_info: mcp_types.Implementation | None = None,
     ):
         """Build a new MCP server.
 
@@ -902,6 +907,7 @@ def __init__(
             cache_resources: Whether to cache the list of resources.
                 See [`MCPServer.cache_resources`][pydantic_ai.mcp.MCPServer.cache_resources].
             id: 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: Information describing the MCP client implementation.
         """
         self.command = command
         self.args = args
@@ -922,6 +928,7 @@ def __init__(
             cache_tools,
             cache_resources,
             id=id,
+            client_info=client_info,
         )
 
     @classmethod
@@ -1042,6 +1049,7 @@ def __init__(
         elicitation_callback: ElicitationFnT | None = None,
         cache_tools: bool = True,
         cache_resources: bool = True,
+        client_info: mcp_types.Implementation | None = None,
         **_deprecated_kwargs: Any,
     ):
         """Build a new MCP server.
@@ -1065,6 +1073,7 @@ def __init__(
                 See [`MCPServer.cache_tools`][pydantic_ai.mcp.MCPServer.cache_tools].
             cache_resources: Whether to cache the list of resources.
                 See [`MCPServer.cache_resources`][pydantic_ai.mcp.MCPServer.cache_resources].
+            client_info: Information describing the MCP client implementation.
         """
         if 'sse_read_timeout' in _deprecated_kwargs:
             if read_timeout is not None:
@@ -1098,6 +1107,7 @@ def __init__(
             cache_tools,
             cache_resources,
             id=id,
+            client_info=client_info,
         )
 
     @property

tests/mcp_server.py

@@ -223,6 +223,25 @@ class UserResponse(BaseModel):
     response: str
 
 
+@mcp.tool()
+async def get_client_info(ctx: Context[ServerSession, None]) -> dict[str, Any] | None:
+    """Get information about the connected MCP client.
+
+    Returns:
+        Dictionary with client info (name, version, etc.) or None if not available.
+    """
+    client_params = ctx.session.client_params
+    if client_params is None:
+        return None
+    client_info = client_params.clientInfo
+    return {
+        'name': client_info.name,
+        'version': client_info.version,
+        'title': getattr(client_info, 'title', None),
+        'websiteUrl': getattr(client_info, 'websiteUrl', None),
+    }
+
+
 @mcp.tool()
 async def use_elicitation(ctx: Context[ServerSession, None], question: str) -> str:
     """Use elicitation callback to ask the user a question."""

tests/test_mcp.py

@@ -54,6 +54,7 @@
         ElicitRequestParams,
         ElicitResult,
         ImageContent,
+        Implementation,
         TextContent,
     )
 
@@ -95,7 +96,7 @@ async def test_stdio_server(run_context: RunContext[int]):
     server = MCPServerStdio('python', ['-m', 'tests.mcp_server'])
     async with server:
         tools = [tool.tool_def for tool in (await server.get_tools(run_context)).values()]
-        assert len(tools) == snapshot(19)
+        assert len(tools) == snapshot(20)
         assert tools[0].name == 'celsius_to_fahrenheit'
         assert isinstance(tools[0].description, str)
         assert tools[0].description.startswith('Convert Celsius to Fahrenheit.')
@@ -156,7 +157,7 @@ async def test_stdio_server_with_cwd(run_context: RunContext[int]):
     server = MCPServerStdio('python', ['mcp_server.py'], cwd=test_dir)
     async with server:
         tools = await server.get_tools(run_context)
-        assert len(tools) == snapshot(19)
+        assert len(tools) == snapshot(20)
 
 
 async def test_process_tool_call(run_context: RunContext[int]) -> int:
@@ -2053,6 +2054,38 @@ async def test_instructions(mcp_server: MCPServerStdio) -> None:
         assert mcp_server.instructions == 'Be a helpful assistant.'
 
 
+async def test_client_info_passed_to_session() -> None:
+    """Test that provided client_info is passed unchanged to ClientSession."""
+    implementation = Implementation(
+        name='MyCustomClient',
+        version='2.5.3',
+        title='Custom MCP client',
+        websiteUrl='https://example.com/client',
+    )
+    server = MCPServerStdio('python', ['-m', 'tests.mcp_server'], client_info=implementation)
+
+    async with server:
+        result = await server.direct_call_tool('get_client_info', {})
+        assert result == {
+            'name': 'MyCustomClient',
+            'version': '2.5.3',
+            'title': 'Custom MCP client',
+            'websiteUrl': 'https://example.com/client',
+        }
+
+
+async def test_client_info_not_set() -> None:
+    """Test that when client_info is not set, the default MCP client info is used."""
+    server = MCPServerStdio('python', ['-m', 'tests.mcp_server'])
+
+    async with server:
+        result = await server.direct_call_tool('get_client_info', {})
+        # When client_info is not set, the MCP library provides default client info
+        assert result is not None
+        assert isinstance(result, dict)
+        assert result['name'] == 'mcp'
+
+
 async def test_agent_run_stream_with_mcp_server_http(allow_model_requests: None, model: Model):
     server = MCPServerStreamableHTTP(url='https://mcp.deepwiki.com/mcp', timeout=30)
     agent = Agent(model, toolsets=[server], instructions='Be concise.')