modern_on_request driver + Client in-process modern path via DirectDispatcher peer-pair

- modern_on_request(server, lifespan_state) returns an OnRequest callback
  that builds Connection.from_envelope per call and drives serve_one — wire
  it into the server side of a DirectDispatcher peer-pair for an in-process
  server on the modern per-request path
- Client(Server|MCPServer, mode!=legacy) enters lifespan once, creates a
  peer-pair, runs the server side with modern_on_request, and hands the
  client side to ClientSession; legacy in-process keeps InMemoryTransport
- Interaction-suite in-memory transport unlocked for 2026-07-28: 71 tests
  now run on [in-memory-2026-07-28], 67 pass; the 5 streamable-http-only
  notify-drop xfails are scoped to that transport; 4 progress-notification
  tests still xfail (peer-pair progress wiring tracked separately)

Author: maxisbey

src/mcp/client/client.py

@@ -2,10 +2,12 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 from contextlib import AsyncExitStack
 from dataclasses import KW_ONLY, dataclass, field
 from typing import Any, Literal
 
+import anyio
 from typing_extensions import deprecated
 
 from mcp import types
@@ -15,6 +17,8 @@
 from mcp.client.streamable_http import streamable_http_client
 from mcp.server import Server
 from mcp.server.mcpserver import MCPServer
+from mcp.server.runner import modern_on_request
+from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
 from mcp.shared.dispatcher import ProgressFnT
 from mcp.shared.exceptions import MCPDeprecationWarning
 from mcp.types import (
@@ -48,6 +52,14 @@ def _synthesize_discover(protocol_version: str) -> types.DiscoverResult:
     )
 
 
+async def _drop_notify(_dctx: Any, _method: str, _params: Mapping[str, Any] | None) -> None:
+    """Server-side ``OnNotify`` for the modern in-process path: client→server notifications are dropped.
+
+    The per-request driver (`serve_one`) has no notification dispatch table; progress and
+    cancellation travel via `CallOptions` on the `DirectDispatcher`, not as JSON-RPC notifies.
+    """
+
+
 @dataclass
 class Client:
     """A high-level MCP client for connecting to MCP servers.
@@ -121,11 +133,14 @@ async def main():
 
     _session: ClientSession | None = field(init=False, default=None)
     _exit_stack: AsyncExitStack | None = field(init=False, default=None)
-    _transport: Transport = field(init=False)
+    _transport: Transport | None = field(init=False, default=None)
+    _inproc_server: Server[Any] | None = field(init=False, default=None)
 
     def __post_init__(self) -> None:
-        if isinstance(self.server, Server | MCPServer):
-            self._transport = InMemoryTransport(self.server, raise_exceptions=self.raise_exceptions)
+        if isinstance(self.server, MCPServer):
+            self._inproc_server = self.server._lowlevel_server  # pyright: ignore[reportPrivateUsage]
+        elif isinstance(self.server, Server):
+            self._inproc_server = self.server
         elif isinstance(self.server, str):
             self._transport = streamable_http_client(self.server)
         else:
@@ -137,10 +152,34 @@ async def __aenter__(self) -> Client:
             raise RuntimeError("Client is already entered; cannot reenter")
 
         async with AsyncExitStack() as exit_stack:
-            read_stream, write_stream = await exit_stack.enter_async_context(self._transport)
-
-            self._session = await exit_stack.enter_async_context(
-                ClientSession(
+            if self._inproc_server is not None and self.mode != "legacy":
+                # Modern in-process path: drive the server through a DirectDispatcher peer-pair
+                # with one `serve_one` per request — no streams, no initialize handshake.
+                lifespan_state = await exit_stack.enter_async_context(self._inproc_server.lifespan(self._inproc_server))
+                client_disp, server_disp = create_direct_dispatcher_pair()
+                tg = await exit_stack.enter_async_context(anyio.create_task_group())
+                exit_stack.callback(server_disp.close)
+                await tg.start(server_disp.run, modern_on_request(self._inproc_server, lifespan_state), _drop_notify)
+                session = ClientSession(
+                    dispatcher=client_disp,
+                    read_timeout_seconds=self.read_timeout_seconds,
+                    sampling_callback=self.sampling_callback,
+                    list_roots_callback=self.list_roots_callback,
+                    logging_callback=self.logging_callback,
+                    message_handler=self.message_handler,
+                    client_info=self.client_info,
+                    elicitation_callback=self.elicitation_callback,
+                )
+            else:
+                if self._inproc_server is not None:
+                    transport: Transport = InMemoryTransport(
+                        self._inproc_server, raise_exceptions=self.raise_exceptions
+                    )
+                else:
+                    assert self._transport is not None
+                    transport = self._transport
+                read_stream, write_stream = await exit_stack.enter_async_context(transport)
+                session = ClientSession(
                     read_stream=read_stream,
                     write_stream=write_stream,
                     read_timeout_seconds=self.read_timeout_seconds,
@@ -151,7 +190,8 @@ async def __aenter__(self) -> Client:
                     client_info=self.client_info,
                     elicitation_callback=self.elicitation_callback,
                 )
-            )
+
+            self._session = await exit_stack.enter_async_context(session)
 
             if self.mode == "legacy":
                 await self._session.initialize()

src/mcp/server/runner.py

@@ -36,11 +36,14 @@
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher, handler_exception_to_error_data
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.shared.transport_context import TransportContext
-from mcp.shared.version import HANDSHAKE_PROTOCOL_VERSIONS
+from mcp.shared.version import HANDSHAKE_PROTOCOL_VERSIONS, MODERN_PROTOCOL_VERSIONS
 from mcp.types import (
+    CLIENT_CAPABILITIES_META_KEY,
+    CLIENT_INFO_META_KEY,
     INTERNAL_ERROR,
     INVALID_PARAMS,
     METHOD_NOT_FOUND,
+    PROTOCOL_VERSION_META_KEY,
     ErrorData,
     Implementation,
     InitializeRequestParams,
@@ -62,6 +65,7 @@
     "ServerMiddleware",
     "ServerRunner",
     "aclose_shielded",
+    "modern_on_request",
     "otel_middleware",
     "serve_connection",
     "serve_loop",
@@ -512,3 +516,34 @@ async def serve_one(
         return await to_jsonrpc_response(request.id, runner.on_request(dctx, request.method, request.params))
     finally:
         await aclose_shielded(connection)
+
+
+def modern_on_request(server: Server[LifespanT], lifespan_state: LifespanT) -> OnRequest:
+    """Return an `OnRequest` callback that serves each call via `serve_one` with a fresh per-request `Connection`.
+
+    Wire this into the server side of a `DirectDispatcher` peer-pair to drive an
+    in-process server on the modern per-request-envelope path (each request
+    carries protocol version, client info, and capabilities in `params._meta`;
+    no `initialize` handshake).
+    """
+
+    async def handle(
+        dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
+    ) -> dict[str, Any]:
+        meta = (params or {}).get("_meta", {})
+        connection = Connection.from_envelope(
+            meta.get(PROTOCOL_VERSION_META_KEY, MODERN_PROTOCOL_VERSIONS[-1]),
+            meta.get(CLIENT_INFO_META_KEY),
+            meta.get(CLIENT_CAPABILITIES_META_KEY),
+        )
+        # `OnRequest` is invoked for requests only, so `request_id` is always set.
+        assert dctx.request_id is not None
+        req = JSONRPCRequest(
+            jsonrpc="2.0", id=dctx.request_id, method=method, params=dict(params) if params is not None else None
+        )
+        msg = await serve_one(server, req, connection=connection, dctx=dctx, lifespan_state=lifespan_state)
+        if isinstance(msg, JSONRPCError):
+            raise MCPError(code=msg.error.code, message=msg.error.message, data=msg.error.data)
+        return msg.result
+
+    return handle

tests/interaction/_connect.py

@@ -88,9 +88,15 @@ async def connect_in_memory(
     elicitation_callback: ElicitationFnT | None = None,
     spec_version: str = LATEST_PROTOCOL_VERSION,
 ) -> AsyncIterator[Client]:
-    """Yield a Client connected to the server over the in-memory transport."""
+    """Yield a Client connected to the server over the in-memory transport.
+
+    When `spec_version` is a modern (2026-07-28+) revision the Client is opened with
+    `mode=<version>`, which drives the server through the DirectDispatcher peer-pair
+    (per-request `serve_one`, no initialize handshake) instead of the legacy stream pair.
+    """
     async with Client(
         server,
+        mode=spec_version if spec_version in MODERN_PROTOCOL_VERSIONS else "legacy",
         read_timeout_seconds=read_timeout_seconds,
         sampling_callback=sampling_callback,
         list_roots_callback=list_roots_callback,

tests/interaction/_requirements.py

@@ -63,9 +63,7 @@
 
 TRANSPORT_SPEC_VERSIONS: dict[Transport, tuple[SpecVersion, ...]] = {
     "sse": ("2025-11-25",),
-    # Temporary lock: the in-memory transport has no modern entry point yet, so it cannot
-    # negotiate the newer revision. Remove once an in-memory factory for the modern path lands.
-    "in-memory": ("2025-11-25",),
+    "in-memory": ("2025-11-25", "2026-07-28"),
     # At the newer revision the protocol-version header check runs before the stateless branch is
     # taken, so a stateless connection at that revision behaves identically to the stateful one.
     # Locked to avoid a redundant matrix column; revisit if the header/stateless ordering changes.
@@ -745,7 +743,9 @@ def __post_init__(self) -> None:
             "Log notifications emitted by a tool handler during execution reach the client's logging "
             "callback before the tool result returns."
         ),
-        known_failures=(KnownFailure(spec_version="2026-07-28", note=_MODERN_NOTIFY_DROP, issue=None),),
+        known_failures=(
+            KnownFailure(spec_version="2026-07-28", transport="streamable-http", note=_MODERN_NOTIFY_DROP, issue=None),
+        ),
     ),
     "tools:call:progress": Requirement(
         source=f"{SPEC_BASE_URL}/basic/utilities/progress#progress-flow",
@@ -974,7 +974,9 @@ def __post_init__(self) -> None:
             "The Context logging helpers (debug/info/warning/error) send log message notifications at the "
             "corresponding severity."
         ),
-        known_failures=(KnownFailure(spec_version="2026-07-28", note=_MODERN_NOTIFY_DROP, issue=None),),
+        known_failures=(
+            KnownFailure(spec_version="2026-07-28", transport="streamable-http", note=_MODERN_NOTIFY_DROP, issue=None),
+        ),
     ),
     "mcpserver:context:progress": Requirement(
         source="sdk",
@@ -1339,15 +1341,19 @@ def __post_init__(self) -> None:
     "logging:message:all-levels": Requirement(
         source=f"{SPEC_BASE_URL}/server/utilities/logging#log-levels",
         behavior="All eight RFC 5424 severity levels are deliverable as log message notifications.",
-        known_failures=(KnownFailure(spec_version="2026-07-28", note=_MODERN_NOTIFY_DROP, issue=None),),
+        known_failures=(
+            KnownFailure(spec_version="2026-07-28", transport="streamable-http", note=_MODERN_NOTIFY_DROP, issue=None),
+        ),
     ),
     "logging:message:fields": Requirement(
         source=f"{SPEC_BASE_URL}/server/utilities/logging#log-message-notifications",
         behavior=(
             "A log message sent by a server handler is delivered to the client's logging callback with its "
             "severity level, logger name, and data."
         ),
-        known_failures=(KnownFailure(spec_version="2026-07-28", note=_MODERN_NOTIFY_DROP, issue=None),),
+        known_failures=(
+            KnownFailure(spec_version="2026-07-28", transport="streamable-http", note=_MODERN_NOTIFY_DROP, issue=None),
+        ),
     ),
     "logging:message:filtered": Requirement(
         source=f"{SPEC_BASE_URL}/server/utilities/logging#setting-log-level",
@@ -1970,6 +1976,7 @@ def __post_init__(self) -> None:
         known_failures=(
             KnownFailure(
                 spec_version="2026-07-28",
+                transport="streamable-http",
                 note=(
                     "List-mutation assertions hold; only the sentinel ctx.info() never reaches the client. "
                     + _MODERN_NOTIFY_DROP

tests/interaction/test_coverage.py

@@ -301,6 +301,7 @@ def test_compute_cells_drops_era_locked_transport_outside_its_versions() -> None
         "sse-2025-11-25",
         "streamable-http-2025-11-25",
         "streamable-http-stateless-2025-11-25",
+        "in-memory-2026-07-28",
         "streamable-http-2026-07-28",
     ]