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])