Skip to content

Commit 26eabbc

Browse files
authored
Merge pull request #632 from koic/authorize_the_resource_subscriptions_of_a_listen_stream
Consult the `resources_subscribe_handler` for listen resource URIs
2 parents fc41dae + 52edb18 commit 26eabbc

7 files changed

Lines changed: 254 additions & 5 deletions

File tree

‎docs/_server/resources.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -283,6 +283,9 @@ Three methods are provided:
283283
- `Server#resources_unsubscribe_handler` - registers a handler for `resources/unsubscribe` requests
284284
- `ServerContext#notify_resources_updated` - sends a `notifications/resources/updated` notification to the subscribing client
285285

286+
The subscribe handler is also consulted for the resource URIs of a 2026-07-28 `subscriptions/listen` stream;
287+
see [Subscriptions](/server/subscriptions/).
288+
286289
```ruby
287290
subscribed_uris = Set.new
288291

‎docs/_server/subscriptions.md‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,35 @@ the server acknowledges the honored subset with `notifications/subscriptions/ack
1818
and every delivered notification carries the correlating `io.modelcontextprotocol/subscriptionId` in `_meta`.
1919
The honored subset follows the capabilities the server declares (`listChanged` and `subscribe` flags).
2020

21+
`subscriptions/listen` takes the place of `resources/subscribe` in this lifecycle, so the URIs in `resourceSubscriptions`
22+
are passed to `Server#resources_subscribe_handler` one at a time, each as `{ uri: }`,
23+
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`.
24+
A URI the handler raises for is left out of the acknowledgement, and the stream receives no `notifications/resources/updated` for it.
25+
Only whether the handler raised counts: its return value, including `false`, `nil`, or a `{ _meta: ... }` Hash, is discarded and accepts the URI.
26+
Unlike a `resources/subscribe` request, the `server_context` has no session, related request id, or cancellation,
27+
so session-scoped helpers such as `notify_resources_updated` do nothing there.
28+
29+
The handler is consulted once, when the stream opens, and its decision is not re-evaluated while the stream stays open,
30+
so a later change in its answer takes effect only on a new stream.
31+
An authenticated stream ends under the stream expiry rules in [Authorization](/server/authorization/),
32+
and any stream ends when either side closes the connection or the transport closes.
33+
34+
Each URI the handler raises for is reported to the exception reporter, one report per denied URI as for a denied `resources/subscribe` request,
35+
so a listen request naming many denied URIs produces as many reports.
36+
37+
Whatever the handler records, such as a watcher or a counter, gets no paired cleanup.
38+
`resources_unsubscribe_handler` is not called when the stream ends, since the subscription lasts as long as the stream,
39+
and the handler may have run for a request that then opens no stream,
40+
such as one refused when two listen requests race for the last slot under `max_listen_subscriptions:`.
41+
42+
```ruby
43+
server.resources_subscribe_handler do |params, server_context:|
44+
unless readable?(server_context.auth_info, params[:uri])
45+
raise MCP::Server::RequestHandlerError.new("Forbidden", params, error_type: :invalid_params)
46+
end
47+
end
48+
```
49+
2150
The client opens the stream with a `subscriptions/listen` request:
2251

2352
```json

‎lib/mcp/server.rb‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -501,6 +501,36 @@ def resources_unsubscribe_handler(&block)
501501
@handlers[Methods::RESOURCES_UNSUBSCRIBE] = block
502502
end
503503

