Skip to content

Commit 1c6bc27

Browse files
committed
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`, 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.
1 parent 9cd1041 commit 1c6bc27

4 files changed

Lines changed: 111 additions & 9 deletions

File tree

docs/_server/subscriptions.md

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,9 @@ The first SSE event on the stream is the acknowledgement:
5252
## Transport Support
5353

5454
The stream is served on the Streamable HTTP modern path; stdio answers `-32601`.
55+
A host that cannot hold an SSE response open (such as the [Rails controller pattern](/server/transports/#rails-controller),
56+
which buffers the body) declines the method with `serve_subscriptions_listen: false`, answering `-32601` and dropping
57+
the `listChanged`/`subscribe` flags from discovery.
5558

5659
## Limits and Keepalives
5760

docs/_server/transports.md

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -153,15 +153,29 @@ class McpController < ActionController::API
153153
prompts: [MyPrompt],
154154
server_context: { user_id: current_user.id },
155155
)
156-
# Since the `MCP-Session-Id` is not shared across requests, `stateless: true` is set.
157-
transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, stateless: true)
156+
# Since the `MCP-Session-Id` is not shared across requests, `stateless: true` is set,
157+
# and since `render` buffers the whole body, `subscriptions/listen` streams are declined.
158+
transport = MCP::Server::Transports::StreamableHTTPTransport.new(
159+
server,
160+
stateless: true,
161+
serve_subscriptions_listen: false,
162+
)
158163
status, headers, body = transport.handle_request(request)
159164

160165
render(json: body.first, status: status, headers: headers)
161166
end
162167
end
163168
```
164169

