Merge pull request #533 from koic/make_subscriptions_listen_refusable

Make `subscriptions/listen` refusable for hosts that cannot stream

Author: koic

docs/_server/subscriptions.md

@@ -52,6 +52,9 @@ The first SSE event on the stream is the acknowledgement:
 ## Transport Support
 
 The stream is served on the Streamable HTTP modern path; stdio answers `-32601`.
+A host that cannot hold an SSE response open (such as the [Rails controller pattern](/server/transports/#rails-controller),
+which buffers the body) declines the method with `serve_subscriptions_listen: false`, answering `-32601` and dropping
+the `listChanged`/`subscribe` flags from discovery.
 
 ## Limits and Keepalives
 

docs/_server/transports.md

@@ -153,15 +153,29 @@ class McpController < ActionController::API
       prompts: [MyPrompt],
       server_context: { user_id: current_user.id },
     )
-    # Since the `MCP-Session-Id` is not shared across requests, `stateless: true` is set.
-    transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, stateless: true)
+    # Since the `MCP-Session-Id` is not shared across requests, `stateless: true` is set,
+    # and since `render` buffers the whole body, `subscriptions/listen` streams are declined.
+    transport = MCP::Server::Transports::StreamableHTTPTransport.new(
+      server,
+      stateless: true,
+      serve_subscriptions_listen: false,
+    )
     status, headers, body = transport.handle_request(request)
 
     render(json: body.first, status: status, headers: headers)
   end
 end
 ```
 
+{: .important }
+> The controller pattern builds a fresh transport per request and `render` buffers the whole body,
+> so it cannot serve the open SSE stream that [`subscriptions/listen`](/server/subscriptions/) needs:
+> without `serve_subscriptions_listen: false`, that request returns a streaming `Proc` body
+> that `body.first` cannot render, and a modern client's notification stream fails with a 500.
+> With the flag, the method answers 404 with JSON-RPC `-32601`, and `server/discover` stops
+> advertising the `listChanged`/`subscribe` capability flags, so well-behaved clients do not ask.
+> To actually serve listen streams, use the [mount approach](#rails-mount).
+
 ### Stateless Mode
 
 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

@@ -129,6 +129,12 @@ class InvalidJsonError < StandardError; end
         #   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 @@ def initialize(
           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 @@ def initialize(
           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 @@ def call(env)
           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 @@ def handle_modern(request, header_version, body_string: nil)
           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 @@ def too_many_listen_subscriptions_response(request_id)
           )
         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) ||

test/mcp/server/transports/streamable_http_transport_test.rb

@@ -6156,6 +6156,56 @@ def string
           transport.close
         end
 
+        test "a buffered read of the listen stream body raises guidance instead of NoMethodError" do
+          response = @transport.handle_request(modern_rack_request(
+            modern_listen_body(id: "listen-1", params: { notifications: { toolsListChanged: true } }),
+          ))
+
+          error = assert_raises(RuntimeError) { response[2].first }
+          assert_includes error.message, "serve_subscriptions_listen: false"
+
+          # The cited page is a published-URL contract, like the kwarg name above.
+          assert_includes error.message, "https://ruby.sdk.modelcontextprotocol.io/server/transports/"
+
+          # Responding to `each` would make Rack classify the body as enumerable and break streaming.
+          refute_respond_to response[2], :each
+        end
+
+        test "subscriptions/listen is refused as unimplemented when the transport does not serve it" do
+          transport = StreamableHTTPTransport.new(@server, serve_subscriptions_listen: false)
+
+          status, headers, body = transport.handle_request(modern_rack_request(
+            modern_listen_body(id: "listen-1", params: { notifications: { toolsListChanged: true } }),
+          ))
+
+          assert_equal 404, status
+          assert_equal "application/json", headers["content-type"]
+
+          # The documented Rails controller pattern renders `body.first`, so the refusal must be an Array body,
+          # never the streaming Proc.
+          assert_kind_of Array, body
+          assert_equal(-32601, JSON.parse(body.first).dig("error", "code"))
+        ensure
+          transport.close
+        end
+
+        test "discover stops advertising subscription capabilities when listen is not served" do
+          server = Server.new(
+            name: "listen_test",
+            capabilities: { tools: { listChanged: true }, resources: { listChanged: true, subscribe: true } },
+          )
+          transport = StreamableHTTPTransport.new(server, serve_subscriptions_listen: false)
+
+          response = transport.handle_request(modern_rack_request(modern_body("server/discover", {})))
+          capabilities = JSON.parse(response[2].first).dig("result", "capabilities")
+
+          assert_nil capabilities.dig("tools", "listChanged")
+          assert_nil capabilities.dig("resources", "listChanged")
+          assert_nil capabilities.dig("resources", "subscribe")
+        ensure
+          transport.close
+        end
+
         test "subscriptions/listen past the concurrent stream cap is rejected with 503" do
           transport = StreamableHTTPTransport.new(@server, max_listen_subscriptions: 1)
           open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true }, transport: transport)