From 34a6f1c665a48f45a8c9727ef96608ba03a21f88 Mon Sep 17 00:00:00 2001 From: Peder Holdgaard Pedersen <127606677+PederHP@users.noreply.github.com> Date: Mon, 17 Mar 2025 17:54:29 +0100 Subject: [PATCH] Fix resource subscription handling (#89) * Fix resource subscription handling Add resource subscription tests * Fix typo * Typo * small fixes * Fix concurrency issue with test * Align with specification schemas --- src/mcpdotnet/Client/McpClientExtensions.cs | 4 +- .../McpServerBuilderExtensions.Handler.cs | 4 +- src/mcpdotnet/Protocol/Types/Capabilities.cs | 4 +- .../Protocol/Types/CompleteResult.cs | 16 +++++ src/mcpdotnet/Protocol/Types/EmptyResult.cs | 10 +++ src/mcpdotnet/Protocol/Types/Reference.cs | 12 ---- .../ResourceUpdatedNotificationParams.cs | 14 ++++ ...estParams.cs => SubscribeRequestParams.cs} | 4 +- .../Types/UnsubscribeRequestParams.cs | 14 ++++ src/mcpdotnet/Server/McpServer.cs | 3 +- src/mcpdotnet/Server/McpServerHandlers.cs | 22 +++--- tests/mcpdotnet.TestServer/Program.cs | 69 ++++++++++++++++++- .../mcpdotnet.Tests/ClientIntegrationTests.cs | 62 +++++++++++++++++ .../McpServerBuilderExtensionsHandlerTests.cs | 4 +- .../Server/McpServerDelegatesTests.cs | 4 +- 15 files changed, 210 insertions(+), 36 deletions(-) create mode 100644 src/mcpdotnet/Protocol/Types/CompleteResult.cs create mode 100644 src/mcpdotnet/Protocol/Types/EmptyResult.cs create mode 100644 src/mcpdotnet/Protocol/Types/ResourceUpdatedNotificationParams.cs rename src/mcpdotnet/Protocol/Types/{SubscribeToResourceRequestParams.cs => SubscribeRequestParams.cs} (73%) create mode 100644 src/mcpdotnet/Protocol/Types/UnsubscribeRequestParams.cs diff --git a/src/mcpdotnet/Client/McpClientExtensions.cs b/src/mcpdotnet/Client/McpClientExtensions.cs index 7e4703c..d10fd4c 100644 --- a/src/mcpdotnet/Client/McpClientExtensions.cs +++ b/src/mcpdotnet/Client/McpClientExtensions.cs @@ -227,7 +227,7 @@ public static class McpClientExtensions { Throw.IfNull(client); - return client.SendRequestAsync( + return client.SendRequestAsync( CreateRequest("resources/subscribe", new() { ["uri"] = uri }), cancellationToken); } @@ -242,7 +242,7 @@ public static class McpClientExtensions { Throw.IfNull(client); - return client.SendRequestAsync( + return client.SendRequestAsync( CreateRequest("resources/unsubscribe", new() { ["uri"] = uri }), cancellationToken); } diff --git a/src/mcpdotnet/Configuration/McpServerBuilderExtensions.Handler.cs b/src/mcpdotnet/Configuration/McpServerBuilderExtensions.Handler.cs index 9ffcf9a..49ea259 100644 --- a/src/mcpdotnet/Configuration/McpServerBuilderExtensions.Handler.cs +++ b/src/mcpdotnet/Configuration/McpServerBuilderExtensions.Handler.cs @@ -107,7 +107,7 @@ public static partial class McpServerBuilderExtensions /// /// The builder instance. /// The handler. - public static IMcpServerBuilder WithSubscribeToResourcesHandler(this IMcpServerBuilder builder, Func, CancellationToken, Task> handler) + public static IMcpServerBuilder WithSubscribeToResourcesHandler(this IMcpServerBuilder builder, Func, CancellationToken, Task> handler) { Throw.IfNull(builder); @@ -120,7 +120,7 @@ public static partial class McpServerBuilderExtensions /// /// The builder instance. /// The handler. - public static IMcpServerBuilder WithUnsubscribeFromResourcesHandler(this IMcpServerBuilder builder, Func, CancellationToken, Task> handler) + public static IMcpServerBuilder WithUnsubscribeFromResourcesHandler(this IMcpServerBuilder builder, Func, CancellationToken, Task> handler) { Throw.IfNull(builder); diff --git a/src/mcpdotnet/Protocol/Types/Capabilities.cs b/src/mcpdotnet/Protocol/Types/Capabilities.cs index b623f29..2513b42 100644 --- a/src/mcpdotnet/Protocol/Types/Capabilities.cs +++ b/src/mcpdotnet/Protocol/Types/Capabilities.cs @@ -126,13 +126,13 @@ public record ResourcesCapability /// Gets or sets the handler for subscribe to resources messages. /// [JsonIgnore] - public Func, CancellationToken, Task>? SubscribeToResourcesHandler { get; init; } + public Func, CancellationToken, Task>? SubscribeToResourcesHandler { get; init; } /// /// Gets or sets the handler for unsubscribe from resources messages. /// [JsonIgnore] - public Func, CancellationToken, Task>? UnsubscribeFromResourcesHandler { get; init; } + public Func, CancellationToken, Task>? UnsubscribeFromResourcesHandler { get; init; } } /// diff --git a/src/mcpdotnet/Protocol/Types/CompleteResult.cs b/src/mcpdotnet/Protocol/Types/CompleteResult.cs new file mode 100644 index 0000000..65f4bac --- /dev/null +++ b/src/mcpdotnet/Protocol/Types/CompleteResult.cs @@ -0,0 +1,16 @@ +using System.Text.Json.Serialization; + +namespace McpDotNet.Protocol.Types; + +/// +/// The server's response to a completion/complete request +/// See the schema for details +/// +public class CompleteResult +{ + /// + /// The completion object containing the completion values. + /// + [JsonPropertyName("completion")] + public Completion Completion { get; set; } = new Completion(); +} diff --git a/src/mcpdotnet/Protocol/Types/EmptyResult.cs b/src/mcpdotnet/Protocol/Types/EmptyResult.cs new file mode 100644 index 0000000..b143031 --- /dev/null +++ b/src/mcpdotnet/Protocol/Types/EmptyResult.cs @@ -0,0 +1,10 @@ +namespace McpDotNet.Protocol.Types; + +/// +/// An empty result object. +/// See the schema for details +/// +public class EmptyResult +{ + +} \ No newline at end of file diff --git a/src/mcpdotnet/Protocol/Types/Reference.cs b/src/mcpdotnet/Protocol/Types/Reference.cs index d313db4..b3cc5e4 100644 --- a/src/mcpdotnet/Protocol/Types/Reference.cs +++ b/src/mcpdotnet/Protocol/Types/Reference.cs @@ -66,15 +66,3 @@ public class Reference return true; } } - -/// -/// The server's response to a completion/complete request -/// -public class CompleteResult -{ - /// - /// The completion object containing the completion values. - /// - [JsonPropertyName("completion")] - public Completion Completion { get; set; } = new Completion(); -} diff --git a/src/mcpdotnet/Protocol/Types/ResourceUpdatedNotificationParams.cs b/src/mcpdotnet/Protocol/Types/ResourceUpdatedNotificationParams.cs new file mode 100644 index 0000000..3d424ed --- /dev/null +++ b/src/mcpdotnet/Protocol/Types/ResourceUpdatedNotificationParams.cs @@ -0,0 +1,14 @@ +namespace McpDotNet.Protocol.Types; + +/// +/// Sent from the server as the payload of "notifications/resources/updated" notifications whenever a subscribed resource changes. +/// See the schema for details +/// +public class ResourceUpdatedNotificationParams +{ + /// + /// The URI of the resource to subscribe to. The URI can use any protocol; it is up to the server how to interpret it. + /// + [System.Text.Json.Serialization.JsonPropertyName("uri")] + public string? Uri { get; init; } +} diff --git a/src/mcpdotnet/Protocol/Types/SubscribeToResourceRequestParams.cs b/src/mcpdotnet/Protocol/Types/SubscribeRequestParams.cs similarity index 73% rename from src/mcpdotnet/Protocol/Types/SubscribeToResourceRequestParams.cs rename to src/mcpdotnet/Protocol/Types/SubscribeRequestParams.cs index 70f4f6c..a6806ee 100644 --- a/src/mcpdotnet/Protocol/Types/SubscribeToResourceRequestParams.cs +++ b/src/mcpdotnet/Protocol/Types/SubscribeRequestParams.cs @@ -1,10 +1,10 @@ namespace McpDotNet.Protocol.Types; /// -/// Sent from the client to request resources/updated notifications from the server whenever a particular resource changes. +/// Sent from the client to request updated notifications from the server whenever a particular primitive changes. /// See the schema for details /// -public class SubscribeToResourceRequestParams +public class SubscribeRequestParams { /// /// The URI of the resource to subscribe to. The URI can use any protocol; it is up to the server how to interpret it. diff --git a/src/mcpdotnet/Protocol/Types/UnsubscribeRequestParams.cs b/src/mcpdotnet/Protocol/Types/UnsubscribeRequestParams.cs new file mode 100644 index 0000000..580a085 --- /dev/null +++ b/src/mcpdotnet/Protocol/Types/UnsubscribeRequestParams.cs @@ -0,0 +1,14 @@ +namespace McpDotNet.Protocol.Types; + +/// +/// Sent from the client to request not receiving updated notifications from the server whenever a primitive resource changes. +/// See the schema for details +/// +public class UnsubscribeRequestParams +{ + /// + /// The URI of the resource to unsubscribe fro. The URI can use any protocol; it is up to the server how to interpret it. + /// + [System.Text.Json.Serialization.JsonPropertyName("uri")] + public string? Uri { get; init; } +} diff --git a/src/mcpdotnet/Server/McpServer.cs b/src/mcpdotnet/Server/McpServer.cs index ddd2be6..e0bd9d7 100644 --- a/src/mcpdotnet/Server/McpServer.cs +++ b/src/mcpdotnet/Server/McpServer.cs @@ -163,7 +163,8 @@ internal sealed class McpServer : McpJsonRpcEndpoint, IMcpServer throw new McpServerException("Resources capability was enabled with subscribe support, but SubscribeToResources and/or UnsubscribeFromResources handlers were not specified."); } - // TODO: Implement Subscribe support + SetRequestHandler("resources/subscribe", request => subscribeHandler(new(this, request), cancellationToken)); + SetRequestHandler("resources/unsubscribe", request => unsubscribeHandler(new(this, request), cancellationToken)); } private void SetPromptsHandler(McpServerOptions options) diff --git a/src/mcpdotnet/Server/McpServerHandlers.cs b/src/mcpdotnet/Server/McpServerHandlers.cs index 507b9b4..a4edc98 100644 --- a/src/mcpdotnet/Server/McpServerHandlers.cs +++ b/src/mcpdotnet/Server/McpServerHandlers.cs @@ -45,12 +45,12 @@ public sealed class McpServerHandlers /// /// Gets or sets the handler for subscribe to resources messages. /// - public Func, CancellationToken, Task>? SubscribeToResourcesHandler { get; set; } + public Func, CancellationToken, Task>? SubscribeToResourcesHandler { get; set; } /// /// Gets or sets the handler for unsubscribe from resources messages. /// - public Func, CancellationToken, Task>? UnsubscribeFromResourcesHandler { get; set; } + public Func, CancellationToken, Task>? UnsubscribeFromResourcesHandler { get; set; } /// /// Overwrite any handlers in McpServerOptions with non-null handlers from this instance. @@ -77,25 +77,29 @@ public sealed class McpServerHandlers ResourcesCapability? resourcesCapability = options.Capabilities?.Resources; if (ListResourcesHandler is not null || - ReadResourceHandler is not null || - SubscribeToResourcesHandler is not null || - UnsubscribeFromResourcesHandler is not null) + ReadResourceHandler is not null) { resourcesCapability = resourcesCapability is null ? new() { ListResourcesHandler = ListResourcesHandler, - ReadResourceHandler = ReadResourceHandler, - SubscribeToResourcesHandler = SubscribeToResourcesHandler, - UnsubscribeFromResourcesHandler = UnsubscribeFromResourcesHandler, + ReadResourceHandler = ReadResourceHandler } : resourcesCapability with { ListResourcesHandler = ListResourcesHandler ?? resourcesCapability.ListResourcesHandler, - ReadResourceHandler = ReadResourceHandler ?? resourcesCapability.ReadResourceHandler, + ReadResourceHandler = ReadResourceHandler ?? resourcesCapability.ReadResourceHandler + }; + + if (SubscribeToResourcesHandler is not null || UnsubscribeFromResourcesHandler is not null) + { + resourcesCapability = resourcesCapability with + { SubscribeToResourcesHandler = SubscribeToResourcesHandler ?? resourcesCapability.SubscribeToResourcesHandler, UnsubscribeFromResourcesHandler = UnsubscribeFromResourcesHandler ?? resourcesCapability.UnsubscribeFromResourcesHandler, + Subscribe = true }; + } } ToolsCapability? toolsCapability = options.Capabilities?.Tools; diff --git a/tests/mcpdotnet.TestServer/Program.cs b/tests/mcpdotnet.TestServer/Program.cs index bc6bd1b..a45a956 100644 --- a/tests/mcpdotnet.TestServer/Program.cs +++ b/tests/mcpdotnet.TestServer/Program.cs @@ -1,4 +1,6 @@ using System.Text; +using System.Text.Json; +using McpDotNet.Protocol.Messages; using McpDotNet.Protocol.Transport; using McpDotNet.Protocol.Types; using McpDotNet.Server; @@ -53,7 +55,27 @@ internal static class Program Log.Logger.Information("Server started."); // Run until process is stopped by the client (parent process) - await Task.Delay(Timeout.Infinite); + while (true) + { + await Task.Delay(5000); + + // Snapshot the subscribed resources, rather than locking while sending notifications + List resources; + lock (_subscribedResourcesLock) + { + resources = _subscribedResources.ToList(); + } + + foreach (var resource in resources) + { + ResourceUpdatedNotificationParams notificationParams = new() { Uri = resource }; + await server.SendMessageAsync(new JsonRpcNotification() + { + Method = NotificationMethods.ResourceUpdatedNotification, + Params = notificationParams + }); + } + } } private static ToolsCapability ConfigureTools() @@ -233,6 +255,9 @@ internal static class Program }; } + private static HashSet _subscribedResources = new(); + private static readonly object _subscribedResourcesLock = new(); + private static ResourcesCapability ConfigureResources() { List resources = []; @@ -320,7 +345,47 @@ internal static class Program { Contents = [contents] }); - } + }, + + SubscribeToResourcesHandler = (request, cancellationToken) => + { + if (request?.Params?.Uri is null) + { + throw new McpServerException("Missing required argument 'uri'"); + } + if (!request.Params.Uri.StartsWith("test://static/resource/")) + { + throw new McpServerException("Invalid resource URI"); + } + + lock (_subscribedResourcesLock) + { + _subscribedResources.Add(request.Params.Uri); + } + + return Task.FromResult(new EmptyResult()); + }, + + UnsubscribeFromResourcesHandler = (request, cancellationToken) => + { + if (request?.Params?.Uri is null) + { + throw new McpServerException("Missing required argument 'uri'"); + } + if (!request.Params.Uri.StartsWith("test://static/resource/")) + { + throw new McpServerException("Invalid resource URI"); + } + + lock (_subscribedResourcesLock) + { + _subscribedResources.Remove(request.Params.Uri); + } + + return Task.FromResult(new EmptyResult()); + }, + + Subscribe = true }; } diff --git a/tests/mcpdotnet.Tests/ClientIntegrationTests.cs b/tests/mcpdotnet.Tests/ClientIntegrationTests.cs index 7e293e5..c115187 100644 --- a/tests/mcpdotnet.Tests/ClientIntegrationTests.cs +++ b/tests/mcpdotnet.Tests/ClientIntegrationTests.cs @@ -3,6 +3,7 @@ using McpDotNet.Configuration; using McpDotNet.Protocol.Messages; using McpDotNet.Protocol.Transport; using McpDotNet.Protocol.Types; +using System.Text.Json; namespace McpDotNet.Tests; @@ -208,6 +209,67 @@ public class ClientIntegrationTests : IClassFixture + { + var notificationParams = JsonSerializer.Deserialize(notification.Params!.ToString() ?? string.Empty); + ++counter; + return Task.CompletedTask; + }); + await client.SubscribeToResourceAsync("test://static/resource/1", CancellationToken.None); + + // notifications happen every 5 seconds, so we wait for 10 seconds to ensure we get at least one notification + await Task.Delay(10000); + + // assert + Assert.True(counter > 0); + } + + // Not supported by "everything" server version on npx + [Fact] + public async Task UnsubscribeResource_Stdio() + { + // arrange + var clientId = "test_server"; + + // act + int counter = 0; + await using var client = await _fixture.CreateClientAsync(clientId); + client.AddNotificationHandler(NotificationMethods.ResourceUpdatedNotification, (notification) => + { + var notificationParams = JsonSerializer.Deserialize(notification.Params!.ToString() ?? string.Empty); + ++counter; + return Task.CompletedTask; + }); + await client.SubscribeToResourceAsync("test://static/resource/1", CancellationToken.None); + + // notifications happen every 5 seconds, so we wait for 10 seconds to ensure we get at least one notification + await Task.Delay(10000); + + // reset counter + int counterAfterSubscribe = counter; + + // unsubscribe + await client.UnsubscribeFromResourceAsync("test://static/resource/1", CancellationToken.None); + counter = 0; + + // notifications happen every 5 seconds, so we wait for 10 seconds to ensure we would've gotten at least one notification + await Task.Delay(10000); + + // assert + Assert.True(counterAfterSubscribe > 0); + Assert.Equal(0, counter); + } + [Theory] [MemberData(nameof(GetClients))] public async Task GetCompletion_Stdio_ResourceReference(string clientId) diff --git a/tests/mcpdotnet.Tests/Configuration/McpServerBuilderExtensionsHandlerTests.cs b/tests/mcpdotnet.Tests/Configuration/McpServerBuilderExtensionsHandlerTests.cs index eeafa67..0b288d2 100644 --- a/tests/mcpdotnet.Tests/Configuration/McpServerBuilderExtensionsHandlerTests.cs +++ b/tests/mcpdotnet.Tests/Configuration/McpServerBuilderExtensionsHandlerTests.cs @@ -113,7 +113,7 @@ public class McpServerBuilderExtensionsHandlerTests [Fact] public void WithSubscribeToResourcesHandler_Sets_Handler() { - Func, CancellationToken, Task> handler = (context, token) => Task.CompletedTask; + Func, CancellationToken, Task> handler = (context, token) => Task.FromResult(new EmptyResult()); _builder.Object.WithSubscribeToResourcesHandler(handler); @@ -126,7 +126,7 @@ public class McpServerBuilderExtensionsHandlerTests [Fact] public void WithUnsubscribeFromResourcesHandler_Sets_Handler() { - Func, CancellationToken, Task> handler = (context, token) => Task.CompletedTask; + Func, CancellationToken, Task> handler = (context, token) => Task.FromResult(new EmptyResult()); _builder.Object.WithUnsubscribeFromResourcesHandler(handler); diff --git a/tests/mcpdotnet.Tests/Server/McpServerDelegatesTests.cs b/tests/mcpdotnet.Tests/Server/McpServerDelegatesTests.cs index aa8d37b..1490fad 100644 --- a/tests/mcpdotnet.Tests/Server/McpServerDelegatesTests.cs +++ b/tests/mcpdotnet.Tests/Server/McpServerDelegatesTests.cs @@ -27,8 +27,8 @@ public class McpServerHandlerTests handlers.ListResourcesHandler = (p, c) => Task.FromResult(new ListResourcesResult()); handlers.ReadResourceHandler = (p, c) => Task.FromResult(new ReadResourceResult()); handlers.GetCompletionHandler = (p, c) => Task.FromResult(new CompleteResult()); - handlers.SubscribeToResourcesHandler = (s, c) => Task.CompletedTask; - handlers.UnsubscribeFromResourcesHandler = (s, c) => Task.CompletedTask; + handlers.SubscribeToResourcesHandler = (s, c) => Task.FromResult(new EmptyResult()); + handlers.UnsubscribeFromResourcesHandler = (s, c) => Task.FromResult(new EmptyResult()); Assert.NotNull(handlers.ListToolsHandler); Assert.NotNull(handlers.CallToolHandler);