Consult the resources_subscribe_handler for listen resource URIs

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
 

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

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<String>] 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<String>]
+    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:,

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

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(

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.

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