Serve subscriptions/listen over the dual-era stream loop

Remove the transport-level rejection of subscriptions/listen on the
stream-pair modern path: the method now dispatches through serve_one to
the registered ListenHandler like every other modern request. Over
stdio the listen request simply stays pending - the ack and every event
are request-scoped notifications on the duplex pipe, and the response
arrives only when the stream ends gracefully. server/discover's
subscription capability advertisement is now truthful on this transport.

Two era-scoped generalizations ride the dispatcher's new per-request
seams, fixing the class rather than the listen instance:

- Era lock at first client-visible success frame. A classified-modern
  request commits the modern lock immediately before its first
  non-error frame is written (notification, progress, or response),
  not when its dispatch returns. Plain requests lock at their response
  exactly as before; a listen locks at its ack, so an initialize
  pipelined behind a live stream is rejected with -32022 instead of
  flipping the connection legacy mid-stream. Classification failures
  and pre-frame cancellations still never lock, and progress written
  for a request that later fails locks correctly (the client that
  stamped a valid envelope already committed to modern).

- Quiet cancellation for every classified-modern request. After an
  inbound notifications/cancelled the server sends nothing further for
  that request - no result, no error - matching the transport spec's
  MUST NOT. Legacy-era requests keep the byte-identical code-0
  "Request cancelled" answer released 2025 clients unblock on.

The single-exchange HTTP and direct in-process paths need no seam:
cancellation is structurally silent there (stream close / caller's own
scope).

Author: maxisbey

docs/handlers/subscriptions.md

@@ -52,12 +52,10 @@ 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.
+`subscriptions/listen` is served on every transport that can stream a request's response: over
+streamable HTTP the response is the SSE stream, and over stdio the listen request simply stays
+pending on the duplex pipe (the acknowledgment and every event are notifications; the request's
+result arrives only when the stream ends gracefully).
 
 ## The client end
 

docs/whats-new.md

@@ -191,7 +191,7 @@ That file is the pitch in one place: one server, one `Resolve`-backed tool, and
 
 ### Change notifications become one stream
 
-At 2026-07-28 the standalone HTTP GET stream and `resources/subscribe` are replaced by `subscriptions/listen`: the client opens one long-lived stream and names the notification kinds it wants. `MCPServer` serves it out of the box; you publish with `await ctx.notify_resource_updated(uri)` (and `notify_tools_changed()`, and so on), and multi-replica deployments plug in a shared `SubscriptionBus`. On the client (since `2.0.0b2`), `async with client.listen(...)` opens the stream: the filter goes in as keyword arguments, typed change events come back, and `sub.honored` is the subset the server agreed to deliver. One honest caveat: over stdio the server does not serve the stream yet.
+At 2026-07-28 the standalone HTTP GET stream and `resources/subscribe` are replaced by `subscriptions/listen`: the client opens one long-lived stream and names the notification kinds it wants. `MCPServer` serves it out of the box; you publish with `await ctx.notify_resource_updated(uri)` (and `notify_tools_changed()`, and so on), and multi-replica deployments plug in a shared `SubscriptionBus`. On the client (since `2.0.0b2`), `async with client.listen(...)` opens the stream: the filter goes in as keyword arguments, typed change events come back, and `sub.honored` is the subset the server agreed to deliver. This works on both transports: over streamable HTTP the stream is the request's SSE response, and over stdio the events ride the duplex pipe while the listen request stays pending.
 
 **[Subscriptions](handlers/subscriptions.md)** covers publishing and serving, **[its Clients twin](client/subscriptions.md)** the watching end, and **[Deploy & scale](run/deploy.md)** the bus.
 

src/mcp/server/runner.py

@@ -59,7 +59,12 @@
 from mcp.shared.dispatcher import CallOptions, DispatchContext, Dispatcher, OnNotify, OnRequest
 from mcp.shared.exceptions import MCPError, NoBackChannelError
 from mcp.shared.inbound import InboundLadderRejection, classify_inbound_request
-from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher, handler_exception_to_error_data
+from mcp.shared.jsonrpc_dispatcher import (
+    JSONRPCDispatcher,
+    handler_exception_to_error_data,
+    observe_success_frames,
+    suppress_cancel_answer,
+)
 from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage
 from mcp.shared.transport_context import TransportContext
 
