feat: allow custom clientInfo when connecting to mcp servers

docs/mcp/client.md

@@ -473,6 +473,43 @@ 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 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 import Agent
+from pydantic_ai.mcp import MCPServerSSE
+
+server = MCPServerSSE(
+    'http://localhost:3001/sse',
+    client_info=mcp_types.Implementation(  # (1)!
+        name='MyApplication',
+        version='2.1.0',
+    ),
+)
+agent = Agent('openai:gpt-5', toolsets=[server])
+
+
+async def main():
+    result = await agent.run('How many days between 2000-01-01 and 2025-03-18?')
+    print(result.output)
+    #> There are 9,208 days between January 1, 2000, and March 18, 2025.
+```
+
+The `client_info` parameter is sent to the MCP server as part of the `clientInfo` during initialization.
+
+When `client_info` is provided, it's sent to the server as an [Implementation](https://modelcontextprotocol.io/specification/2025-11-25/schema#implementation) object. If `client_info` is not specified, no client information is sent.
+
 ## MCP Sampling
 
 !!! info "What is MCP Sampling?"

pydantic_ai_slim/pydantic_ai/mcp.py

@@ -346,6 +346,7 @@ def __init__(
         elicitation_callback: ElicitationFnT | None = None,
         *,
         id: str | None = None,
+        client_info: mcp_types.Implementation | None = None,
     ):
         self.tool_prefix = tool_prefix
         self.log_level = log_level
@@ -357,6 +358,7 @@ def __init__(
         self.sampling_model = sampling_model
         self.max_retries = max_retries
         self.elicitation_callback = elicitation_callback
+        self.client_info = client_info
 
         self._id = id or tool_prefix
 
@@ -621,13 +623,15 @@ 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,
                         sampling_callback=self._sampling_callback if self.allow_sampling else None,
                         elicitation_callback=self.elicitation_callback,
                         logging_callback=self.log_handler,
                         read_timeout_seconds=timedelta(seconds=self.read_timeout),
+                        client_info=self.client_info,
                     )
                     self._client = await exit_stack.enter_async_context(client)
 
@@ -795,6 +799,7 @@ def __init__(
         max_retries: int = 1,
         elicitation_callback: ElicitationFnT | None = None,
         id: str | None = None,
+        client_info: mcp_types.Implementation | None = None,
     ):
         """Build a new MCP server.
 
@@ -814,6 +819,7 @@ def __init__(
             max_retries: The maximum number of times to retry a tool call.
             elicitation_callback: Callback function to handle elicitation requests from the server.
             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
@@ -832,6 +838,7 @@ def __init__(
             max_retries,
             elicitation_callback,
             id=id,
+            client_info=client_info,
         )
 
     @classmethod
@@ -948,6 +955,7 @@ def __init__(
         sampling_model: models.Model | None = None,
         max_retries: int = 1,
         elicitation_callback: ElicitationFnT | None = None,
+        client_info: mcp_types.Implementation | None = None,
         **_deprecated_kwargs: Any,
     ):
         """Build a new MCP server.
@@ -967,6 +975,7 @@ def __init__(
             sampling_model: The model to use for sampling.
             max_retries: The maximum number of times to retry a tool call.
             elicitation_callback: Callback function to handle elicitation requests from the server.
+            client_info: Information describing the MCP client implementation.
         """
         if 'sse_read_timeout' in _deprecated_kwargs:
             if read_timeout is not None:
@@ -998,6 +1007,7 @@ def __init__(
             max_retries,
             elicitation_callback,
             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(18)
+        assert len(tools) == snapshot(19)
         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(18)
+        assert len(tools) == snapshot(19)
 
 
 async def test_process_tool_call(run_context: RunContext[int]) -> int:
@@ -1981,6 +1982,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.')