Add public subscriptions/listen server handler (SEP-2575) (#1775)

Co-authored-by: Tarek Mahmoud Sayed <tarekms@ntdev.microsoft.com>
Copilot-Session: 80653343-5dcb-43cb-89b0-8ac8e572c4f7

Author: tarekgh

src/ModelContextProtocol.Core/Server/McpServerHandlers.cs

@@ -191,6 +191,60 @@ public McpRequestHandler<CallToolRequestParams, ResultOrAlternate<CallToolResult
     /// </remarks>
     public McpRequestHandler<UnsubscribeRequestParams, EmptyResult>? UnsubscribeFromResourcesHandler { get; set; }
 
+    /// <summary>
+    /// Gets or sets the handler for <see cref="RequestMethods.SubscriptionsListen"/> requests (SEP-2575).
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <c>subscriptions/listen</c> is a long-lived request introduced by the 2026-07-28 protocol revision. The
+    /// held-open response is a solicited server-to-client stream: the server first acknowledges which
+    /// subscriptions it will honor and then streams matching notifications until the request is cancelled.
+    /// Setting this handler lets a server author own that stream directly to implement custom subscription
+    /// kinds, application-driven <c>resources/updated</c> delivery, or subscriptions backed by their own event
+    /// source. It is especially useful for stateless Streamable HTTP, where unsolicited notifications are
+    /// dropped (there is no session-wide channel) but the listen request's response stream can still carry
+    /// notifications for the duration of the request.
+    /// </para>
+    /// <para>
+    /// This is a <b>full replacement</b> for the built-in <c>subscriptions/listen</c> handler. When set, the
+    /// SDK does not track the subscription, does not send the acknowledgement, and does not perform any
+    /// automatic <c>*/list_changed</c> fan-out for the request; the handler is solely responsible for the
+    /// entire lifetime of the stream. The SDK still enforces protocol-version gating: the handler is only
+    /// reached when the negotiated protocol revision is 2026-07-28 or later, and is otherwise rejected with
+    /// <see cref="McpErrorCode.MethodNotFound"/>.
+    /// </para>
+    /// <para>
+    /// An implementation of this handler is responsible for:
+    /// </para>
+    /// <list type="bullet">
+    /// <item><description>
+    /// Sending exactly one <see cref="NotificationMethods.SubscriptionsAcknowledgedNotification"/> before any
+    /// subscription events, reporting only the filters it actually honors. Advertised server capabilities must
+    /// match what the handler will actually deliver.
+    /// </description></item>
+    /// <item><description>
+    /// Tagging every streamed notification with the listen request id under
+    /// <c>_meta[<see cref="MetaKeys.SubscriptionId"/>]</c> so clients sharing a channel can demultiplex it.
+    /// </description></item>
+    /// <item><description>
+    /// Remaining active for the subscription lifetime and cleaning up when the supplied
+    /// <see cref="CancellationToken"/> is cancelled (client disconnect on HTTP, or
+    /// <c>notifications/cancelled</c> on stdio).
+    /// </description></item>
+    /// <item><description>
+    /// Returning <see cref="EmptyResult"/> when it deliberately completes the stream.
+    /// </description></item>
+    /// </list>
+    /// <para>
+    /// Notifications are sent through the request's server (for example <c>request.Server.SendMessageAsync</c>),
+    /// which routes them over the request's own response stream. For extension filters not represented by
+    /// <see cref="SubscriptionsListenRequestParams"/>, the handler can inspect
+    /// <c>request.JsonRpcRequest.Params</c>. Application services and event buses can be resolved from
+    /// <c>request.Services</c> or captured by the handler delegate.
+    /// </para>
+    /// </remarks>
+    public McpRequestHandler<SubscriptionsListenRequestParams, EmptyResult>? SubscriptionsListenHandler { get; set; }
+
     /// <summary>
     /// Gets or sets the handler for <see cref="RequestMethods.LoggingSetLevel"/> requests.
     /// </summary>

src/ModelContextProtocol.Core/Server/McpServerImpl.cs

@@ -117,8 +117,11 @@ public McpServerImpl(ITransport transport, McpServerOptions options, ILoggerFact
         }
 
         // A stateful session can push unsolicited list-changed notifications, so subscribe to the
-        // collection change events. A stateless HTTP server cannot send unsolicited notifications, so
-        // instead suppress the listChanged capability it would otherwise advertise.
+        // collection change events. A stateless HTTP server cannot push unsolicited notifications; whether it
+        // may still advertise the listChanged capability (over a custom subscriptions/listen stream to a
+        // 2026-07-28+ client) is decided per response in GetAdvertisedCapabilities rather than cleared here,
+        // because the same ServerCapabilities feeds both the legacy initialize handshake (which can never
+        // deliver it) and server/discover (which can, given a custom handler).
         if (HasStatefulTransport())
         {
             Register(ServerOptions.ToolCollection, NotificationMethods.ToolListChangedNotification);
@@ -136,15 +139,6 @@ void Register<TPrimitive>(McpServerPrimitiveCollection<TPrimitive>? collection,
                 }
             }
         }