504+
# Returns the distinct members of `uris` that the `resources_subscribe_handler` accepts, calling it once per URI
505+
# with `{ uri: }`, plus `_meta` when given. `subscriptions/listen` takes the place of `resources/subscribe`
506+
# in the 2026-07-28 lifecycle, so the handler keeps deciding which resources a client may follow there too.
507+
# A URI the handler raises for is reported and left out, as the spec lets the acknowledgement honor a subset of
508+
# what was requested; any return value accepts the URI and is otherwise discarded. The `server_context` has no session,
509+
# related request id, or cancellation. Public only so that `StreamableHTTPTransport` can call it.
510+
#
511+
# @api private
512+
# @param uris [Array<String>] The `resourceSubscriptions` of a `subscriptions/listen` request.
513+
# @param meta [Hash, nil] The `_meta` of that request.
514+
# @param envelope [MCP::RequestEnvelope, nil] The envelope of that request.
515+
# @param auth_info [MCP::Server::OAuth::AccessToken, nil] The token that authenticated that request.
516+
# @return [Array<String>]
517+
def subscribable_resource_uris(uris, meta: nil, envelope: nil, auth_info: nil)
518+
handler = @handlers[Methods::RESOURCES_SUBSCRIBE]
519+
520+
uris.uniq.select do |uri|
521+
params = { uri: uri }
522+
params[:_meta] = meta if meta
523+
dispatch_optional_context_handler(handler, params, envelope: envelope, auth_info: auth_info)
524+
true
525+
rescue RequestHandlerError => e
526+
report_exception(e.original_error || e, { request: params })
527+
false
528+
rescue => e
529+
report_exception(e, { request: params })
530+
false
531+
end
532+
end
533+
504534
def build_sampling_params(
505535
capabilities,
506536
messages:,

‎lib/mcp/server/transports/streamable_http_transport.rb‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1051,7 +1051,7 @@ def handle_subscriptions_listen(body, auth_info: nil)
10511051

10521052
begin
10531053
if RequestEnvelope.modern?(params)
1054-
RequestEnvelope.parse!(params, request: params)
1054+
envelope = RequestEnvelope.parse!(params, request: params)
10551055
else
10561056
return invalid_request_response("Invalid Request: modern requests require the SEP-2575 `_meta` envelope")
10571057
end
@@ -1091,7 +1091,7 @@ def handle_subscriptions_listen(body, auth_info: nil)
10911091
return too_many_listen_subscriptions_response(request_id)
10921092
end
10931093

1094-
honored = honored_filter(filter)
1094+
honored = honored_filter(filter, meta: params[:_meta], envelope: envelope, auth_info: auth_info)
10951095

10961096
# The acknowledgement echoes the honored filter, whose resource URIs are client-supplied strings,
10971097
# 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)
13381338

13391339
# Per SEP-2575, the server MUST NOT send notification types the client has not requested,
13401340
# and the acknowledgement only includes types the server actually supports
1341-
# (derived from its declared capabilities).
1342-
def honored_filter(filter)
1341+
# (derived from its declared capabilities). Resource URIs are further narrowed to those
1342+
# the server's `resources_subscribe_handler` accepts; the field is omitted when none remain.
1343+
def honored_filter(filter, meta: nil, envelope: nil, auth_info: nil)
13431344
capabilities = @server.capabilities
13441345
honored = {}
13451346
honored[:toolsListChanged] = true if filter[:toolsListChanged] && capability_flag?(capabilities, :tools, :listChanged)
@@ -1348,7 +1349,8 @@ def honored_filter(filter)
13481349

13491350
subscriptions = filter[:resourceSubscriptions]
13501351
if capability_flag?(capabilities, :resources, :subscribe) && subscriptions.is_a?(Array) && !subscriptions.empty?
1351-
honored[:resourceSubscriptions] = subscriptions.uniq
1352+
uris = @server.subscribable_resource_uris(subscriptions, meta: meta, envelope: envelope, auth_info: auth_info)
1353+
honored[:resourceSubscriptions] = uris unless uris.empty?
13521354
end
13531355

13541356
honored

‎test/mcp/server/transports/streamable_http_transport_oauth_test.rb‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -735,6 +735,28 @@ def initialize(message = "rejected by policy")
735735
refute wait_until(timeout: 0.1) { stream.closed? }, "the stream was closed although its token is valid"
736736
end
737737

738+
test "the resources_subscribe_handler sees the token of a subscriptions/listen request" do
739+
@server.capabilities = { resources: { subscribe: true } }
740+
@server.resources_subscribe_handler do |params, server_context:|
741+
owner = params[:uri].delete_prefix("file:///").delete_suffix(".txt")
742+
raise RequestHandlerError.new("Forbidden", params, error_type: :invalid_params) unless server_context.auth_info.subject == owner
743+
end
744+
the_transport = transport
745+
746+
response = the_transport.handle_request(modern_rack_request(
747+
modern_body("subscriptions/listen", { notifications: { resourceSubscriptions: ["file:///alice.txt", "file:///bob.txt"] } }),
748+
headers: { "HTTP_AUTHORIZATION" => "Bearer alice-token" },
749+
))
750+
751+
assert_equal 200, response[0]
752+
753+
stream = TestStream.new
754+
response[2].call(stream)
755+
acknowledgement = JSON.parse(stream.buffer[/^data: (.+)$/, 1])
756+
757+
assert_equal ["file:///alice.txt"], acknowledgement.dig("params", "notifications", "resourceSubscriptions")
758+
end
759+
738760
test "a subscriptions/listen stream is closed at the keepalive once its token expires" do
739761
expired = Time.now.to_i - 1
740762
@verifier = StubVerifier.new(

‎test/mcp/server/transports/streamable_http_transport_test.rb‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7121,6 +7121,84 @@ def string
71217121
transport.close
71227122
end
71237123

7124+
test "subscriptions/listen honors only the resource URIs the resources_subscribe_handler accepts" do
7125+
server = Server.new(name: "listen_test", capabilities: { resources: { subscribe: true } })
7126+
requested = []
7127+
server.resources_subscribe_handler do |params|
7128+
requested << params[:uri]
7129+
raise RequestHandlerError.new("Forbidden", params, error_type: :invalid_params) if params[:uri] == "file:///private.txt"
7130+
end
7131+
transport = StreamableHTTPTransport.new(server)
7132+
7133+
io = open_listen_stream(
7134+
id: "listen-1",
7135+
notifications: { resourceSubscriptions: ["file:///public.txt", "file:///private.txt", "file:///public.txt"] },
7136+
transport: transport,
7137+
)
7138+
7139+
assert_equal ["file:///public.txt", "file:///private.txt"], requested
7140+
assert_equal ["file:///public.txt"], sse_events(io).first.dig("params", "notifications", "resourceSubscriptions")
7141+
7142+
transport.send_notification("notifications/resources/updated", { uri: "file:///private.txt" }, **{})
7143+
transport.send_notification("notifications/resources/updated", { uri: "file:///public.txt" }, **{})
7144+
7145+
assert_equal ["file:///public.txt"], sse_events(io).drop(1).map { |event| event.dig("params", "uri") }
7146+
ensure
7147+
transport.close
7148+
end
7149+
7150+
test "subscriptions/listen passes its _meta to the resources_subscribe_handler" do
7151+
server = Server.new(name: "listen_test", capabilities: { resources: { subscribe: true } })
7152+
seen = []
7153+
server.resources_subscribe_handler do |params, server_context:|
7154+
seen << [params.dig(:_meta, :"io.modelcontextprotocol/clientInfo", :name), server_context[:_meta].equal?(params[:_meta])]
7155+
end
7156+
transport = StreamableHTTPTransport.new(server)
7157+
7158+
io = open_listen_stream(id: "listen-1", notifications: { resourceSubscriptions: ["file:///a.txt"] }, transport: transport)
7159+
7160+
assert_equal [["modern_client", true]], seen
7161+
assert_equal ["file:///a.txt"], sse_events(io).first.dig("params", "notifications", "resourceSubscriptions")
7162+
ensure
7163+
transport.close
7164+
end
7165+
7166+
test "subscriptions/listen does not call the resources_subscribe_handler without the subscribe capability" do
7167+
server = Server.new(name: "listen_test", capabilities: { resources: { listChanged: true } })
7168+
server.resources_subscribe_handler { |_params| flunk "the handler must not be called" }
7169+
transport = StreamableHTTPTransport.new(server)
7170+
7171+
io = open_listen_stream(
7172+
id: "listen-1",
7173+
notifications: { resourcesListChanged: true, resourceSubscriptions: ["file:///a.txt"] },
7174+
transport: transport,
7175+
)
7176+
7177+
assert_equal({ "resourcesListChanged" => true }, sse_events(io).first.dig("params", "notifications"))
7178+
ensure
7179+
transport.close
7180+
end
7181+
7182+
test "subscriptions/listen omits resourceSubscriptions when the resources_subscribe_handler accepts none" do
7183+
server = Server.new(
7184+
name: "listen_test",
7185+
capabilities: { resources: { listChanged: true, subscribe: true } },
7186+
)
7187+
server.resources_subscribe_handler { |_params| raise "denied" }
7188+
transport = StreamableHTTPTransport.new(server)
7189+
7190+
io = open_listen_stream(
7191+
id: "listen-1",
7192+
notifications: { resourcesListChanged: true, resourceSubscriptions: ["file:///private.txt"] },
7193+
transport: transport,
7194+
)
7195+
7196+
assert_equal({ "resourcesListChanged" => true }, sse_events(io).first.dig("params", "notifications"))
7197+
assert_empty transport.instance_variable_get(:@listen_subscriptions).values.first[:resource_uris]
7198+
ensure
7199+
transport.close
7200+
end
7201+
71247202
test "notifications are delivered only after the listen acknowledgement" do
71257203
# Inject an entry in the registered-but-not-yet-acknowledged state: the window between
71267204
# the registry insert and the acknowledgement write, which happens outside the lock.

‎test/mcp/server_test.rb‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4062,6 +4062,91 @@ def handle_subscription(server, method)
40624062
refute handler_called
40634063
end
40644064

4065+
test "#subscribable_resource_uris reports a resources_subscribe_handler error and leaves its URI out" do
4066+
reported = []
4067+
configuration = MCP::Configuration.new
4068+
configuration.exception_reporter = ->(e, server_context) { reported << [e.message, server_context] }
4069+
server = Server.new(name: "test_server", configuration: configuration)
4070+
server.resources_subscribe_handler do |params|
4071+
raise "denied" if params[:uri] == "file:///private.txt"
4072+
end
4073+
4074+
uris = server.subscribable_resource_uris(["file:///public.txt", "file:///private.txt"])
4075+
4076+
assert_equal(["file:///public.txt"], uris)
4077+
assert_equal([["denied", { request: { uri: "file:///private.txt" } }]], reported)
4078+
end
4079+
4080+
test "#subscribable_resource_uris calls the resources_subscribe_handler once per distinct URI and accepts any return value" do
4081+
server = Server.new(name: "test_server")
4082+
requested = []
4083+
server.resources_subscribe_handler do |params|
4084+
requested << params
4085+
false
4086+
end
4087+
4088+
uris = server.subscribable_resource_uris(["file:///a.txt", "file:///a.txt"], meta: { trace: "t1" })
4089+
4090+
assert_equal(["file:///a.txt"], uris)
4091+
assert_equal([{ uri: "file:///a.txt", _meta: { trace: "t1" } }], requested)
4092+
end
4093+
4094+
test "#subscribable_resource_uris reports the original error of a RequestHandlerError" do
4095+
reported = []
4096+
configuration = MCP::Configuration.new
4097+
configuration.exception_reporter = ->(e, _server_context) { reported << e }
4098+
server = Server.new(name: "test_server", configuration: configuration)
4099+
original = ArgumentError.new("lookup failed")
4100+
server.resources_subscribe_handler do |params|
4101+
raise Server::RequestHandlerError.new("Forbidden", params, error_type: :invalid_params, original_error: original)
4102+
end
4103+
4104+
assert_empty(server.subscribable_resource_uris(["file:///a.txt"]))
4105+
assert_equal([original], reported)
4106+
end
4107+
4108+
test "#subscribable_resource_uris reports a RequestHandlerError without an original error itself" do
4109+
reported = []
4110+
configuration = MCP::Configuration.new
4111+
configuration.exception_reporter = ->(e, _server_context) { reported << e }
4112+
server = Server.new(name: "test_server", configuration: configuration)
4113+
error = Server::RequestHandlerError.new("Forbidden", {}, error_type: :invalid_params)
4114+
server.resources_subscribe_handler { |_params| raise error }
4115+
4116+
assert_empty(server.subscribable_resource_uris(["file:///a.txt"]))
4117+
assert_equal([error], reported)
4118+
end
4119+
4120+
test "#subscribable_resource_uris keeps an accepted URI after a denied one and accepts a nil return value" do
4121+
server = Server.new(name: "test_server")
4122+
server.resources_subscribe_handler do |params|
4123+
raise "denied" if params[:uri] == "file:///denied.txt"
4124+
4125+
nil
4126+
end
4127+
4128+
assert_equal(["file:///accepted.txt"], server.subscribable_resource_uris(["file:///denied.txt", "file:///accepted.txt"]))
4129+
end
4130+
4131+
test "#subscribable_resource_uris passes the envelope to the resources_subscribe_handler" do
4132+
server = Server.new(name: "test_server")
4133+
seen = nil
4134+
server.resources_subscribe_handler do |_params, server_context:|
4135+
seen = server_context.envelope
4136+
end
4137+
envelope = RequestEnvelope.new(protocol_version: "2026-07-28", client_info: { name: "c" }, client_capabilities: {}, log_level: nil)
4138+
4139+
server.subscribable_resource_uris(["file:///a.txt"], envelope: envelope)
4140+
4141+
assert_same(envelope, seen)
4142+
end
4143+
4144+
test "#subscribable_resource_uris keeps every URI with the default resources_subscribe_handler" do
4145+
server = Server.new(name: "test_server")
4146+
4147+
assert_equal(["file:///a.txt", "file:///b.txt"], server.subscribable_resource_uris(["file:///a.txt", "file:///b.txt"]))
4148+
end
4149+
40654150
test "tools/call with no args" do
40664151
server = Server.new(tools: [@tool_with_no_args])
40674152

0 commit comments

Comments
 (0)