feat(mcp): add client property into MCPServer

docs/mcp/client.md

@@ -618,3 +618,46 @@ MCP elicitation supports string, number, boolean, and enum types with flat objec
 ### Security
 
 MCP Elicitation requires careful handling - servers must not request sensitive information, and clients must implement user approval controls with clear explanations. See [security considerations](https://modelcontextprotocol.io/specification/2025-06-18/client/elicitation#security-considerations) for details.
+
+## Accessing the underlying MCP client
+
+For advanced use cases, you can access the underlying `ClientSession` to manually retrieve prompts or resources from
+the MCP server. This gives you direct access to all MCP protocol methods.
+
+The [`MCPServer.client`][pydantic_ai.mcp.MCPServer.client] property provides access to the underlying `ClientSession`,
+which is only available when the server is running (within an async context manager).
+
+```python {title="manual_client_access.py"}
+from pydantic_ai.mcp import MCPServerStdio
+
+server = MCPServerStdio('python', args=['my_mcp_server.py'])
+
+async def main():
+    async with server:  # Server must be running to access client
+        # List all available prompts
+        prompts = await server.client.list_prompts()
+        print(f"Available prompts: {[p.name for p in prompts.prompts]}")
+
+        # Get a specific prompt
+        if prompts.prompts:
+            prompt = await server.client.get_prompt(
+                prompts.prompts[0].name,
+                arguments={"topic": "example"}
+            )
+            print(f"Prompt: {prompt}")
+
+        # List all available resources
+        resources = await server.client.list_resources()
+        print(f"Available resources: {[r.uri for r in resources.resources]}")
+
+        # Read a specific resource
+        if resources.resources:
+            resource = await server.client.read_resource(resources.resources[0].uri)
+            print(f"Resource content: {resource}")
+```
+
+_(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_
+
+!!! note
+    The `client` property is only available when the MCP server is running within an async context manager (`async with server:`).
+    Attempting to access it outside this context will raise an `AttributeError`.

pydantic_ai_slim/pydantic_ai/mcp.py

@@ -287,15 +287,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(
+                    self._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),
                     )
-                    self._client = await exit_stack.enter_async_context(client)
+                    self._client = await exit_stack.enter_async_context(self._client)
 
                     with anyio.fail_after(self.timeout):
                         await self._client.initialize()
@@ -307,6 +307,15 @@ async def __aenter__(self) -> Self:
             self._running_count += 1
         return self
 
+    @property
+    def client(self) -> ClientSession:
+        """Access the underlying `ClientSession`."""
+        if getattr(self, '_client', None) is None:
+            raise AttributeError(
+                f'The `{self.__class__.__name__}.client` is only instantiated when entering the async context manager.'
+            )
+        return self._client
+
     async def __aexit__(self, *args: Any) -> bool | None:
         if self._running_count == 0:
             raise ValueError('MCPServer.__aexit__ called more times than __aenter__')

tests/test_mcp.py

@@ -11,6 +11,8 @@
 
 import pytest
 from inline_snapshot import snapshot
+from inline_snapshot.extra import raises
+from pydantic import AnyUrl
 
 from pydantic_ai.agent import Agent
 from pydantic_ai.exceptions import ModelRetry, UnexpectedModelBehavior, UserError
@@ -1408,3 +1410,68 @@ def test_load_mcp_servers(tmp_path: Path):
 
     with pytest.raises(FileNotFoundError):
         load_mcp_servers(tmp_path / 'does_not_exist.json')
+
+
+async def test_mcp_server_access_client(mcp_server: MCPServerStdio):
+    with raises(
+        snapshot(
+            'AttributeError: The `MCPServerStdio.client` is only instantiated when entering the async context manager.'
+        )
+    ):
+        mcp_server.client
+
+    async with mcp_server:
+        resources = await mcp_server.client.list_resources()
+        assert resources.model_dump() == snapshot(
+            {
+                'meta': None,
+                'nextCursor': None,
+                'resources': [
+                    {
+                        'name': 'kiwi_resource',
+                        'title': None,
+                        'uri': AnyUrl('resource://kiwi.png'),
+                        'description': '',
+                        'mimeType': 'image/png',
+                        'size': None,
+                        'annotations': None,
+                        'meta': None,
+                    },
+                    {
+                        'name': 'marcelo_resource',
+                        'title': None,
+                        'uri': AnyUrl('resource://marcelo.mp3'),
+                        'description': '',
+                        'mimeType': 'audio/mpeg',
+                        'size': None,
+                        'annotations': None,
+                        'meta': None,
+                    },
+                    {
+                        'name': 'product_name_resource',
+                        'title': None,
+                        'uri': AnyUrl('resource://product_name.txt'),
+                        'description': '',
+                        'mimeType': 'text/plain',
+                        'size': None,
+                        'annotations': None,
+                        'meta': None,
+                    },
+                ],
+            }
+        )
+
+        resource = await mcp_server.client.read_resource(AnyUrl('resource://product_name.txt'))
+        assert resource.model_dump() == snapshot(
+            {
+                'meta': None,
+                'contents': [
+                    {
+                        'uri': AnyUrl('resource://product_name.txt'),
+                        'mimeType': 'text/plain',
+                        'meta': None,
+                        'text': 'Pydantic AI\n',
+                    }
+                ],
+            }
+        )