-        else
-        {
-            if (ServerCapabilities.Tools is not null)
-                ServerCapabilities.Tools.ListChanged = null;
-            if (ServerCapabilities.Prompts is not null)
-                ServerCapabilities.Prompts.ListChanged = null;
-            if (ServerCapabilities.Resources is not null)
-                ServerCapabilities.Resources.ListChanged = null;
-        }
 
         // And initialize the session. The built-in protocol metadata filters run ahead of any
         // user-supplied message filters.
@@ -516,6 +510,46 @@ private void SetNegotiatedProtocolVersion(string protocolVersion)
     /// <inheritdoc/>
     public ServerCapabilities ServerCapabilities { get; }
 
+    /// <summary>
+    /// Returns the <see cref="ServerCapabilities"/> to advertise in a specific response, suppressing the
+    /// <c>listChanged</c> flags the server has no way to honor.
+    /// </summary>
+    /// <param name="listenStreamCanDeliverListChanged">
+    /// <see langword="true"/> when the client this response targets can receive <c>*/list_changed</c>
+    /// notifications over a <c>subscriptions/listen</c> stream.
+    /// </param>
+    /// <remarks>
+    /// A stateless HTTP server has no session-wide channel to push unsolicited <c>*/list_changed</c>
+    /// notifications. It can only deliver them over a <c>subscriptions/listen</c> stream, which requires both
+    /// a 2026-07-28+ client (so the request is reachable at all) and a custom
+    /// <see cref="McpServerHandlers.SubscriptionsListenHandler"/> to own that stream (the built-in stateless
+    /// handler grants no notifications). When neither the transport is stateful nor that stream can carry
+    /// them, the <c>listChanged</c> flags are dropped so the server never advertises a capability it cannot
+    /// deliver. Everything else (for example <c>resources.subscribe</c>) is preserved.
+    /// </remarks>
+    private ServerCapabilities GetAdvertisedCapabilities(bool listenStreamCanDeliverListChanged)
+    {
+        if (HasStatefulTransport() || listenStreamCanDeliverListChanged)
+        {
+            return ServerCapabilities;
+        }
+
+        // Copy onto a fresh instance so the shared ServerCapabilities keeps the authored listChanged flags;
+        // server/discover with a custom listen handler may still advertise them.
+        return new ServerCapabilities
+        {
+            Experimental = ServerCapabilities.Experimental,
+            Logging = ServerCapabilities.Logging,
+            Completions = ServerCapabilities.Completions,
+            Extensions = ServerCapabilities.Extensions,
+            Prompts = ServerCapabilities.Prompts is null ? null : new PromptsCapability { ListChanged = null },
+            Resources = ServerCapabilities.Resources is { } resources
+                ? new ResourcesCapability { Subscribe = resources.Subscribe, ListChanged = null }
+                : null,
+            Tools = ServerCapabilities.Tools is null ? null : new ToolsCapability { ListChanged = null },
+        };
+    }
+
     /// <inheritdoc />
     public override ClientCapabilities? ClientCapabilities => _clientCapabilities;
 
@@ -667,7 +701,11 @@ private void ConfigureInitialize(McpServerOptions options)
                     ProtocolVersion = negotiatedProtocolVersion,
                     Instructions = options.ServerInstructions,
                     ServerInfo = options.ServerInfo ?? DefaultImplementation,
