Fix resource subscription handling (#89)

PederHP · web-flow · commit 34a6f1c665a4 · 2025-03-17T17:54:29.000+01:00
* Fix resource subscription handling
Add resource subscription tests

* Fix typo

* Typo

* small fixes

* Fix concurrency issue with test

* Align with specification schemas
diff --git a/src/mcpdotnet/Client/McpClientExtensions.cs b/src/mcpdotnet/Client/McpClientExtensions.cs
@@ -227,7 +227,7 @@ public static Task SubscribeToResourceAsync(this IMcpClient client, string uri,
     {
         Throw.IfNull(client);
 
-        return client.SendRequestAsync<dynamic>(
+        return client.SendRequestAsync<EmptyResult>(
             CreateRequest("resources/subscribe", new() { ["uri"] = uri }),
             cancellationToken);
     }
@@ -242,7 +242,7 @@ public static Task UnsubscribeFromResourceAsync(this IMcpClient client, string u
     {
         Throw.IfNull(client);
 
-        return client.SendRequestAsync<dynamic>(
+        return client.SendRequestAsync<EmptyResult>(
             CreateRequest("resources/unsubscribe", new() { ["uri"] = uri }),
             cancellationToken);
     }
diff --git a/src/mcpdotnet/Configuration/McpServerBuilderExtensions.Handler.cs b/src/mcpdotnet/Configuration/McpServerBuilderExtensions.Handler.cs
@@ -107,7 +107,7 @@ public static IMcpServerBuilder WithGetCompletionHandler(this IMcpServerBuilder
     /// </summary>
     /// <param name="builder">The builder instance.</param>
     /// <param name="handler">The handler.</param>
-    public static IMcpServerBuilder WithSubscribeToResourcesHandler(this IMcpServerBuilder builder, Func<RequestContext<string>, CancellationToken, Task> handler)
+    public static IMcpServerBuilder WithSubscribeToResourcesHandler(this IMcpServerBuilder builder, Func<RequestContext<SubscribeRequestParams>, CancellationToken, Task<EmptyResult>> handler)
     {
         Throw.IfNull(builder);
 
@@ -120,7 +120,7 @@ public static IMcpServerBuilder WithSubscribeToResourcesHandler(this IMcpServerB
     /// </summary>
     /// <param name="builder">The builder instance.</param>
     /// <param name="handler">The handler.</param>
-    public static IMcpServerBuilder WithUnsubscribeFromResourcesHandler(this IMcpServerBuilder builder, Func<RequestContext<string>, CancellationToken, Task> handler)
+    public static IMcpServerBuilder WithUnsubscribeFromResourcesHandler(this IMcpServerBuilder builder, Func<RequestContext<UnsubscribeRequestParams>, CancellationToken, Task<EmptyResult>> handler)
     {
         Throw.IfNull(builder);
 
diff --git 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.
     /// </summary>
     [JsonIgnore]
-    public Func<RequestContext<string>, CancellationToken, Task>? SubscribeToResourcesHandler { get; init; }
+    public Func<RequestContext<SubscribeRequestParams>, CancellationToken, Task<EmptyResult>>? SubscribeToResourcesHandler { get; init; }
 
     /// <summary>
     /// Gets or sets the handler for unsubscribe from resources messages.
     /// </summary>
     [JsonIgnore]
-    public Func<RequestContext<string>, CancellationToken, Task>? UnsubscribeFromResourcesHandler { get; init; }
+    public Func<RequestContext<UnsubscribeRequestParams>, CancellationToken, Task<EmptyResult>>? UnsubscribeFromResourcesHandler { get; init; }
 }
 
 /// <summary>
diff --git a/src/mcpdotnet/Protocol/Types/CompleteResult.cs b/src/mcpdotnet/Protocol/Types/CompleteResult.cs
@@ -0,0 +1,16 @@
+﻿using System.Text.Json.Serialization;
+
+namespace McpDotNet.Protocol.Types;
+
+/// <summary>
+/// The server's response to a completion/complete request
+/// <see href="https://github.com/modelcontextprotocol/specification/blob/main/schema/2024-11-05/schema.json">See the schema for details</see>
+/// </summary>
+public class CompleteResult
+{
+    /// <summary>
+    /// The completion object containing the completion values.
+    /// </summary>
+    [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
@@ -0,0 +1,10 @@
+﻿namespace McpDotNet.Protocol.Types;
+
+/// <summary>
+/// An empty result object.
+/// <see href="https://github.com/modelcontextprotocol/specification/blob/main/schema/2024-11-05/schema.json">See the schema for details</see>
+/// </summary>
+public class EmptyResult
+{
+
+}
diff --git a/src/mcpdotnet/Protocol/Types/Reference.cs b/src/mcpdotnet/Protocol/Types/Reference.cs
@@ -66,15 +66,3 @@ public bool Validate([NotNullWhen(false)] out string? validationMessage)
         return true;
     }
 }
-
-/// <summary>
-/// The server's response to a completion/complete request
-/// </summary>
-public class CompleteResult
-{
-    /// <summary>
-    /// The completion object containing the completion values.
-    /// </summary>
-    [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
@@ -0,0 +1,14 @@
+﻿namespace McpDotNet.Protocol.Types;
+
+/// <summary>
+/// Sent from the server as the payload of "notifications/resources/updated" notifications whenever a subscribed resource changes.
+/// <see href="https://github.com/modelcontextprotocol/specification/blob/main/schema/2024-11-05/schema.json">See the schema for details</see>
+/// </summary>
+public class ResourceUpdatedNotificationParams
+{
+    /// <summary>
+    /// The URI of the resource to subscribe to. The URI can use any protocol; it is up to the server how to interpret it.
+    /// </summary>
+    [System.Text.Json.Serialization.JsonPropertyName("uri")]
+    public string? Uri { get; init; }
+}
diff --git a/src/mcpdotnet/Protocol/Types/SubscribeRequestParams.cs b/src/mcpdotnet/Protocol/Types/SubscribeRequestParams.cs
@@ -1,10 +1,10 @@
 ﻿namespace McpDotNet.Protocol.Types;
 
 /// <summary>
-/// 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 href="https://github.com/modelcontextprotocol/specification/blob/main/schema/2024-11-05/schema.json">See the schema for details</see>
 /// </summary>
-public class SubscribeToResourceRequestParams
+public class SubscribeRequestParams
 {
     /// <summary>
     /// 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
@@ -0,0 +1,14 @@
+﻿namespace McpDotNet.Protocol.Types;
+
+/// <summary>
+/// Sent from the client to request not receiving updated notifications from the server whenever a primitive resource changes.
+/// <see href="https://github.com/modelcontextprotocol/specification/blob/main/schema/2024-11-05/schema.json">See the schema for details</see>
+/// </summary>
+public class UnsubscribeRequestParams
+{
+    /// <summary>
+    /// The URI of the resource to unsubscribe fro. The URI can use any protocol; it is up to the server how to interpret it.
+    /// </summary>
+    [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
@@ -163,7 +163,8 @@ private void SetResourcesHandler(McpServerOptions options)
             throw new McpServerException("Resources capability was enabled with subscribe support, but SubscribeToResources and/or UnsubscribeFromResources handlers were not specified.");
         }
 
-        // TODO: Implement Subscribe support
+        SetRequestHandler<SubscribeRequestParams, EmptyResult>("resources/subscribe", request => subscribeHandler(new(this, request), cancellationToken));
+        SetRequestHandler<UnsubscribeRequestParams, EmptyResult>("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
@@ -45,12 +45,12 @@ public sealed class McpServerHandlers
     /// <summary>
     /// Gets or sets the handler for subscribe to resources messages.
     /// </summary>
-    public Func<RequestContext<string>, CancellationToken, Task>? SubscribeToResourcesHandler { get; set; }
+    public Func<RequestContext<SubscribeRequestParams>, CancellationToken, Task<EmptyResult>>? SubscribeToResourcesHandler { get; set; }
 
     /// <summary>
     /// Gets or sets the handler for unsubscribe from resources messages.
     /// </summary>
-    public Func<RequestContext<string>, CancellationToken, Task>? UnsubscribeFromResourcesHandler { get; set; }
+    public Func<RequestContext<UnsubscribeRequestParams>, CancellationToken, Task<EmptyResult>>? UnsubscribeFromResourcesHandler { get; set; }
 
     /// <summary>
     /// Overwrite any handlers in McpServerOptions with non-null handlers from this instance.
@@ -77,25 +77,29 @@ promptsCapability with
 
         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
@@ -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 @@ private static async Task Main(string[] args)
         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<string> 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 @@ private static PromptsCapability ConfigurePrompts()
         };
     }
 
+    private static HashSet<string> _subscribedResources = new();
+    private static readonly object _subscribedResourcesLock = new();
+
     private static ResourcesCapability ConfigureResources()
     {
         List<Resource> resources = [];
@@ -320,7 +345,47 @@ private static ResourcesCapability ConfigureResources()
                 {
                     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
@@ -3,6 +3,7 @@
 using McpDotNet.Protocol.Messages;
 using McpDotNet.Protocol.Transport;
 using McpDotNet.Protocol.Types;
+using System.Text.Json;
 
 namespace McpDotNet.Tests;
 
@@ -208,6 +209,67 @@ public async Task ReadResource_Stdio_BinaryResource(string clientId)
         Assert.NotNull(result.Contents[0].Blob);
     }
 
+    // Not supported by "everything" server version on npx
+    [Fact]
+    public async Task SubscribeResource_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<ResourceUpdatedNotificationParams>(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<ResourceUpdatedNotificationParams>(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
@@ -113,7 +113,7 @@ public void WithGetCompletionHandler_Sets_Handler()
     [Fact]
     public void WithSubscribeToResourcesHandler_Sets_Handler()
     {
-        Func<RequestContext<string>, CancellationToken, Task> handler = (context, token) => Task.CompletedTask;
+        Func<RequestContext<SubscribeRequestParams>, CancellationToken, Task<EmptyResult>> handler = (context, token) => Task.FromResult(new EmptyResult());
 
         _builder.Object.WithSubscribeToResourcesHandler(handler);
 
@@ -126,7 +126,7 @@ public void WithSubscribeToResourcesHandler_Sets_Handler()
     [Fact]
     public void WithUnsubscribeFromResourcesHandler_Sets_Handler()
     {
-        Func<RequestContext<string>, CancellationToken, Task> handler = (context, token) => Task.CompletedTask;
+        Func<RequestContext<UnsubscribeRequestParams>, CancellationToken, Task<EmptyResult>> 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

Original file line number	Diff line number	Diff line change
`@@ -227,7 +227,7 @@ public static Task SubscribeToResourceAsync(this IMcpClient client, string uri,`
`227`	`227`	`{`
`228`	`228`	`Throw.IfNull(client);`
`229`	`229`
`230`		`- return client.SendRequestAsync<dynamic>(`
	`230`	`+ return client.SendRequestAsync<EmptyResult>(`
`231`	`231`	`CreateRequest("resources/subscribe", new() { ["uri"] = uri }),`
`232`	`232`	`cancellationToken);`
`233`	`233`	`}`
`@@ -242,7 +242,7 @@ public static Task UnsubscribeFromResourceAsync(this IMcpClient client, string u`
`242`	`242`	`{`
`243`	`243`	`Throw.IfNull(client);`
`244`	`244`
`245`		`- return client.SendRequestAsync<dynamic>(`
	`245`	`+ return client.SendRequestAsync<EmptyResult>(`
`246`	`246`	`CreateRequest("resources/unsubscribe", new() { ["uri"] = uri }),`
`247`	`247`	`cancellationToken);`
`248`	`248`	`}`
Original file line number	Diff line number	Diff line change
`@@ -163,7 +163,8 @@ private void SetResourcesHandler(McpServerOptions options)`
`163`	`163`	`throw new McpServerException("Resources capability was enabled with subscribe support, but SubscribeToResources and/or UnsubscribeFromResources handlers were not specified.");`
`164`	`164`	`}`
`165`	`165`
`166`		`- // TODO: Implement Subscribe support`
	`166`	`+ SetRequestHandler<SubscribeRequestParams, EmptyResult>("resources/subscribe", request => subscribeHandler(new(this, request), cancellationToken));`
	`167`	`+ SetRequestHandler<UnsubscribeRequestParams, EmptyResult>("resources/unsubscribe", request => unsubscribeHandler(new(this, request), cancellationToken));`
`167`	`168`	`}`
`168`	`169`
`169`	`170`	`private void SetPromptsHandler(McpServerOptions options)`