From 3e92253e6b8a784570e452b789188af85c95432a Mon Sep 17 00:00:00 2001 From: Koichi ITO Date: Sat, 8 Aug 2026 18:53:28 +0900 Subject: [PATCH] Serve the `subscriptions/listen` notification stream per SEP-2575 ## Motivation and Context Companion to the stateless lifecycle work (SEP-2575, modelcontextprotocol/modelcontextprotocol#2575) for the 2026-07-28 MCP spec release. The modern lifecycle removed the HTTP GET listening stream; `subscriptions/listen` replaces it as a long-lived POST that opts in to server change notifications. The Python SDK ships the server side of this as its SEP-2575 event-bus work (python-sdk PR /modelcontextprotocol/python-sdk#3035); the TypeScript SDK landed it in PR modelcontextprotocol/typescript-sdk#2321. Wire behavior, per the draft schema: - The request carries a REQUIRED `notifications` filter (`SubscriptionFilter`): `toolsListChanged`, `promptsListChanged`, `resourcesListChanged`, and `resourceSubscriptions` (URI list, replacing the legacy `resources/subscribe` RPC). Every type is opt-in; the server MUST NOT send types the client did not request. - The first stream message is `notifications/subscriptions/acknowledged`, reporting the subset of requested types the server agreed to honor. Honoring reads the capability FLAGS that promise delivery (`listChanged`, `subscribe`), the same derivation `server/discover` uses for its era-aware capability stripping; the mere presence of a primitive's capability is not enough. - Every notification delivered on the stream carries the correlating `io.modelcontextprotocol/subscriptionId` (the listen request id) in `_meta`. - A graceful teardown (transport `close`) sends a `SubscriptionsListenResult` response before closing the stream, stamped with the REQUIRED 2026-07-28 `resultType` at its construction site (it never passes through the dispatch path); an abrupt disconnect sends nothing. Implementation: - `StreamableHTTPTransport` intercepts `subscriptions/listen` on the modern path (after header and envelope validation) and serves it as a long-lived SSE stream, using the same register-and-return body proc pattern as the legacy GET stream. Subscriptions live in an in-process registry keyed by the listen request id; fan-out hooks into `send_notification` ahead of the legacy delivery, so a resource updated by one session's tool call also reaches modern subscribers. The matching snapshot is taken under the transport mutex, but stream writes happen outside it, matching the legacy delivery paths: a slow or stalled subscriber must not block the transport. Duplicate subscription ids close the new stream instead of double-registering. - Concurrent listen streams are capped (`max_listen_subscriptions:`, default 1000, `nil` to opt out); a listen request past the cap is rejected with HTTP 503, like the `max_sessions` guard against session floods. Each stream holds an open connection for its lifetime, so without a bound an unauthenticated client can retain unbounded connections. - stdio does not serve the stream, matching the Python SDK's stream-pair behavior: no server handler is registered, so the method answers `-32601`. - `Server#discover` becomes era-aware about notification delivery: `listChanged`/`subscribe` capability flags promise delivery over `subscriptions/listen` streams in the modern lifecycle, so they are stripped when the transport does not serve that RPC (the new `Transport#serves_subscriptions_listen?` seam, true for `StreamableHTTPTransport`). - The conformance fixture defines the diagnostic triggers the `server-stateless` scenario calls (`test_trigger_tool_change` / `test_trigger_prompt_change`): each broadcasts its list-changed notification to the listen streams and returns, mirroring the suite's TypeScript reference fixture, which mutates nothing either. Out of scope, noted for follow-ups: SSE keepalive pings on listen streams, an external event bus for multi-worker deployments, and the client-side listen driver (the Python reference is python-sdk PR modelcontextprotocol/python-sdk#3047). Refs #389. ## How Has This Been Tested? New tests in `test/mcp/server/transports/streamable_http_transport_test.rb` cover: the SSE response with the acknowledgement as the first event (including the honored-subset reduction for unsupported types and for capability entries lacking the delivery flag), the `notifications` filter and envelope requirements (400 responses), opt-in-only delivery with the correlating `subscriptionId`, URI-scoped `resources/updated` delivery, per-subscription ids across concurrent streams, the graceful close result carrying `resultType: "complete"`, duplicate-id rejection, and the 503 past the concurrent stream cap. `test/mcp/server/transports/stdio_transport_test.rb` asserts `-32601` over stdio, and `test/mcp/server_test.rb` covers the era-aware capability stripping in `server/discover` for both transport kinds. `bundle exec rake` (tests, RuboCop, and conformance baseline) passes. Against the conformance fixture server at `--spec-version 2026-07-28`, the `server-stateless` subscription checks all report SUCCESS: the acknowledgement, `subscriptionId` tagging, and filter-containment MUSTs, plus both list-changed SHOULD checks driven by the new trigger tools. The `--requirements 2025-11-25` server leg passes 78/78, unchanged. ## Breaking Changes None. The method was previously unhandled (`-32601` everywhere); the only observable change to existing responses is that `server/discover` no longer advertises `listChanged`/`subscribe` flags on transports that cannot deliver those notifications in the modern lifecycle, which `server/discover` has not shipped in a gem release with anyway. --- README.md | 5 + conformance/server.rb | 30 +++ lib/mcp/methods.rb | 7 + lib/mcp/request_envelope.rb | 4 + lib/mcp/server.rb | 17 +- .../transports/streamable_http_transport.rb | 254 +++++++++++++++++- lib/mcp/transport.rb | 7 + .../server/transports/stdio_transport_test.rb | 21 ++ .../streamable_http_transport_test.rb | 213 +++++++++++++++ test/mcp/server_test.rb | 27 +- 10 files changed, 582 insertions(+), 3 deletions(-) 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: {})