feat: allow custom clientInfo when connecting to mcp servers

docs/mcp/client.md

@@ -473,6 +473,40 @@ 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 a client name and version 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_name` and `client_version` parameters:
+
+```python {title="mcp_client_with_name.py"}
+from pydantic_ai import Agent
+from pydantic_ai.mcp import MCPServerStdio
+
+server = MCPServerStdio(
+    'uv',
+    args=['run', 'mcp-run-python', 'stdio'],
+    client_name='MyApplication',  # (1)!
+    client_version='2.1.0',  # (2)!
+)
+agent = Agent('openai:gpt-5', toolsets=[server])
+
+async def main():
+    result = await agent.run('What is 2 + 2?')
+    print(result.output)
+    #> The answer is 4.
+```
+
+1. The `client_name` parameter is sent to the MCP server as part of the `clientInfo` during initialization.
+2. The `client_version` parameter specifies the version string. Defaults to `'1.0.0'` if not provided.
+
+When `client_name` is provided, it's sent to the server as an [Implementation](https://modelcontextprotocol.io/specification/2025-11-25/schema#implementation) object with the specified version (or `'1.0.0'` by default). If `client_name` 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,8 @@ def __init__(
         elicitation_callback: ElicitationFnT | None = None,
         *,
         id: str | None = None,
+        client_name: str | None = None,
+        client_version: str = '1.0.0',
     ):
         self.tool_prefix = tool_prefix
         self.log_level = log_level
@@ -357,6 +359,8 @@ def __init__(
         self.sampling_model = sampling_model
         self.max_retries = max_retries
         self.elicitation_callback = elicitation_callback
+        self.client_name = client_name
+        self.client_version = client_version
 
         self._id = id or tool_prefix
 
@@ -621,13 +625,23 @@ 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())
+
+                    # Build client_info if client_name is provided
+                    client_info = None
+                    if self.client_name:
+                        client_info = mcp_types.Implementation(
+                            name=self.client_name,
+                            version=self.client_version,
+                        )
+
                     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=client_info,
                     )
                     self._client = await exit_stack.enter_async_context(client)
 
@@ -795,6 +809,8 @@ def __init__(
         max_retries: int = 1,
         elicitation_callback: ElicitationFnT | None = None,
         id: str | None = None,
+        client_name: str | None = None,
+        client_version: str = '1.0.0',
     ):
         """Build a new MCP server.
 
@@ -814,6 +830,8 @@ 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_name: An optional client name to send to the MCP server in the clientInfo during initialization.
+            client_version: The version string to send with the client name. Defaults to '1.0.0'.
         """
         self.command = command
         self.args = args
@@ -832,6 +850,8 @@ def __init__(
             max_retries,
             elicitation_callback,
             id=id,
+            client_name=client_name,
+            client_version=client_version,
         )
 
     @classmethod
@@ -948,6 +968,8 @@ def __init__(
         sampling_model: models.Model | None = None,
         max_retries: int = 1,
         elicitation_callback: ElicitationFnT | None = None,
+        client_name: str | None = None,
+        client_version: str = '1.0.0',
         **_deprecated_kwargs: Any,
     ):
         """Build a new MCP server.
@@ -967,6 +989,8 @@ 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_name: An optional client name to send to the MCP server in the clientInfo during initialization.
+            client_version: The version string to send with the client name. Defaults to '1.0.0'.
         """
         if 'sse_read_timeout' in _deprecated_kwargs:
             if read_timeout is not None:
@@ -998,6 +1022,8 @@ def __init__(
             max_retries,
             elicitation_callback,
             id=id,
+            client_name=client_name,
+            client_version=client_version,
         )
 
     @property

tests/test_mcp.py

@@ -54,6 +54,7 @@
         ElicitRequestParams,
         ElicitResult,
         ImageContent,
+        Implementation,
         TextContent,
     )
 
@@ -1981,6 +1982,89 @@ async def test_instructions(mcp_server: MCPServerStdio) -> None:
         assert mcp_server.instructions == 'Be a helpful assistant.'
 
 
+async def test_client_name_passed_to_session() -> None:
+    """Test that client_name is passed to ClientSession as clientInfo."""
+    server = MCPServerStdio('python', ['-m', 'tests.mcp_server'], client_name='MyCustomClient')
+
+    with patch.object(ClientSession, '__init__', return_value=None) as mock_init:
+        with patch.object(ClientSession, '__aenter__', new_callable=AsyncMock) as mock_aenter:
+            with patch.object(ClientSession, '__aexit__', new_callable=AsyncMock):
+                with patch.object(ClientSession, 'initialize', new_callable=AsyncMock) as mock_initialize:
+                    # Mock the initialize response
+                    mock_initialize.return_value = AsyncMock(
+                        serverInfo=Implementation(name='test', version='1.0.0'),
+                        capabilities=ServerCapabilities(tools=True),
+                        instructions=None,
+                    )
+                    mock_aenter.return_value = AsyncMock(initialize=mock_initialize)
+
+                    async with server:
+                        # Check that ClientSession.__init__ was called with client_info
+                        assert mock_init.called
+                        call_kwargs = mock_init.call_args.kwargs
+
+                        # Verify client_info was passed
+                        assert 'client_info' in call_kwargs
+                        client_info = call_kwargs['client_info']
+                        assert client_info is not None
+                        assert client_info.name == 'MyCustomClient'
+                        assert client_info.version == '1.0.0'
+
+
+async def test_client_name_and_version_passed_to_session() -> None:
+    """Test that client_name and client_version are passed to ClientSession as clientInfo."""
+    server = MCPServerStdio('python', ['-m', 'tests.mcp_server'], client_name='MyCustomClient', client_version='2.5.3')
+
+    with patch.object(ClientSession, '__init__', return_value=None) as mock_init:
+        with patch.object(ClientSession, '__aenter__', new_callable=AsyncMock) as mock_aenter:
+            with patch.object(ClientSession, '__aexit__', new_callable=AsyncMock):
+                with patch.object(ClientSession, 'initialize', new_callable=AsyncMock) as mock_initialize:
+                    # Mock the initialize response
+                    mock_initialize.return_value = AsyncMock(
+                        serverInfo=Implementation(name='test', version='1.0.0'),
+                        capabilities=ServerCapabilities(tools=True),
+                        instructions=None,
+                    )
+                    mock_aenter.return_value = AsyncMock(initialize=mock_initialize)
+
+                    async with server:
+                        # Check that ClientSession.__init__ was called with client_info
+                        assert mock_init.called
+                        call_kwargs = mock_init.call_args.kwargs
+
+                        # Verify client_info was passed with custom version
+                        assert 'client_info' in call_kwargs
+                        client_info = call_kwargs['client_info']
+                        assert client_info is not None
+                        assert client_info.name == 'MyCustomClient'
+                        assert client_info.version == '2.5.3'
+
+
+async def test_client_name_not_set() -> None:
+    """Test that when client_name is not set, client_info is None."""
+    server = MCPServerStdio('python', ['-m', 'tests.mcp_server'])
+
+    with patch.object(ClientSession, '__init__', return_value=None) as mock_init:
+        with patch.object(ClientSession, '__aenter__', new_callable=AsyncMock) as mock_aenter:
+            with patch.object(ClientSession, '__aexit__', new_callable=AsyncMock):
+                with patch.object(ClientSession, 'initialize', new_callable=AsyncMock) as mock_initialize:
+                    # Mock the initialize response
+                    mock_initialize.return_value = AsyncMock(
+                        serverInfo=Implementation(name='test', version='1.0.0'),
+                        capabilities=ServerCapabilities(tools=True),
+                        instructions=None,
+                    )
+                    mock_aenter.return_value = AsyncMock(initialize=mock_initialize)
+
+                    async with server:
+                        # Check that ClientSession.__init__ was called with client_info=None
+                        assert mock_init.called
+                        call_kwargs = mock_init.call_args.kwargs
+
+                        assert 'client_info' in call_kwargs
+                        assert call_kwargs['client_info'] is None
+
+
 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.')