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

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

@@ -722,9 +722,53 @@ 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);
+                    }
+
+                    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>

tests/ModelContextProtocol.AspNetCore.Tests/StatelessServerTests.cs

@@ -380,6 +380,113 @@ await client.SendRequestAsync(listenRequest, TestContext.Current.CancellationTok
         Assert.Null(grantedNotifications["resourceSubscriptions"]);
     }
 
+    [Fact]
+    public async Task SubscriptionsListen_WithCustomHandler_InStatelessMode_StreamsNotificationOverHeldOpenPost()
+    {
+        // The built-in stateless handler grants nothing and returns immediately because there is no
+        // session-wide channel. A custom SubscriptionsListenHandler can instead stream notifications over the
+        // held-open POST response (the listen request's RelatedTransport), which is the solicited
+        // server-to-client stream. This is the core scenario of issue #1662.
+        const string subscribedUri = "resource://test";
+
+        Builder.Services.AddMcpServer()
+            .WithHttpTransport(options =>
+            {
+                options.Stateless = true;
+            })
+            .WithSubscriptionsListenHandler(async (request, cancellationToken) =>
+            {
+                var subscriptionId = request.JsonRpcRequest.Id;
+
+                var ack = new JsonRpcNotification
+                {
+                    Method = NotificationMethods.SubscriptionsAcknowledgedNotification,
+                    Params = JsonSerializer.SerializeToNode(
+                        new SubscriptionsAcknowledgedNotificationParams
+                        {
+                            Notifications = new SubscriptionsListenNotifications
+                            {
+                                ResourceSubscriptions = request.Params.Notifications.ResourceSubscriptions,
+                            },
+                        },
+                        McpJsonUtilities.DefaultOptions),
+                };
+                TagWithSubscriptionId(ack, subscriptionId);
+                await request.Server.SendMessageAsync(ack, cancellationToken);
+
+                var updated = new JsonRpcNotification
+                {
+                    Method = NotificationMethods.ResourceUpdatedNotification,
+                    Params = new JsonObject { ["uri"] = subscribedUri },
+                };
+                TagWithSubscriptionId(updated, subscriptionId);
+                await request.Server.SendMessageAsync(updated, cancellationToken);
+
+                // Complete the stream so the POST response finishes; the buffered notifications flush to the client.
+                return new EmptyResult();
+            });
+
+        _app = Builder.Build();
+        _app.MapMcp();
+        await _app.StartAsync(TestContext.Current.CancellationToken);
+
+        HttpClient.DefaultRequestHeaders.Accept.Add(new("application/json"));
+        HttpClient.DefaultRequestHeaders.Accept.Add(new("text/event-stream"));
+
+        await using var client = await ConnectMcpClientAsync();
+
+        var ackChannel = Channel.CreateUnbounded<JsonRpcNotification>();
+        var updatedChannel = Channel.CreateUnbounded<JsonRpcNotification>();
+        await using var ackReg = client.RegisterNotificationHandler(NotificationMethods.SubscriptionsAcknowledgedNotification,
+            (notification, _) => { ackChannel.Writer.TryWrite(notification); return default; });
+        await using var updatedReg = client.RegisterNotificationHandler(NotificationMethods.ResourceUpdatedNotification,
+            (notification, _) => { updatedChannel.Writer.TryWrite(notification); return default; });
+
+        var listenRequest = new JsonRpcRequest
+        {
+            Method = RequestMethods.SubscriptionsListen,
+            Params = JsonSerializer.SerializeToNode(
+                new SubscriptionsListenRequestParams
+                {
+                    Notifications = new SubscriptionsListenNotifications { ResourceSubscriptions = [subscribedUri] },
+                },
+                McpJsonUtilities.DefaultOptions),
+        };
+
+        await client.SendRequestAsync(listenRequest, TestContext.Current.CancellationToken)
+            .WaitAsync(TestConstants.DefaultTimeout, TestContext.Current.CancellationToken);
+
+        var ack = await ackChannel.Reader.ReadAsync(TestContext.Current.CancellationToken);
+        var subscriptionId = GetSubscriptionId(ack);
+        Assert.NotNull(subscriptionId);
+
+        var updated = await updatedChannel.Reader.ReadAsync(TestContext.Current.CancellationToken);
+        Assert.Equal(subscriptionId, GetSubscriptionId(updated));
+        Assert.Equal(subscribedUri, Assert.IsType<JsonObject>(updated.Params)["uri"]?.GetValue<string>());
+    }
+
+    private static string? GetSubscriptionId(JsonRpcNotification notification)
+        => ((notification.Params as JsonObject)?["_meta"] as JsonObject)?[MetaKeys.SubscriptionId]?.ToJsonString();
+
+    private static void TagWithSubscriptionId(JsonRpcNotification notification, RequestId subscriptionId)
+    {
+        var paramsObject = notification.Params as JsonObject ?? new JsonObject();
+        if (paramsObject["_meta"] is not JsonObject meta)
+        {
+            meta = new JsonObject();
+            paramsObject["_meta"] = meta;
+        }
+
+        meta[MetaKeys.SubscriptionId] = subscriptionId.Id switch
+        {
+            string stringId => JsonValue.Create(stringId),
+            long longId => JsonValue.Create(longId),
+            _ => null,
+        };
+
+        notification.Params = paramsObject;
+    }
+
     [McpServerTool(Name = "testSamplingErrors")]
     public static async Task<string> TestSamplingErrors(McpServer server)
     {