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

@@ -227,7 +227,7 @@ public static class McpClientExtensions
     {
         Throw.IfNull(client);
 
-        return client.SendRequestAsync<dynamic>(
+        return client.SendRequestAsync<EmptyResult>(
             CreateRequest("resources/subscribe", new() { ["uri"] = uri }),
             cancellationToken);
     }
@@ -242,7 +242,7 @@ public static class McpClientExtensions
     {
         Throw.IfNull(client);
 
-        return client.SendRequestAsync<dynamic>(
+        return client.SendRequestAsync<EmptyResult>(
             CreateRequest("resources/unsubscribe", new() { ["uri"] = uri }),
             cancellationToken);
     }

src/mcpdotnet/Configuration/McpServerBuilderExtensions.Handler.cs

@@ -107,7 +107,7 @@ public static partial class McpServerBuilderExtensions
     /// </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 partial class McpServerBuilderExtensions
     /// </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);
 

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>

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();
+}

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
+{
+
+}

src/mcpdotnet/Protocol/Types/Reference.cs

@@ -66,15 +66,3 @@ public class Reference
         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();
-}

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; }
+}

src/mcpdotnet/Protocol/Types/SubscribeRequestParams.cs

@@ -0,0 +1,14 @@
+namespace McpDotNet.Protocol.Types;
+
+/// <summary>
+/// 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 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.
+    /// </summary>
+    [System.Text.Json.Serialization.JsonPropertyName("uri")]
+    public string? Uri { get; init; }
+}

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; }
+}

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<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)

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 @@ 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;

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<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 @@ internal static class Program
         };
     }
 
+    private static HashSet<string> _subscribedResources = new();
+    private static readonly object _subscribedResourcesLock = new();
+
     private static ResourcesCapability ConfigureResources()
     {
         List<Resource> 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
         };
     }
 

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<ClientIntegrationTestFixture
         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)

tests/mcpdotnet.Tests/Configuration/McpServerBuilderExtensionsHandlerTests.cs

@@ -113,7 +113,7 @@ public class McpServerBuilderExtensionsHandlerTests
     [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 class McpServerBuilderExtensionsHandlerTests
     [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);
 

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);