170+
{: .important }
171+
> The controller pattern builds a fresh transport per request and `render` buffers the whole body,
172+
> so it cannot serve the open SSE stream that [`subscriptions/listen`](/server/subscriptions/) needs:
173+
> without `serve_subscriptions_listen: false`, that request returns a streaming `Proc` body
174+
> that `body.first` cannot render, and a modern client's notification stream fails with a 500.
175+
> With the flag, the method answers 404 with JSON-RPC `-32601`, and `server/discover` stops
176+
> advertising the `listChanged`/`subscribe` capability flags, so well-behaved clients do not ask.
177+
> To actually serve listen streams, use the [mount approach](#rails-mount).
178+
165179
### Stateless Mode
166180

167181
You can use Stateless Streamable HTTP, where notifications are not supported and all calls are request/response interactions.

lib/mcp/server/transports/streamable_http_transport.rb

Lines changed: 45 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,12 @@ class InvalidJsonError < StandardError; end
129129
# on a `subscriptions/listen` stream; the periodic write frees the stream's slot when the peer
130130
# has gone away. Defaults to `DEFAULT_LISTEN_KEEPALIVE_INTERVAL` (15); pass `nil` to disable
131131
# when an upstream proxy already keeps the stream alive.
132+
# @param serve_subscriptions_listen [Boolean] whether `subscriptions/listen` opens a stream.
133+
# A host that buffers responses and cannot serve an open SSE stream (e.g. the Rails controller pattern,
134+
# which builds a fresh transport per request and renders the body) passes `false`:
135+
# the method then answers 404 with JSON-RPC `-32601` like any unimplemented method,
136+
# and `Server#discover` stops advertising the `listChanged`/`subscribe` capability flags,
137+
# keeping the advertisement and the actual behavior in agreement. Defaults to `true`.
132138
# @param server_to_client_request_timeout [Numeric] seconds a server-to-client request waits for its
133139
# response before the transport stops waiting and raises `MCP::Server::RequestTimeoutError`.
134140
# Defaults to `DEFAULT_SERVER_TO_CLIENT_REQUEST_TIMEOUT` (600); individual calls override it with `timeout:`.
@@ -145,6 +151,7 @@ def initialize(
145151
max_request_bytes: DEFAULT_MAX_REQUEST_BYTES,
146152
max_listen_subscriptions: DEFAULT_MAX_LISTEN_SUBSCRIPTIONS,
147153
listen_keepalive_interval: DEFAULT_LISTEN_KEEPALIVE_INTERVAL,
154+
serve_subscriptions_listen: true,
148155
server_to_client_request_timeout: DEFAULT_SERVER_TO_CLIENT_REQUEST_TIMEOUT
149156
)
150157
super(server)
@@ -211,6 +218,7 @@ def initialize(
211218
end
212219

213220
@listen_keepalive_interval = listen_keepalive_interval
221+
@serve_subscriptions_listen = serve_subscriptions_listen
214222

215223
unless server_to_client_request_timeout.is_a?(Numeric) && server_to_client_request_timeout.positive?
216224
raise ArgumentError, "server_to_client_request_timeout must be a positive number"
@@ -260,10 +268,12 @@ def call(env)
260268
handle_request(Rack::Request.new(env))
261269
end
262270

263-
# The `subscriptions/listen` notification stream (SEP-2575) is served on the modern path,
264-
# so `Server#discover` may advertise `listChanged`/`subscribe` capability flags.
271+
# Whether this transport serves the `subscriptions/listen` notification stream (SEP-2575).
272+
# Gates both the route (a refusing transport answers the method as unimplemented) and
273+
# the `listChanged`/`subscribe` capability flags `Server#discover` advertises,
274+
# so the two always agree. Set via the `serve_subscriptions_listen:` constructor keyword.
265275
def serves_subscriptions_listen?
266-
true
276+
@serve_subscriptions_listen
267277
end
268278

269279
def handle_request(request)
@@ -723,8 +733,13 @@ def handle_modern(request, header_version, body_string: nil)
723733
return mismatch_error if mismatch_error
724734

725735
# `subscriptions/listen` is a long-lived notification stream served at the transport layer;
726-
# it never dispatches through `Server#handle`.
727-
return handle_subscriptions_listen(body) if body[:method] == Methods::SUBSCRIPTIONS_LISTEN
736+
# it never dispatches through `Server#handle`. A transport constructed with
737+
# `serve_subscriptions_listen: false` skips the interception, so the method falls through
738+
# to the dispatcher as unimplemented (404 with `-32601`) - the refusal a host that cannot
739+
# serve an open SSE stream needs, instead of a `Proc` body it can never call.
740+
if body[:method] == Methods::SUBSCRIPTIONS_LISTEN && serves_subscriptions_listen?
741+
return handle_subscriptions_listen(body)
742+
end
728743

729744
session = modern_session
730745
notifications = @mutex.synchronize { @modern_request_sinks[session.session_id] = [] }
@@ -881,10 +896,33 @@ def too_many_listen_subscriptions_response(request_id)
881896
)
882897
end
883898

884-
# The proc registers the stream and returns, leaving the response open like
899+
# The Rack streaming body of a `subscriptions/listen` response. It responds to `call`
900+
# and deliberately not to `each`, so Rack keeps classifying it as a streaming body;
901+
# `first` exists only to turn the buffered-host mistake (e.g. `render(json: body.first)`
902+
# in the Rails controller pattern) from a bare NoMethodError into guidance naming the fix.
903+
class ListenStreamBody
904+
def initialize(&block)
905+
@block = block
906+
end
907+
908+
def call(stream)
909+
@block.call(stream)
910+
end
911+
912+
def first
913+
raise <<~MESSAGE
914+
subscriptions/listen returned a streaming SSE body, which cannot be buffered into a JSON response. \
915+
A host that cannot hold an SSE response open should construct the transport with `serve_subscriptions_listen: false`, \
916+
so the method is answered as unimplemented instead.
917+
MESSAGE
918+
end
919+
end
920+
private_constant :ListenStreamBody
921+
922+
# The body registers the stream and returns, leaving the response open like
885923
# the legacy GET stream (`create_sse_body`).
886924
def listen_sse_body(request_id, honored)
887-
proc do |stream|
925+
ListenStreamBody.new do |stream|
888926
rejected = false
889927
@mutex.synchronize do
890928
if @listen_subscriptions.key?(request_id) ||

test/mcp/server/transports/streamable_http_transport_test.rb

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6156,6 +6156,53 @@ def string
61566156
transport.close
61576157
end
61586158

6159+
test "a buffered read of the listen stream body raises guidance instead of NoMethodError" do
6160+
response = @transport.handle_request(modern_rack_request(
6161+
modern_listen_body(id: "listen-1", params: { notifications: { toolsListChanged: true } }),
6162+
))
6163+
6164+
error = assert_raises(RuntimeError) { response[2].first }
6165+
assert_includes error.message, "serve_subscriptions_listen: false"
6166+
6167+
# Responding to `each` would make Rack classify the body as enumerable and break streaming.
6168+
refute_respond_to response[2], :each
6169+
end
6170+
6171+
test "subscriptions/listen is refused as unimplemented when the transport does not serve it" do
6172+
transport = StreamableHTTPTransport.new(@server, serve_subscriptions_listen: false)
6173+
6174+
status, headers, body = transport.handle_request(modern_rack_request(
6175+
modern_listen_body(id: "listen-1", params: { notifications: { toolsListChanged: true } }),
6176+
))
6177+
6178+
assert_equal 404, status
6179+
assert_equal "application/json", headers["content-type"]
6180+
6181+
# The documented Rails controller pattern renders `body.first`, so the refusal must be an Array body,
6182+
# never the streaming Proc.
6183+
assert_kind_of Array, body
6184+
assert_equal(-32601, JSON.parse(body.first).dig("error", "code"))
6185+
ensure
6186+
transport.close
6187+
end
6188+
6189+
test "discover stops advertising subscription capabilities when listen is not served" do
6190+
server = Server.new(
6191+
name: "listen_test",
6192+
capabilities: { tools: { listChanged: true }, resources: { listChanged: true, subscribe: true } },
6193+
)
6194+
transport = StreamableHTTPTransport.new(server, serve_subscriptions_listen: false)
6195+
6196+
response = transport.handle_request(modern_rack_request(modern_body("server/discover", {})))
6197+
capabilities = JSON.parse(response[2].first).dig("result", "capabilities")
6198+
6199+
assert_nil capabilities.dig("tools", "listChanged")
6200+
assert_nil capabilities.dig("resources", "listChanged")
6201+
assert_nil capabilities.dig("resources", "subscribe")
6202+
ensure
6203+
transport.close
6204+
end
6205+
61596206
test "subscriptions/listen past the concurrent stream cap is rejected with 503" do
61606207
transport = StreamableHTTPTransport.new(@server, max_listen_subscriptions: 1)
61616208
open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true }, transport: transport)

0 commit comments

Comments
 (0)