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
This commit is contained in:
Tarek Mahmoud Sayed
GitHub
parent
79e13b3e2c
commit
81ae6ec6c7
6 changed files with 689 additions and 13 deletions
|
|
@ -191,6 +191,60 @@ public sealed class McpServerHandlers
|
|||
/// </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>
|
||||
|
|
|
|||
|
|
@ -117,8 +117,11 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
}
|
||||
|
||||
// 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 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
}
|
||||
}
|
||||
}
|
||||
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 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
/// <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 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
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 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
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 @@ internal sealed partial class McpServerImpl : McpServer
|
|||
/// 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) =>
|
||||
{
|
||||
|
|
|
|||
|
|
@ -859,6 +859,51 @@ public static partial class McpServerBuilderExtensions
|
|||
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>
|
||||
|
|
|
|||
|
|
@ -380,6 +380,196 @@ public class StatelessServerTests(ITestOutputHelper outputHelper) : KestrelInMem
|
|||
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>());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SubscriptionsListen_WithCustomHandler_InStatelessMode_AdvertisesAndStreamsListChanged()
|
||||
{
|
||||
// resources/updated rides on resources.subscribe, which is never suppressed, so it cannot prove the
|
||||
// listChanged capability is advertised. A custom SubscriptionsListenHandler gives a stateless server a
|
||||
// way to deliver */list_changed over the held-open POST, so server/discover (the only path a
|
||||
// 2026-07-28+ client uses) must advertise tools.listChanged rather than dropping it (issue #1662).
|
||||
Builder.Services.AddMcpServer()
|
||||
.WithHttpTransport(options =>
|
||||
{
|
||||
options.Stateless = true;
|
||||
})
|
||||
.WithTools([McpServerTool.Create(() => "result", new() { Name = "myTool" })])
|
||||
.WithSubscriptionsListenHandler(async (request, cancellationToken) =>
|
||||
{
|
||||
var subscriptionId = request.JsonRpcRequest.Id;
|
||||
|
||||
var ack = new JsonRpcNotification
|
||||
{
|
||||
Method = NotificationMethods.SubscriptionsAcknowledgedNotification,
|
||||
Params = JsonSerializer.SerializeToNode(
|
||||
new SubscriptionsAcknowledgedNotificationParams
|
||||
{
|
||||
Notifications = new SubscriptionsListenNotifications
|
||||
{
|
||||
ToolsListChanged = request.Params.Notifications.ToolsListChanged,
|
||||
},
|
||||
},
|
||||
McpJsonUtilities.DefaultOptions),
|
||||
};
|
||||
TagWithSubscriptionId(ack, subscriptionId);
|
||||
await request.Server.SendMessageAsync(ack, cancellationToken);
|
||||
|
||||
var listChanged = new JsonRpcNotification { Method = NotificationMethods.ToolListChangedNotification };
|
||||
TagWithSubscriptionId(listChanged, subscriptionId);
|
||||
await request.Server.SendMessageAsync(listChanged, cancellationToken);
|
||||
|
||||
return new EmptyResult();
|
||||
});
|
||||
|
||||
// Advertise tools.listChanged so the per-response capability decision has something to preserve.
|
||||
Builder.Services.Configure<McpServerOptions>(options =>
|
||||
{
|
||||
options.Capabilities ??= new();
|
||||
options.Capabilities.Tools ??= new();
|
||||
options.Capabilities.Tools.ListChanged = true;
|
||||
});
|
||||
|
||||
_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();
|
||||
|
||||
// The stateless server can now deliver tools/list_changed over the custom listen stream, so the
|
||||
// capability must survive on the server/discover response instead of being cleared.
|
||||
Assert.True(client.ServerCapabilities.Tools?.ListChanged);
|
||||
|
||||
var listChangedChannel = Channel.CreateUnbounded<JsonRpcNotification>();
|
||||
await using var listChangedReg = client.RegisterNotificationHandler(NotificationMethods.ToolListChangedNotification,
|
||||
(notification, _) => { listChangedChannel.Writer.TryWrite(notification); return default; });
|
||||
|
||||
var listenRequest = new JsonRpcRequest
|
||||
{
|
||||
Method = RequestMethods.SubscriptionsListen,
|
||||
Params = JsonSerializer.SerializeToNode(
|
||||
new SubscriptionsListenRequestParams
|
||||
{
|
||||
Notifications = new SubscriptionsListenNotifications { ToolsListChanged = true },
|
||||
},
|
||||
McpJsonUtilities.DefaultOptions),
|
||||
};
|
||||
|
||||
await client.SendRequestAsync(listenRequest, TestContext.Current.CancellationToken)
|
||||
.WaitAsync(TestConstants.DefaultTimeout, TestContext.Current.CancellationToken);
|
||||
|
||||
var listChangedNotification = await listChangedChannel.Reader.ReadAsync(TestContext.Current.CancellationToken);
|
||||
Assert.NotNull(GetSubscriptionId(listChangedNotification));
|
||||
}
|
||||
|
||||
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)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -147,4 +147,17 @@ public class McpServerBuilderExtensionsHandlerTests
|
|||
|
||||
Assert.Equal(handler, options.Handlers.UnsubscribeFromResourcesHandler);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void WithSubscriptionsListenHandler_Sets_Handler()
|
||||
{
|
||||
McpRequestHandler<SubscriptionsListenRequestParams, EmptyResult> handler = async (context, token) => new EmptyResult();
|
||||
|
||||
_builder.Object.WithSubscriptionsListenHandler(handler);
|
||||
|
||||
var serviceProvider = _services.BuildServiceProvider();
|
||||
var options = serviceProvider.GetRequiredService<IOptions<McpServerOptions>>().Value;
|
||||
|
||||
Assert.Equal(handler, options.Handlers.SubscriptionsListenHandler);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,278 @@
|
|||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Options;
|
||||
using ModelContextProtocol.Client;
|
||||
using ModelContextProtocol.Protocol;
|
||||
using ModelContextProtocol.Server;
|
||||
using ModelContextProtocol.Tests.Utils;
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Nodes;
|
||||
using System.Threading.Channels;
|
||||
|
||||
namespace ModelContextProtocol.Tests.Server;
|
||||
|
||||
/// <summary>
|
||||
/// Tests for the custom <see cref="McpServerHandlers.SubscriptionsListenHandler"/> (SEP-2575, issue #1662).
|
||||
/// A custom handler is a full replacement for the built-in <c>subscriptions/listen</c> handling: it owns the
|
||||
/// stream, sends the acknowledgement itself, streams application-defined notifications, and receives no
|
||||
/// automatic <c>*/list_changed</c> fan-out. These tests run over the in-memory stream transport exercised by
|
||||
/// <see cref="ClientServerTestBase"/>.
|
||||
/// </summary>
|
||||
public class SubscriptionsListenHandlerTests : ClientServerTestBase
|
||||
{
|
||||
private const string CustomResourceUri = "custom://event/1";
|
||||
|
||||
// Signalled after the custom handler has sent its acknowledgement and first notification and is waiting
|
||||
// for cancellation. Lets cancellation tests wait until the handler is actually holding the stream open.
|
||||
private readonly TaskCompletionSource<bool> _handlerHoldingStream = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
|
||||
// Signalled from the custom handler's finally block, proving it observed cancellation and cleaned up.
|
||||
private readonly TaskCompletionSource<bool> _handlerCleanedUp = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
|
||||
public SubscriptionsListenHandlerTests(ITestOutputHelper testOutputHelper)
|
||||
: base(testOutputHelper)
|
||||
{
|
||||
}
|
||||
|
||||
protected override void ConfigureServices(ServiceCollection services, IMcpServerBuilder mcpServerBuilder)
|
||||
{
|
||||
// Register a tool so the server advertises tools.listChanged; this lets the replacement test below
|
||||
// trigger a collection change and prove the built-in fan-out no longer runs for the custom handler.
|
||||
mcpServerBuilder.WithTools<ListenTools>();
|
||||
|
||||
mcpServerBuilder.WithSubscriptionsListenHandler(async (request, cancellationToken) =>
|
||||
{
|
||||
var subscriptionId = request.JsonRpcRequest.Id;
|
||||
|
||||
// SEP-2575 requires the acknowledgement to be the first message on the stream. The custom handler
|
||||
// owns this: it echoes back the requested filters as granted and tags the ack with the id.
|
||||
var ack = new JsonRpcNotification
|
||||
{
|
||||
Method = NotificationMethods.SubscriptionsAcknowledgedNotification,
|
||||
Params = JsonSerializer.SerializeToNode(
|
||||
new SubscriptionsAcknowledgedNotificationParams { Notifications = request.Params.Notifications },
|
||||
McpJsonUtilities.DefaultOptions),
|
||||
};
|
||||
TagWithSubscriptionId(ack, subscriptionId);
|
||||
await request.Server.SendMessageAsync(ack, cancellationToken);
|
||||
|
||||
// Stream one application-defined notification the built-in handler could never produce on its own,
|
||||
// proving the handler drives the stream. Tagged with the same subscription id.
|
||||
var updated = new JsonRpcNotification
|
||||
{
|
||||
Method = NotificationMethods.ResourceUpdatedNotification,
|
||||
Params = new JsonObject { ["uri"] = CustomResourceUri },
|
||||
};
|
||||
TagWithSubscriptionId(updated, subscriptionId);
|
||||
await request.Server.SendMessageAsync(updated, cancellationToken);
|
||||
|
||||
_handlerHoldingStream.TrySetResult(true);
|
||||
|
||||
try
|
||||
{
|
||||
// Remain active for the subscription lifetime, exactly like the built-in handler, until the
|
||||
// request-scoped token is cancelled (notifications/cancelled on stdio, client disconnect on HTTP).
|
||||
var cancelled = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
using var registration = cancellationToken.Register(
|
||||
static state => ((TaskCompletionSource<bool>)state!).TrySetResult(true), cancelled);
|
||||
await cancelled.Task;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_handlerCleanedUp.TrySetResult(true);
|
||||
}
|
||||
|
||||
return new EmptyResult();
|
||||
});
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task CustomHandler_July2026_SendsAcknowledgementThenTaggedNotification()
|
||||
{
|
||||
await using McpClient client = await CreateMcpClientForServer(new McpClientOptions
|
||||
{
|
||||
ProtocolVersion = McpProtocolVersions.July2026ProtocolVersion,
|
||||
});
|
||||
|
||||
// Capture the acknowledgement and the streamed notification on separate channels. Both must arrive
|
||||
// tagged with the subscription id. The test does not assert cross-notification arrival order because
|
||||
// the client dispatches incoming messages concurrently (see McpSessionHandler.ProcessMessagesCoreAsync),
|
||||
// so handler invocation order is not observable; acknowledgement-first is a server-side/wire guarantee.
|
||||
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; });
|
||||
|
||||
using var listenCts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
|
||||
var listenTask = SendSubscriptionsListenAsync(
|
||||
client, new SubscriptionsListenNotifications { ResourcesListChanged = true }, listenCts.Token);
|
||||
|
||||
// The acknowledgement carries the subscription id.
|
||||
var ack = await ackChannel.Reader.ReadAsync(TestContext.Current.CancellationToken);
|
||||
var subscriptionId = GetSubscriptionId(ack);
|
||||
Assert.NotNull(subscriptionId);
|
||||
|
||||
// The custom application notification is tagged with the same subscription id.
|
||||
var updated = await updatedChannel.Reader.ReadAsync(TestContext.Current.CancellationToken);
|
||||
Assert.Equal(subscriptionId, GetSubscriptionId(updated));
|
||||
Assert.Equal(CustomResourceUri, (updated.Params as JsonObject)?["uri"]?.GetValue<string>());
|
||||
|
||||
await CancelSubscriptionAsync(listenCts, listenTask);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task CustomHandler_ReplacesBuiltIn_SuppressesAutomaticListChangedFanOut()
|
||||
{
|
||||
await using McpClient client = await CreateMcpClientForServer(new McpClientOptions
|
||||
{
|
||||
ProtocolVersion = McpProtocolVersions.July2026ProtocolVersion,
|
||||
});
|
||||
|
||||
var ackChannel = Channel.CreateUnbounded<JsonRpcNotification>();
|
||||
var toolsChannel = Channel.CreateUnbounded<JsonRpcNotification>();
|
||||
|
||||
await using var ackReg = client.RegisterNotificationHandler(NotificationMethods.SubscriptionsAcknowledgedNotification,
|
||||
(notification, _) => { ackChannel.Writer.TryWrite(notification); return default; });
|
||||
await using var toolsReg = client.RegisterNotificationHandler(NotificationMethods.ToolListChangedNotification,
|
||||
(notification, _) => { toolsChannel.Writer.TryWrite(notification); return default; });
|
||||
|
||||
using var listenCts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
|
||||
// Request tools/list_changed. The built-in handler would fan these out; the custom handler replaces it
|
||||
// and never tracks the subscription, so the SDK must deliver nothing automatically for this change.
|
||||
var listenTask = SendSubscriptionsListenAsync(
|
||||
client, new SubscriptionsListenNotifications { ToolsListChanged = true }, listenCts.Token);
|
||||
|
||||
// Wait until the custom handler is holding the stream open (ack + first notification already sent).
|
||||
await _handlerHoldingStream.Task.WaitAsync(TestContext.Current.CancellationToken);
|
||||
await ackChannel.Reader.ReadAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
// Mutate the tool collection. With the built-in handler this would deliver a tagged tools/list_changed;
|
||||
// under the custom replacement handler it must not, because _activeSubscriptions is never populated.
|
||||
// The list-changed fan-out iterates that (empty) set and completes synchronously during Add, so it
|
||||
// buffers nothing to send. The ListToolsAsync round-trip then flushes the client read pipeline.
|
||||
var serverOptions = ServiceProvider.GetRequiredService<IOptions<McpServerOptions>>().Value;
|
||||
serverOptions.ToolCollection!.Add(McpServerTool.Create([McpServerTool(Name = "AddedTool")] () => "42"));
|
||||
|
||||
await client.ListToolsAsync(cancellationToken: TestContext.Current.CancellationToken);
|
||||
|
||||
await CancelSubscriptionAsync(listenCts, listenTask);
|
||||
|
||||
// Completed-and-empty proves nothing was ever delivered, not merely "nothing buffered right now":
|
||||
// WaitToReadAsync returns false only when the channel is both empty and completed.
|
||||
toolsChannel.Writer.Complete();
|
||||
Assert.False(await toolsChannel.Reader.WaitToReadAsync(TestContext.Current.CancellationToken));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task CustomHandler_PreJuly2026_IsRejectedWithMethodNotFound()
|
||||
{
|
||||
await using McpClient client = await CreateMcpClientForServer(new McpClientOptions
|
||||
{
|
||||
ProtocolVersion = McpProtocolVersions.November2025ProtocolVersion,
|
||||
});
|
||||
|
||||
var request = new JsonRpcRequest
|
||||
{
|
||||
Method = RequestMethods.SubscriptionsListen,
|
||||
Params = JsonSerializer.SerializeToNode(
|
||||
new SubscriptionsListenRequestParams { Notifications = new SubscriptionsListenNotifications { ResourcesListChanged = true } },
|
||||
McpJsonUtilities.DefaultOptions),
|
||||
};
|
||||
|
||||
var ex = await Assert.ThrowsAsync<McpProtocolException>(() =>
|
||||
client.SendRequestAsync(request, TestContext.Current.CancellationToken));
|
||||
|
||||
Assert.Equal(McpErrorCode.MethodNotFound, ex.ErrorCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task CustomHandler_OnCancellation_ObservesTokenAndCleansUp()
|
||||
{
|
||||
await using McpClient client = await CreateMcpClientForServer(new McpClientOptions
|
||||
{
|
||||
ProtocolVersion = McpProtocolVersions.July2026ProtocolVersion,
|
||||
});
|
||||
|
||||
// Use an explicit request id so the test can address it in a notifications/cancelled message,
|
||||
// which is the transport-level cancellation signal for a long-lived subscriptions/listen on stdio
|
||||
// (an HTTP client would instead disconnect, which cancels the same request-scoped token).
|
||||
var subscriptionId = new RequestId("listen-cancel-1");
|
||||
using var listenCts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
|
||||
var listenTask = SendSubscriptionsListenAsync(
|
||||
client, new SubscriptionsListenNotifications { ResourcesListChanged = true }, listenCts.Token, subscriptionId);
|
||||
|
||||
// Ensure the handler is actively holding the stream before cancelling.
|
||||
await _handlerHoldingStream.Task.WaitAsync(TestContext.Current.CancellationToken);
|
||||
|
||||
// Deterministically drive server-side cancellation by sending notifications/cancelled for the request
|
||||
// id. The server routes this to the request-scoped token the handler is observing.
|
||||
await client.SendMessageAsync(new JsonRpcNotification
|
||||
{
|
||||
Method = NotificationMethods.CancelledNotification,
|
||||
Params = JsonSerializer.SerializeToNode(
|
||||
new CancelledNotificationParams { RequestId = subscriptionId },
|
||||
McpJsonUtilities.DefaultOptions),
|
||||
}, TestContext.Current.CancellationToken);
|
||||
|
||||
// The handler observed the token: its cancellation registration ran and its finally block completed.
|
||||
await _handlerCleanedUp.Task.WaitAsync(TestConstants.DefaultTimeout, TestContext.Current.CancellationToken);
|
||||
|
||||
// A cancelled request sends no response, so unblock the client side and confirm it observes cancellation.
|
||||
await CancelSubscriptionAsync(listenCts, listenTask);
|
||||
}
|
||||
|
||||
private static Task SendSubscriptionsListenAsync(
|
||||
McpClient client, SubscriptionsListenNotifications notifications, CancellationToken cancellationToken, RequestId requestId = default)
|
||||
{
|
||||
var request = new JsonRpcRequest
|
||||
{
|
||||
Method = RequestMethods.SubscriptionsListen,
|
||||
Params = JsonSerializer.SerializeToNode(
|
||||
new SubscriptionsListenRequestParams { Notifications = notifications },
|
||||
McpJsonUtilities.DefaultOptions),
|
||||
};
|
||||
|
||||
if (requestId.Id is not null)
|
||||
{
|
||||
request.Id = requestId;
|
||||
}
|
||||
|
||||
return client.SendRequestAsync(request, cancellationToken);
|
||||
}
|
||||
|
||||
private static async Task CancelSubscriptionAsync(CancellationTokenSource listenCts, Task listenTask)
|
||||
{
|
||||
await listenCts.CancelAsync();
|
||||
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => listenTask);
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
private sealed class ListenTools
|
||||
{
|
||||
[McpServerTool, System.ComponentModel.Description("Echoes the input back to the caller.")]
|
||||
public static string Echo([System.ComponentModel.Description("The message to echo.")] string message) => message;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue