Address review: keep sender contexts through the replay, bound the opening peek

- The opening-request replay now carries each frame's captured sender
  context, so the streams the SSE transport hands the loop keep their
  contextvars propagation; pinned by a test on the dual-era loop.
- Bound how many frames may precede the client's first request; past
  the limit the loop stops looking for an opening request and serves a
  handshake connection, so a peer streaming pre-request frames cannot
  grow the buffer without bound.
- Drop the stale docs admonition that stdio still rejects listen, and
  trim the era-decision docstring to the rule it now implements.

Author: maxisbey

docs/handlers/subscriptions.md

@@ -52,13 +52,6 @@ Two things the stream is *not*:
     handler on the low-level `Server` and acknowledge a smaller filter than the client asked
     for; the acknowledgment is how the client learns what it actually got.
 
-!!! warning "Streamable HTTP only, for now"
-    `subscriptions/listen` needs a transport that can stream a request's response, which today
-    means streamable HTTP. Over stdio a 2026-07-28 connection rejects the method with
-    METHOD_NOT_FOUND, even though `server/discover` advertises the subscription capabilities
-    there. Serving it over stdio is planned; the open-stream semantics for that transport are
-    not built yet.
-
 ## The client end
 
 Here is a client on the other side of that stream, following the board:

src/mcp/server/runner.py

@@ -13,6 +13,7 @@
 
 from __future__ import annotations
 
+import contextvars
 import logging
 from collections.abc import AsyncIterator, Awaitable, Mapping
 from contextlib import asynccontextmanager
@@ -59,6 +60,7 @@
 from mcp.server.context import CallNext, HandlerResult, ServerMiddleware, ServerRequestContext
 from mcp.server.models import InitializationOptions
 from mcp.server.session import ServerSession
+from mcp.shared._context_streams import ContextReceiveStream
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.dispatcher import CallOptions, DispatchContext, Dispatcher, OnNotify, OnRequest
 from mcp.shared.exceptions import MCPError, NoBackChannelError
@@ -499,15 +501,12 @@ async def serve_loop(
 def _has_modern_envelope(params: Mapping[str, Any] | None) -> bool:
     """Whether `params._meta` carries the reserved protocol-version key.
 
-    Era evidence is the client's explicit version declaration: the
-    `io.modelcontextprotocol/protocolVersion` key exists only in 2026-07-28+
-    envelopes, and the `io.modelcontextprotocol/` prefix is spec-reserved, so
-    legacy traffic never mints it (bare `_meta` is NOT evidence - legacy
-    requests carry `progressToken` there). Presence of the version key alone
-    is the rule, not the full required pair, so a half-built envelope
-    (version present, capabilities missing) still routes modern and gets the
-    classifier's INVALID_PARAMS naming the missing key instead of the legacy
-    path's generic one - and, like every failed classification, locks no era.
+    The `io.modelcontextprotocol/protocolVersion` key exists only in
+    2026-07-28+ envelopes and its prefix is spec-reserved, so legacy traffic
+    never mints it (a bare `_meta` is not evidence - legacy requests carry
+    `progressToken` there). The version key alone is the signal, not the full
+    required pair, so a half-built envelope still routes modern and gets the
+    classifier's INVALID_PARAMS naming the missing key.
     """
     if not params:
         return False
@@ -644,6 +643,17 @@ async def serve_dual_era_loop(
         await write_stream.aclose()
 
 
+_OPENING_PEEK_LIMIT: int = 32
+"""How many frames may precede the client's first request before the loop
+stops looking for one. Every legitimate client opens with a request, so this
+only bounds a peer that streams other frames without ever sending one."""
+
+
+def _sender_context(stream: ReadStream[Any]) -> contextvars.Context:
+    """The per-message sender context a context-aware stream carries, else the current one."""
+    return getattr(stream, "last_context", None) or contextvars.copy_context()
+
+
 @asynccontextmanager
 async def _replay_from_opening_request(
     read_stream: ReadStream[SessionMessage | Exception],
@@ -652,37 +662,43 @@ async def _replay_from_opening_request(
 
     Reads frames until the first JSON-RPC request arrives, then yields that
     request together with a stream that replays every frame read so far and
-    relays the rest of `read_stream` behind it. The request is `None` if the
-    client closed the channel before sending any request.
+    relays the rest of `read_stream` behind it, sender contexts included. The
+    request is `None` if the channel closes - or `_OPENING_PEEK_LIMIT` frames
+    pass - before any request appears.
     """
-    peeked: list[SessionMessage | Exception] = []
+    peeked: list[tuple[contextvars.Context, SessionMessage | Exception]] = []
     opening: JSONRPCRequest | None = None
-    replay_send, replayed = anyio.create_memory_object_stream[SessionMessage | Exception]()
+    replay_send, replay_receive = anyio.create_memory_object_stream[
+        tuple[contextvars.Context, SessionMessage | Exception]
+    ]()
+    replayed = ContextReceiveStream(replay_receive)
 
     async def replay_then_relay() -> None:
         async with replay_send:
-            for item in peeked:
-                await replay_send.send(item)
+            for envelope in peeked:
+                await replay_send.send(envelope)
             async for item in read_stream:
-                await replay_send.send(item)
+                await replay_send.send((_sender_context(read_stream), item))
 
     # This helper takes ownership of `read_stream` from the serving loop, so
     # every exit - including cancellation while awaiting the first request -
     # closes it and the replay channel.
     try:
         async for item in read_stream:
-            peeked.append(item)
+            peeked.append((_sender_context(read_stream), item))
             if isinstance(item, SessionMessage) and isinstance(item.message, JSONRPCRequest):
                 opening = item.message
                 break
+            if len(peeked) >= _OPENING_PEEK_LIMIT:
+                break
         async with anyio.create_task_group() as tg:
             tg.start_soon(replay_then_relay)
             yield opening, replayed
             tg.cancel_scope.cancel()
     finally:
         await read_stream.aclose()
         replay_send.close()
-        replayed.close()
+        replay_receive.close()
 
 
 async def _serve_legacy_stream(

tests/server/test_runner.py

@@ -7,6 +7,7 @@
 `aclose_shielded`) follow at the bottom.
 """
 
+import contextvars
 from collections.abc import AsyncIterator, Mapping
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field, replace
@@ -57,6 +58,7 @@
 from mcp.server.lowlevel.server import NotificationOptions, Server
 from mcp.server.models import InitializationOptions
 from mcp.server.runner import (
+    _OPENING_PEEK_LIMIT,
     ServerRunner,
     _extract_meta,
     _has_modern_envelope,
@@ -69,6 +71,7 @@
 )
 from mcp.server.session import ServerSession
 from mcp.server.subscriptions import SUBSCRIPTION_ID_META_KEY, InMemorySubscriptionBus, ListenHandler
+from mcp.shared._context_streams import create_context_streams
 from mcp.shared.dispatcher import CallOptions
 from mcp.shared.exceptions import MCPError, NoBackChannelError
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
@@ -1356,6 +1359,8 @@ async def _append_async(dst: list[int], v: int) -> None:
 
 _LIFESPAN: dict[str, Any] = {}
 
+_SENDER_VAR: contextvars.ContextVar[str] = contextvars.ContextVar("dual_era_sender_var", default="unset")
+
 
 @pytest.mark.anyio
 async def test_serve_one_runs_handler_and_returns_result_dict(server: SrvT):
@@ -1972,6 +1977,49 @@ async def test_notify_only_outbound_forwards_notifications_and_refuses_requests(
         await outbound.send_raw_request("ping", None)
 
 
+@pytest.mark.anyio
+async def test_dual_era_loop_carries_the_sender_context_through_the_replay():
+    """A context-aware read stream's per-message sender context still reaches
+    handlers over the dual-era loop (the SSE transport delivers requests this
+    way): the opening-request replay forwards each frame's captured context."""
+    seen: list[str] = []
+
+    async def probe(ctx: Ctx, params: RequestParams | None) -> dict[str, Any]:
+        seen.append(_SENDER_VAR.get())
+        return {}
+
+    server = Server(name="ctx-server", version="0.0.1")
+    server.add_request_handler("x/probe", RequestParams, probe)
+    c2s_send, c2s_recv = create_context_streams[SessionMessage | Exception](8)
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](8)
+    frame = JSONRPCRequest(jsonrpc="2.0", id=1, method="x/probe", params=_modern_params())
+    async with anyio.create_task_group() as tg, s2c_recv:
+        tg.start_soon(partial(serve_dual_era_loop, server, c2s_recv, s2c_send, lifespan_state=_LIFESPAN))
+        token = _SENDER_VAR.set("from-the-sender")
+        try:
+            await c2s_send.send(SessionMessage(message=frame))
+        finally:
+            _SENDER_VAR.reset(token)
+        with anyio.fail_after(5):
+            await s2c_recv.receive()
+        tg.cancel_scope.cancel()
+        await c2s_send.aclose()
+    assert seen == ["from-the-sender"]
+
+
+@pytest.mark.anyio
+async def test_dual_era_loop_stops_peeking_after_a_flood_of_pre_request_frames(server: SrvT):
+    """A peer that streams notifications and never opens with a request only
+    gets the first `_OPENING_PEEK_LIMIT` frames buffered: past that the loop
+    stops looking for an opening request and serves the rest as an ordinary
+    handshake connection, so the handshake that follows still lands."""
+    async with dual_era_client(server) as (client, _):
+        for _ in range(_OPENING_PEEK_LIMIT + 5):
+            await client.notify("notifications/flood", None)
+        init = await client.send_raw_request("initialize", _initialize_params())
+        assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION
+
+
 @pytest.mark.anyio
 async def test_dual_era_client_propagates_body_exception_unwrapped(server: SrvT):
     """The harness re-raises body exceptions as-is, not as `ExceptionGroup`."""