-                    Capabilities = ServerCapabilities ?? new(),
+
+                    // The initialize handshake only serves pre-2026-07-28 clients, which cannot open a
+                    // subscriptions/listen stream, so a stateless server has no way to deliver list-changed
+                    // notifications to them regardless of any custom handler.
+                    Capabilities = GetAdvertisedCapabilities(listenStreamCanDeliverListChanged: false),
 
                     // resultType is a 2026-07-28 result field. The initialize handshake is only available on
                     // 2025-11-25 and earlier revisions (2026-07-28+ negotiate via server/discover and throw
@@ -694,7 +732,13 @@ private void ConfigureDiscover(McpServerOptions options)
                 return new ValueTask<DiscoverResult>(new DiscoverResult
                 {
                     SupportedVersions = [.. _perRequestMetadataProtocolVersions],
-                    Capabilities = ServerCapabilities ?? new(),
+
+                    // server/discover only serves 2026-07-28+ clients, which can open a subscriptions/listen
+                    // stream. A stateless server can therefore still deliver list-changed notifications if the
+                    // author supplied a custom handler to own that stream (the built-in stateless handler
+                    // grants nothing, so it cannot).
+                    Capabilities = GetAdvertisedCapabilities(
+                        listenStreamCanDeliverListChanged: options.Handlers.SubscriptionsListenHandler is not null),
                     Instructions = options.ServerInstructions,
                     // Spec PR #2855 makes ttlMs and cacheScope required on DiscoverResult. Default to
                     // the safest values (immediately stale, not shareable) so existing servers keep
@@ -722,9 +766,61 @@ private void ConfigureDiscover(McpServerOptions options)
     /// Subscription-bound notifications carry the listen request's id in their
     /// <c>_meta/io.modelcontextprotocol/subscriptionId</c> field per SEP-2575 so clients can demultiplex.
     /// </para>
+    /// <para>
+    /// A server author may supply a custom <see cref="McpServerHandlers.SubscriptionsListenHandler"/> to take
+    /// over the stream entirely; see the design notes at the top of this method for the behavior.
+    /// </para>
     /// </remarks>
     private void ConfigureSubscriptions(McpServerOptions options)
     {
+        // Design decision 1 of issue #1662 (replacement vs. additive handler): a custom
+        // SubscriptionsListenHandler is a FULL REPLACEMENT for the built-in subscriptions/listen handler, not
+        // an additive/composed one. When one is set, that handler exclusively owns the stream: the SDK does
+        // not track the subscription in _activeSubscriptions, does not send the acknowledgement, and performs
+        // no automatic */list_changed fan-out for the request. This keeps the SEP-2575 contract trivial to
+        // honor (exactly one acknowledgement, no duplicate delivery) and mirrors the existing low-level
+        // replacement handlers such as CallToolWithAlternateHandler. An additive design was rejected because
+        // two writers on one stream create ambiguity over who sends the single acknowledgement, force the two
+        // lifetimes to be coordinated, and risk double-tagging the subscription id.
+        if (options.Handlers.SubscriptionsListenHandler is { } subscriptionsListenHandler)
+        {
+            // Route the custom handler through SetHandler so it receives the same DestinationBoundMcpServer as
+            // every other typed handler. That server sends notifications over this request's own response
+            // stream (its RelatedTransport), which is what lets the handler stream even under stateless
+            // Streamable HTTP, where the held-open POST response is the only solicited server-to-client
+            // channel (the core scenario of issue #1662). Going through SetHandler also applies the standard
+            // 2026-07-28 resultType stamping and provides the request-scoped service provider via
+            // request.Services.
+            SetHandler(RequestMethods.SubscriptionsListen,
+                (request, cancellationToken) =>
+                {
+                    // Protocol-version gating stays in the SDK rather than the custom handler, so a custom
+                    // handler can never be reached on a revision that predates SEP-2575. subscriptions/listen
+                    // is a 2026-07-28 feature; on older negotiated revisions it is rejected as an unknown
+                    // method, exactly as the built-in handler below does.
+                    if (!IsJuly2026OrLaterProtocolRequest(request.JsonRpcRequest))
+                    {
+                        throw new McpProtocolException(
+                            $"The method '{RequestMethods.SubscriptionsListen}' requires a newer protocol revision that supports per-request subscriptions; " +
+                            $"the negotiated protocol version is '{NegotiatedProtocolVersion ?? "(none)"}'.",
+                            McpErrorCode.MethodNotFound);
+                    }
+
+                    // Notifications is 'required', but that only enforces presence during deserialization,
+                    // not non-nullness: a '{"notifications": null}' payload produces a non-null params object
+                    // with a null Notifications (DefaultOptions does not set RespectNullableAnnotations).
+                    // Normalize null to empty so a custom handler can dereference request.Params.Notifications
+                    // without an NRE, matching the built-in handler's request?.Notifications guard below.
+                    request.Params ??= new SubscriptionsListenRequestParams { Notifications = new() };
+                    request.Params.Notifications ??= new SubscriptionsListenNotifications();
+
+                    return subscriptionsListenHandler(request, cancellationToken);
+                },
+                McpJsonUtilities.JsonContext.Default.SubscriptionsListenRequestParams,
+                McpJsonUtilities.JsonContext.Default.EmptyResult);
+            return;
+        }
+
         _requestHandlers.Set(RequestMethods.SubscriptionsListen,
             async (request, jsonRpcRequest, cancellationToken) =>
             {

src/ModelContextProtocol/McpServerBuilderExtensions.cs

@@ -859,6 +859,51 @@ public static IMcpServerBuilder WithUnsubscribeFromResourcesHandler(this IMcpSer
         return builder;
     }
 
+    /// <summary>
+    /// Configures a handler for <c>subscriptions/listen</c> requests (SEP-2575), taking over the long-lived
+    /// subscription stream introduced by the 2026-07-28 protocol revision.
+    /// </summary>
+    /// <param name="builder">The server builder instance.</param>
+    /// <param name="handler">The handler that owns the subscription stream for the lifetime of the request.</param>
+    /// <returns>The builder provided in <paramref name="builder"/>.</returns>
+    /// <exception cref="ArgumentNullException"><paramref name="builder"/> is <see langword="null"/>.</exception>
+    /// <remarks>
+    /// <para>
+    /// <c>subscriptions/listen</c> is a long-lived request whose held-open response is a solicited
+    /// server-to-client stream. Providing a handler here lets a server author own that stream to implement
+    /// custom subscription kinds, application-driven <c>resources/updated</c> delivery, or subscriptions backed
+    /// by their own event source. It is especially useful for stateless Streamable HTTP, where unsolicited
+    /// notifications are dropped but the listen request's response stream can still carry notifications for the
+    /// duration of the request.
+    /// </para>
+    /// <para>
+    /// This is a full replacement for the SDK's built-in <c>subscriptions/listen</c> handling. When set, the
+    /// handler alone is responsible for sending exactly one
+    /// <see cref="NotificationMethods.SubscriptionsAcknowledgedNotification"/> before any events, tagging every
+    /// streamed notification with the listen request id under <c>_meta[<see cref="MetaKeys.SubscriptionId"/>]</c>,
+    /// staying active until the supplied <see cref="CancellationToken"/> is cancelled, and returning
+    /// <see cref="EmptyResult"/> when it completes. See <see cref="McpServerHandlers.SubscriptionsListenHandler"/>
+    /// for the full contract.
+    /// </para>
+    /// <para>
+    /// Unlike <see cref="WithSubscribeToResourcesHandler"/>, this method intentionally does not advertise any
+    /// server capabilities on the author's behalf. The set of notifications a listen handler honors is decided
+    /// at runtime and can include custom kinds not represented by a capability flag, so the author must
+    /// configure only the capabilities their handler will actually deliver. Advertised capabilities must match
+    /// what the handler delivers.
+    /// </para>
+    /// </remarks>
+    public static IMcpServerBuilder WithSubscriptionsListenHandler(this IMcpServerBuilder builder, McpRequestHandler<SubscriptionsListenRequestParams, EmptyResult> handler)
+    {
+        Throw.IfNull(builder);
+
+        builder.Services.Configure<McpServerOptions>(options =>
+        {
+            options.Handlers.SubscriptionsListenHandler = handler;
+        });
+        return builder;
+    }
+
     /// <summary>
     /// Configures a handler for processing logging level change requests from clients.
     /// </summary>