feat: add ergonomic Client class for testing MCP servers

Add a high-level Client class that wraps ClientSession with transport
management, making it easier to test MCP servers.

Key changes:
- Add Client class with async context manager support
- Add InMemoryTransport for in-memory server connections
- Migrate test files to use new Client(server) API
- Update testing documentation with Client usage examples

Usage:
    async with Client(server) as client:
        result = await client.call_tool('my_tool', {'arg': 'value'})

Author: felixweinberger

docs/testing.md

@@ -1,10 +1,11 @@
 # Testing MCP Servers
 
-If you call yourself a developer, you will want to test your MCP server.
-The Python SDK offers the `create_connected_server_and_client_session` function to create a session
-using an in-memory transport. I know, I know, the name is too long... We are working on improving it.
+The Python SDK provides a `Client` class for testing MCP servers with an in-memory transport.
+This makes it easy to write tests without network overhead.
 
-Anyway, let's assume you have a simple server with a single tool:
+## Basic Usage
+
+Let's assume you have a simple server with a single tool:
 
 ```python title="server.py"
 from mcp.server import FastMCP
@@ -40,12 +41,9 @@ To run the below test, you'll need to install the following dependencies:
     server - you don't need to use it, but we are spreading the word for best practices.
 
 ```python title="test_server.py"
-from collections.abc import AsyncGenerator
-
 import pytest
 from inline_snapshot import snapshot
-from mcp.client.session import ClientSession
-from mcp.shared.memory import create_connected_server_and_client_session
+from mcp import Client
 from mcp.types import CallToolResult, TextContent
 
 from server import app
@@ -57,14 +55,14 @@ def anyio_backend():  # (1)!
 
 
 @pytest.fixture
-async def client_session() -> AsyncGenerator[ClientSession]:
-    async with create_connected_server_and_client_session(app, raise_exceptions=True) as _session:
-        yield _session
+async def client():  # (2)!
+    async with Client(app, raise_exceptions=True) as c:
+        yield c
 
 
 @pytest.mark.anyio
-async def test_call_add_tool(client_session: ClientSession):
-    result = await client_session.call_tool("add", {"a": 1, "b": 2})
+async def test_call_add_tool(client: Client):
+    result = await client.call_tool("add", {"a": 1, "b": 2})
     assert result == snapshot(
         CallToolResult(
             content=[TextContent(type="text", text="3")],
@@ -74,5 +72,6 @@ async def test_call_add_tool(client_session: ClientSession):
 ```
 
 1. If you are using `trio`, you should set `"trio"` as the `anyio_backend`. Check more information in the [anyio documentation](https://anyio.readthedocs.io/en/stable/testing.html#specifying-the-backends-to-run-on).
+2. The `client` fixture creates a connected client that can be reused across multiple tests.
 
 There you go! You can now extend your tests to cover more scenarios.

examples/fastmcp/weather_structured.py

@@ -14,8 +14,8 @@
 
 from pydantic import BaseModel, Field
 
+from mcp.client import Client
 from mcp.server.fastmcp import FastMCP
-from mcp.shared.memory import create_connected_server_and_client_session as client_session
 
 # Create server
 mcp = FastMCP("Weather Service")
@@ -157,7 +157,7 @@ async def test() -> None:
         print("Testing Weather Service Tools (via MCP protocol)\n")
         print("=" * 80)
 
-        async with client_session(mcp._mcp_server) as client:
+        async with Client(mcp) as client:
             # Test get_weather
             result = await client.call_tool("get_weather", {"city": "London"})
             print("\nWeather in London:")

src/mcp/__init__.py

@@ -1,3 +1,4 @@
+from .client.client import Client
 from .client.session import ClientSession
 from .client.session_group import ClientSessionGroup
 from .client.stdio import StdioServerParameters, stdio_client
@@ -66,6 +67,7 @@
 
 __all__ = [
     "CallToolRequest",
+    "Client",
     "ClientCapabilities",
     "ClientNotification",
     "ClientRequest",

src/mcp/client/__init__.py

@@ -0,0 +1,9 @@
+"""MCP Client module."""
+
+from mcp.client.client import Client
+from mcp.client.session import ClientSession
+
+__all__ = [
+    "Client",
+    "ClientSession",
+]

src/mcp/client/_memory.py

@@ -0,0 +1,97 @@
+"""In-memory transport for testing MCP servers without network overhead."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+from typing import Any
+
+import anyio
+from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+
+from mcp.server import Server
+from mcp.server.fastmcp import FastMCP
+from mcp.shared.memory import create_client_server_memory_streams
+from mcp.shared.message import SessionMessage
+
+
+class InMemoryTransport:
+    """
+    In-memory transport for testing MCP servers without network overhead.
+
+    This transport starts the server in a background task and provides
+    streams for client-side communication. The server is automatically
+    stopped when the context manager exits.
+
+    Example:
+        server = FastMCP("test")
+        transport = InMemoryTransport(server)
+
+        async with transport.connect() as (read_stream, write_stream):
+            async with ClientSession(read_stream, write_stream) as session:
+                await session.initialize()
+                # Use the session...
+
+    Or more commonly, use with Client:
+        async with Client(server) as client:
+            result = await client.call_tool("my_tool", {...})
+    """
+
+    def __init__(
+        self,
+        server: Server[Any] | FastMCP,
+        *,
+        raise_exceptions: bool = False,
+    ) -> None:
+        """
+        Initialize the in-memory transport.
+
+        Args:
+            server: The MCP server to connect to (Server or FastMCP instance)
+            raise_exceptions: Whether to raise exceptions from the server
+        """
+        self._server = server
+        self._raise_exceptions = raise_exceptions
+
+    @asynccontextmanager
+    async def connect(
+        self,
+    ) -> AsyncGenerator[
+        tuple[
+            MemoryObjectReceiveStream[SessionMessage | Exception],
+            MemoryObjectSendStream[SessionMessage],
+        ],
+        None,
+    ]:
+        """
+        Connect to the server and return streams for communication.
+
+        Yields:
+            A tuple of (read_stream, write_stream) for bidirectional communication
+        """
+        # Unwrap FastMCP to get underlying Server
+        actual_server: Server[Any]
+        if isinstance(self._server, FastMCP):
+            actual_server = self._server._mcp_server  # type: ignore[reportPrivateUsage]
+        else:
+            actual_server = self._server
+
+        async with create_client_server_memory_streams() as (client_streams, server_streams):
+            client_read, client_write = client_streams
+            server_read, server_write = server_streams
+
+            async with anyio.create_task_group() as tg:
+                # Start server in background
+                tg.start_soon(
+                    lambda: actual_server.run(
+                        server_read,
+                        server_write,
+                        actual_server.create_initialization_options(),
+                        raise_exceptions=self._raise_exceptions,
+                    )
+                )
+
+                try:
+                    yield client_read, client_write
+                finally:
+                    tg.cancel_scope.cancel()