Serve the 2026-07-28 protocol over stdio: decide the era from the opening message#3151
Serve the 2026-07-28 protocol over stdio: decide the era from the opening message#3151maxisbey wants to merge 2 commits into
Conversation
📚 Documentation preview
|
… opening message The 2026-07-28 protocol did not work over stdio: subscriptions/listen was hard-refused, a legacy initialize arriving during an in-flight modern request was accepted and re-locked the connection, and a peer cancel produced a trailing "Request cancelled" frame. All three share one root cause: the connection's era was derived from which requests had completed instead of being decided once, in wire order, from how the client opened the connection. Replace serve_dual_era_loop and serve_loop with a single serve_stream driver that decides the era synchronously in the dispatcher's read loop, before the request body is spawned. initialize (or any envelope-less request) opens the legacy era, an enveloped request opens the modern era, server/discover is answered without pinning, and a stray leading notification opens nothing. A conflicting era claim on a committed connection is refused (-32022 or -32600) rather than silently switching. Cancel silence is structural: each request answers through a one-shot channel whose write target becomes a powerless void on a peer cancel, so there is no cancelled-check at any write site. The generic JSON-RPC dispatcher loses the code-0 cancel frame, the code-0 str(exc) catch-all, and the inline_methods knob, and documents that handlers are invoked synchronously in receive order with the returned awaitable as the body. Add a Posture enum (DUAL default, LEGACY_ONLY, MODERN_ONLY) on the Server and MCPServer constructors, honoured by the stream driver and the streamable-HTTP manager alike. Server.run(read, write) now stands alone, Server.lifespan() is a bound context manager, and serve_listener / newline_json_transport / close_subscriptions() give a straightforward path for custom transports. See docs/migration.md for the full list of observable changes.
Delete the Posture enum, its export, and the posture= constructor parameter on Server and MCPServer. The lowlevel Server is a registry of handlers and carries no serving configuration; a server offers both the 2025 handshake era and the 2026 per-request era on every transport, and each stream connection's era is decided by the client's opening message, as the default already did. _StreamConnection no longer takes a starting era; it starts undecided and pins on the first era-distinctive message. The streamable-HTTP manager's header routing is unchanged: streams it has already routed to the handshake era are entered born-legacy through the transport- internal serve_legacy_stream, so the legacy leg keeps refusing enveloped requests with -32600 exactly as before. The MODERN_ONLY / LEGACY_ONLY tests and doc sections go with it, and the streamable-HTTP orphan test file is renamed now that its posture half is gone.
There was a problem hiding this comment.
10 issues found across 68 files
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="examples/stories/legacy_routing/server.py">
<violation number="1" location="examples/stories/legacy_routing/server.py:36">
P1: A modern-envelope `initialize` is silently routed to the legacy backend because this excludes every `initialize` before checking `_meta`. Route based solely on envelope intent so this request reaches the ladder and is rejected as modern, preserving the documented no-downgrade behavior.</violation>
</file>
<file name="docs/handlers/subscriptions.md">
<violation number="1" location="docs/handlers/subscriptions.md:143">
P2: Stdio EOF does not guarantee a subscription's final result reaches the peer: draining stops after the two-second `_EOF_DRAIN_WINDOW`, and the peer may already be unable to read. Describe this shutdown as best-effort so callers do not rely on that final frame.</violation>
</file>
<file name="src/mcp/server/runner.py">
<violation number="1" location="src/mcp/server/runner.py:461">
P2: Modern in-process handlers can issue server-initiated requests when `modern_on_request` is wired to the default direct-dispatcher pair, violating the modern no-backchannel rule. Preserve the denying context wrapper here, rather than relying on every caller to configure `can_send_request=False`.</violation>
</file>
<file name="tests/server/test_http_orphan.py">
<violation number="1" location="tests/server/test_http_orphan.py:223">
P2: The regression test does not enforce the required empty `202 Accepted` response: a `200` response containing a resurrected cancellation frame still passes the `< 500` check. Asserting both the exact status and empty body would make this test catch the behavior described in its contract.</violation>
</file>
<file name="tests/server/test_stdio_ordering.py">
<violation number="1" location="tests/server/test_stdio_ordering.py:312">
P2: The pipelined legacy-first case does not verify the era of the request behind the handshake: it discards request 2's response, so an implementation could accept the handshake but incorrectly serve the following modern-enveloped request and this regression test would remain green. Retaining the response and asserting `JSONRPCError` with `INVALID_REQUEST` would cover the stated legacy-commitment behavior.</violation>
</file>
<file name="tests/server/test_serving.py">
<violation number="1" location="tests/server/test_serving.py:200">
P2: This test does not exercise the behavior named in `test_modern_client_notification_during_in_flight_work`: the awaited `tools/list` call has already completed before the notification is sent. A regression that drops or misroutes notifications while a modern request body is running would still pass, so a blocked handler plus an explicit release after the notification would make this coverage meaningful.</violation>
<violation number="2" location="tests/server/test_serving.py:539">
P2: This listen test has a delivery race: `bus.publish()` can yield to the client dispatcher before the event replacement, causing the delivered-event wait to wait on a fresh event that was never set and time out. Waiting for the recorded notification count or using a per-method event avoids replacing a shared signal after publishing.</violation>
</file>
<file name="tests/shared/test_jsonrpc_dispatcher.py">
<violation number="1" location="tests/shared/test_jsonrpc_dispatcher.py:2198">
P3: This test assumes request id 1 is written before id 2 even though both handlers run concurrently and JSON-RPC responses may arrive out of order. Scheduler differences can make the test flaky; collect both frames and assert them by id instead.</violation>
</file>
<file name="src/mcp/server/serving.py">
<violation number="1" location="src/mcp/server/serving.py:242">
P2: Closing one client input terminates `subscriptions/listen` streams owned by other clients sharing this `Server`, producing unsolicited completion responses. `_drain` calls global `close_subscriptions()`; scope stream registration and closure to this `_StreamConnection` so only its streams drain.</violation>
</file>
<file name="tests/interaction/lowlevel/test_cancellation.py">
<violation number="1" location="tests/interaction/lowlevel/test_cancellation.py:103">
P2: The assertion `assert all(outcome == f"error:{CONNECTION_CLOSED}" for outcome in outcomes)` is vacuously true when `outcomes` is empty. `all()` on an empty iterable returns `True`, so the assertion passes without recording any outcome.
When `task_group.cancel_scope.cancel()` is called, `CancelledError` is injected into `call_and_record_outcome` at the `await client.call_tool(...)` point inside the try block. Since only `MCPError` is caught, `CancelledError` propagates uncaught through the except and out of `call_and_record_outcome` — the task never appends to `outcomes`. The task group swallows the `CancelledError`, the test continues, and `all(...)` on the empty list returns `True`.
Add `assert outcomes` before the `all()` check to verify at least one outcome was recorded, or switch to an exact-length assertion like `assert outcomes == [f"error:{CONNECTION_CLOSED}"]`.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| return verdict | ||
| raw_params = body.get("params") | ||
| params = cast(Mapping[str, Any], raw_params) if isinstance(raw_params, Mapping) else None | ||
| if body.get("method") != "initialize" and parse_envelope(params) is not None: |
There was a problem hiding this comment.
P1: A modern-envelope initialize is silently routed to the legacy backend because this excludes every initialize before checking _meta. Route based solely on envelope intent so this request reaches the ladder and is rejected as modern, preserving the documented no-downgrade behavior.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At examples/stories/legacy_routing/server.py, line 36:
<comment>A modern-envelope `initialize` is silently routed to the legacy backend because this excludes every `initialize` before checking `_meta`. Route based solely on envelope intent so this request reaches the ladder and is rejected as modern, preserving the documented no-downgrade behavior.</comment>
<file context>
@@ -23,18 +22,21 @@
- return verdict
+ raw_params = body.get("params")
+ params = cast(Mapping[str, Any], raw_params) if isinstance(raw_params, Mapping) else None
+ if body.get("method") != "initialize" and parse_envelope(params) is not None:
+ verdict = classify_inbound_request(body, headers=headers)
+ return "modern" if isinstance(verdict, InboundModernRoute) else verdict
</file context>
| if body.get("method") != "initialize" and parse_envelope(params) is not None: | |
| if parse_envelope(params) is not None: |
| * You own the bus, so you publish to it directly: `await bus.publish(ResourceUpdated(uri=...))`. Put it wherever your handlers can reach it: module scope here, the lifespan in a bigger app. | ||
| * `ListenHandler(bus)` is the same handler `MCPServer` registers, and `on_subscriptions_listen=` is an ordinary handler slot. Put your own callable in that slot for different semantics, and the spec obligations move to you: acknowledge first, stamp every frame with the subscription id, deliver nothing outside the filter. | ||
| * `ListenHandler.close()` ends every open stream gracefully. Each one receives the listen request's result as its final frame, which is the spec's way of saying the server ended the subscription deliberately. It returns before those streams finish flushing, so give them a moment before you tear the transport down. Without it, streams end when the client disconnects. | ||
| * `ListenHandler.close()` ends every open stream gracefully, and `Server.close_subscriptions()` is the same verb on the server that registered the handler. Each stream drains its buffered events and then receives the listen request's result as its final frame, which is the spec's way of saying the server ended the subscription deliberately. The call returns before those streams finish flushing, so give them a moment before you tear the transport down yourself. Over stdio you rarely need to: when the client's input ends, the driver closes your open streams for you inside a short bounded window, in which each stream's final result is guaranteed to reach the departing peer and any events it still had buffered flush as time allows. Without any close, streams end when the client disconnects. |
There was a problem hiding this comment.
P2: Stdio EOF does not guarantee a subscription's final result reaches the peer: draining stops after the two-second _EOF_DRAIN_WINDOW, and the peer may already be unable to read. Describe this shutdown as best-effort so callers do not rely on that final frame.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At docs/handlers/subscriptions.md, line 143:
<comment>Stdio EOF does not guarantee a subscription's final result reaches the peer: draining stops after the two-second `_EOF_DRAIN_WINDOW`, and the peer may already be unable to read. Describe this shutdown as best-effort so callers do not rely on that final frame.</comment>
<file context>
@@ -132,7 +140,7 @@ Down on the low-level `Server` there is no pre-wired anything, and the same part
* You own the bus, so you publish to it directly: `await bus.publish(ResourceUpdated(uri=...))`. Put it wherever your handlers can reach it: module scope here, the lifespan in a bigger app.
* `ListenHandler(bus)` is the same handler `MCPServer` registers, and `on_subscriptions_listen=` is an ordinary handler slot. Put your own callable in that slot for different semantics, and the spec obligations move to you: acknowledge first, stamp every frame with the subscription id, deliver nothing outside the filter.
-* `ListenHandler.close()` ends every open stream gracefully. Each one receives the listen request's result as its final frame, which is the spec's way of saying the server ended the subscription deliberately. It returns before those streams finish flushing, so give them a moment before you tear the transport down. Without it, streams end when the client disconnects.
+* `ListenHandler.close()` ends every open stream gracefully, and `Server.close_subscriptions()` is the same verb on the server that registered the handler. Each stream drains its buffered events and then receives the listen request's result as its final frame, which is the spec's way of saying the server ended the subscription deliberately. The call returns before those streams finish flushing, so give them a moment before you tear the transport down yourself. Over stdio you rarely need to: when the client's input ends, the driver closes your open streams for you inside a short bounded window, in which each stream's final result is guaranteed to reach the departing peer and any events it still had buffered flush as time allows. Without any close, streams end when the client disconnects.
## Recap
</file context>
| * `ListenHandler.close()` ends every open stream gracefully, and `Server.close_subscriptions()` is the same verb on the server that registered the handler. Each stream drains its buffered events and then receives the listen request's result as its final frame, which is the spec's way of saying the server ended the subscription deliberately. The call returns before those streams finish flushing, so give them a moment before you tear the transport down yourself. Over stdio you rarely need to: when the client's input ends, the driver closes your open streams for you inside a short bounded window, in which each stream's final result is guaranteed to reach the departing peer and any events it still had buffered flush as time allows. Without any close, streams end when the client disconnects. | |
| * `ListenHandler.close()` ends every open stream gracefully, and `Server.close_subscriptions()` is the same verb on the server that registered the handler. Each stream drains its buffered events and then receives the listen request's result as its final frame, which is the spec's way of saying the server ended the subscription deliberately. The call returns before those streams finish flushing, so give them a moment before you tear the transport down yourself. Over stdio you rarely need to: when the client's input ends, the driver starts closing your open streams inside a short bounded window. Draining is best-effort: the peer may already be gone, and a blocked stream can be stopped before its final result or buffered events are written. Without any close, streams end when the client disconnects. |
| connection=connection, | ||
| lifespan_state=lifespan_state, | ||
| ) | ||
| return await serve_one(server, dctx, method, params, connection=connection, lifespan_state=lifespan_state) |
There was a problem hiding this comment.
P2: Modern in-process handlers can issue server-initiated requests when modern_on_request is wired to the default direct-dispatcher pair, violating the modern no-backchannel rule. Preserve the denying context wrapper here, rather than relying on every caller to configure can_send_request=False.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/mcp/server/runner.py, line 461:
<comment>Modern in-process handlers can issue server-initiated requests when `modern_on_request` is wired to the default direct-dispatcher pair, violating the modern no-backchannel rule. Preserve the denying context wrapper here, rather than relying on every caller to configure `can_send_request=False`.</comment>
<file context>
@@ -751,13 +458,6 @@ async def handle(
- connection=connection,
- lifespan_state=lifespan_state,
- )
+ return await serve_one(server, dctx, method, params, connection=connection, lifespan_state=lifespan_state)
return handle
</file context>
| # and the client's transport then resolves the still-pending await locally | ||
| # with CONNECTION_CLOSED - a client-side synthesis, not a server frame - so the | ||
| # only outcome ever recorded is that local resolution; a served result never appears. | ||
| assert all(outcome == f"error:{CONNECTION_CLOSED}" for outcome in outcomes) |
There was a problem hiding this comment.
P2: The assertion assert all(outcome == f"error:{CONNECTION_CLOSED}" for outcome in outcomes) is vacuously true when outcomes is empty. all() on an empty iterable returns True, so the assertion passes without recording any outcome.
When task_group.cancel_scope.cancel() is called, CancelledError is injected into call_and_record_outcome at the await client.call_tool(...) point inside the try block. Since only MCPError is caught, CancelledError propagates uncaught through the except and out of call_and_record_outcome — the task never appends to outcomes. The task group swallows the CancelledError, the test continues, and all(...) on the empty list returns True.
Add assert outcomes before the all() check to verify at least one outcome was recorded, or switch to an exact-length assertion like assert outcomes == [f"error:{CONNECTION_CLOSED}"].
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/interaction/lowlevel/test_cancellation.py, line 103:
<comment>The assertion `assert all(outcome == f"error:{CONNECTION_CLOSED}" for outcome in outcomes)` is vacuously true when `outcomes` is empty. `all()` on an empty iterable returns `True`, so the assertion passes without recording any outcome.
When `task_group.cancel_scope.cancel()` is called, `CancelledError` is injected into `call_and_record_outcome` at the `await client.call_tool(...)` point inside the try block. Since only `MCPError` is caught, `CancelledError` propagates uncaught through the except and out of `call_and_record_outcome` — the task never appends to `outcomes`. The task group swallows the `CancelledError`, the test continues, and `all(...)` on the empty list returns `True`.
Add `assert outcomes` before the `all()` check to verify at least one outcome was recorded, or switch to an exact-length assertion like `assert outcomes == [f"error:{CONNECTION_CLOSED}"]`.</comment>
<file context>
@@ -70,24 +76,31 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
+ # and the client's transport then resolves the still-pending await locally
+ # with CONNECTION_CLOSED - a client-side synthesis, not a server frame - so the
+ # only outcome ever recorded is that local resolution; a served result never appears.
+ assert all(outcome == f"error:{CONNECTION_CLOSED}" for outcome in outcomes)
</file context>
| init = await wire.response_to(1) | ||
| # Drain the pipelined request: every request is owed an answer, but its content is | ||
| # not era-diagnostic here (a legacy-served tools/list and a modern one look alike). | ||
| await wire.response_to(2) |
There was a problem hiding this comment.
P2: The pipelined legacy-first case does not verify the era of the request behind the handshake: it discards request 2's response, so an implementation could accept the handshake but incorrectly serve the following modern-enveloped request and this regression test would remain green. Retaining the response and asserting JSONRPCError with INVALID_REQUEST would cover the stated legacy-commitment behavior.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/server/test_stdio_ordering.py, line 312:
<comment>The pipelined legacy-first case does not verify the era of the request behind the handshake: it discards request 2's response, so an implementation could accept the handshake but incorrectly serve the following modern-enveloped request and this regression test would remain green. Retaining the response and asserting `JSONRPCError` with `INVALID_REQUEST` would cover the stated legacy-commitment behavior.</comment>
<file context>
@@ -0,0 +1,477 @@
+ init = await wire.response_to(1)
+ # Drain the pipelined request: every request is owed an answer, but its content is
+ # not era-diagnostic here (a legacy-served tools/list and a modern one look alike).
+ await wire.response_to(2)
+ assert isinstance(init, JSONRPCResponse), f"the handshake that opened the connection must be accepted, got {init}"
+ assert init.result["protocolVersion"] == LATEST_HANDSHAKE_VERSION
</file context>
| await client.send_raw_request("initialize", _initialize_params()) | ||
| await client.notify("notifications/initialized", None) | ||
| with pytest.raises(MCPError) as exc: | ||
| await client.send_raw_request("tools/list", _modern_params()) |
There was a problem hiding this comment.
P2: This test does not exercise the behavior named in test_modern_client_notification_during_in_flight_work: the awaited tools/list call has already completed before the notification is sent. A regression that drops or misroutes notifications while a modern request body is running would still pass, so a blocked handler plus an explicit release after the notification would make this coverage meaningful.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/server/test_serving.py, line 200:
<comment>This test does not exercise the behavior named in `test_modern_client_notification_during_in_flight_work`: the awaited `tools/list` call has already completed before the notification is sent. A regression that drops or misroutes notifications while a modern request body is running would still pass, so a blocked handler plus an explicit release after the notification would make this coverage meaningful.</comment>
<file context>
@@ -0,0 +1,633 @@
+ await client.send_raw_request("initialize", _initialize_params())
+ await client.notify("notifications/initialized", None)
+ with pytest.raises(MCPError) as exc:
+ await client.send_raw_request("tools/list", _modern_params())
+ # ...and the connection is still the same legacy connection afterwards.
+ result = await client.send_raw_request("tools/list", None)
</file context>
| await bus.publish(ToolsListChanged()) | ||
| # The stream flushes the event, then the graceful close ends it: the | ||
| # response only arrives once the handler returns its result. | ||
| recorder.notified = anyio.Event() |
There was a problem hiding this comment.
P2: This listen test has a delivery race: bus.publish() can yield to the client dispatcher before the event replacement, causing the delivered-event wait to wait on a fresh event that was never set and time out. Waiting for the recorded notification count or using a per-method event avoids replacing a shared signal after publishing.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/server/test_serving.py, line 539:
<comment>This listen test has a delivery race: `bus.publish()` can yield to the client dispatcher before the event replacement, causing the delivered-event wait to wait on a fresh event that was never set and time out. Waiting for the recorded notification count or using a per-method event avoids replacing a shared signal after publishing.</comment>
<file context>
@@ -0,0 +1,633 @@
+ await bus.publish(ToolsListChanged())
+ # The stream flushes the event, then the graceful close ends it: the
+ # response only arrives once the handler returns its result.
+ recorder.notified = anyio.Event()
+ await recorder.notified.wait() # the delivered event
+ server.close_subscriptions() # server-initiated graceful close
</file context>
| first = await s2c_recv.receive() | ||
| second = await s2c_recv.receive() | ||
| assert isinstance(first, SessionMessage) and isinstance(second, SessionMessage) | ||
| assert first.message == JSONRPCResponse(jsonrpc="2.0", id=1, result={"served_by": "alternate"}) | ||
| assert second.message == JSONRPCResponse(jsonrpc="2.0", id=2, result={"echoed": "t", "params": {}}) |
There was a problem hiding this comment.
P3: This test assumes request id 1 is written before id 2 even though both handlers run concurrently and JSON-RPC responses may arrive out of order. Scheduler differences can make the test flaky; collect both frames and assert them by id instead.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/shared/test_jsonrpc_dispatcher.py, line 2198:
<comment>This test assumes request id 1 is written before id 2 even though both handlers run concurrently and JSON-RPC responses may arrive out of order. Scheduler differences can make the test flaky; collect both frames and assert them by id instead.</comment>
<file context>
@@ -2037,6 +2132,147 @@ def builder(_meta: MessageMetadata) -> TransportContext:
+ )
+ await c2s_send.send(SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=2, method="t", params=None)))
+ with anyio.fail_after(5):
+ first = await s2c_recv.receive()
+ second = await s2c_recv.receive()
+ assert isinstance(first, SessionMessage) and isinstance(second, SessionMessage)
</file context>
| first = await s2c_recv.receive() | |
| second = await s2c_recv.receive() | |
| assert isinstance(first, SessionMessage) and isinstance(second, SessionMessage) | |
| assert first.message == JSONRPCResponse(jsonrpc="2.0", id=1, result={"served_by": "alternate"}) | |
| assert second.message == JSONRPCResponse(jsonrpc="2.0", id=2, result={"echoed": "t", "params": {}}) | |
| responses = [await s2c_recv.receive(), await s2c_recv.receive()] | |
| assert all(isinstance(response, SessionMessage) for response in responses) | |
| by_id = { | |
| response.message.id: response.message | |
| for response in responses | |
| if isinstance(response, SessionMessage) | |
| } | |
| assert by_id == { | |
| 1: JSONRPCResponse(jsonrpc="2.0", id=1, result={"served_by": "alternate"}), | |
| 2: JSONRPCResponse(jsonrpc="2.0", id=2, result={"echoed": "t", "params": {}}), | |
| } |
| # blocked senders and revoke open channels. | ||
| self._running = False | ||
| self._closed = True | ||
| await self._on_read_eof() | ||
| self._revoke_in_flight() | ||
| self._fan_out_closed() |
There was a problem hiding this comment.
🔴 In run()'s EOF arm, self._closed = True is set before await self._on_read_eof(), so during serve_stream's drain window every buffered listen event (and any progress notification) is silently dropped by notify()'s _closed gate — only the final SubscriptionsListenResult lands, because reply() bypasses the gate. This contradicts the behavior this PR documents in docs/handlers/subscriptions.md ("any events it still had buffered flush as time allows"); fix by running the drain hook before flipping _closed, or gating notify() on a separate peer-gone flag set after the drain.
Extended reasoning...
The bug. The EOF arm of JSONRPCDispatcher.run() executes in this order (jsonrpc_dispatcher.py:557-561):
self._running = False
self._closed = True # <-- flipped first
await self._on_read_eof() # serve_stream's _drain runs here
self._revoke_in_flight()
self._fan_out_closed()and notify() short-circuits unconditionally once _closed is set (jsonrpc_dispatcher.py:491-493, "dropped %s: dispatcher closed"). So the entire drain window — the new _EOF_DRAIN_WINDOW machinery whose whole purpose is to flush graceful-close traffic "while the output is still up" (_StreamConnection._drain's docstring) — runs with all notification writes disabled.
The code path that triggers it. At stdin EOF, _StreamConnection._drain (serving.py) calls server.close_subscriptions(), which closes each open ListenHandler stream's send side. Each handler's async for event in recv loop (subscriptions.py:226-229) then drains its remaining buffered events, delivering each via ctx.session.send_notification(event_to_notification(...), related_request_id=subscription_id). That routes through the request's dispatch context — whose target is still a live _Wire, since _revoke_in_flight() runs only after the drain hook — into _Wire.notify -> dispatcher.notify -> the _closed gate, where every event is silently discarded. The same gate also drops progress notifications from in-flight requests finishing inside the shielded 2s wait_for_in_flight() window.
Why the final result still lands (and why the tests miss it). The listen handler's SubscriptionsListenResult goes dctx.reply() -> _spend() -> _Wire.write -> _write_frame -> _write_stream.send, which never consults _closed, and the write stream stays open until the task-group join. So the result is delivered while every buffered event is not. The existing EOF test (test_stdin_eof_ends_open_listen_streams_gracefully...) publishes and delivers its one event before closing stdin, so it only ever observes ack + result — the exact frames that survive — and cannot catch this. One verifier also reproduced it empirically: driving a dispatcher with an on_read_eof hook mimicking serve_stream's drain, a handler that flushed a notification then returned its result left only the JSONRPCResponse on the wire; the notification was dropped.
Why this is wrong, not just suboptimal. The delivery premise is sound: at stdin EOF the peer closed only its write side and can still read stdout — that is the entire reason the shielded drain window exists. And this PR itself documents the intended behavior in docs/handlers/subscriptions.md: "when the client's input ends, the driver closes your open streams for you inside a short bounded window, in which each stream's final result is guaranteed to reach the departing peer and any events it still had buffered flush as time allows." In reality zero buffered events ever flush — not "as time allows" but never. ListenHandler.close()'s docstring ("drains its buffered events and sends its SubscriptionsListenResult") makes the same promise.
Step-by-step proof.
- Client opens
subscriptions/listen(idsub-1) over stdio; server acks. - Server publishes two
ToolsListChangedevents while the client's reader is briefly slow, so both sit buffered in the listen stream. - Client closes its write side (stdin EOF at the server) while still reading stdout.
run()exits the read loop, sets_closed = True(line 558), then awaits_on_read_eof()(line 559)._draincallsclose_subscriptions(); the handler'sasync for event in recvyields both buffered events and callssend_notification(..., related_request_id="sub-1")for each.- Both calls reach
dispatcher.notify, hitif self._closed:at line 491, and are dropped with a debug log. - The handler returns;
reply()writes theSubscriptionsListenResultthrough_write_frame, which lands. The departing peer receives the result but neither event — the documented "flush as time allows" delivered nothing.
Fix. A small reorder: flip _closed (or the part of it that gates notify()) after await self._on_read_eof() — e.g. run the drain first, or introduce a separate peer-gone flag set after the drain — keeping the post-drain _revoke_in_flight()/_fan_out_closed() sequence as-is. All three verifiers confirmed independently (one with an empirical wire-level reproduction); there were no refutations.
| async def _drain(self) -> None: | ||
| """The peer closed our input: close listen streams, then let in-flight work end | ||
| gracefully within `_EOF_DRAIN_WINDOW`, while the output is still up.""" | ||
| self._server.close_subscriptions() | ||
| with anyio.move_on_after(_EOF_DRAIN_WINDOW, shield=True): | ||
| await self._dispatcher.wait_for_in_flight() |
There was a problem hiding this comment.
🔴 Any legacy streamable-HTTP session ending now closes every open subscriptions/listen stream server-wide: serve_legacy_stream builds a _StreamConnection whose on_read_eof hook (_drain) unconditionally calls self._server.close_subscriptions() on the shared application-wide Server, so a routine legacy client DELETE (or any session teardown) gracefully ends all modern 2026-07-28 clients' listen streams on unrelated connections — a regression, since the old serve_loop had no EOF hook. serve_legacy_stream should skip (or scope to the connection) the close_subscriptions() call; the drain hook was designed for stdio, where one process is one connection.
Extended reasoning...
The bug
_StreamConnection.__init__ unconditionally installs on_read_eof=self._drain on its JSONRPCDispatcher (src/mcp/server/serving.py:168), and _drain (serving.py:239-244) calls self._server.close_subscriptions(). That hook was designed for the stdio wind-down ("the peer closed our input... close listen streams"), where one process = one connection and closing the server's streams only affects the departing peer.
But serve_legacy_stream (serving.py:295-318) builds the same _StreamConnection, and StreamableHTTPSessionManager.run_server now drives every stateful legacy HTTP session through serve_legacy_stream(self.app, ...) (streamable_http_manager.py:324) — where self.app is the one application-wide Server. Server.close_subscriptions() reaches the registered ListenHandler and calls ListenHandler.close(), which ends every stream in the handler's stream set — including the listen streams that handle_modern_request serves for 2026-07-28 clients through the same handler on the same Server, on completely unrelated HTTP connections.
Step-by-step proof
- A mixed-era deployment: modern client M holds an open
subscriptions/listenstream over streamable HTTP (served byhandle_modern_request(self.app, ...)throughself.app'sListenHandler); legacy client L has an ordinary stateful 2025-era session on the same server. - L closes cleanly. The SDK's own legacy client defaults
terminate_on_close=Trueand sendsDELETE(src/mcp/client/streamable_http.py:652,715). - The server's
_handle_delete_requestcallsterminate()(streamable_http.py:777,785), which closes_read_stream_writer/_read_stream(lines ~805-807) while L's session serve task is parked in the dispatcher's read loop. JSONRPCDispatcher.run()treatsEndOfStream/ClosedResourceErroras EOF and awaitsself._on_read_eof()(jsonrpc_dispatcher.py:552-559) →_drain→self._server.close_subscriptions()→ListenHandler.close().- M's listen stream — on an unrelated connection — is gracefully ended: M receives its listen request's result and must re-listen and refetch, losing any events published between the close and the re-listen (there is no replay). Every other modern subscriber is churned the same way, on every legacy DELETE or session teardown.
Note the close is pure collateral: a legacy-era connection cannot itself own listen streams (subscriptions/listen is modern-only; the legacy era refuses enveloped requests and a bare one is METHOD_NOT_FOUND), so the terminating session gains nothing from the call.
Why this is a regression and why nothing else prevents it
On main, the manager drove legacy sessions through serve_loop, whose dispatcher had no EOF hook, and close_subscriptions() did not exist — legacy session teardown never touched listen streams. The PR's docs document the shared-ListenHandler caveat only for stdio ("It never bites on stdio... behind a listener it does") and serve_listener socket hosts; nothing covers this streamable-HTTP trigger, and no test pins it for HTTP (the shared-server test in test_serving.py covers two serve_stream connections, not the HTTP manager path). Each legacy session teardown additionally gains a shielded up-to-2s wait_for_in_flight drain (_EOF_DRAIN_WINDOW) that HTTP teardown never had.
The PR description's "still landing" note (moving listen-stream ownership per-connection and dropping the server-level close_subscriptions() verb) acknowledges the general shared-handler coupling, but that follow-up is not in this diff — as it stands, a routine legacy client disconnect in any mixed-era streamable-HTTP deployment churns all modern subscribers server-wide.
Fix
serve_legacy_stream should not close subscriptions on EOF: either pass a flag into _StreamConnection so its _drain skips close_subscriptions() for transport-internal legacy HTTP sessions (a legacy session can never own a listen stream, so nothing is lost), or scope the close to streams owned by the ending connection. The stdio/serve_stream behavior can stay as documented.
| if msg.method == "notifications/cancelled": | ||
| rid = cancelled_request_id_from_params(msg.params) | ||
| if rid is not None and (in_flight := self._in_flight.get(coerce_request_id(rid))) is not None: | ||
| in_flight.dctx.cancel_requested.set() | ||
| if self._peer_cancel_mode == "interrupt": | ||
| # The peer withdrew the request: revoke its channel, then interrupt the work. | ||
| in_flight.dctx.close() | ||
| in_flight.scope.cancel() |
There was a problem hiding this comment.
🔴 Peer-cancelled requests never leave the dispatcher's _in_flight table: the interrupt path (_dispatch_notification, jsonrpc_dispatcher.py:679-686) only revokes the channel and cancels the scope, and the sole removal point — _spend(), reached via reply()/reply_error() — never runs because the body's cancellation is absorbed at the with scope: exit. Each peer cancel therefore leaks one _InFlight entry (CancelScope + full dispatch context, including message_metadata, which on legacy streamable HTTP holds the Starlette Request) for the connection's lifetime, growing without bound over repeated listen open/cancel cycles on a long-lived stdio connection — the exact flow this PR enables. The fix is to pop the entry on the interrupt path (or after scope.cancelled_caught); _spend()'s entry.dctx is self identity guard already makes that safe for the deferred cancel-as-wakeup case.
Extended reasoning...
The bug
This PR restructures request cleanup: the old _handle_request had an inner finally that always popped the request's _in_flight entry (with an identity guard against evicting a reused id's newer entry). That finally is gone, and removal now lives solely in _JSONRPCDispatchContext._spend() (src/mcp/shared/jsonrpc_dispatcher.py:238-246), which is reachable only through reply() / reply_error().
The peer-cancel interrupt path never gets there. In _dispatch_notification (lines 679-686), a notifications/cancelled for an in-flight id does:
in_flight.dctx.cancel_requested.set()
if self._peer_cancel_mode == "interrupt":
in_flight.dctx.close() # swaps the write target for _VOID — does NOT touch _in_flight
in_flight.scope.cancel() # interrupts the bodyclose() (lines 234-236) only does self._target = _VOID. The scope cancel then interrupts the body inside _handle_request (line 777), and because it is the scope's own cancellation, anyio absorbs it at the with scope: exit (cancelled_caught semantics). Execution falls straight through to the finally, which only calls _release_request_task() — a counter decrement. Neither reply() nor reply_error() runs (that silence is the PR's cancel design, and it is correct on the wire), so _spend() never runs, and self._in_flight[coerce_request_id(req.id)] is never deleted. _revoke_in_flight() at EOF (lines 817-820) also only calls close() on each entry without clearing the dict.
Step-by-step proof
- Client sends request id 1 (
subscriptions/listen, or any slow handler)._dispatch_requestinserts_in_flight[1] = _InFlight(scope, dctx)and spawns_handle_request; the body parks (e.g. inanyio.sleep_forever()or waiting on the bus). - Client sends
notifications/cancelledfor id 1._dispatch_notificationfinds the entry, callsdctx.close()(target →_VOID) andscope.cancel(). - The body's await raises the cancelled exception; it propagates to the
with scope:exit in_handle_request, where anyio absorbs it (scope.cancelled_caught = True). No exception escapes, so theexcept anyio.get_cancelled_exc_class()arm — which fires only for outer cancels that re-raise past the scope — does not run. - The
finallydecrements_request_tasksto 0._in_flightstill contains key 1, holding theCancelScopeand the full_JSONRPCDispatchContext— includingmessage_metadata, which for the legacy streamable-HTTP path is aServerMessageMetadatacarrying the entire StarletteRequest. - Repeat: every open/cancel cycle adds one entry. A verifier ran this empirically against in-memory streams: after 5 cancels,
_in_flightheld 5 entries while_request_taskswas 0. The only escapes are the client reusing the same request id (blind overwrite) or connection teardown.
Why this matters and why nothing catches it
Repeated listen open/cancel cycles on a long-lived stdio connection are precisely the flow this PR ships (client.listen(...) cancels via notifications/cancelled over stdio now), so the leak accrues on the PR's own headline path — unbounded memory growth in the core dispatcher, on every seat and transport that uses peer_cancel_mode="interrupt" (the server default). The 18-case cancel-race suite asserts wire frames only; no test asserts table state after a cancel, so it merges green.
This is a regression, not a design necessity. The _InFlight docstring itself says the entry is "keyed until its answer commits", and a cancelled request never commits. Popping the entry in the interrupt branch (right next to dctx.close()), or after scope.cancelled_caught in _handle_request with the old identity guard, is safe: _spend() already guards with entry.dctx is self, so the deferred cancel-as-wakeup case (a handler that completes anyway and replies into the void) cleans up correctly either way.
All three verifiers confirmed independently, one with an empirical reproduction; there were no refutations.
| * `server.streamable_http_app()` returns the same Starlette app `MCPServer`'s does; deploy it the way **[Running your server](../run/index.md)** deploys any other ASGI app. There is no `server.run(transport=...)` down here: `server.run(read_stream, write_stream, server.create_initialization_options())` drives one connection over a pair of streams, and that one line is the whole story. | ||
| * `server.streamable_http_app()` returns the same Starlette app `MCPServer`'s does; deploy it the way **[Running your server](../run/index.md)** deploys any other ASGI app. There is no `server.run(transport=...)` down here: `server.run(read_stream, write_stream)` drives one connection over a pair of duplex streams (stdio, a socket you framed, an in-memory pair in a test), and that one line is the whole story. Every other hosting shape is the next section. | ||
|
|
||
| ## Serving over your own transport |
There was a problem hiding this comment.
why even add this, this isn't needed int his PR
| """Return the registered entry for a notification method, or `None`.""" | ||
| return self._notification_handlers.get(method) | ||
|
|
||
| def close_subscriptions(self) -> None: |
| def subscriptions(self) -> SubscriptionBus: | ||
| """The server's subscription bus; publish here to reach open listen streams: | ||
| `await mcp.subscriptions.publish(ToolsListChanged())`.""" | ||
| return self._subscriptions | ||
|
|
||
| def close_subscriptions(self) -> None: | ||
| """Gracefully close every open `subscriptions/listen` stream (all connections). | ||
|
|
||
| Streams drain their pending events and end with the listen request's | ||
| result while the connection carries on; returns before they finish flushing. | ||
| """ | ||
| self._lowlevel_server.close_subscriptions() |
There was a problem hiding this comment.
neither of these should be here. this is making state available on the MCPServer instead of the context object, bad
| @property | ||
| def lowlevel_server(self) -> Server[LifespanResultT]: | ||
| """The low-level `Server` this instance is built on. | ||
|
|
||
| Hand it to the stream drivers `run()` does not name: | ||
|
|
||
| listener = await anyio.create_unix_listener("/tmp/mcp.sock") | ||
| await serve_listener(mcp.lowlevel_server, listener) | ||
| """ | ||
| return self._lowlevel_server |
There was a problem hiding this comment.
why are you making this public? this shouldn't be public I see no reason for it relating to what this PR is trying to achieve
| "serve_listener", | ||
| "serve_stream", |
There was a problem hiding this comment.
what are these why are they public now?
|
|
||
|
|
||
| @asynccontextmanager | ||
| async def newline_json_transport( |
There was a problem hiding this comment.
why make this public? we're not wanting to support custom transports. this PR should just be about making stdio work for 2026 subscriptions. refactoring to make the internals is better, but we're not wanting to add a bunch of new functionality here. that isn't the goal
| @dataclass(frozen=True, slots=True) | ||
| class Admission: | ||
| """The receive-order verdict for one inbound request. | ||
|
|
||
| `handler` serves it; when `hold` is true the receive loop parks until the | ||
| request finishes answering, so its body must not await the peer. | ||
| """ | ||
|
|
||
| handler: OnRequest | ||
| hold: bool = False | ||
|
|
||
|
|
||
| Admit = Callable[[str, Mapping[str, Any] | None], Admission] | ||
| """Synchronous receive-order gate for inbound requests: `(method, params) -> Admission`. |
There was a problem hiding this comment.
what are these for and whya re they in this file if they're not used here
e74eee3 to
1e4d87e
Compare
There was a problem hiding this comment.
Beyond the earlier inline findings, this pass examined and ruled out a few more candidates: the removal of the denying dispatch-context wrapper around modern_on_request (the SDK's only wiring builds the direct pair with can_send_request=False, and the direct dispatcher's context refuses requests off that flag, so the modern no-backchannel rule still holds on that entry); MCPServer.get_prompt's catch-all mapping to -32602 with the exception text (the messages surfaced there are MCPServer's own authored prompt errors, which is this PR's documented intent, not a handler-internals leak); and three suspected test races/gaps in tests/server/test_serving.py and tests/shared/test_jsonrpc_dispatcher.py (each is ordered by the dispatcher's sequential read loop or by explicit events, not the scheduler).
Extended reasoning...
This run found no new bugs. The three inline findings from the earlier pass, the unaddressed maintainer comments, and the pending rebase already make clear a human review is in progress, so this note only records what was additionally examined and refuted this run — it is informational, not a correctness guarantee, and does not restate the inline findings.
Motivation
The 2026-07-28 protocol did not work over stdio, and when I dug in it came down to one refused
feature plus a handful of conformance defects that all shared a root cause. A connection's
protocol era (the 2025
initializehandshake versus the 2026 self-contained per-requestenvelope) was being derived after the fact from which requests had completed, in several
places, instead of being decided once, in wire order, from how the client opened the connection.
That model couldn't host a request that never returns, so
subscriptions/listenwashard-refused over stdio with
-32601even though the same connection advertisedtools.listChanged: true; a slow modern request left the connection "unlocked" so a legacyhandshake arriving mid-flight was accepted and re-locked the connection out from under the
running call; and a peer cancel produced a trailing
{"code": 0, "message": "Request cancelled"}frame that the stdio binding forbids ("MUST NOT send any further messages" for acancelled request).
The listen handler was never the problem. Served through a driver that decides the era from
the opening message, the unmodified handler produces the exact spec flow over a pipe (ack first,
id-stamped events, graceful close, silence after a client cancel). So this change is not "make
listen work on stdio"; it is "decide the era once, in the right place, and delete the code that
was compensating for deciding it wrong."
What changed
One stream driver, era decided in the read loop.
serve_dual_era_loopandserve_looparereplaced by a single
serve_stream(server, read, write). The connection's era is decidedsynchronously, inside the JSON-RPC dispatcher's sequential read loop, before the request body is
spawned and before the next frame is read, so it is ordered by the wire and by nothing else (no
appeal to task scheduling anywhere).
initializeor any envelope-less request opens the legacyera; an enveloped request opens the modern era; an enveloped
server/discoveris a probe thatis answered but pins nothing, so a client that probes and falls back to
initializestillfinds the handshake available; a stray leading notification opens nothing. A second,
conflicting era claim on a committed connection is refused rather than silently switching (a
legacy
initializeon a modern connection gets-32022naming the served versions; anenveloped request on a handshake connection gets
-32600). Both eras are always served; thereis no server-level switch to disable one, matching the other SDKs.
Cancel silence by construction. Each request now answers through a one-shot channel whose
write target is swapped for a powerless void by a peer cancel, by the request's own terminal
write, and by the peer going away. There is no
if cancelledat any write site. The commitpoint is one synchronous step (leave the cancellable table and take the write target, no
checkpoint between), which gives a clean rule for the race: if the cancel is read before the
body returned, nothing is written; if the body had already returned, the owed answer is written
exactly once even when parked on backpressure. That satisfies both the stdio MUST NOT and the
cancellation pattern's "MAY ignore a cancel whose processing has completed" at the same time.
The dispatcher knows less MCP than before. The generic JSON-RPC dispatcher loses the code-0
cancel frame and the code-0
str(exc)catch-all (both legacy-era vocabulary in plumbing thatshouldn't know eras exist) and the
inline_methods={"initialize"}knob. It gains twoMCP-agnostic mechanisms: an optional receive-order admission gate on
run(admit=)and therequest channel. The
Dispatchercontract'sOnRequest/OnNotifyare now documented as invokedsynchronously in receive order, with the returned awaitable as the body; every existing
async defhandler already satisfies that, since calling anasync defdoes no work. This isthe same shape as the go-sdk's jsonrpc2
Preempter(a pre-queue admission hook), so it'sgeneric JSON-RPC prior art rather than an MCP intrusion.
Author-facing surface.
Server.run(read, write)now stands alone(
initialization_optionsis optional).Server.lifespan()is a bound zero-argument contextmanager, so the odd
server.lifespan(server)self-reference is gone. New:serve_listener(server, listener), the socket-shaped host that enters the lifespan once and drives every acceptedconnection, and
mcp.server.stdio.newline_json_transport(byte_stream), the newline-delimitedJSON-RPC framer for any byte stream. The custom-transport docs are now a four-rung ladder from
server.run()up to composingserver.lifespan()+serve_stream(..., lifespan_state=)yourself, with the enter-the-lifespan-once rule stated on the rung where it bites.
Other observable changes (all in the migration guide with before/after): the discovery
probe no longer pins; a modern request needs only the required
_metapair (clientInfoisoptional); unmapped handler exceptions become
-32603 "Internal server error"with the textkept in the server log rather than on the wire;
MCPServerprompt errors are-32602with theauthored message; at stdin EOF the server drains gracefully (open listens end with their result)
instead of writing
CONNECTION_CLOSEDper id onto a closing pipe; over legacy streamable HTTPin JSON mode a peer-cancelled POST now completes as an empty
202 Acceptedinstead of lingeringuntil the client disconnects.
Migration
Everything is grouped in
docs/migration.md; the short version:serve_loop(...)/serve_dual_era_loop(...)becomeserver.run(read, write)for a singleconnection, or
serve_stream(..., lifespan_state=)when connections share a lifespan youentered yourself.
server.lifespan(server)becomesasync with server.lifespan() as state:(aTypeErrorotherwise). The constructor's
lifespan=callable is unchanged.Dispatcher, note the documented obligation that handlers areinvoked synchronously in receive order and return the body awaitable, and that
JSONRPCDispatcher(inline_methods=)is gone (statehold=(method == "initialize")in yourown admission via
run(admit=)if you need the handshake to hold the loop).notifications/cancelledwhile still awaiting the request now gets noresponse for the cancelled id (that's the spec); code that read exception text off
code=0errors should raise
MCPErrorto put a deliberate message on the wire.MCPServer.get_prompt()raisesMCPError(wasValueError) for its own prompt errors.Testing
asyncioandtrio: a wire-orderingsuite (era decided by the opening message, the straddle refused, probe-then-fallback,
stray-notification and claim-less-request rules, cancel silence), and an 18-case cancel-race
suite pinning every direction of the cancel-versus-answer race, the EOF cases, and the
owner-teardown cases, with no fixed sleeps. Plus a streamable-HTTP suite for the JSON-mode
cancelled-POST completion, and the existing stream-driver, stdio, and dispatcher suites
parametrized over both anyio backends.
strict-no-coverclean; pyright and ruff clean;no new
# pragma: no cover,# type: ignore, or# noqainsrc/.concurrent traffic, published event, client cancel with no trailing frame, server-initiated
graceful close, legacy handshake during in-flight modern work, pair-only envelope routed
modern, late handshake refused
-32022, and probe-then-fallback.against this server (probe to
DiscoverResultto modern requests, ack-first stamped listen andcancel silence, and a forced probe timeout landing on a working legacy handshake on the same
connection), and this client against its server.
Still landing on this branch
mainnow that Align with spec #3002: optional clientInfo, serverInfo in result _meta #3143 has merged (the envelope work overlaps the sameclassification code), which also retires the interop caveat about the previously required
clientInfo.ListenHandlerand dropping the server-levelclose_subscriptions()verb: streams belong to the connection that opened them (and end withit), the handler is stateless, and a graceful close-everything is the subscription bus's job.
Today, as on
main, aServershared across connections ends every connection's listenstreams when one connection closes; this fixes that by scope rather than by caveat.
Out of scope, deliberately: making one publish reach a 2026 listen stream and a 2025 client
(feeding legacy peers from the subscription bus) is a separate follow-up, since it changes what
a legacy handshake advertises and delivers.
AI Disclaimer