Refactor to InMemoryTransport class and delete create_connected_server_and_client_session

Breaking change for v2:
- Replace create_connected_server_and_client_session() function with InMemoryTransport class
- Add Client.from_server() for ergonomic testing
- Migrate all tests to use new API patterns
- Update stream spy fixture to patch both module locations

High-level tests now use:
  async with Client.from_server(server) as client:
      result = await client.call_tool(...)

Low-level tests use:
  transport = InMemoryTransport(server)
  async with transport.connect() as (read, write):
      async with ClientSession(read, write) as session:
          ...

Github-Issue:#1728

Author: felixweinberger

src/mcp/client/__init__.py

@@ -2,10 +2,10 @@
 
 from mcp.client.client import Client
 from mcp.client.session import ClientSession
-from mcp.client.transports.memory import create_in_memory_transport
+from mcp.client.transports.memory import InMemoryTransport
 
 __all__ = [
     "Client",
     "ClientSession",
-    "create_in_memory_transport",
+    "InMemoryTransport",
 ]

src/mcp/client/client.py

@@ -19,7 +19,7 @@
     MessageHandlerFnT,
     SamplingFnT,
 )
-from mcp.client.transports.memory import create_in_memory_transport
+from mcp.client.transports.memory import InMemoryTransport
 from mcp.server import Server
 from mcp.server.fastmcp import FastMCP
 from mcp.shared.message import SessionMessage
@@ -132,9 +132,8 @@ def my_tool(arg: str) -> str:
             async with Client.from_server(server) as client:
                 result = await client.call_tool("my_tool", {"arg": "value"})
         """
-        async with create_in_memory_transport(
-            server, raise_exceptions=raise_exceptions
-        ) as (read_stream, write_stream):
+        transport = InMemoryTransport(server, raise_exceptions=raise_exceptions)
+        async with transport.connect() as (read_stream, write_stream):
             client = cls(
                 read_stream=read_stream,
                 write_stream=write_stream,

src/mcp/client/transports/__init__.py

@@ -1,5 +1,5 @@
 """Transport implementations for MCP clients."""
 
-from mcp.client.transports.memory import create_in_memory_transport
+from mcp.client.transports.memory import InMemoryTransport
 
-__all__ = ["create_in_memory_transport"]
+__all__ = ["InMemoryTransport"]

src/mcp/client/transports/memory.py

@@ -15,63 +15,83 @@
 from mcp.shared.message import SessionMessage
 
 
-@asynccontextmanager
-async def create_in_memory_transport(
-    server: Server[Any] | FastMCP,
-    *,
-    raise_exceptions: bool = False,
-) -> AsyncGenerator[
-    tuple[
-        MemoryObjectReceiveStream[SessionMessage | Exception],
-        MemoryObjectSendStream[SessionMessage],
-    ],
-    None,
-]:
+class InMemoryTransport:
     """
-    Create an in-memory transport connected to a server.
+    In-memory transport for testing MCP servers without network overhead.
 