@@ -575,15 +580,29 @@ async def serve_dual_era_loop(
     for genuine version negotiation or for `initialize` on an
     already-modern connection.
 
-    The era lock rides the request's own dispatch. For the inline methods
-    (`initialize`, `server/discover`) that completes before the next frame is
-    read, so the canonical probe-then-go flow is race-free; a pinned-modern
-    client that pipelines frames ahead of its first response should expect
-    envelope-less notifications sent in that window to be dropped. The lock
-    settles exactly once: a request from the other era that was already in
-    flight when the lock committed may still complete and its response
-    stands, but the era does not move; and a success the peer cancelled away
-    (it sees "Request cancelled", not the result) does not lock either.
+    The modern lock commits on the FIRST client-visible success frame the
+    classified request produces - a request-scoped notification (progress and
+    the `subscriptions/listen` ack ride those) or its result - immediately
+    before that frame is written, with no checkpoint in between. One
+    definition for every method: a plain request locks at its response
+    exactly as before, and a streaming request (`subscriptions/listen`, whose
+    response arrives only at close) locks at its ack, so a pipelined
+    `initialize` behind the ack can never hijack a live stream back to
+    legacy. For the inline methods (`initialize`, `server/discover`) the
+    dispatch completes before the next frame is read, so the canonical
+    probe-then-go flow is race-free; a pinned-modern client that pipelines
+    frames ahead of its first response should expect envelope-less
+    notifications sent in that window to be dropped. The lock settles exactly
+    once: a request from the other era that was already in flight when the
+    lock committed may still complete and its response stands, but the era
+    does not move; and a success the peer cancelled away before any frame
+    went out does not lock either.
+
+    Classified-modern requests are also governed by the 2026 cancellation
+    rule for their whole lifetime: after an inbound `notifications/cancelled`
+    the server sends nothing further for that request - no result, no error
+    (the transport spec's MUST NOT). Legacy-era requests keep the
+    byte-identical "Request cancelled" answer released 2025 clients expect.
     """
     dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
         read_stream,
@@ -601,36 +620,45 @@ async def serve_dual_era_loop(
     modern_version = LATEST_MODERN_VERSION
 
     def era_settles(dctx: DispatchContext[TransportContext]) -> bool:
-        # The one definition of "this request may lock the era": it settled as
+        # The one definition of "this request may lock the era": it produced
         # a client-visible success on a still-unlocked connection. The lock is
         # monotone - the first success wins, so a straggling request from the
         # other era can never overwrite a committed lock. A pending peer
-        # cancel means the dispatcher is about to replace this response with
-        # "Request cancelled": the client never sees the success, no lock.
+        # cancel means the client already abandoned this request and will
+        # never read the frame as a success: no lock.
         return era == "unlocked" and not dctx.cancel_requested.is_set()
 
+    def commit_modern(dctx: DispatchContext[TransportContext], version: str) -> None:
+        # The success-frame observer for classified-modern requests: the
+        # dispatcher invokes it immediately before each non-error frame for
+        # the request (ack/event notifications, progress, or the result),
+        # with no checkpoint before the write - so by the time the client can
+        # read any modern output, the era is already locked and a pipelined
+        # `initialize` is rejected instead of hijacking the connection.
+        nonlocal era, modern_version
+        if era_settles(dctx):
+            era, modern_version = "modern", version
+
     async def serve_modern(
         dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
     ) -> dict[str, Any]:
-        nonlocal era, modern_version
         route = classify_inbound_request({"method": method, "params": params})
         if isinstance(route, InboundLadderRejection):
             raise MCPError(code=route.code, message=route.message, data=route.data)
-        if method == "subscriptions/listen":
-            # The registered listen handler assumes the HTTP entry's stream
-            # semantics; served over a stream pair it would wedge. Reject until
-            # this transport grows its own listen design.
-            raise MCPError(
-                code=METHOD_NOT_FOUND, message="subscriptions/listen is not served over this transport", data=method
-            )
+        # Classification succeeded: 2026 wire semantics govern this request's
+        # lifecycle from here. Both facts attach synchronously - no checkpoint
+        # since the dispatcher entered the handler scope - so a peer cancel
+        # can never interleave and observe the legacy defaults.
+        suppress_cancel_answer(dctx)
+        observe_success_frames(dctx, partial(commit_modern, dctx, route.protocol_version))
         connection = Connection.from_envelope(
             route.protocol_version,
             route.client_info,
             route.client_capabilities,
             outbound=standalone_outbound,
         )
         try:
-            result = await serve_one(
+            return await serve_one(
                 server,
                 _NoServerRequestsDispatchContext(dctx),
                 method,
@@ -647,9 +675,6 @@ async def serve_modern(
                 raise
             error = modern_error_data(exc)
             raise MCPError(code=error.code, message=error.message, data=error.data) from exc
-        if era_settles(dctx):
-            era, modern_version = "modern", route.protocol_version
-        return result
 
     async def on_request(
         dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None

src/mcp/server/subscriptions.py

@@ -164,8 +164,9 @@ class ListenHandler:
     cancels the handler; the stream just ends, per the spec's abrupt-close
     contract) or `close` ends all streams gracefully.
 
-    Requires a transport that can stream a request's response (streamable
-    HTTP's SSE mode).
+    Requires a transport that can stream a request's response: streamable
+    HTTP's SSE mode, or a duplex stream pair (stdio) where the listen request
+    simply stays pending while notifications ride the pipe.
 
     `max_subscriptions` bounds concurrent streams (further listen requests are
     rejected with `INTERNAL_ERROR`, before the ack). `max_buffered_events`