From ffcc0576c09422b687639f684a0bbed16253a06c Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:38:45 +0000 Subject: [PATCH 1/6] Fail fast on request-scoped server requests in JSON-response mode In stateful JSON-response mode a tool handler calling ctx.elicit() (or any server-to-client request tied to the in-flight call) hung forever: the loop dispatcher reported can_send_request=True, the request was routed into the call's own per-request queue, and the JSON POST drain loop silently discarded it, so the parked waiter never woke. The transport now supplies each message's TransportContext (StreamableHTTPServerTransport.transport_context_for): it owns what a request's response can carry, so can_send_request is False in JSON-response mode and the request raises NoBackChannelError at once - the same fail-fast the stateless path and the 2026-07-28 entries already have. serve_loop takes the builder as a keyword and the session manager passes it for both the stateful and stateless legs. The session's standalone GET stream is untouched and still carries unrelated messages. --- docs/migration.md | 13 +++++--- docs/run/index.md | 2 +- docs/run/legacy-clients.md | 7 +++++ docs/troubleshooting.md | 15 +++++---- src/mcp/server/runner.py | 13 +++++++- src/mcp/server/streamable_http.py | 31 +++++++++++++++++-- src/mcp/server/streamable_http_manager.py | 19 +++++++----- src/mcp/shared/exceptions.py | 15 +++++---- src/mcp/shared/transport_context.py | 17 +++++++--- tests/interaction/_requirements.py | 10 ++++++ .../transports/test_streamable_http.py | 16 +++++++++- tests/server/test_streamable_http_router.py | 19 +++++++++++- 12 files changed, 141 insertions(+), 36 deletions(-) diff --git a/docs/migration.md b/docs/migration.md index 11569680e0..516826c365 100644 --- a/docs/migration.md +++ b/docs/migration.md @@ -749,7 +749,7 @@ Transport-specific parameters have been moved off the `MCPServer` constructor an - `host`, `port` - HTTP server binding, on `run()` only. The app factories have no `port` (`streamable_http_app(port=...)` raises `TypeError`; a mounted app binds wherever the outer ASGI server does) but do take `host` (default `"127.0.0.1"`), used only to decide whether DNS rebinding protection auto-enables (see the note below) - `sse_path`, `message_path` - SSE transport paths, on `run(transport="sse", ...)` and `sse_app()` - `streamable_http_path` - StreamableHTTP endpoint path, on `run(transport="streamable-http", ...)` and `streamable_http_app()` -- `json_response`, `stateless_http` - StreamableHTTP behavior, same two places +- `json_response`, `stateless_http` - StreamableHTTP behavior, same two places; each also removes a server-to-client channel, see [Server-initiated sampling, elicitation, and roots raise `NoBackChannelError`](#server-initiated-sampling-elicitation-and-roots-raise-nobackchannelerror) - `max_request_body_size` - StreamableHTTP request-body limit, same two places - `event_store`, `retry_interval` - StreamableHTTP event handling, same two places - `transport_security` - DNS rebinding protection, on `run()` for both HTTP transports and on both app methods @@ -963,8 +963,11 @@ enforce the spec's egress rule: an undeclared capability (form-mode `elicitation or `tool_choice`) fails the call with a `-32021` `MISSING_REQUIRED_CLIENT_CAPABILITY` JSON-RPC error instead of sending a request the client cannot handle. This applies on 2025-11-25 sessions with a -live back-channel too; a session with no back-channel keeps failing with its -no-back-channel error. To migrate, declare the capability: the SDK client +live back-channel too; a pre-`2026-07-28` session with no back-channel +(stateless HTTP, or streamable HTTP with `json_response=True`) keeps failing +with its no-back-channel error. At `2026-07-28` a resolver never uses a +back-channel — it answers with an `InputRequiredResult` — so the `-32021` +check applies there unconditionally. To migrate, declare the capability: the SDK client declares `elicitation`, `sampling`, and `roots` when the matching callback is set, and `sampling.tools` needs an explicit `Client(sampling_capabilities=SamplingCapability(tools=...))`. Direct @@ -1617,7 +1620,7 @@ Nothing changes for callers of the built-in client: abandoning a call (cancellin The `stateless: bool` parameter on the lowlevel `Server.run()` has been removed. Stateless serving is now a property of how the connection is constructed (the streamable-HTTP manager builds a born-ready `Connection` per request), not a flag the loop driver inspects. -Server-initiated requests that have no channel to travel on — a legacy session against a `stateless_http=True` server, or any connection negotiated at 2026-07-28 — now raise `NoBackChannelError` instead of stalling as they did in v1 (the transport silently dropped the outbound message), so a stateless-HTTP `ctx.elicit()` that used to hang now fails fast; see [Server-initiated sampling, elicitation, and roots raise `NoBackChannelError`](#server-initiated-sampling-elicitation-and-roots-raise-nobackchannelerror) for the exception and the migration paths. +Server-initiated requests that have no channel to travel on — a legacy session against a `stateless_http=True` server, the request-scoped channel of a stateful legacy session against a `json_response=True` server, or any connection negotiated at 2026-07-28 — now raise `NoBackChannelError` instead of stalling as they did in v1 (the transport silently dropped the outbound message), so a stateless-HTTP or JSON-mode `ctx.elicit()` that used to hang now fails fast; see [Server-initiated sampling, elicitation, and roots raise `NoBackChannelError`](#server-initiated-sampling-elicitation-and-roots-raise-nobackchannelerror) for the exception and the migration paths. ### Lowlevel `Server`: `request_context` property removed @@ -2801,7 +2804,7 @@ rest of this guide stays focused on the v1-to-v2 upgrade itself. The 2026-07-28 protocol has no server-initiated requests, so a handler that reaches back to the client mid-request — `ctx.elicit()`, `ctx.elicit_url()`, `ctx.session.create_message()`, `ctx.session.list_roots()`, or any other `ServerSession` request helper — raises `NoBackChannelError` on such a connection instead of sending. An in-process `Client(server)` negotiates 2026-07-28 by default (see [`Client` defaults to `mode='auto'`](#client-defaults-to-modeauto)), so the first smoke test of an unchanged v1 sampling or elicitation tool fails, and setting `sampling_callback=` / `elicitation_callback=` on the client changes nothing because no request ever reaches the client. -`NoBackChannelError` lives in `mcp.shared.exceptions` and subclasses `MCPError` (code `-32600`, message `Cannot send '': this transport context has no back-channel for server-initiated requests.`). Raised inside an `@mcp.tool()` it reaches the client as a top-level JSON-RPC error, not `CallToolResult(is_error=True)` — see [`MCPError` raised from an `@mcp.tool()` handler now surfaces as a JSON-RPC error](#mcperror-raised-from-an-mcptool-handler-now-surfaces-as-a-json-rpc-error) — and the [Troubleshooting](troubleshooting.md) page walks through the client-side traceback. The same exception is raised on a legacy session against a `stateless_http=True` server, where v1 dropped the message and stalled ([`Server.run()` no longer takes a `stateless` flag](#serverrun-no-longer-takes-a-stateless-flag)). Notifications never raise it: `send_log_message()`, `send_tool_list_changed()`, and the other notification helpers are dropped with a debug log where no channel exists, and `UrlElicitationRequiredError` from a tool is unaffected (it is an error response, not a request). +`NoBackChannelError` lives in `mcp.shared.exceptions` and subclasses `MCPError` (code `-32600`, message `Cannot send '': this transport context has no back-channel for server-initiated requests.`). Raised inside an `@mcp.tool()` it reaches the client as a top-level JSON-RPC error, not `CallToolResult(is_error=True)` — see [`MCPError` raised from an `@mcp.tool()` handler now surfaces as a JSON-RPC error](#mcperror-raised-from-an-mcptool-handler-now-surfaces-as-a-json-rpc-error) — and the [Troubleshooting](troubleshooting.md) page walks through the client-side traceback. The same exception is raised on a legacy session against a `stateless_http=True` server, and on the request-scoped channel of a stateful legacy session against a `json_response=True` server (a JSON body carries exactly one response, so a mid-request `ctx.elicit()` cannot ride it; the session's standalone `GET` stream still carries unrelated messages) — both places v1 dropped the message and stalled ([`Server.run()` no longer takes a `stateless` flag](#serverrun-no-longer-takes-a-stateless-flag)). Notifications never raise it: `send_log_message()`, `send_tool_list_changed()`, and the other notification helpers are dropped with a debug log where no channel exists, and `UrlElicitationRequiredError` from a tool is unaffected (it is an error response, not a request). Two ways to migrate: diff --git a/docs/run/index.md b/docs/run/index.md index adaa0c85d3..dbea20d0fe 100644 --- a/docs/run/index.md +++ b/docs/run/index.md @@ -65,7 +65,7 @@ Each transport has its own keyword arguments, all on `run()`: * `host` / `port`: where to listen. Defaults `127.0.0.1` and `8000`. * `streamable_http_path`: where the MCP endpoint lives. Default `/mcp`. -* `json_response=True`: answer with plain JSON instead of an SSE stream. +* `json_response=True`: answer each POST with a single JSON body instead of an SSE stream. That body has room for the response and nothing else, so a tool that calls back into the client mid-request (`ctx.elicit()`, sampling) raises `NoBackChannelError` on this leg, and notifications tied to the in-flight call (progress from `ctx.report_progress()`, per-call log messages) are dropped; the standalone `GET` stream still carries unrelated ones. * `stateless_http=True`: a fresh transport per request, no session tracking. * `max_request_body_size`: largest accepted POST body in bytes. Defaults to 4 MiB; larger requests receive HTTP 413 before parsing or session creation. Raise it only when legitimate MCP messages diff --git a/docs/run/legacy-clients.md b/docs/run/legacy-clients.md index c7a1096db6..d65b7e0803 100644 --- a/docs/run/legacy-clients.md +++ b/docs/run/legacy-clients.md @@ -72,6 +72,13 @@ Two things about it matter more than what it does. **It costs both server-to-client channels on that leg.** A session that lives for one `POST` has no stream for the server to push a request down and no standalone stream for it to push notifications down. Every server-initiated request raises `NoBackChannelError`: `ctx.elicit()`, the retired sampling and roots calls (**[Deprecated features](../deprecated.md)**), and, yes, `Resolve` asking a *legacy* client its question. Notifications don't even get an error; they are silently dropped. +!!! note + `json_response=True` is not that knob, but it takes half the same cost on *every* legacy + session: a `POST` answered with one JSON body has no stream for the request-scoped channel, + so a mid-request `ctx.elicit()` raises the same `NoBackChannelError` and notifications tied to + the request are dropped. The session's standalone stream is untouched: unrelated notifications + still arrive. + !!! check Do the wrong thing. `reserve` is the exact tool that just served both clients. Deploy it with `stateless_http=True`, connect the same two clients over HTTP, and call it from each. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index fda7e4bc30..75a6652ecc 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -305,7 +305,7 @@ You see this one from `ctx.elicit()` on a legacy connection, and on any connecti ## `MCPError: Cannot send 'elicitation/create': this transport context has no back-channel for server-initiated requests.` -Your handler tried to reach the client mid-request, on a connection where nothing can carry a request from the server. There are exactly two ways to be on one. +Your handler tried to reach the client mid-request, on a connection whose call has no channel that can carry a request from the server. There are three server configurations that put a call there. **A `2026-07-28` connection: any transport, always.** The modern protocol has no server-initiated requests at all, so the server refuses before anything is sent. `ctx.elicit()` inside a tool is the classic way to meet this (on the very first in-memory test, since `Client(server)` negotiates `2026-07-28` without being asked), and passing `elicitation_callback=` changes nothing, because no request ever reaches the client for it to answer: @@ -329,20 +329,23 @@ mcp.shared.exceptions.MCPError: Cannot send 'elicitation/create': this transport --8<-- "docs_src/troubleshooting/tutorial008.py" ``` +**A legacy connection on a `json_response=True` server.** The `POST` is answered with one JSON body, and one body carries only the response, so the request-scoped stream a mid-request `ctx.elicit()` needs does not exist here either. The session, its `Mcp-Session-Id`, and its standalone stream are all still there; only the request-scoped channel is gone. + The message names the method it could not send. `NoBackChannelError` is the class the server raises, but the wire carries only the base `MCPError`, so the sentence above is your traceback's last line, not the class name. -The fix is the same for both: don't reach back mid-call. Move the question into a **resolver** (or return an `InputRequiredResult` yourself) and it becomes part of the *response*, which every connection can carry: +For a `2026-07-28` client the fix is the same on all three: don't reach back mid-call. Move the question into a **resolver** (or return an `InputRequiredResult` yourself) and it becomes part of the *response*, which every connection can carry: ```python title="server.py" hl_lines="15-17 21" --8<-- "docs_src/troubleshooting/tutorial007.py" ``` -Same question, same `elicitation_callback` on the client. The difference is under the hood: a resolver lets the server *return* the question from the call instead of pushing it, so nothing ever flows server-to-client. **[Elicitation](handlers/elicitation.md)** covers resolvers; **[Multi-round-trip requests](handlers/multi-round-trip.md)** covers what happens on the wire. +Same question, same `elicitation_callback` on the client. The difference is under the hood: a resolver lets the server *return* the question from the call instead of pushing it, so nothing ever flows server-to-client. That rescues every `2026-07-28` client, whichever of the three configurations the server is in. A *legacy* client is not rescued by the rewrite alone: `2025-11-25` has no way to return a question, so on a legacy connection the resolver still sends `elicitation/create` down the request-scoped channel, and still needs a server that keeps it — neither `stateless_http=True` nor `json_response=True`. **[Elicitation](handlers/elicitation.md)** covers resolvers; **[Multi-round-trip requests](handlers/multi-round-trip.md)** covers what happens on the wire. !!! check The tool with `ctx.elicit()` is not wrong, it is *pre-2026*. Connect with `mode="legacy"` - (the classic `initialize` handshake, spec `2025-11-25` and earlier) to a server that is not - `stateless_http=True`, and it works, because the server-to-client channel exists there. + (the classic `initialize` handshake, spec `2025-11-25` and earlier) to a server that is neither + `stateless_http=True` nor `json_response=True`, and it works, because the server-to-client + channel exists there. **[Protocol versions](protocol-versions.md)** is the page on what each version has. ## `MCPError: Invalid or expired requestState` @@ -407,6 +410,6 @@ mcp = MCPServer("Weather", request_state_security=RequestStateSecurity(keys=[key * One 421, three spellings: `Server returned an error response` (the python `Client`), `421 Misdirected Request` / `Invalid Host header` (everything else), `Invalid Host header: ` (the server log). Fix: `transport_security=TransportSecuritySettings(allowed_hosts=[...])`. * `Task group is not initialized` -> a mounted app whose host lifespan never entered `mcp.session_manager.run()`. * `Session not found` -> the server restarted; reconnect. -* `Cannot send 'elicitation/create': ... no back-channel ...` -> `ctx.elicit()` needs a server-to-client channel: a `2026-07-28` connection never has one, and `stateless_http=True` takes away the legacy one. Use a resolver. Its neighbour `Method not found` is a request for a method the other side's protocol revision doesn't have. +* `Cannot send 'elicitation/create': ... no back-channel ...` -> `ctx.elicit()` needs a server-to-client channel: a `2026-07-28` connection never has one, `stateless_http=True` takes away the legacy one, and `json_response=True` takes away the request-scoped one. Use a resolver (a legacy client also needs a server that keeps the channel). Its neighbour `Method not found` is a request for a method the other side's protocol revision doesn't have. * `Client did not declare the form elicitation capability ...` and `Elicitation not supported` -> the client is missing `elicitation_callback=`. * `Invalid or expired requestState` never says why on the wire. The server log does; `unknown key` means share `RequestStateSecurity(keys=[...])` across workers. diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py index 6f9f7a8f74..4f3f4e91dd 100644 --- a/src/mcp/server/runner.py +++ b/src/mcp/server/runner.py @@ -15,7 +15,7 @@ import contextvars import logging -from collections.abc import AsyncIterator, Awaitable, Mapping +from collections.abc import AsyncIterator, Awaitable, Callable, Mapping from contextlib import asynccontextmanager from dataclasses import KW_ONLY, dataclass, replace from functools import cached_property, partial @@ -473,6 +473,7 @@ async def serve_loop( session_id: str | None = None, init_options: InitializationOptions | None = None, raise_exceptions: bool = False, + transport_builder: Callable[[MessageMetadata], TransportContext] | None = None, ) -> None: """Drive ``server`` in handshake-only loop mode over a stream pair until the channel closes. @@ -482,10 +483,20 @@ async def serve_loop( this; `Server.run` drives `serve_dual_era_loop`, which extends the same dispatcher recipe (notably the `inline_methods={"initialize"}` rule) with era routing. + + Args: + transport_builder: Builds each inbound message's `TransportContext` from + its `SessionMessage.metadata`. Omitted, every message reads as a full + duplex pipe (`can_send_request=True`) - the truth for a plain stream + pair; a transport whose response cannot carry a server-initiated + request (streamable HTTP in JSON-response mode) passes its own + `StreamableHTTPServerTransport.transport_context_for` so the + request-scoped channel refuses instead of stalling. """ dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher( read_stream, write_stream, + transport_builder=transport_builder, raise_handler_exceptions=raise_exceptions, # Handle `initialize` inline so a client that pipelines it with the # next request (spec: SHOULD NOT, not MUST NOT) sees the initialized diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 324fc3e04b..8aad403c4a 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -44,7 +44,8 @@ from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams from mcp.shared._stream_protocols import ReadStream, WriteStream from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER -from mcp.shared.message import ServerMessageMetadata, SessionMessage +from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage +from mcp.shared.transport_context import TransportContext logger = logging.getLogger(__name__) @@ -173,8 +174,13 @@ def __init__( Args: mcp_session_id: Optional session identifier for this connection. Must contain only visible ASCII characters (0x21-0x7E). - is_json_response_enabled: If True, return JSON responses for requests - instead of SSE streams. Default is False. + is_json_response_enabled: If True, answer each request POST with a single + JSON body instead of an SSE stream. The body carries + only the response, so `transport_context_for` reports + `can_send_request=False` for it: a server-initiated + request on the request-scoped channel raises + `NoBackChannelError` and request-scoped notifications + are dropped. Default is False. event_store: Event store for resumability support. If provided, resumability will be enabled, allowing clients to reconnect and resume messages. @@ -212,6 +218,25 @@ def is_terminated(self) -> bool: """Check if this transport has been explicitly terminated.""" return self._terminated + def transport_context_for(self, metadata: MessageMetadata) -> TransportContext: + """Build the `TransportContext` for an inbound message this transport delivered. + + The transport owns what a request's response can carry, so it supplies + the loop dispatcher's `transport_builder`. A JSON body holds exactly one + JSON-RPC response, so in JSON-response mode the request-scoped channel has + no room for a server-initiated request and `can_send_request` is `False`: + `ctx.elicit()` raises `NoBackChannelError` at once instead of parking a + waiter no reply can reach. An SSE response streams whatever the handler + emits. The connection's standalone GET stream is a separate channel and + is not described here. + """ + request = metadata.request_context if isinstance(metadata, ServerMessageMetadata) else None + return TransportContext( + kind="streamable-http", + can_send_request=not self.is_json_response_enabled, + headers=request.headers if isinstance(request, Request) else None, + ) + def close_sse_stream(self, request_id: RequestId) -> None: """Close SSE connection for a specific request without terminating the stream. diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py index 31f587ee66..e23f0ee76a 100644 --- a/src/mcp/server/streamable_http_manager.py +++ b/src/mcp/server/streamable_http_manager.py @@ -6,6 +6,7 @@ import logging from collections import deque from collections.abc import AsyncIterator +from dataclasses import replace from typing import TYPE_CHECKING, Any, Final from uuid import uuid4 @@ -214,12 +215,14 @@ async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STA read_stream, write_stream, inline_methods=frozenset({"initialize"}), - # No session ID means a server-to-client request can be - # written to this POST's response stream, but the client's - # reply has nowhere to land — `can_send_request=False` - # makes the per-request channel raise `NoBackChannelError` - # for requests while still allowing notifications. - transport_builder=lambda _md: TransportContext(kind="streamable-http", can_send_request=False), + # The transport says what its response stream can carry; on + # top of that, no session ID means the client's reply to a + # server request has nowhere to land, so the request-scoped + # channel refuses (`NoBackChannelError`) even in SSE mode + # while still forwarding notifications. + transport_builder=lambda md: replace( + http_transport.transport_context_for(md), can_send_request=False + ), ) # Born-ready, no standalone channel: the legacy stateless path # never opens a GET stream and need not see `initialize`. The @@ -320,13 +323,15 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE with idle_scope: # Drive via `serve_loop` (not `Server.run()`) so the # manager's already-entered lifespan is reused - # rather than re-entered per session. + # rather than re-entered per session; the transport + # rules on what each request's response can carry. await serve_loop( self.app, read_stream, write_stream, lifespan_state=self._lifespan_state, session_id=http_transport.mcp_session_id, + transport_builder=http_transport.transport_context_for, ) if idle_scope.cancelled_caught: diff --git a/src/mcp/shared/exceptions.py b/src/mcp/shared/exceptions.py index 2f8a539dab..99ca0d1ae4 100644 --- a/src/mcp/shared/exceptions.py +++ b/src/mcp/shared/exceptions.py @@ -53,12 +53,15 @@ def __str__(self) -> str: class NoBackChannelError(MCPError): - """Raised when sending a server-initiated request over a transport that cannot deliver it. - - Stateless HTTP and JSON-response-mode HTTP have no channel for the server to - push requests (sampling, elicitation, roots/list) to the client. This is - raised by `DispatchContext.send_raw_request` when `can_send_request` is - `False`, and serializes to an `INVALID_REQUEST` error response. + """Raised when a server-initiated request has no channel that can deliver it. + + The request-scoped channel is a dead end when the response has no room + (streamable HTTP in JSON-response mode, the 2026-07-28 single-exchange + entry), when the client's reply has nowhere to land (stateless HTTP), or + when the protocol forbids server-initiated requests (2026-07-28). Raised by + `DispatchContext.send_raw_request` when `can_send_request` is `False`, and + by a connection's standalone channel when it has none; serializes to an + `INVALID_REQUEST` error response. """ def __init__(self, method: str): diff --git a/src/mcp/shared/transport_context.py b/src/mcp/shared/transport_context.py index 55e5f6bc5f..8d15a2eaa2 100644 --- a/src/mcp/shared/transport_context.py +++ b/src/mcp/shared/transport_context.py @@ -23,11 +23,18 @@ class TransportContext: """Short identifier for the transport (e.g. `"stdio"`, `"streamable-http"`).""" can_send_request: bool - """Whether the transport can deliver server-initiated requests to the peer. - - `False` for stateless HTTP and HTTP with JSON response mode; `True` for - stdio, SSE, and stateful streamable HTTP. When `False`, - `DispatchContext.send_raw_request` raises `NoBackChannelError`. + """Whether this message's request-scoped channel can deliver a server-initiated request. + + `False` for any of three reasons: the response has no room (streamable + HTTP in JSON-response mode and the 2026-07-28 single-exchange entry answer + with one JSON-RPC reply), the client's reply has nowhere to land (stateless + HTTP, no session), or the protocol forbids server-initiated requests (any + 2026-07-28 connection, whose dispatch masks the flag off). `True` for a + plain duplex pipe (stdio, SSE) and stateful streamable HTTP with SSE + responses, all pre-2026-07-28. When `False`, + `DispatchContext.send_raw_request` raises `NoBackChannelError` instead of + parking a waiter no reply can reach. Says nothing about the connection's + standalone channel, which refuses separately. """ headers: Mapping[str, str] | None = None diff --git a/tests/interaction/_requirements.py b/tests/interaction/_requirements.py index a26f7d9983..a7bfcf5bb5 100644 --- a/tests/interaction/_requirements.py +++ b/tests/interaction/_requirements.py @@ -2577,6 +2577,16 @@ def __post_init__(self) -> None: "overtake anything already queued for the request." ), ), + "transport:streamable-http:json-response-restrictions": Requirement( + source="sdk", + behavior=( + "In JSON-response mode a handler's request-scoped server-initiated request fails fast with an " + "INVALID_REQUEST protocol error and request-scoped notifications are not delivered, because the " + "single JSON body carries only the response; the connection's standalone stream is unaffected." + ), + transports=("streamable-http",), + note="Only observable over streamable HTTP: JSON-response mode is an HTTP framing option.", + ), "transport:streamable-http:stateless": Requirement( source=f"{SPEC_BASE_URL}/basic/transports#streamable-http", behavior=( diff --git a/tests/interaction/transports/test_streamable_http.py b/tests/interaction/transports/test_streamable_http.py index e1f16dc5f0..3e796b07e4 100644 --- a/tests/interaction/transports/test_streamable_http.py +++ b/tests/interaction/transports/test_streamable_http.py @@ -64,7 +64,8 @@ class Confirmation(BaseModel): async def ask(ctx: Context) -> str: """Elicit a confirmation from the client and report the outcome.""" answer = await ctx.elicit("Proceed?", Confirmation) - # In stateless mode the elicit raises before this point: there is no session to call back through. + # In stateless and JSON-response modes the elicit raises before this point: there is no + # request-scoped channel to call back through. assert isinstance(answer, AcceptedElicitation) return f"confirmed={answer.data.confirmed}" @@ -120,6 +121,19 @@ async def test_stateless_streamable_http_rejects_server_initiated_requests() -> assert exc_info.value.error.code == INVALID_REQUEST +@requirement("transport:streamable-http:json-response-restrictions") +async def test_json_response_streamable_http_rejects_request_scoped_server_requests() -> None: + """A handler that calls back to the client mid-request fails fast when the server answers with + JSON: the one response body cannot carry the nested `elicitation/create`, so the request-scoped + channel raises `NoBackChannelError` (a top-level `MCPError`) instead of parking a waiter no reply + could ever reach. Bounded, because before the fix this call hung until it timed out.""" + async with connect_over_streamable_http(_smoke_server(), json_response=True) as client: + with anyio.fail_after(5), pytest.raises(MCPError) as exc_info: + await client.call_tool("ask", {}) + + assert exc_info.value.error.code == INVALID_REQUEST + + @requirement("transport:streamable-http:notifications") @requirement("transport:streamable-http:unrelated-messages") @requirement("hosting:http:standalone-sse") diff --git a/tests/server/test_streamable_http_router.py b/tests/server/test_streamable_http_router.py index 3086dca990..9fb75ee4ef 100644 --- a/tests/server/test_streamable_http_router.py +++ b/tests/server/test_streamable_http_router.py @@ -3,6 +3,7 @@ import anyio import pytest from mcp_types import JSONRPCMessage, JSONRPCResponse +from starlette.requests import Request from starlette.types import Message, Scope from mcp.server.streamable_http import ( @@ -14,7 +15,8 @@ StreamableHTTPServerTransport, StreamId, ) -from mcp.shared.message import SessionMessage +from mcp.shared.message import ServerMessageMetadata, SessionMessage +from mcp.shared.transport_context import TransportContext class _PrimingFailingStore(EventStore): @@ -114,3 +116,18 @@ async def asgi_send(message: Message) -> None: assert sent[0]["status"] == 500 body = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body") assert b"backend unavailable" not in body + + +def test_transport_context_reports_response_mode_and_request_headers() -> None: + """The transport's own verdict: no request-scoped requests in JSON mode, headers off the carried Request.""" + json_transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=True) + assert json_transport.transport_context_for(None) == TransportContext( + kind="streamable-http", can_send_request=False + ) + sse_transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=False) + assert sse_transport.transport_context_for(None) == TransportContext(kind="streamable-http", can_send_request=True) + + request = Request({"type": "http", "method": "POST", "path": "/", "headers": [(b"x-test", b"1")]}) + context = sse_transport.transport_context_for(ServerMessageMetadata(request_context=request)) + assert context.headers is not None + assert context.headers["x-test"] == "1" From e20be4db136dec0df322d447271b454111264dd7 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:39:12 +0000 Subject: [PATCH 2/6] Drop request-scoped messages a JSON response cannot carry at the router In JSON-response mode a message tied to an in-flight request has no wire form and no POST could replay it, so message_router now drops it before storing or queueing rather than leaving it for a consumer to discard. The per-request queue then carries exactly the one response, and the JSON POST handler collapses from a discard loop to a single receive with a real arm for a session torn down mid-request. This removes the four coverage pragmas the discard loop needed. --- src/mcp/server/streamable_http.py | 57 ++++++++----------- .../transports/test_streamable_http.py | 29 ++++++++++ tests/server/test_streamable_http_router.py | 42 ++++++++++++++ 3 files changed, 96 insertions(+), 32 deletions(-) diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 8aad403c4a..02d4144f13 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -612,43 +612,24 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re session_message = SessionMessage(message, metadata=metadata) await writer.send(session_message) try: - # Process messages from the request-specific stream - # We need to collect all messages until we get a response - response_message = None - - # Use similar approach to SSE writer for consistency - async for event_message in request_stream_reader: # pragma: no branch - # If it's a response, this is what we're waiting for - if isinstance(event_message.message, JSONRPCResponse | JSONRPCError): - response_message = event_message.message - break - # For notifications and requests, keep waiting - else: # pragma: no cover - logger.debug(f"received: {event_message.message.method}") - - # At this point we should have a response - if response_message: - # Create JSON response - response = self._create_json_response(response_message) - await response(scope, receive, send) - else: # pragma: no cover - # This shouldn't happen in normal operation - logger.error("No response message received before stream closed") - response = self._create_error_response( - "Error processing request: No response received", - HTTPStatus.INTERNAL_SERVER_ERROR, - ) - await response(scope, receive, send) - except Exception: # pragma: no cover - logger.exception("Error processing JSON response") + # `message_router` deposits only this request's own response + # here: anything else scoped to the request has no wire in + # JSON-response mode. + event_message = await request_stream_reader.receive() + except (anyio.EndOfStream, anyio.ClosedResourceError): + # The stream closed with no response: the session was + # terminated while this request was in flight. + logger.debug(f"Session terminated with request {request_id} in flight; no response to send") response = self._create_error_response( - "Error processing request", + "Session terminated before the request completed", HTTPStatus.INTERNAL_SERVER_ERROR, INTERNAL_ERROR, ) - await response(scope, receive, send) + else: + response = self._create_json_response(event_message.message) finally: await self._clean_up_memory_streams(request_id) + await response(scope, receive, send) else: # Mint the priming event before any per-request state exists: # `EventStore.store_event` is user code and may raise, in which @@ -1049,7 +1030,19 @@ async def message_router(): ) and session_message.metadata.related_request_id is not None ): - target_request_id = str(session_message.metadata.related_request_id) + related_request_id = session_message.metadata.related_request_id + if self.is_json_response_enabled: + # A JSON body carries exactly the one response: a + # message that only rides this request's stream + # has no wire form (and no POST could replay it), + # so drop it before storing or queueing - never + # park it in a queue nothing drains (#1764). + logger.debug( + f"Dropped message related to request {related_request_id}: " + "a JSON response carries only its own response" + ) + continue + target_request_id = str(related_request_id) request_stream_id = target_request_id if target_request_id is not None else GET_STREAM_KEY diff --git a/tests/interaction/transports/test_streamable_http.py b/tests/interaction/transports/test_streamable_http.py index 3e796b07e4..2176b27823 100644 --- a/tests/interaction/transports/test_streamable_http.py +++ b/tests/interaction/transports/test_streamable_http.py @@ -134,6 +134,35 @@ async def test_json_response_streamable_http_rejects_request_scoped_server_reque assert exc_info.value.error.code == INVALID_REQUEST +@requirement("transport:streamable-http:json-response-restrictions") +@requirement("transport:streamable-http:unrelated-messages") +@requirement("hosting:http:standalone-sse") +async def test_json_response_streamable_http_delivers_only_unrelated_notifications() -> None: + """In JSON-response mode the call's own log notification has no stream to ride and never + reaches the client, while the tool result comes back as the JSON body and the unrelated + resource-updated notification arrives on the standalone stream. The handler writes both + notifications before returning, so once the result and the unrelated message are in, no + request-scoped message can still be in flight.""" + received: list[IncomingMessage] = [] + server_message_seen = anyio.Event() + + async def collect(message: IncomingMessage) -> None: + received.append(message) + server_message_seen.set() + + async with connect_over_streamable_http(_smoke_server(), json_response=True, message_handler=collect) as client: + with anyio.fail_after(5): + result = await client.call_tool("announce", {}) + await server_message_seen.wait() + + assert result == snapshot( + CallToolResult(content=[TextContent(text="announced")], structured_content={"result": "announced"}) + ) + assert received == snapshot( + [ResourceUpdatedNotification(params=ResourceUpdatedNotificationParams(uri="file:///watched.txt"))] + ) + + @requirement("transport:streamable-http:notifications") @requirement("transport:streamable-http:unrelated-messages") @requirement("hosting:http:standalone-sse") diff --git a/tests/server/test_streamable_http_router.py b/tests/server/test_streamable_http_router.py index 9fb75ee4ef..81d51b959c 100644 --- a/tests/server/test_streamable_http_router.py +++ b/tests/server/test_streamable_http_router.py @@ -118,6 +118,48 @@ async def asgi_send(message: Message) -> None: assert b"backend unavailable" not in body +@pytest.mark.anyio +async def test_json_post_answers_500_when_session_terminates_mid_request() -> None: + """A JSON-mode POST whose session is torn down before the handler answers gets a 500, not a stall.""" + transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=True) + body = b'{"jsonrpc":"2.0","id":"req-1","method":"tools/list","params":{}}' + scope: Scope = { + "type": "http", + "method": "POST", + "path": "/", + "query_string": b"", + "headers": [ + (b"accept", b"application/json"), + (b"content-type", b"application/json"), + (b"mcp-session-id", b"sid"), + (b"mcp-protocol-version", b"2025-11-25"), + ], + } + body_sent = False + + async def receive() -> Message: + nonlocal body_sent + if not body_sent: + body_sent = True + return {"type": "http.request", "body": body, "more_body": False} + raise NotImplementedError + + sent: list[Message] = [] + + async def asgi_send(message: Message) -> None: + sent.append(message) + + async with transport.connect() as (read_stream, _write_stream): + async with anyio.create_task_group() as tg: + tg.start_soon(transport.handle_request, scope, receive, asgi_send) + with anyio.fail_after(5): + await read_stream.receive() # the request reached the session; the POST is parked + await transport.terminate() + + assert sent[0]["type"] == "http.response.start" + assert sent[0]["status"] == 500 + + def test_transport_context_reports_response_mode_and_request_headers() -> None: """The transport's own verdict: no request-scoped requests in JSON mode, headers off the carried Request.""" json_transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=True) From c8a4bb1bb9a2f48a29ddbedf616382cfa714cbec Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:54:42 +0000 Subject: [PATCH 3/6] Document that connect() drivers must pass the transport's context builder The JSON-response-mode refusal of request-scoped server requests takes effect through the dispatcher's transport_builder, so a caller driving the transport's streams by hand needs to pass transport_context_for the way the session manager does. No-Verification-Needed: docstring-only edit --- src/mcp/server/streamable_http.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 02d4144f13..2bcf737be8 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -992,6 +992,12 @@ async def connect( ]: """Context manager that provides read and write streams for a connection. + Drive the yielded streams through a `JSONRPCDispatcher` built with + `transport_context_for` as its `transport_builder`, as the session + manager does: that is what makes the request-scoped channel refuse + server-initiated requests in JSON-response mode instead of parking + them where nothing can deliver them. + Yields: Tuple of (read_stream, write_stream) for bidirectional communication """ From 08638b8e0b6f2c5316c9dde5fd6eecc69e0275f7 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Mon, 27 Jul 2026 23:33:06 +0000 Subject: [PATCH 4/6] Move the JSON-mode back-channel verdict onto the message metadata Instead of the transport supplying a `transport_builder` that the session manager has to ferry into the dispatcher, the transport now states its verdict on the message it already emits: ServerMessageMetadata grows `can_send_request`, which StreamableHTTPServerTransport stamps as `mcp_session_id is not None and not is_json_response_enabled`, and the dispatcher's default builder honors it. This drops the public transport_context_for method, the serve_loop keyword, and the manager's stateless special-case builder: both legs are now correct with no wiring at all, and so is a transport driven by hand. The metadata already carries transport facts the dispatcher consumes (related_request_id, on_request_unanswered), so the verdict follows an existing channel rather than opening a new one. --- src/mcp/server/runner.py | 13 +----- src/mcp/server/streamable_http.py | 51 +++++++++------------ src/mcp/server/streamable_http_manager.py | 17 +++---- src/mcp/shared/jsonrpc_dispatcher.py | 17 ++++++- src/mcp/shared/message.py | 4 ++ tests/server/test_streamable_http_router.py | 19 +------- tests/shared/test_jsonrpc_dispatcher.py | 39 ++++++++++++++++ 7 files changed, 87 insertions(+), 73 deletions(-) diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py index 4f3f4e91dd..6f9f7a8f74 100644 --- a/src/mcp/server/runner.py +++ b/src/mcp/server/runner.py @@ -15,7 +15,7 @@ import contextvars import logging -from collections.abc import AsyncIterator, Awaitable, Callable, Mapping +from collections.abc import AsyncIterator, Awaitable, Mapping from contextlib import asynccontextmanager from dataclasses import KW_ONLY, dataclass, replace from functools import cached_property, partial @@ -473,7 +473,6 @@ async def serve_loop( session_id: str | None = None, init_options: InitializationOptions | None = None, raise_exceptions: bool = False, - transport_builder: Callable[[MessageMetadata], TransportContext] | None = None, ) -> None: """Drive ``server`` in handshake-only loop mode over a stream pair until the channel closes. @@ -483,20 +482,10 @@ async def serve_loop( this; `Server.run` drives `serve_dual_era_loop`, which extends the same dispatcher recipe (notably the `inline_methods={"initialize"}` rule) with era routing. - - Args: - transport_builder: Builds each inbound message's `TransportContext` from - its `SessionMessage.metadata`. Omitted, every message reads as a full - duplex pipe (`can_send_request=True`) - the truth for a plain stream - pair; a transport whose response cannot carry a server-initiated - request (streamable HTTP in JSON-response mode) passes its own - `StreamableHTTPServerTransport.transport_context_for` so the - request-scoped channel refuses instead of stalling. """ dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher( read_stream, write_stream, - transport_builder=transport_builder, raise_handler_exceptions=raise_exceptions, # Handle `initialize` inline so a client that pipelines it with the # next request (spec: SHOULD NOT, not MUST NOT) sees the initialized diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 2bcf737be8..4d2b16b4dc 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -44,8 +44,7 @@ from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams from mcp.shared._stream_protocols import ReadStream, WriteStream from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER -from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage -from mcp.shared.transport_context import TransportContext +from mcp.shared.message import ServerMessageMetadata, SessionMessage logger = logging.getLogger(__name__) @@ -176,9 +175,9 @@ def __init__( Must contain only visible ASCII characters (0x21-0x7E). is_json_response_enabled: If True, answer each request POST with a single JSON body instead of an SSE stream. The body carries - only the response, so `transport_context_for` reports - `can_send_request=False` for it: a server-initiated - request on the request-scoped channel raises + only the response, so this transport marks its messages + as having no request-scoped back-channel: a + server-initiated request on that channel raises `NoBackChannelError` and request-scoped notifications are dropped. Default is False. event_store: Event store for resumability support. If provided, @@ -218,24 +217,14 @@ def is_terminated(self) -> bool: """Check if this transport has been explicitly terminated.""" return self._terminated - def transport_context_for(self, metadata: MessageMetadata) -> TransportContext: - """Build the `TransportContext` for an inbound message this transport delivered. - - The transport owns what a request's response can carry, so it supplies - the loop dispatcher's `transport_builder`. A JSON body holds exactly one - JSON-RPC response, so in JSON-response mode the request-scoped channel has - no room for a server-initiated request and `can_send_request` is `False`: - `ctx.elicit()` raises `NoBackChannelError` at once instead of parking a - waiter no reply can reach. An SSE response streams whatever the handler - emits. The connection's standalone GET stream is a separate channel and - is not described here. + @property + def _request_channel_can_send_request(self) -> bool: + """Whether the request-scoped channel of a message this transport delivers can carry a + server-initiated request. It cannot in JSON-response mode (the POST is answered with one + JSON body) nor with no session (the client's reply would have no session to land on); + stamped on each message's `ServerMessageMetadata` so any dispatcher's builder reads it. """ - request = metadata.request_context if isinstance(metadata, ServerMessageMetadata) else None - return TransportContext( - kind="streamable-http", - can_send_request=not self.is_json_response_enabled, - headers=request.headers if isinstance(request, Request) else None, - ) + return self.mcp_session_id is not None and not self.is_json_response_enabled def close_sse_stream(self, request_id: RequestId) -> None: """Close SSE connection for a specific request without terminating the stream. @@ -312,9 +301,14 @@ async def close_standalone_stream_callback() -> None: close_sse_stream=close_stream_callback, close_standalone_sse_stream=close_standalone_stream_callback, on_request_unanswered=end_stream, + can_send_request=self._request_channel_can_send_request, ) else: - metadata = ServerMessageMetadata(request_context=request, on_request_unanswered=end_stream) + metadata = ServerMessageMetadata( + request_context=request, + on_request_unanswered=end_stream, + can_send_request=self._request_channel_can_send_request, + ) return SessionMessage(message, metadata=metadata) @@ -582,7 +576,9 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re await response(scope, receive, send) # Process the message after sending the response - metadata = ServerMessageMetadata(request_context=request) + metadata = ServerMessageMetadata( + request_context=request, can_send_request=self._request_channel_can_send_request + ) session_message = SessionMessage(message, metadata=metadata) await writer.send(session_message) @@ -608,6 +604,7 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re metadata = ServerMessageMetadata( request_context=request, on_request_unanswered=partial(self._terminate_unanswered_request, message.id), + can_send_request=self._request_channel_can_send_request, ) session_message = SessionMessage(message, metadata=metadata) await writer.send(session_message) @@ -992,12 +989,6 @@ async def connect( ]: """Context manager that provides read and write streams for a connection. - Drive the yielded streams through a `JSONRPCDispatcher` built with - `transport_context_for` as its `transport_builder`, as the session - manager does: that is what makes the request-scoped channel refuse - server-initiated requests in JSON-response mode instead of parking - them where nothing can deliver them. - Yields: Tuple of (read_stream, write_stream) for bidirectional communication """ diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py index e23f0ee76a..415be5511b 100644 --- a/src/mcp/server/streamable_http_manager.py +++ b/src/mcp/server/streamable_http_manager.py @@ -6,7 +6,6 @@ import logging from collections import deque from collections.abc import AsyncIterator -from dataclasses import replace from typing import TYPE_CHECKING, Any, Final from uuid import uuid4 @@ -215,14 +214,11 @@ async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STA read_stream, write_stream, inline_methods=frozenset({"initialize"}), - # The transport says what its response stream can carry; on - # top of that, no session ID means the client's reply to a - # server request has nowhere to land, so the request-scoped - # channel refuses (`NoBackChannelError`) even in SSE mode - # while still forwarding notifications. - transport_builder=lambda md: replace( - http_transport.transport_context_for(md), can_send_request=False - ), + # No `transport_builder`: with no session ID the transport + # marks each message's request-scoped channel unable to + # carry a server-initiated request (the client's reply has + # nowhere to land), so requests raise `NoBackChannelError` + # while notifications still flow. ) # Born-ready, no standalone channel: the legacy stateless path # never opens a GET stream and need not see `initialize`. The @@ -324,14 +320,13 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE # Drive via `serve_loop` (not `Server.run()`) so the # manager's already-entered lifespan is reused # rather than re-entered per session; the transport - # rules on what each request's response can carry. + # marks each message with what its channel can carry. await serve_loop( self.app, read_stream, write_stream, lifespan_state=self._lifespan_state, session_id=http_transport.mcp_session_id, - transport_builder=http_transport.transport_context_for, ) if idle_scope.cancelled_caught: diff --git a/src/mcp/shared/jsonrpc_dispatcher.py b/src/mcp/shared/jsonrpc_dispatcher.py index 0d9467ffeb..1c8c997b66 100644 --- a/src/mcp/shared/jsonrpc_dispatcher.py +++ b/src/mcp/shared/jsonrpc_dispatcher.py @@ -184,8 +184,21 @@ def close(self) -> None: self._closed = True -def _default_transport_builder(_meta: MessageMetadata) -> TransportContext: - return TransportContext(kind="jsonrpc", can_send_request=True) +def _default_transport_builder(metadata: MessageMetadata) -> TransportContext: + """The `TransportContext` for a message, honoring the transport's own verdict when it stamps one. + + A message reads as riding a full duplex pipe (`can_send_request=True`) + unless the transport that framed it says otherwise on the metadata it + attached: streamable HTTP marks JSON-response-mode and sessionless + messages `ServerMessageMetadata(can_send_request=False)`, so their + request-scoped channel raises `NoBackChannelError` instead of parking a + waiter no reply can reach - with no wiring needed from whoever drives the + streams. + """ + can_send_request = True + if isinstance(metadata, ServerMessageMetadata) and metadata.can_send_request is not None: + can_send_request = metadata.can_send_request + return TransportContext(kind="jsonrpc", can_send_request=can_send_request) def _shielded_progress(fn: ProgressFnT) -> ProgressFnT: diff --git a/src/mcp/shared/message.py b/src/mcp/shared/message.py index 5ce255d15e..d148a1037b 100644 --- a/src/mcp/shared/message.py +++ b/src/mcp/shared/message.py @@ -45,6 +45,10 @@ class ServerMessageMetadata: # (e.g. it was cancelled), for a transport whose wire must still end the # request even though no response is written. on_request_unanswered: Callable[[], Awaitable[None]] | None = None + # The transport's verdict on whether this message's request-scoped channel + # can deliver a server-initiated request; `None` when the transport does + # not say, which the default `TransportContext` builder reads as True. + can_send_request: bool | None = None MessageMetadata = ClientMessageMetadata | ServerMessageMetadata | None diff --git a/tests/server/test_streamable_http_router.py b/tests/server/test_streamable_http_router.py index 81d51b959c..e14ed424a2 100644 --- a/tests/server/test_streamable_http_router.py +++ b/tests/server/test_streamable_http_router.py @@ -3,7 +3,6 @@ import anyio import pytest from mcp_types import JSONRPCMessage, JSONRPCResponse -from starlette.requests import Request from starlette.types import Message, Scope from mcp.server.streamable_http import ( @@ -15,8 +14,7 @@ StreamableHTTPServerTransport, StreamId, ) -from mcp.shared.message import ServerMessageMetadata, SessionMessage -from mcp.shared.transport_context import TransportContext +from mcp.shared.message import SessionMessage class _PrimingFailingStore(EventStore): @@ -158,18 +156,3 @@ async def asgi_send(message: Message) -> None: assert sent[0]["type"] == "http.response.start" assert sent[0]["status"] == 500 - - -def test_transport_context_reports_response_mode_and_request_headers() -> None: - """The transport's own verdict: no request-scoped requests in JSON mode, headers off the carried Request.""" - json_transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=True) - assert json_transport.transport_context_for(None) == TransportContext( - kind="streamable-http", can_send_request=False - ) - sse_transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=False) - assert sse_transport.transport_context_for(None) == TransportContext(kind="streamable-http", can_send_request=True) - - request = Request({"type": "http", "method": "POST", "path": "/", "headers": [(b"x-test", b"1")]}) - context = sse_transport.transport_context_for(ServerMessageMetadata(request_context=request)) - assert context.headers is not None - assert context.headers["x-test"] == "1" diff --git a/tests/shared/test_jsonrpc_dispatcher.py b/tests/shared/test_jsonrpc_dispatcher.py index 51f7f7c820..9bee8b2c3b 100644 --- a/tests/shared/test_jsonrpc_dispatcher.py +++ b/tests/shared/test_jsonrpc_dispatcher.py @@ -1329,6 +1329,45 @@ async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> assert seen[0] is metadata # the exact object, passed through verbatim +@pytest.mark.anyio +async def test_transport_stamped_can_send_request_makes_the_request_channel_refuse(): + """A transport that marks a message `can_send_request=False` on its metadata gets a request-scoped + channel that raises `NoBackChannelError` immediately - the default builder reads the transport's + verdict off the message, so no driver has to wire it.""" + c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) + s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) + server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send) + outcomes: list[bool | str] = [] + + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + outcomes.append(ctx.can_send_request) + try: + await ctx.send_raw_request("elicitation/create", {}) + except NoBackChannelError as exc: + outcomes.append(exc.method) + return {} + + async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None: + raise NotImplementedError + + try: + async with anyio.create_task_group() as tg: + await tg.start(server.run, on_request, on_notify) + await c2s_send.send( + SessionMessage( + message=JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params=None), + metadata=ServerMessageMetadata(can_send_request=False), + ) + ) + with anyio.fail_after(5): + await s2c_recv.receive() # response sent => the handler has run + tg.cancel_scope.cancel() + finally: + for s in (c2s_send, c2s_recv, s2c_send, s2c_recv): + s.close() + assert outcomes == [False, "elicitation/create"] + + @pytest.mark.anyio async def test_ctx_message_metadata_carries_inbound_notification_metadata(): """Notifications get the same metadata pass-through as requests.""" From 8d7695967913ab5ed0e7377d7f1621810a7ad704 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Mon, 27 Jul 2026 23:47:22 +0000 Subject: [PATCH 5/6] Narrow the transport's back-channel verdict to JSON-response mode The previous commit had the transport also refuse whenever it holds no session id, but a missing session id is not what makes a client's reply unroutable - the session manager creating a fresh transport per request is. A single long-lived sessionless transport still receives the reply on the same instance, so server-initiated requests over SSE kept working there and must continue to. The transport now states only the fact it owns (a JSON body carries one response), and the manager keeps its own can_send_request=False builder for the stateless leg. The manager module is unchanged from main. --- src/mcp/server/streamable_http.py | 9 +++++---- src/mcp/server/streamable_http_manager.py | 14 +++++++------- src/mcp/shared/jsonrpc_dispatcher.py | 9 ++++----- 3 files changed, 16 insertions(+), 16 deletions(-) diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 4d2b16b4dc..78addd3fe2 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -220,11 +220,12 @@ def is_terminated(self) -> bool: @property def _request_channel_can_send_request(self) -> bool: """Whether the request-scoped channel of a message this transport delivers can carry a - server-initiated request. It cannot in JSON-response mode (the POST is answered with one - JSON body) nor with no session (the client's reply would have no session to land on); - stamped on each message's `ServerMessageMetadata` so any dispatcher's builder reads it. + server-initiated request. It cannot in JSON-response mode: the POST is answered with one + JSON body, which holds only the response. Stamped on each message's + `ServerMessageMetadata` so any dispatcher's builder reads it; whether a reply has a session + to land on is the session manager's fact, not the transport's, so it is not decided here. """ - return self.mcp_session_id is not None and not self.is_json_response_enabled + return not self.is_json_response_enabled def close_sse_stream(self, request_id: RequestId) -> None: """Close SSE connection for a specific request without terminating the stream. diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py index 415be5511b..31f587ee66 100644 --- a/src/mcp/server/streamable_http_manager.py +++ b/src/mcp/server/streamable_http_manager.py @@ -214,11 +214,12 @@ async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STA read_stream, write_stream, inline_methods=frozenset({"initialize"}), - # No `transport_builder`: with no session ID the transport - # marks each message's request-scoped channel unable to - # carry a server-initiated request (the client's reply has - # nowhere to land), so requests raise `NoBackChannelError` - # while notifications still flow. + # No session ID means a server-to-client request can be + # written to this POST's response stream, but the client's + # reply has nowhere to land — `can_send_request=False` + # makes the per-request channel raise `NoBackChannelError` + # for requests while still allowing notifications. + transport_builder=lambda _md: TransportContext(kind="streamable-http", can_send_request=False), ) # Born-ready, no standalone channel: the legacy stateless path # never opens a GET stream and need not see `initialize`. The @@ -319,8 +320,7 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE with idle_scope: # Drive via `serve_loop` (not `Server.run()`) so the # manager's already-entered lifespan is reused - # rather than re-entered per session; the transport - # marks each message with what its channel can carry. + # rather than re-entered per session. await serve_loop( self.app, read_stream, diff --git a/src/mcp/shared/jsonrpc_dispatcher.py b/src/mcp/shared/jsonrpc_dispatcher.py index 1c8c997b66..fa143a15a7 100644 --- a/src/mcp/shared/jsonrpc_dispatcher.py +++ b/src/mcp/shared/jsonrpc_dispatcher.py @@ -189,11 +189,10 @@ def _default_transport_builder(metadata: MessageMetadata) -> TransportContext: A message reads as riding a full duplex pipe (`can_send_request=True`) unless the transport that framed it says otherwise on the metadata it - attached: streamable HTTP marks JSON-response-mode and sessionless - messages `ServerMessageMetadata(can_send_request=False)`, so their - request-scoped channel raises `NoBackChannelError` instead of parking a - waiter no reply can reach - with no wiring needed from whoever drives the - streams. + attached: streamable HTTP marks JSON-response-mode messages + `ServerMessageMetadata(can_send_request=False)`, so their request-scoped + channel raises `NoBackChannelError` instead of parking a waiter no reply + can reach - with no wiring needed from whoever drives the streams. """ can_send_request = True if isinstance(metadata, ServerMessageMetadata) and metadata.can_send_request is not None: From ca1a848b535b4eb2f5094c964d18b565f6c4f76d Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Tue, 28 Jul 2026 00:03:58 +0000 Subject: [PATCH 6/6] Stamp the JSON-mode verdict in one metadata factory Route every ServerMessageMetadata the transport frames through a single _message_metadata factory so the can_send_request stamp has one home a future construction site cannot forget. The field becomes a plain bool defaulting to True, dropping the None sentinel and the branch that read it. Docstrings now defer to TransportContext.can_send_request for the full rule instead of restating it, and the two direct-transport tests share one fake-ASGI scaffold. --- src/mcp/server/streamable_http.py | 77 +++++++++---------- src/mcp/shared/exceptions.py | 11 +-- src/mcp/shared/jsonrpc_dispatcher.py | 11 +-- src/mcp/shared/message.py | 7 +- tests/server/test_streamable_http_router.py | 83 +++++++++------------ 5 files changed, 83 insertions(+), 106 deletions(-) diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 78addd3fe2..1a4e9939a4 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -44,7 +44,7 @@ from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams from mcp.shared._stream_protocols import ReadStream, WriteStream from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER -from mcp.shared.message import ServerMessageMetadata, SessionMessage +from mcp.shared.message import CloseSSEStreamCallback, ServerMessageMetadata, SessionMessage logger = logging.getLogger(__name__) @@ -174,12 +174,11 @@ def __init__( mcp_session_id: Optional session identifier for this connection. Must contain only visible ASCII characters (0x21-0x7E). is_json_response_enabled: If True, answer each request POST with a single - JSON body instead of an SSE stream. The body carries - only the response, so this transport marks its messages - as having no request-scoped back-channel: a - server-initiated request on that channel raises - `NoBackChannelError` and request-scoped notifications - are dropped. Default is False. + JSON body instead of an SSE stream, which removes + the request-scoped back-channel: a server-initiated + request tied to the call raises `NoBackChannelError` + and its notifications are dropped (see + `TransportContext.can_send_request`). Default is False. event_store: Event store for resumability support. If provided, resumability will be enabled, allowing clients to reconnect and resume messages. @@ -217,15 +216,28 @@ def is_terminated(self) -> bool: """Check if this transport has been explicitly terminated.""" return self._terminated - @property - def _request_channel_can_send_request(self) -> bool: - """Whether the request-scoped channel of a message this transport delivers can carry a - server-initiated request. It cannot in JSON-response mode: the POST is answered with one - JSON body, which holds only the response. Stamped on each message's - `ServerMessageMetadata` so any dispatcher's builder reads it; whether a reply has a session - to land on is the session manager's fact, not the transport's, so it is not decided here. + def _message_metadata( + self, + request: Request, + *, + close_sse_stream: CloseSSEStreamCallback | None = None, + close_standalone_sse_stream: CloseSSEStreamCallback | None = None, + on_request_unanswered: Callable[[], Awaitable[None]] | None = None, + ) -> ServerMessageMetadata: + """The metadata this transport frames every inbound message with. + + The one place `can_send_request` is stamped, so no construction site can + forget it: a JSON body carries only the response, so in JSON-response mode + the request-scoped channel cannot carry a server-initiated request (see + `TransportContext.can_send_request`). """ - return not self.is_json_response_enabled + return ServerMessageMetadata( + request_context=request, + close_sse_stream=close_sse_stream, + close_standalone_sse_stream=close_standalone_sse_stream, + on_request_unanswered=on_request_unanswered, + can_send_request=not self.is_json_response_enabled, + ) def close_sse_stream(self, request_id: RequestId) -> None: """Close SSE connection for a specific request without terminating the stream. @@ -297,19 +309,14 @@ async def close_stream_callback() -> None: async def close_standalone_stream_callback() -> None: self.close_standalone_sse_stream() - metadata = ServerMessageMetadata( - request_context=request, + metadata = self._message_metadata( + request, close_sse_stream=close_stream_callback, close_standalone_sse_stream=close_standalone_stream_callback, on_request_unanswered=end_stream, - can_send_request=self._request_channel_can_send_request, ) else: - metadata = ServerMessageMetadata( - request_context=request, - on_request_unanswered=end_stream, - can_send_request=self._request_channel_can_send_request, - ) + metadata = self._message_metadata(request, on_request_unanswered=end_stream) return SessionMessage(message, metadata=metadata) @@ -577,10 +584,7 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re await response(scope, receive, send) # Process the message after sending the response - metadata = ServerMessageMetadata( - request_context=request, can_send_request=self._request_channel_can_send_request - ) - session_message = SessionMessage(message, metadata=metadata) + session_message = SessionMessage(message, metadata=self._message_metadata(request)) await writer.send(session_message) return @@ -602,10 +606,8 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re ) request_stream_reader = self._request_streams[request_id][1] # Process the message - metadata = ServerMessageMetadata( - request_context=request, - on_request_unanswered=partial(self._terminate_unanswered_request, message.id), - can_send_request=self._request_channel_can_send_request, + metadata = self._message_metadata( + request, on_request_unanswered=partial(self._terminate_unanswered_request, message.id) ) session_message = SessionMessage(message, metadata=metadata) await writer.send(session_message) @@ -1030,15 +1032,10 @@ async def message_router(): ): related_request_id = session_message.metadata.related_request_id if self.is_json_response_enabled: - # A JSON body carries exactly the one response: a - # message that only rides this request's stream - # has no wire form (and no POST could replay it), - # so drop it before storing or queueing - never - # park it in a queue nothing drains (#1764). - logger.debug( - f"Dropped message related to request {related_request_id}: " - "a JSON response carries only its own response" - ) + # A JSON body carries only the response: this message + # has no wire form (nor a replay), so drop it before + # storing or queueing rather than park it (#1764). + logger.debug(f"Dropped message related to request {related_request_id} in JSON mode") continue target_request_id = str(related_request_id) diff --git a/src/mcp/shared/exceptions.py b/src/mcp/shared/exceptions.py index 99ca0d1ae4..c2a7fd44e7 100644 --- a/src/mcp/shared/exceptions.py +++ b/src/mcp/shared/exceptions.py @@ -55,13 +55,10 @@ def __str__(self) -> str: class NoBackChannelError(MCPError): """Raised when a server-initiated request has no channel that can deliver it. - The request-scoped channel is a dead end when the response has no room - (streamable HTTP in JSON-response mode, the 2026-07-28 single-exchange - entry), when the client's reply has nowhere to land (stateless HTTP), or - when the protocol forbids server-initiated requests (2026-07-28). Raised by - `DispatchContext.send_raw_request` when `can_send_request` is `False`, and - by a connection's standalone channel when it has none; serializes to an - `INVALID_REQUEST` error response. + Raised by `DispatchContext.send_raw_request` when its request-scoped channel + reports `TransportContext.can_send_request` as `False` (the cases are + documented on that field), and by a connection's standalone channel when it + has none; serializes to an `INVALID_REQUEST` error response. """ def __init__(self, method: str): diff --git a/src/mcp/shared/jsonrpc_dispatcher.py b/src/mcp/shared/jsonrpc_dispatcher.py index fa143a15a7..87bdf31ceb 100644 --- a/src/mcp/shared/jsonrpc_dispatcher.py +++ b/src/mcp/shared/jsonrpc_dispatcher.py @@ -189,14 +189,11 @@ def _default_transport_builder(metadata: MessageMetadata) -> TransportContext: A message reads as riding a full duplex pipe (`can_send_request=True`) unless the transport that framed it says otherwise on the metadata it - attached: streamable HTTP marks JSON-response-mode messages - `ServerMessageMetadata(can_send_request=False)`, so their request-scoped - channel raises `NoBackChannelError` instead of parking a waiter no reply - can reach - with no wiring needed from whoever drives the streams. + attached, so a transport whose response has no room for a server request + (streamable HTTP in JSON-response mode) needs no wiring from whoever drives + its streams. """ - can_send_request = True - if isinstance(metadata, ServerMessageMetadata) and metadata.can_send_request is not None: - can_send_request = metadata.can_send_request + can_send_request = metadata.can_send_request if isinstance(metadata, ServerMessageMetadata) else True return TransportContext(kind="jsonrpc", can_send_request=can_send_request) diff --git a/src/mcp/shared/message.py b/src/mcp/shared/message.py index d148a1037b..31e51e7128 100644 --- a/src/mcp/shared/message.py +++ b/src/mcp/shared/message.py @@ -46,9 +46,10 @@ class ServerMessageMetadata: # request even though no response is written. on_request_unanswered: Callable[[], Awaitable[None]] | None = None # The transport's verdict on whether this message's request-scoped channel - # can deliver a server-initiated request; `None` when the transport does - # not say, which the default `TransportContext` builder reads as True. - can_send_request: bool | None = None + # can deliver a server-initiated request (see + # `TransportContext.can_send_request`); a transport that says nothing leaves + # it True. + can_send_request: bool = True MessageMetadata = ClientMessageMetadata | ServerMessageMetadata | None diff --git a/tests/server/test_streamable_http_router.py b/tests/server/test_streamable_http_router.py index e14ed424a2..07aa063499 100644 --- a/tests/server/test_streamable_http_router.py +++ b/tests/server/test_streamable_http_router.py @@ -25,6 +25,25 @@ async def replay_events_after(self, last_event_id: EventId, send_callback: Event raise NotImplementedError +class _AsgiPost: + """A one-shot POST driven straight at `handle_request`, capturing what the transport sends.""" + + def __init__(self, body: bytes, headers: list[tuple[bytes, bytes]]) -> None: + self.scope: Scope = {"type": "http", "method": "POST", "path": "/", "query_string": b"", "headers": headers} + self.sent: list[Message] = [] + self._body = body + self._body_sent = False + + async def receive(self) -> Message: + if not self._body_sent: + self._body_sent = True + return {"type": "http.request", "body": self._body, "more_body": False} + raise NotImplementedError + + async def send(self, message: Message) -> None: + self.sent.append(message) + + @pytest.mark.anyio async def test_router_unconsumed_request_stream_does_not_block_siblings() -> None: """A response whose `sse_writer` is not yet receiving must not park the router (#1764). @@ -73,35 +92,18 @@ async def test_priming_store_failure_leaves_no_per_request_state() -> None: event_store=_PrimingFailingStore(), ) - body = b'{"jsonrpc":"2.0","id":"req-1","method":"tools/list","params":{}}' - scope: Scope = { - "type": "http", - "method": "POST", - "path": "/", - "query_string": b"", - "headers": [ + post = _AsgiPost( + b'{"jsonrpc":"2.0","id":"req-1","method":"tools/list","params":{}}', + [ (b"accept", b"application/json, text/event-stream"), (b"content-type", b"application/json"), (b"mcp-protocol-version", b"2025-11-25"), ], - } - body_sent = False - - async def receive() -> Message: - nonlocal body_sent - if not body_sent: - body_sent = True - return {"type": "http.request", "body": body, "more_body": False} - raise NotImplementedError - - sent: list[Message] = [] - - async def asgi_send(message: Message) -> None: - sent.append(message) + ) async with transport.connect() as (read_stream, _write_stream): async with anyio.create_task_group() as tg: - tg.start_soon(transport.handle_request, scope, receive, asgi_send) + tg.start_soon(transport.handle_request, post.scope, post.receive, post.send) with anyio.fail_after(5): forwarded = await read_stream.receive() assert isinstance(forwarded, Exception) @@ -110,9 +112,9 @@ async def asgi_send(message: Message) -> None: assert transport._request_streams == {} assert transport._sse_stream_writers == {} - assert sent[0]["type"] == "http.response.start" - assert sent[0]["status"] == 500 - body = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body") + assert post.sent[0]["type"] == "http.response.start" + assert post.sent[0]["status"] == 500 + body = b"".join(m.get("body", b"") for m in post.sent if m["type"] == "http.response.body") assert b"backend unavailable" not in body @@ -120,39 +122,22 @@ async def asgi_send(message: Message) -> None: async def test_json_post_answers_500_when_session_terminates_mid_request() -> None: """A JSON-mode POST whose session is torn down before the handler answers gets a 500, not a stall.""" transport = StreamableHTTPServerTransport(mcp_session_id="sid", is_json_response_enabled=True) - body = b'{"jsonrpc":"2.0","id":"req-1","method":"tools/list","params":{}}' - scope: Scope = { - "type": "http", - "method": "POST", - "path": "/", - "query_string": b"", - "headers": [ + post = _AsgiPost( + b'{"jsonrpc":"2.0","id":"req-1","method":"tools/list","params":{}}', + [ (b"accept", b"application/json"), (b"content-type", b"application/json"), (b"mcp-session-id", b"sid"), (b"mcp-protocol-version", b"2025-11-25"), ], - } - body_sent = False - - async def receive() -> Message: - nonlocal body_sent - if not body_sent: - body_sent = True - return {"type": "http.request", "body": body, "more_body": False} - raise NotImplementedError - - sent: list[Message] = [] - - async def asgi_send(message: Message) -> None: - sent.append(message) + ) async with transport.connect() as (read_stream, _write_stream): async with anyio.create_task_group() as tg: - tg.start_soon(transport.handle_request, scope, receive, asgi_send) + tg.start_soon(transport.handle_request, post.scope, post.receive, post.send) with anyio.fail_after(5): await read_stream.receive() # the request reached the session; the POST is parked await transport.terminate() - assert sent[0]["type"] == "http.response.start" - assert sent[0]["status"] == 500 + assert post.sent[0]["type"] == "http.response.start" + assert post.sent[0]["status"] == 500