diff --git a/README.md b/README.md index 3e3709cd..401dd1f7 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,11 @@ It implements the Model Context Protocol specification, handling model context r and the Streamable HTTP transport serves them on a sessionless single-exchange path. On the client, `MCP::Client#connect` negotiates the lifecycle automatically by default (probe `server/discover`, fall back to the `initialize` handshake), `connect(mode: :modern)` skips the handshake entirely, `connect(mode: :legacy)` forces the classic handshake, and `MCP::Client#discover` exposes the raw discovery result +- `subscriptions/listen` - Long-lived notification subscription stream (MCP 2026-07-28, SEP-2575), replacing the legacy HTTP GET listening stream: + the client opts in via the `notifications` filter (`toolsListChanged` / `promptsListChanged` / `resourcesListChanged` / `resourceSubscriptions`), + the server acknowledges the honored subset with `notifications/subscriptions/acknowledged` as the first stream message, + and every delivered notification carries the correlating `io.modelcontextprotocol/subscriptionId` in `_meta`. Served on the Streamable HTTP modern path; + stdio answers `-32601` - Multi round-trip `input_required` results (MCP 2026-07-28, SEP-2322): a `tools/call`, `prompts/get`, or `resources/read` handler that opts in to `server_context:` may return `MCP::Server::InputRequiredResult.new(input_requests:, request_state:)` to ask the client for additional input (`elicitation/create`, `sampling/createMessage`, or `roots/list` shapes) instead of performing a server-initiated request, diff --git a/conformance/server.rb b/conformance/server.rb index dca807db..69e928a2 100644 --- a/conformance/server.rb +++ b/conformance/server.rb @@ -633,6 +633,34 @@ def call(server_context:, **_args) end end end + + class TestTriggerToolChange < MCP::Tool + tool_name "test_trigger_tool_change" + description "A diagnostic tool that broadcasts notifications/tools/list_changed to listen streams (SEP-2575)" + + class << self + # The broadcast alone exercises the `subscriptions/listen` delivery, so no actual + # tool-list mutation is needed, matching the suite's TypeScript reference fixture. + def call(server_context:, **_args) + server_context.notify_tools_list_changed + + MCP::Tool::Response.new([MCP::Content::Text.new("Mutation triggered").to_h]) + end + end + end + + class TestTriggerPromptChange < MCP::Tool + tool_name "test_trigger_prompt_change" + description "A diagnostic tool that broadcasts notifications/prompts/list_changed to listen streams (SEP-2575)" + + class << self + def call(server_context:, **_args) + server_context.notify_prompts_list_changed + + MCP::Tool::Response.new([MCP::Content::Text.new("Mutation triggered").to_h]) + end + end + end end module Prompts @@ -872,6 +900,8 @@ def build_server Tools::TestInputRequiredResultCapabilities, Tools::TestStreamingElicitation, Tools::TestLoggingTool, + Tools::TestTriggerToolChange, + Tools::TestTriggerPromptChange, ], prompts: [ Prompts::TestSimplePrompt, diff --git a/lib/mcp/methods.rb b/lib/mcp/methods.rb index f4c7df68..60af7084 100644 --- a/lib/mcp/methods.rb +++ b/lib/mcp/methods.rb @@ -7,6 +7,10 @@ module Methods LOGGING_SET_LEVEL = "logging/setLevel" # Sessionless capability discovery (MCP 2026-07-28 draft, SEP-2575). SERVER_DISCOVER = "server/discover" + # Long-lived notification subscription stream (MCP 2026-07-28, SEP-2575), + # replacing the legacy HTTP GET listening stream. Served at the transport layer + # (Streamable HTTP modern path); transports without streaming support answer `-32601`. + SUBSCRIPTIONS_LISTEN = "subscriptions/listen" PROMPTS_GET = "prompts/get" PROMPTS_LIST = "prompts/list" @@ -50,6 +54,9 @@ module Methods NOTIFICATIONS_PROGRESS = "notifications/progress" NOTIFICATIONS_CANCELLED = "notifications/cancelled" NOTIFICATIONS_ELICITATION_COMPLETE = "notifications/elicitation/complete" + # First message on a `subscriptions/listen` stream (SEP-2575): reports the subset + # of requested notification types the server agreed to honor. + NOTIFICATIONS_SUBSCRIPTIONS_ACKNOWLEDGED = "notifications/subscriptions/acknowledged" class MissingRequiredCapabilityError < StandardError attr_reader :method diff --git a/lib/mcp/request_envelope.rb b/lib/mcp/request_envelope.rb index 0a345490..3f1b19a7 100644 --- a/lib/mcp/request_envelope.rb +++ b/lib/mcp/request_envelope.rb @@ -16,6 +16,10 @@ class RequestEnvelope # Optional per-request log level, replacing the `logging/setLevel` RPC in the modern lifecycle. # Deprecated as of 2026-07-28 (SEP-2577) but still part of the wire format. LOG_LEVEL_META_KEY = "io.modelcontextprotocol/logLevel" + # Notification-side reserved key (SEP-2575): correlates a notification delivered on + # a `subscriptions/listen` stream (and the stream's closing result) with the JSON-RPC id of + # the `subscriptions/listen` request that opened it. Not part of the request envelope triple. + SUBSCRIPTION_ID_META_KEY = "io.modelcontextprotocol/subscriptionId" # Result-side counterpart of the request envelope: the server's identity rides in # the result's `_meta` as an optional stamp, not as a top-level field, since the SEP was diff --git a/lib/mcp/server.rb b/lib/mcp/server.rb index df68ef7b..567a8057 100644 --- a/lib/mcp/server.rb +++ b/lib/mcp/server.rb @@ -867,7 +867,7 @@ def validate_initialize_params!(params) def discover(_request) { supportedVersions: Configuration::SUPPORTED_MODERN_PROTOCOL_VERSIONS, - capabilities: capabilities, + capabilities: discover_capabilities, instructions: instructions, _meta: { RequestEnvelope::SERVER_INFO_META_KEY => server_info }, }.compact.merge( @@ -879,6 +879,21 @@ def discover(_request) ) end + # Capabilities as advertised by `server/discover`. In the modern lifecycle, `listChanged` and `subscribe` flags + # promise delivery over `subscriptions/listen` streams, so they are stripped when the transport does not serve that RPC + # (e.g. stdio), matching the Python SDK's era-aware capability derivation. + def discover_capabilities + return capabilities if @transport.respond_to?(:serves_subscriptions_listen?) && @transport.serves_subscriptions_listen? + + capabilities.each_with_object({}) do |(name, value), stripped| + stripped[name] = if value.is_a?(Hash) + value.reject { |flag, _| ["listChanged", "subscribe"].include?(flag.to_s) } + else + value + end + end + end + def configure_logging_level(request, session: nil) if capabilities[:logging].nil? raise RequestHandlerError.new("Server does not support logging", request, error_type: :internal_error) diff --git a/lib/mcp/server/transports/streamable_http_transport.rb b/lib/mcp/server/transports/streamable_http_transport.rb index f49dba6c..7782a8c3 100644 --- a/lib/mcp/server/transports/streamable_http_transport.rb +++ b/lib/mcp/server/transports/streamable_http_transport.rb @@ -1,6 +1,7 @@ # frozen_string_literal: true require "json" +require_relative "../../result_type" require_relative "../../transport" # This file is autoloaded only when `StreamableHTTPTransport` is referenced, @@ -35,6 +36,12 @@ class InvalidJsonError < StandardError; end DEFAULT_SESSION_IDLE_TIMEOUT = 1800 DEFAULT_MAX_SESSIONS = 10_000 + # Cap on concurrent `subscriptions/listen` streams (SEP-2575). Each stream holds an open SSE connection + # for its lifetime, so without a bound an unauthenticated client can retain unbounded connections, + # like the session-flood case `DEFAULT_MAX_SESSIONS` guards. A listen request past the cap is rejected with HTTP 503; + # pass `max_listen_subscriptions: nil` to opt out. + DEFAULT_MAX_LISTEN_SUBSCRIPTIONS = 1_000 + # Distinguishes "argument omitted, apply the secure default" from an explicit `nil` (opt out of expiry). UNSET_IDLE_TIMEOUT = Object.new.freeze private_constant :UNSET_IDLE_TIMEOUT @@ -87,6 +94,9 @@ class InvalidJsonError < StandardError; end # ownership is not enforced. # @param max_request_bytes [Integer] upper bound in bytes on a POST request body; larger # requests are rejected with HTTP 413. Defaults to 4 MiB. + # @param max_listen_subscriptions [Integer, nil] cap on concurrent `subscriptions/listen` + # streams; a listen request past the cap is rejected with HTTP 503, and `nil` disables + # the cap. def initialize( server, stateless: false, @@ -97,7 +107,8 @@ def initialize( allowed_hosts: nil, dns_rebinding_protection: true, session_request_validator: nil, - max_request_bytes: DEFAULT_MAX_REQUEST_BYTES + max_request_bytes: DEFAULT_MAX_REQUEST_BYTES, + max_listen_subscriptions: DEFAULT_MAX_LISTEN_SUBSCRIPTIONS ) super(server) # Maps `session_id` to `{ get_sse_stream: stream_object, server_session: ServerSession, last_active_at: float_from_monotonic_clock, origin: origin_header }`. @@ -114,6 +125,11 @@ def initialize( @allowed_origins = Array(allowed_origins).map(&:downcase).freeze @pending_responses = {} + # Maps a `subscriptions/listen` request id to `{ stream: stream_object, filter: honored_subscription_filter }` + # (SEP-2575). In-process only; a multi-worker deployment needs an external event bus to fan notifications out across processes, + # which is a follow-up. + @listen_subscriptions = {} + # Maps a modern request's ephemeral session id to the Array collecting the notifications its handler emits; # `handle_modern` registers the sink and flushes it as SSE frames ahead of the final response (SEP-2575). @modern_request_sinks = {} @@ -147,6 +163,12 @@ def initialize( @max_request_bytes = max_request_bytes + if !max_listen_subscriptions.nil? && !(max_listen_subscriptions.is_a?(Integer) && max_listen_subscriptions > 0) + raise ArgumentError, "max_listen_subscriptions must be a positive Integer or nil" + end + + @max_listen_subscriptions = max_listen_subscriptions + start_reaper_thread if @session_idle_timeout end @@ -163,6 +185,15 @@ def initialize( # JSON-RPC methods whose target name is mirrored into the `Mcp-Name` header (SEP-2575). NAME_BEARING_METHODS = [Methods::TOOLS_CALL, Methods::RESOURCES_READ, Methods::PROMPTS_GET].freeze + # Maps broadcast notification methods to the `SubscriptionFilter` field that opts in to them on + # a `subscriptions/listen` stream (SEP-2575). `notifications/resources/updated` is matched by URI + # against `resourceSubscriptions` instead. + LISTEN_FILTER_FIELDS = { + Methods::NOTIFICATIONS_TOOLS_LIST_CHANGED => :toolsListChanged, + Methods::NOTIFICATIONS_PROMPTS_LIST_CHANGED => :promptsListChanged, + Methods::NOTIFICATIONS_RESOURCES_LIST_CHANGED => :resourcesListChanged, + }.freeze + # JSON-RPC error codes that surface as HTTP 400 on the modern path. `-32601` maps to 404 # (disambiguating an unknown method from a legacy HTTP+SSE 404) and everything else, including internal errors, # stays 200, matching the Python SDK's status ladder. @@ -180,6 +211,12 @@ def call(env) handle_request(Rack::Request.new(env)) end + # The `subscriptions/listen` notification stream (SEP-2575) is served on the modern path, + # so `Server#discover` may advertise `listChanged`/`subscribe` capability flags. + def serves_subscriptions_listen? + true + end + def handle_request(request) rebinding_error = validate_dns_rebinding(request) return rebinding_error if rebinding_error @@ -235,6 +272,8 @@ def close @reaper_thread&.kill @reaper_thread = nil + teardown_listen_subscriptions + removed_sessions = @mutex.synchronize do @sessions.each_key.filter_map { |session_id| cleanup_session_unsafe(session_id) } end @@ -246,6 +285,12 @@ def close end def send_notification(method, params = nil, session_id: nil, related_request_id: nil) + # `subscriptions/listen` streams (SEP-2575) receive matching change notifications regardless of the delivery below: + # a resource updated by one session's tool call changed globally, so modern subscribers hear about it too. + # Runs before the per-request sink and the stateless guard because the listen registry does not depend on sessions, + # and a sink capturing the notification for its own response stream must not hide it from other subscriptions. + deliver_to_listen_subscriptions(method, params) + notification = { jsonrpc: "2.0", method: method, @@ -588,6 +633,10 @@ def handle_modern(request, header_version, body_string: nil) mismatch_error = validate_modern_headers(request, body, header_version) return mismatch_error if mismatch_error + # `subscriptions/listen` is a long-lived notification stream served at the transport layer; + # it never dispatches through `Server#handle`. + return handle_subscriptions_listen(body) if body[:method] == Methods::SUBSCRIPTIONS_LISTEN + session = modern_session notifications = @mutex.synchronize { @modern_request_sinks[session.session_id] = [] } begin @@ -677,6 +726,209 @@ def validate_modern_headers(request, body, header_version) nil end + # Serves `subscriptions/listen` (SEP-2575): opens a long-lived SSE stream whose first message is + # `notifications/subscriptions/acknowledged` with the subset of requested notification types + # the server agreed to honor. Notifications delivered on the stream carry `io.modelcontextprotocol/subscriptionId` + # (= the listen request id) in `_meta`. A graceful teardown (transport `close`) sends a `SubscriptionsListenResult` + # response; an abrupt disconnect sends nothing. SSE keepalive pings are not sent yet. + def handle_subscriptions_listen(body) + request_id = body[:id] + params = body[:params] + + # A listen frame without an id could never receive stream teardown correlation. + unless request_id + return invalid_request_response("Invalid Request: subscriptions/listen requires an id") + end + + begin + if RequestEnvelope.modern?(params) + RequestEnvelope.parse!(params, request: params) + else + return invalid_request_response("Invalid Request: modern requests require the SEP-2575 `_meta` envelope") + end + rescue Server::RequestHandlerError => e + return json_rpc_error_response( + status: 400, + code: e.error_code || JsonRpcHandler::ErrorCode::INVALID_REQUEST, + message: e.message, + data: e.error_data, + id: request_id, + ) + end + + filter = params[:notifications] + unless filter.is_a?(Hash) + return json_rpc_error_response( + status: 400, + code: JsonRpcHandler::ErrorCode::INVALID_PARAMS, + message: "Invalid params: subscriptions/listen requires a `notifications` filter object", + id: request_id, + ) + end + + # Best-effort cap check before committing to the SSE response; the registration inside + # `listen_sse_body` re-checks atomically for the race between two concurrent listens + # crossing the cap together. + if listen_subscriptions_full? + return too_many_listen_subscriptions_response(request_id) + end + + [200, SSE_HEADERS.dup, listen_sse_body(request_id, honored_filter(filter))] + end + + def listen_subscriptions_full? + return false unless @max_listen_subscriptions + + @mutex.synchronize { @listen_subscriptions.size >= @max_listen_subscriptions } + end + + def too_many_listen_subscriptions_response(request_id) + json_rpc_error_response( + status: 503, + code: JsonRpcHandler::ErrorCode::INTERNAL_ERROR, + message: "Service unavailable: maximum concurrent subscriptions/listen streams (#{@max_listen_subscriptions}) reached", + id: request_id, + ) + end + + # The proc registers the stream and returns, leaving the response open like + # the legacy GET stream (`create_sse_body`). + def listen_sse_body(request_id, honored) + proc do |stream| + rejected = false + @mutex.synchronize do + if @listen_subscriptions.key?(request_id) || + (@max_listen_subscriptions && @listen_subscriptions.size >= @max_listen_subscriptions) + rejected = true + else + @listen_subscriptions[request_id] = { stream: stream, filter: honored } + end + end + + if rejected + close_stream_safely(stream) + else + acknowledgement = { + jsonrpc: "2.0", + method: Methods::NOTIFICATIONS_SUBSCRIPTIONS_ACKNOWLEDGED, + params: { + notifications: honored, + _meta: { RequestEnvelope::SUBSCRIPTION_ID_META_KEY.to_sym => request_id }, + }, + } + + begin + send_to_stream(stream, acknowledgement) + rescue *STREAM_WRITE_ERRORS + remove_listen_subscription(request_id) + close_stream_safely(stream) + end + end + end + end + + # 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) + capabilities = @server.capabilities + honored = {} + honored[:toolsListChanged] = true if filter[:toolsListChanged] && capability_flag?(capabilities, :tools, :listChanged) + honored[:promptsListChanged] = true if filter[:promptsListChanged] && capability_flag?(capabilities, :prompts, :listChanged) + honored[:resourcesListChanged] = true if filter[:resourcesListChanged] && capability_flag?(capabilities, :resources, :listChanged) + + subscriptions = filter[:resourceSubscriptions] + if capability_flag?(capabilities, :resources, :subscribe) && subscriptions.is_a?(Array) && !subscriptions.empty? + honored[:resourceSubscriptions] = subscriptions + end + + honored + end + + # Reads a nested capability flag tolerating both symbol and string keys, since user-supplied capability hashes arrive + # in either form. The flag that promises delivery (`listChanged` / `subscribe`) decides honoring, the same derivation + # `Server#discover` uses for its era-aware capability stripping; the mere presence of the primitive's capability is not enough. + def capability_flag?(capabilities, name, flag) + value = capabilities[name] || capabilities[name.to_s] + return false unless value.is_a?(Hash) + + !!(value[flag] || value[flag.to_s]) + end + + # Fans a notification out to every `subscriptions/listen` stream whose honored filter opted in to it, + # stamping the correlating `subscriptionId` into `_meta`. Matching against the honored filter + # (not the requested one) enforces the MUST NOT-send-unrequested-types rule. + def deliver_to_listen_subscriptions(method, params) + field = LISTEN_FILTER_FIELDS[method] + return if field.nil? && method != Methods::NOTIFICATIONS_RESOURCES_UPDATED + + # The matching snapshot is taken under `@mutex`, but stream writes happen outside it: + # a slow or stalled subscriber must not block the transport, matching the legacy delivery paths. + matched = @mutex.synchronize do + @listen_subscriptions.filter_map do |request_id, subscription| + hit = if field + subscription[:filter][field] + else + uris = subscription[:filter][:resourceSubscriptions] + uri = params.is_a?(Hash) ? params[:uri] || params["uri"] : nil + uris.is_a?(Array) && uris.include?(uri) + end + + [request_id, subscription[:stream]] if hit + end + end + + matched.each do |request_id, stream| + meta = { RequestEnvelope::SUBSCRIPTION_ID_META_KEY.to_sym => request_id } + notification_params = (params || {}).merge(_meta: meta) + notification = { jsonrpc: "2.0", method: method, params: notification_params } + + begin + send_to_stream(stream, notification) + rescue *STREAM_WRITE_ERRORS => e + MCP.configuration.exception_reporter.call( + e, + { subscription_id: request_id, error: "Failed to send notification" }, + ) + remove_listen_subscription(request_id) + close_stream_safely(stream) + end + end + end + + def remove_listen_subscription(request_id) + @mutex.synchronize { @listen_subscriptions.delete(request_id) } + end + + # Graceful teardown (SEP-2575): each open listen stream receives its `SubscriptionsListenResult` response + # before the stream closes. + def teardown_listen_subscriptions + removed = @mutex.synchronize do + subscriptions = @listen_subscriptions.dup + @listen_subscriptions.clear + subscriptions + end + + removed.each do |request_id, subscription| + begin + send_to_stream(subscription[:stream], { + jsonrpc: "2.0", + id: request_id, + result: { + # `SubscriptionsListenResult` is served at the transport layer and never + # passes through the dispatch path, so the REQUIRED 2026-07-28 `resultType` is + # stamped at its construction site. + resultType: ResultType::COMPLETE, + _meta: { RequestEnvelope::SUBSCRIPTION_ID_META_KEY.to_sym => request_id }, + }, + }) + rescue *STREAM_WRITE_ERRORS + nil + end + close_stream_safely(subscription[:stream]) + end + end + def header_mismatch_response(message, id) json_rpc_error_response( status: 400, diff --git a/lib/mcp/transport.rb b/lib/mcp/transport.rb index 9f94c601..c48e3c9f 100644 --- a/lib/mcp/transport.rb +++ b/lib/mcp/transport.rb @@ -50,6 +50,13 @@ def send_request(method, params = nil) raise NotImplementedError, "Subclasses must implement send_request" end + # Whether the transport serves the `subscriptions/listen` notification stream (MCP 2026-07-28, SEP-2575). + # `Server#discover` strips the `listChanged`/`subscribe` capability flags when the transport cannot deliver + # those notifications in the modern lifecycle. + def serves_subscriptions_listen? + false + end + private def generate_request_id diff --git a/test/mcp/server/transports/stdio_transport_test.rb b/test/mcp/server/transports/stdio_transport_test.rb index 699bcb7a..5b76a747 100644 --- a/test/mcp/server/transports/stdio_transport_test.rb +++ b/test/mcp/server/transports/stdio_transport_test.rb @@ -634,6 +634,27 @@ class StdioTransportTest < ActiveSupport::TestCase 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)]) diff --git a/test/mcp/server/transports/streamable_http_transport_test.rb b/test/mcp/server/transports/streamable_http_transport_test.rb index 6edce06e..5a60c128 100644 --- a/test/mcp/server/transports/streamable_http_transport_test.rb +++ b/test/mcp/server/transports/streamable_http_transport_test.rb @@ -5772,6 +5772,184 @@ def string 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 "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 + private def initialize_test_session(id: "init") @@ -5885,6 +6063,41 @@ def modern_body(method, params, version: "2026-07-28", capabilities: {}) ), }.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 diff --git a/test/mcp/server_test.rb b/test/mcp/server_test.rb index fa99de63..446401dd 100644 --- a/test/mcp/server_test.rb +++ b/test/mcp/server_test.rb @@ -130,7 +130,9 @@ class ServerTest < ActiveSupport::TestCase 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 @@ class ServerTest < ActiveSupport::TestCase 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: {})