Rebuild the streamable HTTP server transport around per-request dispatch - #3192
Rebuild the streamable HTTP server transport around per-request dispatch#3192maxisbey wants to merge 7 commits into
Conversation
Moves outbound request correlation, the inbound in-flight table, and the exception-to-wire policy into shared/_correlation.py so a transport without a stream pair can share the semantics. JSONRPCDispatcher delegates; behaviour unchanged.
The per-session message router and its stream fan-out are gone, so the tests pinning that mechanism are rewritten against the behaviour they guarded (priming-store failure returns 500 with no leaked state; standalone stream teardown via close_standalone_sse_stream logs no error; the manager evicts sessions whose task exits or crashes; stateless requests leave no channels behind) or deleted where the guarded race is now unrepresentable (#1764 router head-of-line blocking, the standalone writer between-dequeues window). The close_sse_stream protocol-version gating tests move to the renamed metadata builder.
Removes now-dead code (the attach take-over branch, redundant containment around the notification handler, an unreachable None-connection arm, the channel-closing loop in run() teardown) and adds tests for the behaviours that were untested: terminating a session with a request in flight, client-posted progress for a server-initiated request, containment of a raising event store per request, concurrent POSTs sharing a request id, the channel attach/detach identity guard, and the serve_loop driver.
Records the removal of StreamableHTTPServerTransport.connect() (the transport is now driven per request by the session manager) and the behaviours clarified alongside it, and refreshes the docstrings that still described the old serve_loop wiring.
📚 Documentation preview
|
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
1 issue found across 13 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="src/mcp/server/streamable_http.py">
<violation number="1" location="src/mcp/server/streamable_http.py:314">
P1: Standalone server requests in JSON-response mode can hang instead of raising `NoBackChannelError`. `ServerSession.send_request()` without `related_request_id` selects `connection.outbound`, and this outbound implementation always writes to the GET channel even though JSON mode is documented and otherwise modeled as having no request back-channel. Make the connection's standalone outbound refuse requests in JSON-response mode (while retaining its best-effort notification behavior), so this path has the same contract as request-scoped sends.</violation>
</file>
Tip: cubic can generate docs of your entire codebase and keep them up to date. Try it here.
Re-trigger cubic
| ) -> dict[str, Any]: | ||
| if not self.can_send_request: | ||
| raise NoBackChannelError(method) | ||
| return await _call_over_channel(self._corr, self._channel, method, params, opts) |
There was a problem hiding this comment.
P1: Standalone server requests in JSON-response mode can hang instead of raising NoBackChannelError. ServerSession.send_request() without related_request_id selects connection.outbound, and this outbound implementation always writes to the GET channel even though JSON mode is documented and otherwise modeled as having no request back-channel. Make the connection's standalone outbound refuse requests in JSON-response mode (while retaining its best-effort notification behavior), so this path has the same contract as request-scoped sends.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/mcp/server/streamable_http.py, line 314:
<comment>Standalone server requests in JSON-response mode can hang instead of raising `NoBackChannelError`. `ServerSession.send_request()` without `related_request_id` selects `connection.outbound`, and this outbound implementation always writes to the GET channel even though JSON mode is documented and otherwise modeled as having no request back-channel. Make the connection's standalone outbound refuse requests in JSON-response mode (while retaining its best-effort notification behavior), so this path has the same contract as request-scoped sends.</comment>
<file context>
@@ -139,18 +160,237 @@ async def replay_events_after(
+ ) -> dict[str, Any]:
+ if not self.can_send_request:
+ raise NoBackChannelError(method)
+ return await _call_over_channel(self._corr, self._channel, method, params, opts)
+
+ async def progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
</file context>
- Serialize store-then-forward per channel: concurrent writers on one channel (the standalone GET stream) could put frames on the wire out of event-store order, breaking Last-Event-ID resumption. A per-channel lock restores the store-order == wire-order invariant the serial router used to provide. - Refuse a request that arrives across session termination instead of running it: dispatch now branches on the transport's identity (stateful vs stateless) rather than on session-task presence, re-checks liveness after the request's awaits, and answers 404 for an ended session; queued work for an ended session is dropped. - Contain event-store failures at the channel: a raising store_event was reaching the correlator as a handler error and leaking its text onto the wire; write() now never raises (a broken store ends that stream), which also protects the courtesy-cancel write. - Close the replay reader when priming fails; drop dead exception arms in the SSE pump. - Correct the migration note's JSON-mode claim to request-scoped requests, drop a claim handlers cannot observe, fix a vacuous test assertion, and add regression tests for each fix.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
There was a problem hiding this comment.
All reported issues were addressed across 5 files (changes from recent commits).
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
The second review round found several defects that traced back to a few structural gaps rather than isolated bugs; this addresses the gaps. - Channel delivery is now a value: write() returns whether the message reached the event store or an attached response, an EventStore failure is contained inside the write (resumability degrades, the stream survives, no store text on the wire), and attach() returns None on a dead channel so nothing can stream from one. A server-to-client request that cannot reach any client fails the caller with CONNECTION_CLOSED instead of parking it, and the JSON-response body derives from the channel's recorded outcome (result, terminated-404, or 500) rather than a "cannot happen" branch. - Stream ids handed to the EventStore are minted by the transport in a session-scoped namespace, so a client-chosen request id can no longer name the standalone GET stream and two sessions on one store can no longer replay each other's frames. - One SSE-response runner owns the pump, error containment, and cleanup for the POST, GET, and replay responses (the POST path had lost the guard its siblings kept). - A session-level gate holds requests that arrive while an initialize is still being served until the handshake commits, restoring the ordering the stream-pair driver's parked read loop used to guarantee. - The POSTed client message is delivered even when the 202 could not be written back, and the correlator marks the single site where the cancelled-request answer policy lives for every transport. Adds regression tests for each of the above.
There was a problem hiding this comment.
2 issues found across 7 files (changes from recent commits).
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="src/mcp/server/streamable_http.py">
<violation number="1" location="src/mcp/server/streamable_http.py:512">
P2: Concurrent POSTs that reuse a JSON-RPC id still share the same stored stream and `_streams` entry. Their direct SSE responses are separate, but once either response is closed, replay/live-tail and `ctx.close_sse_stream()` can target the other request (or lose the still-running request entirely). A per-dispatch stream token, with the request's close callback bound to that channel, would keep resumability and stream control isolated as well.</violation>
<violation number="2" location="src/mcp/server/streamable_http.py:977">
P2: Requests queued behind a slow initialization remain hung after DELETE instead of reaching `_start_request()` and returning its terminated-session 404. Release the current initialization gate when termination begins.</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
| initialize_gate = anyio.Event() | ||
| self._initializing = initialize_gate | ||
| elif stateful and (gate := self._initializing) is not None: | ||
| await gate.wait() |
There was a problem hiding this comment.
P2: Requests queued behind a slow initialization remain hung after DELETE instead of reaching _start_request() and returning its terminated-session 404. Release the current initialization gate when termination begins.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/mcp/server/streamable_http.py, line 977:
<comment>Requests queued behind a slow initialization remain hung after DELETE instead of reaching `_start_request()` and returning its terminated-session 404. Release the current initialization gate when termination begins.</comment>
<file context>
@@ -922,7 +961,52 @@ async def _serve_request(
+ initialize_gate = anyio.Event()
+ self._initializing = initialize_gate
+ elif stateful and (gate := self._initializing) is not None:
+ await gate.wait()
+
+ # From here the gate must be released whatever becomes of this
</file context>
| neither reachable from another session sharing the store nor equal to | ||
| the standalone stream's id, whatever the client picks as request id. | ||
| """ | ||
| return f"{self._stream_scope}:request:{request_id}" |
There was a problem hiding this comment.
P2: Concurrent POSTs that reuse a JSON-RPC id still share the same stored stream and _streams entry. Their direct SSE responses are separate, but once either response is closed, replay/live-tail and ctx.close_sse_stream() can target the other request (or lose the still-running request entirely). A per-dispatch stream token, with the request's close callback bound to that channel, would keep resumability and stream control isolated as well.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/mcp/server/streamable_http.py, line 512:
<comment>Concurrent POSTs that reuse a JSON-RPC id still share the same stored stream and `_streams` entry. Their direct SSE responses are separate, but once either response is closed, replay/live-tail and `ctx.close_sse_stream()` can target the other request (or lose the still-running request entirely). A per-dispatch stream token, with the request's close callback bound to that channel, would keep resumability and stream control isolated as well.</comment>
<file context>
@@ -479,6 +498,19 @@ def is_terminated(self) -> bool:
+ neither reachable from another session sharing the store nor equal to
+ the standalone stream's id, whatever the client picks as request id.
+ """
+ return f"{self._stream_scope}:request:{request_id}"
+
def close_sse_stream(self, request_id: RequestId) -> None:
</file context>
There was a problem hiding this comment.
Beyond the inline finding on the initialize gate, two candidates were examined and ruled out this run: (1) the buffered Last-Event-ID replay in replay_then_tail materializing the stored backlog before sending — the buffering is what enables the session-ownership check on the store-returned stream id, and the backlog is bounded by what the store retains for one stream; (2) the interaction README's lax-pragma paragraph omitting the two new non-exception lax markers (_respond_json's no-terminal arm, _handle_get_request's attach race) — that paragraph inventories only the bridge-timing-dependent except Exception handlers, which it does cover accurately.
Extended reasoning...
This run's bug hunt produced one nit (the initialize-gate residual gaps), delivered as an inline comment. Two further candidate issues were investigated and refuted: the replay-path memory-materialization concern (the replayed list buffers store output before the wire, but this is a deliberate consequence of validating stream ownership before releasing frames, and is bounded by the store's per-stream retention) and a claimed documentation gap in tests/interaction/README.md's lax-pragma inventory (the paragraph scopes itself to timing-dependent exception handlers under the in-process bridge and matches the markers in that category). Recording these here so a later review pass has a trail of what was already examined; this is informational only, not a correctness guarantee. The PR itself remains a large, behaviour-sensitive transport rewrite that warrants human review regardless.
| # From here the gate must be released whatever becomes of this | ||
| # request: by the handler task once it exists, else by this frame. | ||
| handler_started = False | ||
| try: | ||
| handler_started = await self._start_request( | ||
| scope, request, receive, send, message, protocol_version, stream_id, initialize_gate | ||
| ) | ||
| finally: | ||
| if initialize_gate is not None and not handler_started: | ||
| self._release_initialize_gate(initialize_gate) |
There was a problem hiding this comment.
🟡 The init-ordering gate added after review has two residual gaps: (1) POSTed notifications and responses go through _deliver_client_message, which never consults self._initializing — so a notification pipelined behind initialize is dispatched concurrently with the handshake and dropped by ServerRunner._on_notify, and a notifications/cancelled for a request still parked at gate.wait() finds no in_flight entry and is silently lost; (2) handler_started only becomes True when _start_request returns, after the _respond_sse/_respond_json await — so if the initialize POST's ASGI task is cancelled while parked there (Hypercorn/Daphne disconnect semantics, timeout middleware, graceful shutdown), the finally releases the gate while the uncommitted initialize handler is still running, letting parked requests race the handshake and fail INVALID_PARAMS. Fix: route _deliver_client_message's stateful dispatch through the same gate wait, and record gate ownership by flipping a flag immediately before start_soon(_run_handler) instead of via the return value.
Extended reasoning...
Gap 1 — notifications and cancels bypass the gate. The gate in _serve_request (src/mcp/server/streamable_http.py:1073-1086) applies only to POSTed JSONRPCRequests: a non-initialize request awaits self._initializing before dispatch. But POSTed notifications and responses take the _deliver_client_message path, which resolves/peer-cancels/spawns on_notify immediately without ever consulting self._initializing. Two consequences, both divergences from the ordering the PR description explicitly claims to preserve ("Requests that arrive while an initialize is still being served wait for the handshake to commit — the ordering the stream-pair driver's parked read loop used to provide"):\n\n- A notification pipelined behind initialize is dropped instead of handled: ServerRunner._on_notify (runner.py:273-275) drops any notification with 'received before initialization' while connection.initialize_accepted is still false. Over the old buffer-0 stream + parked read loop (with inline_methods={\"initialize\"}), the notification was not dequeued until initialize had committed. Reachable because the Mcp-Session-Id ships with the SSE response headers before the handler commits client_params. A pipelined notifications/initialized likewise runs concurrently with the initialize handler and sets connection.initialized before client_params is committed — an ordering the old transport made impossible.\n- A notifications/cancelled for a request parked at await gate.wait() cannot land: the request has not yet been enter_inbound'd (that happens later in _start_request), so _corr.peer_cancel finds no in_flight entry and the cancel is silently dropped; the request then runs to completion. On the stream pair, the cancel was queued behind the request and dequeued after it had entered in_flight.\n\nGap 2 — the gate's release protocol can open early. In _serve_request (streamable_http.py:979-988) the frame releases the gate whenever handler_started is falsy. The comment states the intent — released "by the handler task once it exists, else by this frame" — but handler_started only becomes True when _start_request returns, which is after await self._respond_sse(...) / _respond_json(...) completes. The handler task was already handed the gate at session_task_group.start_soon(_run_handler) before those awaits. For an initialize over SSE, the respond await spans the handler's entire runtime, so the cancellation window is wide.\n\nStep-by-step proof of gap 2 (SSE mode, stateful):\n1. Client POSTs initialize. _serve_request installs the gate; _start_request spawns _run_handler on the session task group and starts the SSE response — the Mcp-Session-Id header ships immediately.\n2. The initialize handler suspends in user middleware (the exact window the gate was built for; cf. runner.py:232-241).\n3. The client, holding the session id, POSTs tools/list, which parks at await gate.wait() (line 977).\n4. The initialize POST's connection drops and the ASGI server cancels the request task. Uvicorn's clean http.disconnect path is benign (sse-starlette exits normally, handler_started becomes True), but Hypercorn and Daphne cancel mid-request on disconnect, as does any timeout middleware and uvicorn's graceful-shutdown path — all supported ASGI behaviour.\n5. The cancellation propagates out of _respond_sse before the handler_started = ... assignment; the finally sees handler_started == False and releases the gate (gate.set() + self._initializing = None are synchronous, so they run fine under pending cancellation) — while the initialize handler is still running on the session task group (deliberately not cancelled per the disconnect ≠ cancel rule).\n6. The parked tools/list, and every request arriving afterwards, dispatches concurrently with the uncommitted handshake and hits ServerRunner._on_request's init gate → MCPError(INVALID_PARAMS) (runner.py:211-213) — exactly the pipelined-behind-initialize failure this PR added the gate to prevent. _run_handler's own later release is a no-op because _release_initialize_gate is idempotent, so nothing detects the double release. The JSON-mode variant (cancellation while parked in channel.finished.wait()) releases the gate the same way.\n\nWhy nothing else catches these. The gate is the only ordering mechanism left after the parked read loop was deleted; _deliver_client_message simply predates it in the dispatch flow, and the 404/_accepting check in _start_request doesn't apply (the session is healthy). No test covers cancelling the initialize POST mid-response — test_requests_wait_for_a_handshake_in_progress_to_commit keeps the POST task alive throughout.\n\nImpact and severity. Both gaps only affect clients that pipeline behind initialize (spec: SHOULD NOT, not MUST NOT — but the PR's own gate shows this ordering was judged worth preserving for requests). The failure modes are a silently dropped notification (client still gets its 202; drop is debug-logged), a best-effort cancel that misses (spec-defensible — receivers MAY ignore), and a transient, retryable INVALID_PARAMS for gap 2 that additionally requires a cancellation-propagating ASGI server or middleware. No crash, hang, or state corruption, and conforming clients are unaffected — hence not merge-blocking.\n\nFix direction. (1) Route _deliver_client_message's stateful dispatch through the same gate wait _serve_request applies (or document the divergence in migration.md). (2) Record gate ownership at spawn time: flip a flag visible to _serve_request's finally immediately before session_task_group.start_soon(_run_handler), so a cancellation landing after the spawn leaves the release to the handler task instead of the return value deciding it.
1 participant
Rebuilds the legacy (2025-era, sessionful) streamable HTTP server transport around per-request
direct dispatch, deleting the fabricated duplex message pipe and the router that fanned it back out.
Motivation and Context
StreamableHTTPServerTransportused to reconstruct HTTP request boundaries on top of astdio-shaped seam:
connect()yielded a(read_stream, write_stream)pair for aJSONRPCDispatcher, and amessage_routertask read the session-wide write stream and fannedmessages back out into per-request queues (
_request_streams), inferring "this POST is done"from the response frame passing by. HTTP already knows request boundaries, so all of that
machinery (and its head-of-line and teardown races — #1363, #1764) was reconstructing
information the transport threw away. The 2026 route (
_streamable_http_modern.py) alreadydispatches each POST directly; this brings the legacy path onto the same shape while keeping
everything the 2025 spec needs (sessions, the GET stream, server-to-client requests, event-store
resumability,
close_sse_stream()polling).What changed:
RequestCorrelator(newshared/_correlation.py), extracted fromJSONRPCDispatcher: theoutbound pending table + the whole send/await/abandon policy, the inbound in-flight table +
peer-cancel, and the exception-to-wire boundary for inbound handlers now live in one place.
JSONRPCDispatcherdelegates to it (behaviour unchanged); the HTTP transport shares it. Thecancelled-request answer policy lives at exactly one site in it, which Stop answering cancelled requests #3188 will need to
target when it lands (a suppressed answer will need the response stream terminated another
way — noted in a comment there).
_MessageChannel: one per response stream.write()is store-first (event store) thenforward to the attached SSE response, serialized per channel so wire order always matches
store order; it returns whether the message reached the store or the response, and never
raises. The standalone GET stream is the same object. Attach/detach is the resumability
primitive (
close_sse_stream()detaches; aLast-Event-IDreconnect replays thenre-attaches);
attach()returnsNoneon a channel whose life is over.StreamableHTTPServerTransportis now the per-session core: session id,Connection, oneServerRunner, the correlator, the open channels, and a session task hosting request-handlertasks (so a dropped connection does not cancel a request — the 2025 spec's disconnect ≠ cancel
rule). Each POSTed request is dispatched to the handler kernel directly; a POSTed
response resolves the waiting server-to-client request; a POSTed notification is handled after
the
202. Requests that arrive while aninitializeis still being served wait for thehandshake to commit (the ordering the stream-pair driver's parked read loop used to provide).
connect()and the fabricated stream pair are gone — the one API removal, and nothing indocs/examples used it (see the migration note).
StreamableHTTPSessionManager: identical public surface; it binds a transport to theServerper session and hosts the session task instead of aserve_loop.Behaviour clarified in the same change (all in
docs/migration.md):NoBackChannelErrorinstead of hanging (connection-scoped sends still ride the GET stream).storing it) fails the handler with
CONNECTION_CLOSEDinstead of parking for an answer thatcannot arrive.
EventStore.store_eventdegrades resumability for that message rather than takingthe stream down: it is delivered live and the store's exception is logged, never on the wire.
EventStoreare minted by the transport in a session-scopednamespace (opaque), so a client-chosen request id can no longer name the standalone GET
stream and two sessions sharing a store can no longer replay each other's frames.
Last-Event-IDGET on a server with no event store behaves as a plain GET instead ofreturning no response; two concurrent POSTs sharing a JSON-RPC id each keep their own response
stream instead of the second silently taking over the first's queue.
How Has This Been Tested?
tests/interaction/(the interaction-model suite pinning observable behaviour) passes unchanged— 867 tests, across the in-memory / streamable-http / streamable-http-stateless / sse matrix —
as does
tests/examples/(the story legs over HTTP) and the full suite (5523 tests) at 100%branch coverage with
strict-no-coverclean.mcp-everything-serverlocally: theactivesuite is 42/42,and the full
--suite allrun passes everything except the already-baselinedtasks-*scenarios ("baseline check passed") — including
server-sse-polling(priming-first,retry,Last-Event-IDreplay),server-sse-multiple-streams, DNS-rebinding, and everysampling/elicitation/progress scenario, i.e. against the TypeScript SDK client on the real wire.
stream (and 409 for a second one), tool calls over SSE, an overlapping re-
initializenext toa
tools/liston one session, a request id shaped like the internal GET-stream marker(served on its own stream), DELETE → 404-after-termination, plus the 400/404/405/409 probes.
against the behaviour they guarded (priming-store failure → 500 with no leaked state, standalone
teardown via
close_standalone_sse_stream(), manager eviction of exited/crashed sessions,stateless cleanup, close-callback protocol-version gating) or deleted where the guarded race
is now unrepresentable (
#1764router head-of-line, the standalone writer's between-dequeueswindow). New transport tests cover: terminating a session mid-request (SSE and JSON mode),
client-posted progress for a server-initiated request, per-channel write ordering, a request
suspended across a DELETE (refused, not run), a store failure costing only resumability (per
request and on the standalone stream), an undeliverable server-to-client request failing fast,
the init gate (including overlapping handshakes), shared-store session isolation, a request id
colliding with the GET-stream marker, a replay resumed across termination, and concurrent
POSTs sharing a request id.
Breaking Changes
StreamableHTTPServerTransport.connect()is removed; the transport is created and driven perrequest by
StreamableHTTPSessionManager(new keyword-onlyapp/lifespan_stateconstructorarguments). Code serving through
streamable_http_app()/run(transport="streamable-http")ormounting the session manager is unaffected; only hand-rolled
async with transport.connect()usage needs to move to the manager. Documented in
docs/migration.md.Types of changes
Checklist
Additional context
The transport's public module surface (
EventStore,EventMessage,EventCallback,GET_STREAM_KEY,check_accept_headers, the header/pattern constants,close_sse_stream(),close_standalone_sse_stream(),terminate(),handle_request()) and the manager's surface areunchanged, and no interaction/story/conformance test needed touching. Recorded divergences that
this design would make trivial to change (accepting a second
initializeon a session; anunknown
Last-Event-IDproducing an empty stream) are deliberately left as they are so this staysa behaviour-preserving change.
Follow-ups deliberately out of scope here (both pre-existing at the manager level): a session
that a client DELETEs stays registered until manager shutdown rather than being evicted (the
interaction suite currently pins that), and a session-id-less non-
initializePOST still mints asession before its 400.
Responses to the automated reviews are in the threads; the two rounds of findings that held up
(write ordering per channel, requests racing termination, event-store failures reaching the
wire or hanging a waiting handler, the client-namespace stream ids, and the SSE containment
drift) were fixed by making the underlying state unrepresentable rather than case by case.