From 52edb18aadfcb3d76ff257a71d904941dff27d43 Mon Sep 17 00:00:00 2001 From: Koichi ITO Date: Sun, 25 Oct 2026 01:05:55 +0700 Subject: [PATCH] Consult the `resources_subscribe_handler` for listen resource URIs ## Motivation and Context In the 2026-07-28 lifecycle `subscriptions/listen` takes the place of `resources/subscribe`, but the URIs it names in `resourceSubscriptions` were honored on the `subscribe` capability alone. The `resources_subscribe_handler`, the hook an application uses to decide which resources a client may follow, was never called for them, so a decision it made for `resources/subscribe` did not carry over to a modern client. Each distinct URI is now passed to that handler with the params a `resources/subscribe` request for it would carry, together with the listen request's envelope and `auth_info` when the handler declares `server_context:`. A URI the handler raises for is reported to the exception reporter and left out of the acknowledgement, and the stream receives no `notifications/resources/updated` for it. The spec lets the acknowledgement honor a subset of the requested filter, so this needs no new error on the wire, and the field is omitted when no URI remains. The decision is made once when the stream opens, before it is registered; the stream already closes when its token expires. The params also carry the listen request's `_meta`, as those of a `resources/subscribe` request would, while the `server_context` has no session, so session-scoped helpers do nothing there. Any value the handler returns accepts the URI. The TypeScript and Python SDKs copy the requested URIs into the acknowledgement as well; the TypeScript SDK leaves authorization to the middleware layer. In the Ruby SDK the handler was already the documented place for this decision, so it is reused instead of adding a second hook that existing applications would have to discover. `resources_unsubscribe_handler` is not called when a listen stream ends: the subscription lasts as long as the stream. `Server#subscribable_resource_uris` is public only so that the transport can call it; it is marked `@api private` and is not part of the published API. ## How Has This Been Tested? - test/mcp/server/transports/streamable_http_transport_test.rb - test/mcp/server/transports/streamable_http_transport_oauth_test.rb - test/mcp/server_test.rb The two transport tests and the OAuth test fail against the previous `lib`, which acknowledged every requested URI. The `_meta` transport test fails against it too. ## Breaking Changes A `resources_subscribe_handler` that raises for a URI now narrows the listen acknowledgement for that URI, and a handler that records subscriptions is now also called for listen streams, without a matching `resources_unsubscribe_handler` call. Servers without a custom handler are unaffected. --- docs/_server/resources.md | 3 + docs/_server/subscriptions.md | 29 +++++++ lib/mcp/server.rb | 30 +++++++ .../transports/streamable_http_transport.rb | 12 +-- .../streamable_http_transport_oauth_test.rb | 22 +++++ .../streamable_http_transport_test.rb | 78 +++++++++++++++++ test/mcp/server_test.rb | 85 +++++++++++++++++++ 7 files changed, 254 insertions(+), 5 deletions(-) diff --git a/docs/_server/resources.md b/docs/_server/resources.md index bf6eb140..d8fe8068 100644 --- a/docs/_server/resources.md +++ b/docs/_server/resources.md @@ -283,6 +283,9 @@ Three methods are provided: - `Server#resources_unsubscribe_handler` - registers a handler for `resources/unsubscribe` requests - `ServerContext#notify_resources_updated` - sends a `notifications/resources/updated` notification to the subscribing client +The subscribe handler is also consulted for the resource URIs of a 2026-07-28 `subscriptions/listen` stream; +see [Subscriptions](/server/subscriptions/). + ```ruby subscribed_uris = Set.new diff --git a/docs/_server/subscriptions.md b/docs/_server/subscriptions.md index 397342a7..80185c52 100644 --- a/docs/_server/subscriptions.md +++ b/docs/_server/subscriptions.md @@ -18,6 +18,35 @@ the server acknowledges the honored subset with `notifications/subscriptions/ack and every delivered notification carries the correlating `io.modelcontextprotocol/subscriptionId` in `_meta`. The honored subset follows the capabilities the server declares (`listChanged` and `subscribe` flags). +`subscriptions/listen` takes the place of `resources/subscribe` in this lifecycle, so the URIs in `resourceSubscriptions` +are passed to `Server#resources_subscribe_handler` one at a time, each as `{ uri: }`, +with the listen request's `_meta` added when it carries one, and, when the handler declares it, a `server_context` holding the request's `auth_info`. +A URI the handler raises for is left out of the acknowledgement, and the stream receives no `notifications/resources/updated` for it. +Only whether the handler raised counts: its return value, including `false`, `nil`, or a `{ _meta: ... }` Hash, is discarded and accepts the URI. +Unlike a `resources/subscribe` request, the `server_context` has no session, related request id, or cancellation, +so session-scoped helpers such as `notify_resources_updated` do nothing there. + +The handler is consulted once, when the stream opens, and its decision is not re-evaluated while the stream stays open, +so a later change in its answer takes effect only on a new stream. +An authenticated stream ends under the stream expiry rules in [Authorization](/server/authorization/), +and any stream ends when either side closes the connection or the transport closes. + +Each URI the handler raises for is reported to the exception reporter, one report per denied URI as for a denied `resources/subscribe` request, +so a listen request naming many denied URIs produces as many reports. + +Whatever the handler records, such as a watcher or a counter, gets no paired cleanup. +`resources_unsubscribe_handler` is not called when the stream ends, since the subscription lasts as long as the stream, +and the handler may have run for a request that then opens no stream, +such as one refused when two listen requests race for the last slot under `max_listen_subscriptions:`. + +```ruby +server.resources_subscribe_handler do |params, server_context:| + unless readable?(server_context.auth_info, params[:uri]) + raise MCP::Server::RequestHandlerError.new("Forbidden", params, error_type: :invalid_params) + end +end +``` + The client opens the stream with a `subscriptions/listen` request: ```json diff --git a/lib/mcp/server.rb b/lib/mcp/server.rb index 53f99d8a..3ef08c09 100644 --- a/lib/mcp/server.rb +++ b/lib/mcp/server.rb @@ -501,6 +501,36 @@ def resources_unsubscribe_handler(&block) @handlers[Methods::RESOURCES_UNSUBSCRIBE] = block end + # Returns the distinct members of `uris` that the `resources_subscribe_handler` accepts, calling it once per URI + # with `{ uri: }`, plus `_meta` when given. `subscriptions/listen` takes the place of `resources/subscribe` + # in the 2026-07-28 lifecycle, so the handler keeps deciding which resources a client may follow there too. + # A URI the handler raises for is reported and left out, as the spec lets the acknowledgement honor a subset of + # what was requested; any return value accepts the URI and is otherwise discarded. The `server_context` has no session, + # related request id, or cancellation. Public only so that `StreamableHTTPTransport` can call it. + # + # @api private + # @param uris [Array] The `resourceSubscriptions` of a `subscriptions/listen` request. + # @param meta [Hash, nil] The `_meta` of that request. + # @param envelope [MCP::RequestEnvelope, nil] The envelope of that request. + # @param auth_info [MCP::Server::OAuth::AccessToken, nil] The token that authenticated that request. + # @return [Array] + def subscribable_resource_uris(uris, meta: nil, envelope: nil, auth_info: nil) + handler = @handlers[Methods::RESOURCES_SUBSCRIBE] + + uris.uniq.select do |uri| + params = { uri: uri } + params[:_meta] = meta if meta + dispatch_optional_context_handler(handler, params, envelope: envelope, auth_info: auth_info) + true + rescue RequestHandlerError => e + report_exception(e.original_error || e, { request: params }) + false + rescue => e + report_exception(e, { request: params }) + false + end + end + def build_sampling_params( capabilities, messages:, diff --git a/lib/mcp/server/transports/streamable_http_transport.rb b/lib/mcp/server/transports/streamable_http_transport.rb index a2dafd30..9ea681d6 100644 --- a/lib/mcp/server/transports/streamable_http_transport.rb +++ b/lib/mcp/server/transports/streamable_http_transport.rb @@ -1051,7 +1051,7 @@ def handle_subscriptions_listen(body, auth_info: nil) begin if RequestEnvelope.modern?(params) - RequestEnvelope.parse!(params, request: params) + envelope = RequestEnvelope.parse!(params, request: params) else return invalid_request_response("Invalid Request: modern requests require the SEP-2575 `_meta` envelope") end @@ -1091,7 +1091,7 @@ def handle_subscriptions_listen(body, auth_info: nil) return too_many_listen_subscriptions_response(request_id) end - honored = honored_filter(filter) + honored = honored_filter(filter, meta: params[:_meta], envelope: envelope, auth_info: auth_info) # The acknowledgement echoes the honored filter, whose resource URIs are client-supplied strings, # so it is encoded before the stream is committed to: a URI that cannot be written as JSON is @@ -1338,8 +1338,9 @@ def listen_keepalive_wait(expires_at) # Per SEP-2575, the server MUST NOT send notification types the client has not requested, # and the acknowledgement only includes types the server actually supports - # (derived from its declared capabilities). - def honored_filter(filter) + # (derived from its declared capabilities). Resource URIs are further narrowed to those + # the server's `resources_subscribe_handler` accepts; the field is omitted when none remain. + def honored_filter(filter, meta: nil, envelope: nil, auth_info: nil) capabilities = @server.capabilities honored = {} honored[:toolsListChanged] = true if filter[:toolsListChanged] && capability_flag?(capabilities, :tools, :listChanged) @@ -1348,7 +1349,8 @@ def honored_filter(filter) subscriptions = filter[:resourceSubscriptions] if capability_flag?(capabilities, :resources, :subscribe) && subscriptions.is_a?(Array) && !subscriptions.empty? - honored[:resourceSubscriptions] = subscriptions.uniq + uris = @server.subscribable_resource_uris(subscriptions, meta: meta, envelope: envelope, auth_info: auth_info) + honored[:resourceSubscriptions] = uris unless uris.empty? end honored diff --git a/test/mcp/server/transports/streamable_http_transport_oauth_test.rb b/test/mcp/server/transports/streamable_http_transport_oauth_test.rb index bac3c97b..6c567d62 100644 --- a/test/mcp/server/transports/streamable_http_transport_oauth_test.rb +++ b/test/mcp/server/transports/streamable_http_transport_oauth_test.rb @@ -735,6 +735,28 @@ def initialize(message = "rejected by policy") refute wait_until(timeout: 0.1) { stream.closed? }, "the stream was closed although its token is valid" end + test "the resources_subscribe_handler sees the token of a subscriptions/listen request" do + @server.capabilities = { resources: { subscribe: true } } + @server.resources_subscribe_handler do |params, server_context:| + owner = params[:uri].delete_prefix("file:///").delete_suffix(".txt") + raise RequestHandlerError.new("Forbidden", params, error_type: :invalid_params) unless server_context.auth_info.subject == owner + end + the_transport = transport + + response = the_transport.handle_request(modern_rack_request( + modern_body("subscriptions/listen", { notifications: { resourceSubscriptions: ["file:///alice.txt", "file:///bob.txt"] } }), + headers: { "HTTP_AUTHORIZATION" => "Bearer alice-token" }, + )) + + assert_equal 200, response[0] + + stream = TestStream.new + response[2].call(stream) + acknowledgement = JSON.parse(stream.buffer[/^data: (.+)$/, 1]) + + assert_equal ["file:///alice.txt"], acknowledgement.dig("params", "notifications", "resourceSubscriptions") + end + test "a subscriptions/listen stream is closed at the keepalive once its token expires" do expired = Time.now.to_i - 1 @verifier = StubVerifier.new( diff --git a/test/mcp/server/transports/streamable_http_transport_test.rb b/test/mcp/server/transports/streamable_http_transport_test.rb index b4d335cd..b721341a 100644 --- a/test/mcp/server/transports/streamable_http_transport_test.rb +++ b/test/mcp/server/transports/streamable_http_transport_test.rb @@ -7121,6 +7121,84 @@ def string transport.close end + test "subscriptions/listen honors only the resource URIs the resources_subscribe_handler accepts" do + server = Server.new(name: "listen_test", capabilities: { resources: { subscribe: true } }) + requested = [] + server.resources_subscribe_handler do |params| + requested << params[:uri] + raise RequestHandlerError.new("Forbidden", params, error_type: :invalid_params) if params[:uri] == "file:///private.txt" + end + transport = StreamableHTTPTransport.new(server) + + io = open_listen_stream( + id: "listen-1", + notifications: { resourceSubscriptions: ["file:///public.txt", "file:///private.txt", "file:///public.txt"] }, + transport: transport, + ) + + assert_equal ["file:///public.txt", "file:///private.txt"], requested + assert_equal ["file:///public.txt"], sse_events(io).first.dig("params", "notifications", "resourceSubscriptions") + + transport.send_notification("notifications/resources/updated", { uri: "file:///private.txt" }, **{}) + transport.send_notification("notifications/resources/updated", { uri: "file:///public.txt" }, **{}) + + assert_equal ["file:///public.txt"], sse_events(io).drop(1).map { |event| event.dig("params", "uri") } + ensure + transport.close + end + + test "subscriptions/listen passes its _meta to the resources_subscribe_handler" do + server = Server.new(name: "listen_test", capabilities: { resources: { subscribe: true } }) + seen = [] + server.resources_subscribe_handler do |params, server_context:| + seen << [params.dig(:_meta, :"io.modelcontextprotocol/clientInfo", :name), server_context[:_meta].equal?(params[:_meta])] + end + transport = StreamableHTTPTransport.new(server) + + io = open_listen_stream(id: "listen-1", notifications: { resourceSubscriptions: ["file:///a.txt"] }, transport: transport) + + assert_equal [["modern_client", true]], seen + assert_equal ["file:///a.txt"], sse_events(io).first.dig("params", "notifications", "resourceSubscriptions") + ensure + transport.close + end + + test "subscriptions/listen does not call the resources_subscribe_handler without the subscribe capability" do + server = Server.new(name: "listen_test", capabilities: { resources: { listChanged: true } }) + server.resources_subscribe_handler { |_params| flunk "the handler must not be called" } + transport = StreamableHTTPTransport.new(server) + + io = open_listen_stream( + id: "listen-1", + notifications: { resourcesListChanged: true, resourceSubscriptions: ["file:///a.txt"] }, + transport: transport, + ) + + assert_equal({ "resourcesListChanged" => true }, sse_events(io).first.dig("params", "notifications")) + ensure + transport.close + end + + test "subscriptions/listen omits resourceSubscriptions when the resources_subscribe_handler accepts none" do + server = Server.new( + name: "listen_test", + capabilities: { resources: { listChanged: true, subscribe: true } }, + ) + server.resources_subscribe_handler { |_params| raise "denied" } + transport = StreamableHTTPTransport.new(server) + + io = open_listen_stream( + id: "listen-1", + notifications: { resourcesListChanged: true, resourceSubscriptions: ["file:///private.txt"] }, + transport: transport, + ) + + assert_equal({ "resourcesListChanged" => true }, sse_events(io).first.dig("params", "notifications")) + assert_empty transport.instance_variable_get(:@listen_subscriptions).values.first[:resource_uris] + ensure + transport.close + end + test "notifications are delivered only after the listen acknowledgement" do # Inject an entry in the registered-but-not-yet-acknowledged state: the window between # the registry insert and the acknowledgement write, which happens outside the lock. diff --git a/test/mcp/server_test.rb b/test/mcp/server_test.rb index 3c89666f..4cfa6456 100644 --- a/test/mcp/server_test.rb +++ b/test/mcp/server_test.rb @@ -4062,6 +4062,91 @@ def handle_subscription(server, method) refute handler_called end + test "#subscribable_resource_uris reports a resources_subscribe_handler error and leaves its URI out" do + reported = [] + configuration = MCP::Configuration.new + configuration.exception_reporter = ->(e, server_context) { reported << [e.message, server_context] } + server = Server.new(name: "test_server", configuration: configuration) + server.resources_subscribe_handler do |params| + raise "denied" if params[:uri] == "file:///private.txt" + end + + uris = server.subscribable_resource_uris(["file:///public.txt", "file:///private.txt"]) + + assert_equal(["file:///public.txt"], uris) + assert_equal([["denied", { request: { uri: "file:///private.txt" } }]], reported) + end + + test "#subscribable_resource_uris calls the resources_subscribe_handler once per distinct URI and accepts any return value" do + server = Server.new(name: "test_server") + requested = [] + server.resources_subscribe_handler do |params| + requested << params + false + end + + uris = server.subscribable_resource_uris(["file:///a.txt", "file:///a.txt"], meta: { trace: "t1" }) + + assert_equal(["file:///a.txt"], uris) + assert_equal([{ uri: "file:///a.txt", _meta: { trace: "t1" } }], requested) + end + + test "#subscribable_resource_uris reports the original error of a RequestHandlerError" do + reported = [] + configuration = MCP::Configuration.new + configuration.exception_reporter = ->(e, _server_context) { reported << e } + server = Server.new(name: "test_server", configuration: configuration) + original = ArgumentError.new("lookup failed") + server.resources_subscribe_handler do |params| + raise Server::RequestHandlerError.new("Forbidden", params, error_type: :invalid_params, original_error: original) + end + + assert_empty(server.subscribable_resource_uris(["file:///a.txt"])) + assert_equal([original], reported) + end + + test "#subscribable_resource_uris reports a RequestHandlerError without an original error itself" do + reported = [] + configuration = MCP::Configuration.new + configuration.exception_reporter = ->(e, _server_context) { reported << e } + server = Server.new(name: "test_server", configuration: configuration) + error = Server::RequestHandlerError.new("Forbidden", {}, error_type: :invalid_params) + server.resources_subscribe_handler { |_params| raise error } + + assert_empty(server.subscribable_resource_uris(["file:///a.txt"])) + assert_equal([error], reported) + end + + test "#subscribable_resource_uris keeps an accepted URI after a denied one and accepts a nil return value" do + server = Server.new(name: "test_server") + server.resources_subscribe_handler do |params| + raise "denied" if params[:uri] == "file:///denied.txt" + + nil + end + + assert_equal(["file:///accepted.txt"], server.subscribable_resource_uris(["file:///denied.txt", "file:///accepted.txt"])) + end + + test "#subscribable_resource_uris passes the envelope to the resources_subscribe_handler" do + server = Server.new(name: "test_server") + seen = nil + server.resources_subscribe_handler do |_params, server_context:| + seen = server_context.envelope + end + envelope = RequestEnvelope.new(protocol_version: "2026-07-28", client_info: { name: "c" }, client_capabilities: {}, log_level: nil) + + server.subscribable_resource_uris(["file:///a.txt"], envelope: envelope) + + assert_same(envelope, seen) + end + + test "#subscribable_resource_uris keeps every URI with the default resources_subscribe_handler" do + server = Server.new(name: "test_server") + + assert_equal(["file:///a.txt", "file:///b.txt"], server.subscribable_resource_uris(["file:///a.txt", "file:///b.txt"])) + end + test "tools/call with no args" do server = Server.new(tools: [@tool_with_no_args])