Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions docs/_server/subscriptions.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
18 changes: 16 additions & 2 deletions docs/_server/transports.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
53 changes: 46 additions & 7 deletions lib/mcp/server/transports/streamable_http_transport.rb
Original file line number Diff line number Diff line change
Expand Up @@ -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:`.
Expand All @@ -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)
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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] = [] }
Expand Down Expand Up @@ -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) ||
Expand Down
50 changes: 50 additions & 0 deletions test/mcp/server/transports/streamable_http_transport_test.rb
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down