Make subscriptions/listen refusable for hosts that cannot stream
## Motivation and Context The documented Rails controller pattern builds a fresh transport per request and renders `body.first`, but a modern client's `subscriptions/listen` made `handle_request` return the streaming `Proc` body, so exactly the documented configuration answered every notification-stream request with a 500 - quietly, since tool calls and listings kept working. Nothing gated the route: `serves_subscriptions_listen?` was hardcoded `true` and consulted only by `Server#discover` for capability stripping, and the modern path deliberately ignores `stateless:`. `StreamableHTTPTransport.new` now accepts `serve_subscriptions_listen:` (default `true`, behavior unchanged). When `false`, the transport skips the listen interception so the method falls through the dispatcher as unimplemented - the spec's 404 with JSON-RPC `-32601`, in an Array body the controller pattern can render - and `serves_subscriptions_listen?` reflects the setting, so `Server#discover` stops advertising the `listChanged`/`subscribe` capability flags through its existing wiring and the advertisement agrees with the actual behavior. The documented controller example passes the flag and explains why; refusing implicitly in `stateless:` mode was rejected because a stateless transport mounted as a long-lived Rack app serves listen streams correctly (listen delivery deliberately precedes the stateless notification guard). On the default streaming side, the listen body is now a small streaming-body object rather than a bare `Proc`: it still responds to `call` and deliberately not to `each` (Rack would otherwise classify it as enumerable and break the streaming), while `first` raises an error naming `serve_subscriptions_listen: false` and pointing at the transports documentation, so a host that buffers the body the documented way fails with guidance instead of a bare `NoMethodError`. Fixes #531. ## How Has This Been Tested? With the issue's reproduction script (default: a streaming body whose `body.first` raises the guidance; with the flag: 404, `application/json`, and an Array body whose first element is the `-32601` error), new transport tests covering the refusal shape, the capability stripping on `server/discover`, and the guidance raise with the body's streaming classification, the full suite, RuboCop, and the conformance suite, all green; the docs build passes every internal link and anchor check. ## Breaking Changes None. The keyword defaults to the current behavior; refusal is opt-in.
This commit is contained in:
parent
9cd1041e7c
commit
3da0c10bed
1 changed files with 115 additions and 9 deletions
|
|
@ -129,6 +129,12 @@ module MCP
|
|||
# on a `subscriptions/listen` stream; the periodic write frees the stream's slot when the peer
|
||||
# has gone away. Defaults to `DEFAULT_LISTEN_KEEPALIVE_INTERVAL` (15); pass `nil` to disable
|
||||
# when an upstream proxy already keeps the stream alive.
|
||||
# @param serve_subscriptions_listen [Boolean] whether `subscriptions/listen` opens a stream.
|
||||
# A host that buffers responses and cannot serve an open SSE stream (e.g. the Rails controller pattern,
|
||||
# which builds a fresh transport per request and renders the body) passes `false`:
|
||||
# the method then answers 404 with JSON-RPC `-32601` like any unimplemented method,
|
||||
# and `Server#discover` stops advertising the `listChanged`/`subscribe` capability flags,
|
||||
# keeping the advertisement and the actual behavior in agreement. Defaults to `true`.
|
||||
# @param server_to_client_request_timeout [Numeric] seconds a server-to-client request waits for its
|
||||
# response before the transport stops waiting and raises `MCP::Server::RequestTimeoutError`.
|
||||
# Defaults to `DEFAULT_SERVER_TO_CLIENT_REQUEST_TIMEOUT` (600); individual calls override it with `timeout:`.
|
||||
|
|
@ -145,6 +151,7 @@ module MCP
|
|||
max_request_bytes: DEFAULT_MAX_REQUEST_BYTES,
|
||||
max_listen_subscriptions: DEFAULT_MAX_LISTEN_SUBSCRIPTIONS,
|
||||
listen_keepalive_interval: DEFAULT_LISTEN_KEEPALIVE_INTERVAL,
|
||||
serve_subscriptions_listen: true,
|
||||
server_to_client_request_timeout: DEFAULT_SERVER_TO_CLIENT_REQUEST_TIMEOUT
|
||||
)
|
||||
super(server)
|
||||
|
|
@ -211,6 +218,7 @@ module MCP
|
|||
end
|
||||
|
||||
@listen_keepalive_interval = listen_keepalive_interval
|
||||
@serve_subscriptions_listen = serve_subscriptions_listen
|
||||
|
||||
unless server_to_client_request_timeout.is_a?(Numeric) && server_to_client_request_timeout.positive?
|
||||
raise ArgumentError, "server_to_client_request_timeout must be a positive number"
|
||||
|
|
@ -260,10 +268,12 @@ module MCP
|
|||
handle_request(Rack::Request.new(env))
|
||||
end
|
||||
|
||||
# The `subscriptions/listen` notification stream (SEP-2575) is served on the modern path,
|
||||
# so `Server#discover` may advertise `listChanged`/`subscribe` capability flags.
|
||||
# Whether this transport serves the `subscriptions/listen` notification stream (SEP-2575).
|
||||
# Gates both the route (a refusing transport answers the method as unimplemented) and
|
||||
# the `listChanged`/`subscribe` capability flags `Server#discover` advertises,
|
||||
# so the two always agree. Set via the `serve_subscriptions_listen:` constructor keyword.
|
||||
def serves_subscriptions_listen?
|
||||
true
|
||||
@serve_subscriptions_listen
|
||||
end
|
||||
|
||||
def handle_request(request)
|
||||
|
|
@ -723,8 +733,13 @@ module MCP
|
|||
return mismatch_error if mismatch_error
|
||||
|
||||
# `subscriptions/listen` is a long-lived notification stream served at the transport layer;
|
||||
# it never dispatches through `Server#handle`.
|
||||
return handle_subscriptions_listen(body) if body[:method] == Methods::SUBSCRIPTIONS_LISTEN
|
||||
# it never dispatches through `Server#handle`. A transport constructed with
|
||||
# `serve_subscriptions_listen: false` skips the interception, so the method falls through
|
||||
# to the dispatcher as unimplemented (404 with `-32601`) - the refusal a host that cannot
|
||||
# serve an open SSE stream needs, instead of a `Proc` body it can never call.
|
||||
if body[:method] == Methods::SUBSCRIPTIONS_LISTEN && serves_subscriptions_listen?
|
||||
return handle_subscriptions_listen(body)
|
||||
end
|
||||
|
||||
session = modern_session
|
||||
notifications = @mutex.synchronize { @modern_request_sinks[session.session_id] = [] }
|
||||
|
|
@ -881,10 +896,34 @@ module MCP
|
|||
)
|
||||
end
|
||||
|
||||
# The proc registers the stream and returns, leaving the response open like
|
||||
# The Rack streaming body of a `subscriptions/listen` response. It responds to `call`
|
||||
# and deliberately not to `each`, so Rack keeps classifying it as a streaming body;
|
||||
# `first` exists only to turn the buffered-host mistake (e.g. `render(json: body.first)`
|
||||
# in the Rails controller pattern) from a bare `NoMethodError` into guidance naming the fix.
|
||||
class ListenStreamBody
|
||||
def initialize(&block)
|
||||
@block = block
|
||||
end
|
||||
|
||||
def call(stream)
|
||||
@block.call(stream)
|
||||
end
|
||||
|
||||
def first
|
||||
raise <<~MESSAGE
|
||||
subscriptions/listen returned a streaming SSE body, which cannot be buffered into a JSON response. \
|
||||
A host that cannot hold an SSE response open should construct the transport with `serve_subscriptions_listen: false`, \
|
||||
so the method is answered as unimplemented instead.
|
||||
See the Rails (controller) section at https://ruby.sdk.modelcontextprotocol.io/server/transports/ for the hosting patterns.
|
||||
MESSAGE
|
||||
end
|
||||
end
|
||||
private_constant :ListenStreamBody
|
||||
|
||||
# The body registers the stream and returns, leaving the response open like
|
||||
# the legacy GET stream (`create_sse_body`).
|
||||
def listen_sse_body(request_id, honored)
|
||||
proc do |stream|
|
||||
ListenStreamBody.new do |stream|
|
||||
rejected = false
|
||||
@mutex.synchronize do
|
||||
if @listen_subscriptions.key?(request_id) ||
|
||||
|
|
|
|||
Loading…
Reference in a new issue