-    This starts the server in a background task and returns streams
-    for client-side communication. The server is automatically stopped
-    when the context manager exits.
-
-    Args:
-        server: The MCP server to connect to (Server or FastMCP instance)
-        raise_exceptions: Whether to raise exceptions from the server
-
-    Yields:
-        A tuple of (read_stream, write_stream) for communicating with the server
+    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 create_in_memory_transport(server) as (read, write):
-            async with ClientSession(read, write) as session:
+        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.from_server(server) as client:
+            result = await client.call_tool("my_tool", {...})
     """
-    # Unwrap FastMCP to get underlying Server
-    actual_server: Server[Any]
-    if isinstance(server, FastMCP):
-        actual_server = server._mcp_server  # type: ignore[reportPrivateUsage]
-    else:
-        actual_server = 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=raise_exceptions,
+
+    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()
+                try:
+                    yield client_read, client_write
+                finally:
+                    tg.cancel_scope.cancel()

src/mcp/shared/memory.py

@@ -4,18 +4,12 @@
 
 from __future__ import annotations
 
-import warnings
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
-from typing import Any
 
 import anyio
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 
-import mcp.types as types
-from mcp.client.session import ClientSession, ElicitationFnT, ListRootsFnT, LoggingFnT, MessageHandlerFnT, SamplingFnT
-from mcp.server import Server
-from mcp.server.fastmcp import FastMCP
 from mcp.shared.message import SessionMessage
 
 MessageStream = tuple[MemoryObjectReceiveStream[SessionMessage | Exception], MemoryObjectSendStream[SessionMessage]]
@@ -44,75 +38,3 @@ async def create_client_server_memory_streams() -> AsyncGenerator[tuple[MessageS
         server_to_client_send,
     ):
         yield client_streams, server_streams
-
-
-@asynccontextmanager
-async def create_connected_server_and_client_session(
-    server: Server[Any] | FastMCP,
-    read_timeout_seconds: float | None = None,
-    sampling_callback: SamplingFnT | None = None,
-    list_roots_callback: ListRootsFnT | None = None,
-    logging_callback: LoggingFnT | None = None,
-    message_handler: MessageHandlerFnT | None = None,
-    client_info: types.Implementation | None = None,
-    raise_exceptions: bool = False,
-    elicitation_callback: ElicitationFnT | None = None,
-) -> AsyncGenerator[ClientSession, None]:
-    """Creates a ClientSession that is connected to a running MCP server.
-
-    .. deprecated::
-        Use :class:`mcp.client.Client` with :meth:`Client.from_server` instead.
-        This function will be removed in a future version.
-
-        Example migration::
-
-            # Before
-            async with create_connected_server_and_client_session(server) as session:
-                result = await session.call_tool("my_tool", {...})
-
-            # After
-            from mcp import Client
-            async with Client.from_server(server) as client:
-                result = await client.call_tool("my_tool", {...})
-    """
-    warnings.warn(
-        "create_connected_server_and_client_session is deprecated. "
-        "Use Client.from_server(server) instead.",
-        DeprecationWarning,
-        stacklevel=2,
-    )
-
-    if isinstance(server, FastMCP):  # pragma: no cover
-        server = server._mcp_server  # type: ignore[reportPrivateUsage]
-
-    async with create_client_server_memory_streams() as (client_streams, server_streams):
-        client_read, client_write = client_streams
-        server_read, server_write = server_streams
-
-        # Create a cancel scope for the server task
-        async with anyio.create_task_group() as tg:
-            tg.start_soon(
-                lambda: server.run(
-                    server_read,
-                    server_write,
-                    server.create_initialization_options(),
-                    raise_exceptions=raise_exceptions,
-                )
-            )
-
-            try:
-                async with ClientSession(
-                    read_stream=client_read,
-                    write_stream=client_write,
-                    read_timeout_seconds=read_timeout_seconds,
-                    sampling_callback=sampling_callback,
-                    list_roots_callback=list_roots_callback,
-                    logging_callback=logging_callback,
-                    message_handler=message_handler,
-                    client_info=client_info,
-                    elicitation_callback=elicitation_callback,
-                ) as client_session:
-                    await client_session.initialize()
-                    yield client_session
-            finally:  # pragma: no cover
-                tg.cancel_scope.cancel()

tests/client/conftest.py

@@ -123,11 +123,13 @@ async def patched_create_streams():
             yield (client_read, spy_client_write), (server_read, spy_server_write)
 
     # Apply the patch for the duration of the test
+    # Patch both locations since InMemoryTransport imports it directly
     with patch("mcp.shared.memory.create_client_server_memory_streams", patched_create_streams):
-        # Return a collection with helper methods
-        def get_spy_collection() -> StreamSpyCollection:
-            assert client_spy is not None, "client_spy was not initialized"
-            assert server_spy is not None, "server_spy was not initialized"
-            return StreamSpyCollection(client_spy, server_spy)
-
-        yield get_spy_collection
+        with patch("mcp.client.transports.memory.create_client_server_memory_streams", patched_create_streams):
+            # Return a collection with helper methods
+            def get_spy_collection() -> StreamSpyCollection:
+                assert client_spy is not None, "client_spy was not initialized"
+                assert server_spy is not None, "server_spy was not initialized"
+                return StreamSpyCollection(client_spy, server_spy)
+
+            yield get_spy_collection