Class InMemoryTransport

src/mcp/client/_memory.py:21–109

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.

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.
    """

    def __init__(self, server: Server[Any] | MCPServer, *, raise_exceptions: bool = False) -> None:
        """Initialize the in-memory transport.

        Args:
            server: The MCP server to connect to (Server or MCPServer instance)
            raise_exceptions: Whether to raise exceptions from the server
        """
        self._server = server
        self._raise_exceptions = raise_exceptions
        self._cm: AbstractAsyncContextManager[TransportStreams] | None = None

    @asynccontextmanager
    async def _connect(self) -> AsyncIterator[TransportStreams]:
        """Connect to the server and yield streams for communication."""
        # Unwrap MCPServer to get underlying Server
        if isinstance(self._server, MCPServer):
            # TODO(Marcelo): Make `lowlevel_server` public.
            actual_server: Server[Any] = self._server._lowlevel_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

            server_done = anyio.Event()

            async def _run_server() -> None:
                try:
                    await actual_server.run(
                        server_read,
                        server_write,
                        actual_server.create_initialization_options(),
                        raise_exceptions=self._raise_exceptions,
                    )
                finally:
                    server_done.set()

            async with anyio.create_task_group() as tg:
                tg.start_soon(_run_server)

                try:
                    yield client_read, client_write
                finally:
                    # EOF the server (and our own read side) instead of
                    # cancelling outright. The dispatcher's run() cancels its
                    # own in-flight handlers on read-stream EOF, so for a
                    # well-behaved server the task exits naturally and the
                    # task-group join below is immediate. Cancelling here
                    # unconditionally would `coro.throw()` into this task,