Serve the subscriptions/listen notification stream per SEP-2575
## Motivation and Context Companion to the stateless lifecycle work (SEP-2575, modelcontextprotocol/modelcontextprotocol#2575) for the 2026-07-28 MCP spec release. The modern lifecycle removed the HTTP GET listening stream; `subscriptions/listen` replaces it as a long-lived POST that opts in to server change notifications. The Python SDK ships the server side of this as its SEP-2575 event-bus work (python-sdk PR /modelcontextprotocol/python-sdk#3035); the TypeScript SDK landed it in PR modelcontextprotocol/typescript-sdk#2321. Wire behavior, per the draft schema: - The request carries a REQUIRED `notifications` filter (`SubscriptionFilter`): `toolsListChanged`, `promptsListChanged`, `resourcesListChanged`, and `resourceSubscriptions` (URI list, replacing the legacy `resources/subscribe` RPC). Every type is opt-in; the server MUST NOT send types the client did not request. - The first stream message is `notifications/subscriptions/acknowledged`, reporting the subset of requested types the server agreed to honor. Honoring reads the capability FLAGS that promise delivery (`listChanged`, `subscribe`), the same derivation `server/discover` uses for its era-aware capability stripping; the mere presence of a primitive's capability is not enough. - Every notification delivered on the stream carries the correlating `io.modelcontextprotocol/subscriptionId` (the listen request id) in `_meta`. - A graceful teardown (transport `close`) sends a `SubscriptionsListenResult` response before closing the stream, stamped with the REQUIRED 2026-07-28 `resultType` at its construction site (it never passes through the dispatch path); an abrupt disconnect sends nothing. Implementation: - `StreamableHTTPTransport` intercepts `subscriptions/listen` on the modern path (after header and envelope validation) and serves it as a long-lived SSE stream, using the same register-and-return body proc pattern as the legacy GET stream. Subscriptions live in an in-process registry keyed by the listen request id; fan-out hooks into `send_notification` ahead of the legacy delivery, so a resource updated by one session's tool call also reaches modern subscribers. The matching snapshot is taken under the transport mutex, but stream writes happen outside it, matching the legacy delivery paths: a slow or stalled subscriber must not block the transport. Duplicate subscription ids close the new stream instead of double-registering. - Concurrent listen streams are capped (`max_listen_subscriptions:`, default 1000, `nil` to opt out); a listen request past the cap is rejected with HTTP 503, like the `max_sessions` guard against session floods. Each stream holds an open connection for its lifetime, so without a bound an unauthenticated client can retain unbounded connections. - Each stream is kept alive by an SSE comment frame written every `listen_keepalive_interval:` seconds (default 15, matching the TypeScript SDK; `nil` to opt out when an upstream proxy pings the stream). A silently dropped connection would otherwise hold its capped slot until the next fan-out write failed, so on a quiet server the cap would ratchet down permanently; the periodic write detects the dead peer and frees the slot. A comment frame cannot corrupt an interleaved notification's JSON, and the write happens outside the transport mutex, reusing the legacy GET stream's keepalive mechanism. - Every SSE response now carries `x-accel-buffering: no`, which the spec asks of SSE streams and both reference SDKs send: a buffering reverse proxy would hold events back instead of delivering them as they are written, and on a listen stream it would also swallow the keepalive frames a dropped peer is detected by. The header rides the shared `SSE_HEADERS`, so the legacy streams gain it too. - stdio does not serve the stream, matching the Python SDK's stream-pair behavior: no server handler is registered, so the method answers `-32601`. - `Server#discover` becomes era-aware about notification delivery: `listChanged`/`subscribe` capability flags promise delivery over `subscriptions/listen` streams in the modern lifecycle, so they are stripped when the transport does not serve that RPC (the new `Transport#serves_subscriptions_listen?` seam, true for `StreamableHTTPTransport`). - The conformance fixture defines the diagnostic triggers the `server-stateless` scenario calls (`test_trigger_tool_change` / `test_trigger_prompt_change`): each broadcasts its list-changed notification to the listen streams and returns, mirroring the suite's TypeScript reference fixture, which mutates nothing either. Out of scope, noted for follow-ups: an external event bus for multi-worker deployments, and the client-side listen driver (the Python reference is python-sdk PR modelcontextprotocol/python-sdk#3047). Refs #389. ## How Has This Been Tested? New tests in `test/mcp/server/transports/streamable_http_transport_test.rb` cover: the SSE response with the acknowledgement as the first event (including the honored-subset reduction for unsupported types and for capability entries lacking the delivery flag), the `notifications` filter and envelope requirements (400 responses), opt-in-only delivery with the correlating `subscriptionId`, URI-scoped `resources/updated` delivery, per-subscription ids across concurrent streams, the graceful close result carrying `resultType: "complete"`, duplicate-id rejection, the 503 past the concurrent stream cap, the keepalive (writing the comment frame outside the mutex, freeing a dead peer's slot, no thread when the interval is `nil`, and the positive-or-nil interval validation), and `x-accel-buffering: no` on both the listen stream and the legacy GET stream. `test/mcp/server/transports/stdio_transport_test.rb` asserts `-32601` over stdio, and `test/mcp/server_test.rb` covers the era-aware capability stripping in `server/discover` for both transport kinds. `bundle exec rake` (tests, RuboCop, and conformance baseline) passes. Against the conformance fixture server at `--spec-version 2026-07-28`, the `server-stateless` subscription checks all report SUCCESS: the acknowledgement, `subscriptionId` tagging, and filter-containment MUSTs, plus both list-changed SHOULD checks driven by the new trigger tools. The `--requirements 2025-11-25` server leg passes 78/78, unchanged. ## Breaking Changes None. The method was previously unhandled (`-32601` everywhere); the only observable change to existing responses is that `server/discover` no longer advertises `listChanged`/`subscribe` flags on transports that cannot deliver those notifications in the modern lifecycle, which `server/discover` has not shipped in a gem release with anyway.
This commit is contained in:
Koichi ITO
parent
b98b783a5f
commit
c40aa9cc59
3 changed files with 719 additions and 4 deletions
|
|
@ -634,6 +634,27 @@ module MCP
|
|||
end
|
||||
end
|
||||
|
||||
test "subscriptions/listen is not served over stdio and answers -32601" do
|
||||
# Like the Python SDK, the stream-pair transport does not serve the SEP-2575 notification subscription stream.
|
||||
listen = {
|
||||
jsonrpc: "2.0",
|
||||
method: "subscriptions/listen",
|
||||
id: "listen-1",
|
||||
params: {
|
||||
notifications: { toolsListChanged: true },
|
||||
_meta: {
|
||||
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
|
||||
"io.modelcontextprotocol/clientInfo": { name: "modern_client", version: "2.0" },
|
||||
"io.modelcontextprotocol/clientCapabilities": {},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
responses = run_transport_session([listen])
|
||||
|
||||
assert_equal JsonRpcHandler::ErrorCode::METHOD_NOT_FOUND, responses[0].dig(:error, :code)
|
||||
end
|
||||
|
||||
test "#send_request raises on a modern-locked session" do
|
||||
run_transport_session([modern_tools_list_request(id: 1)])
|
||||
|
||||
|
|
|
|||
|
|
@ -41,7 +41,9 @@ module MCP
|
|||
prompts: [],
|
||||
resources: [],
|
||||
)
|
||||
@transport = StreamableHTTPTransport.new(@server)
|
||||
# Disable the listen keepalive by default so opening a listen stream in a test does not
|
||||
# spawn a background timer thread; the keepalive is exercised by its own dedicated tests.
|
||||
@transport = StreamableHTTPTransport.new(@server, listen_keepalive_interval: nil)
|
||||
end
|
||||
|
||||
teardown do
|
||||
|
|
@ -5772,6 +5774,248 @@ module MCP
|
|||
transport.close
|
||||
end
|
||||
|
||||
test "subscriptions/listen opens an SSE stream and acknowledges the honored subset first" do
|
||||
# `resourceSubscriptions` honoring requires the `resources.subscribe` capability flag,
|
||||
# which the server defaults do not declare.
|
||||
server = Server.new(
|
||||
name: "listen_test",
|
||||
capabilities: { tools: { listChanged: true }, resources: { listChanged: true, subscribe: true } },
|
||||
)
|
||||
transport = StreamableHTTPTransport.new(server)
|
||||
|
||||
io = open_listen_stream(
|
||||
id: "listen-1",
|
||||
notifications: { toolsListChanged: true, resourceSubscriptions: ["file:///a.txt"] },
|
||||
transport: transport,
|
||||
)
|
||||
|
||||
events = sse_events(io)
|
||||
assert_equal(1, events.size)
|
||||
ack = events[0]
|
||||
assert_equal("notifications/subscriptions/acknowledged", ack["method"])
|
||||
assert_equal(
|
||||
{ "toolsListChanged" => true, "resourceSubscriptions" => ["file:///a.txt"] },
|
||||
ack.dig("params", "notifications"),
|
||||
)
|
||||
assert_equal("listen-1", ack.dig("params", "_meta", "io.modelcontextprotocol/subscriptionId"))
|
||||
ensure
|
||||
transport.close
|
||||
end
|
||||
|
||||
test "SSE responses disable proxy buffering" do
|
||||
# The spec asks servers to send `X-Accel-Buffering: no` on SSE streams so a reverse proxy
|
||||
# delivers events as they are written; on a listen stream, buffering would also swallow
|
||||
# the keepalive frames that detect a dropped peer.
|
||||
listen = @transport.handle_request(modern_rack_request(
|
||||
modern_listen_body(id: "listen-1", params: { notifications: { toolsListChanged: true } }),
|
||||
))
|
||||
|
||||
session_id = initialize_test_session
|
||||
get_stream = @transport.handle_request(create_rack_request("GET", "/", { "HTTP_MCP_SESSION_ID" => session_id }))
|
||||
|
||||
assert_equal("no", listen[1]["x-accel-buffering"])
|
||||
assert_equal("no", get_stream[1]["x-accel-buffering"])
|
||||
end
|
||||
|
||||
test "subscriptions/listen acknowledgement omits notification types the server does not support" do
|
||||
server = Server.new(name: "listen_test", capabilities: { tools: { listChanged: true } })
|
||||
transport = StreamableHTTPTransport.new(server)
|
||||
|
||||
io = open_listen_stream(
|
||||
id: "listen-1",
|
||||
notifications: { toolsListChanged: true, promptsListChanged: true, resourceSubscriptions: ["file:///a.txt"] },
|
||||
transport: transport,
|
||||
)
|
||||
|
||||
ack = sse_events(io)[0]
|
||||
assert_equal({ "toolsListChanged" => true }, ack.dig("params", "notifications"))
|
||||
ensure
|
||||
transport.close
|
||||
end
|
||||
|
||||
test "subscriptions/listen requires a notifications filter object" do
|
||||
response = @transport.handle_request(modern_rack_request(
|
||||
modern_listen_body(id: "listen-1", params: {}),
|
||||
))
|
||||
|
||||
assert_equal 400, response[0]
|
||||
assert_equal(-32602, JSON.parse(response[2][0]).dig("error", "code"))
|
||||
end
|
||||
|
||||
test "subscriptions/listen requires the modern _meta envelope" do
|
||||
response = @transport.handle_request(modern_rack_request(
|
||||
{ jsonrpc: "2.0", method: "subscriptions/listen", id: "listen-1", params: { notifications: {} } }.to_json,
|
||||
))
|
||||
|
||||
assert_equal 400, response[0]
|
||||
assert_equal(-32600, JSON.parse(response[2][0]).dig("error", "code"))
|
||||
end
|
||||
|
||||
test "subscriptions/listen delivers only opted-in notifications with the subscriptionId" do
|
||||
io = open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true })
|
||||
|
||||
@server.notify_tools_list_changed
|
||||
# Not opted in: MUST NOT be delivered.
|
||||
@server.notify_prompts_list_changed
|
||||
|
||||
events = sse_events(io)
|
||||
assert_equal 2, events.size
|
||||
assert_equal "notifications/subscriptions/acknowledged", events[0]["method"]
|
||||
assert_equal "notifications/tools/list_changed", events[1]["method"]
|
||||
assert_equal "listen-1", events[1].dig("params", "_meta", "io.modelcontextprotocol/subscriptionId")
|
||||
end
|
||||
|
||||
test "subscriptions/listen delivers resource updates only for subscribed URIs" do
|
||||
server = Server.new(
|
||||
name: "listen_test",
|
||||
capabilities: { resources: { listChanged: true, subscribe: true } },
|
||||
)
|
||||
transport = StreamableHTTPTransport.new(server)
|
||||
|
||||
io = open_listen_stream(
|
||||
id: "listen-1",
|
||||
notifications: { resourceSubscriptions: ["file:///subscribed.txt"] },
|
||||
transport: transport,
|
||||
)
|
||||
|
||||
# `**{}` keeps the params Hash positional on Ruby 2.7, matching the other `send_notification` tests.
|
||||
transport.send_notification("notifications/resources/updated", { uri: "file:///subscribed.txt" }, **{})
|
||||
transport.send_notification("notifications/resources/updated", { uri: "file:///other.txt" }, **{})
|
||||
|
||||
events = sse_events(io)
|
||||
assert_equal(2, events.size)
|
||||
assert_equal("notifications/resources/updated", events[1]["method"])
|
||||
assert_equal("file:///subscribed.txt", events[1].dig("params", "uri"))
|
||||
assert_equal("listen-1", events[1].dig("params", "_meta", "io.modelcontextprotocol/subscriptionId"))
|
||||
ensure
|
||||
transport.close
|
||||
end
|
||||
|
||||
test "subscriptions/listen streams for different subscriptions receive their own subscriptionId" do
|
||||
first = open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true })
|
||||
second = open_listen_stream(id: "listen-2", notifications: { toolsListChanged: true })
|
||||
|
||||
@server.notify_tools_list_changed
|
||||
|
||||
first_events = sse_events(first)
|
||||
second_events = sse_events(second)
|
||||
assert_equal "listen-1", first_events[1].dig("params", "_meta", "io.modelcontextprotocol/subscriptionId")
|
||||
assert_equal "listen-2", second_events[1].dig("params", "_meta", "io.modelcontextprotocol/subscriptionId")
|
||||
end
|
||||
|
||||
test "subscriptions/listen closes gracefully with a SubscriptionsListenResult on transport close" do
|
||||
io = open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true })
|
||||
|
||||
@transport.close
|
||||
|
||||
events = sse_events(io)
|
||||
result = events.last
|
||||
assert_equal "listen-1", result["id"]
|
||||
assert_equal "listen-1", result.dig("result", "_meta", "io.modelcontextprotocol/subscriptionId")
|
||||
# `SubscriptionsListenResult` is a 2026-07-28 result, so it carries the REQUIRED `resultType`.
|
||||
assert_equal "complete", result.dig("result", "resultType")
|
||||
assert_predicate io, :closed?
|
||||
|
||||
# The subscription is gone: further notifications are not delivered anywhere.
|
||||
@server.notify_tools_list_changed
|
||||
assert_equal events, sse_events(io)
|
||||
end
|
||||
|
||||
test "subscriptions/listen honoring reads the capability flags, not capability presence" do
|
||||
# A server declaring `tools` without `listChanged: true` promises no
|
||||
# list-changed delivery, so the acknowledgement omits the type.
|
||||
server = Server.new(name: "listen_test", capabilities: { tools: {}, resources: { listChanged: true } })
|
||||
transport = StreamableHTTPTransport.new(server)
|
||||
|
||||
io = open_listen_stream(
|
||||
id: "listen-1",
|
||||
notifications: { toolsListChanged: true, resourcesListChanged: true, resourceSubscriptions: ["file:///a.txt"] },
|
||||
transport: transport,
|
||||
)
|
||||
|
||||
ack = sse_events(io)[0]
|
||||
assert_equal({ "resourcesListChanged" => true }, ack.dig("params", "notifications"))
|
||||
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)
|
||||
|
||||
response = transport.handle_request(modern_rack_request(
|
||||
modern_listen_body(id: "listen-2", params: { notifications: { toolsListChanged: true } }),
|
||||
))
|
||||
|
||||
assert_equal(503, response[0])
|
||||
body = JSON.parse(response[2][0])
|
||||
assert_equal("listen-2", body["id"])
|
||||
assert_includes(body.dig("error", "message"), "maximum concurrent subscriptions/listen streams")
|
||||
ensure
|
||||
transport.close
|
||||
end
|
||||
|
||||
test "subscriptions/listen rejects a duplicate subscription id by closing the new stream" do
|
||||
open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true })
|
||||
|
||||
duplicate = StringIO.new
|
||||
response = @transport.handle_request(modern_rack_request(
|
||||
modern_listen_body(id: "listen-1", params: { notifications: { toolsListChanged: true } }),
|
||||
))
|
||||
response[2].call(duplicate)
|
||||
|
||||
assert_predicate duplicate, :closed?
|
||||
end
|
||||
|
||||
test "listen keepalive writes a comment frame outside the mutex" do
|
||||
# Register a subscription backed by a mutex-probing stream, then invoke the ping directly.
|
||||
mutex = @transport.instance_variable_get(:@mutex)
|
||||
writes = { total: 0, under_mutex: 0 }
|
||||
ping = nil
|
||||
stream = Object.new
|
||||
stream.define_singleton_method(:write) do |data|
|
||||
writes[:total] += 1
|
||||
mutex.try_lock ? mutex.unlock : writes[:under_mutex] += 1
|
||||
ping = data
|
||||
end
|
||||
stream.define_singleton_method(:flush) {}
|
||||
@transport.instance_variable_get(:@listen_subscriptions)["listen-1"] = { stream: stream, filter: {} }
|
||||
|
||||
@transport.send(:send_listen_keepalive_ping, "listen-1")
|
||||
|
||||
assert_equal 1, writes[:total]
|
||||
assert_equal 0, writes[:under_mutex], "keepalive write must not hold @mutex"
|
||||
assert_match(/\A: ping /, ping)
|
||||
end
|
||||
|
||||
test "listen keepalive frees the slot and closes the stream when the peer is gone" do
|
||||
transport = StreamableHTTPTransport.new(@server, listen_keepalive_interval: 0.01)
|
||||
io = open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true }, transport: transport)
|
||||
io.define_singleton_method(:write) { |_data| raise Errno::ECONNRESET }
|
||||
|
||||
deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 2
|
||||
until transport.instance_variable_get(:@listen_subscriptions).empty?
|
||||
flunk("keepalive did not free the slot for a dead peer") if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline
|
||||
sleep(0.01)
|
||||
end
|
||||
|
||||
assert_predicate io, :closed?
|
||||
ensure
|
||||
transport.close
|
||||
end
|
||||
|
||||
test "listen keepalive is not started when the interval is nil" do
|
||||
before = Thread.list.size
|
||||
open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true })
|
||||
|
||||
assert_equal before, Thread.list.size, "a nil interval must not spawn a keepalive thread"
|
||||
end
|
||||
|
||||
test "listen_keepalive_interval rejects a non-positive value" do
|
||||
assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, listen_keepalive_interval: 0) }
|
||||
assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, listen_keepalive_interval: -1) }
|
||||
end
|
||||
|
||||
private
|
||||
|
||||
def initialize_test_session(id: "init")
|
||||
|
|
@ -5885,6 +6129,41 @@ module MCP
|
|||
),
|
||||
}.to_json
|
||||
end
|
||||
|
||||
def modern_listen_body(id:, params:)
|
||||
{
|
||||
jsonrpc: "2.0",
|
||||
method: "subscriptions/listen",
|
||||
id: id,
|
||||
params: params.merge(
|
||||
_meta: {
|
||||
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
|
||||
"io.modelcontextprotocol/clientInfo": { name: "modern_client", version: "2.0" },
|
||||
"io.modelcontextprotocol/clientCapabilities": {},
|
||||
},
|
||||
),
|
||||
}.to_json
|
||||
end
|
||||
|
||||
# Opens a `subscriptions/listen` stream on the modern path and returns the StringIO
|
||||
# backing the SSE stream (already carrying the acknowledgement event).
|
||||
def open_listen_stream(id:, notifications:, transport: @transport)
|
||||
response = transport.handle_request(modern_rack_request(
|
||||
modern_listen_body(id: id, params: { notifications: notifications }),
|
||||
))
|
||||
|
||||
assert_equal(200, response[0])
|
||||
assert_equal("text/event-stream", response[1]["content-type"])
|
||||
|
||||
io = StringIO.new
|
||||
response[2].call(io)
|
||||
io
|
||||
end
|
||||
|
||||
# Parses every `data:` event written to an SSE StringIO.
|
||||
def sse_events(io)
|
||||
io.string.scan(/^data: (.+)$/).map { |match| JSON.parse(match[0]) }
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -130,7 +130,9 @@ module MCP
|
|||
result = response[:result]
|
||||
|
||||
assert_equal Configuration::SUPPORTED_MODERN_PROTOCOL_VERSIONS, result[:supportedVersions]
|
||||
assert_equal @server.capabilities, result[:capabilities]
|
||||
# Without a `subscriptions/listen`-serving transport, the `listChanged` flags are
|
||||
# stripped from the advertised capabilities (see the dedicated tests below).
|
||||
assert_equal @server.capabilities.keys, result[:capabilities].keys
|
||||
# Per the finalized spec (PR #3002), the server identity is the optional `_meta` stamp,
|
||||
# not a top-level `serverInfo` field.
|
||||
assert_equal @server_name, result.dig(:_meta, RequestEnvelope::SERVER_INFO_META_KEY, :name)
|
||||
|
|
@ -139,6 +141,29 @@ module MCP
|
|||
assert_equal "Optional instructions for the client", result[:instructions]
|
||||
end
|
||||
|
||||
test "#handle server/discover strips listChanged and subscribe flags without a listen-serving transport" do
|
||||
server = Server.new(name: "discover_test", capabilities: {
|
||||
tools: { listChanged: true },
|
||||
resources: { listChanged: true, subscribe: true },
|
||||
logging: {},
|
||||
})
|
||||
|
||||
result = server.handle({ jsonrpc: "2.0", method: "server/discover", id: 1 })[:result]
|
||||
|
||||
assert_equal({ tools: {}, resources: {}, logging: {} }, result[:capabilities])
|
||||
end
|
||||
|
||||
test "#handle server/discover keeps listChanged flags when the transport serves subscriptions/listen" do
|
||||
server = Server.new(name: "discover_test", capabilities: { tools: { listChanged: true } })
|
||||
transport = mock
|
||||
transport.stubs(:serves_subscriptions_listen?).returns(true)
|
||||
server.transport = transport
|
||||
|
||||
result = server.handle({ jsonrpc: "2.0", method: "server/discover", id: 1 })[:result]
|
||||
|
||||
assert_equal({ tools: { listChanged: true } }, result[:capabilities])
|
||||
end
|
||||
|
||||
test "#handle server/discover responds before initialize and regardless of capabilities" do
|
||||
# Per SEP-2575, discovery is sessionless: no prior `initialize`, no capability gate.
|
||||
server = Server.new(name: "discover_test", capabilities: {})
|
||||
|
|
|
|||
Loading…
Reference in a new issue