From 9ebff0c5d9364c84957fd5da90080234c9c120a1 Mon Sep 17 00:00:00 2001 From: gongwn1 Date: Mon, 26 May 2025 12:08:37 +0800 Subject: [PATCH] feat: implement resource update notifications (#264) This enables servers to notify clients when specific resources are updated, allowing clients to react to individual resource changes rather than just list changes. - Add ResourcesUpdatedNotification schema with uri field - Add METHOD_NOTIFICATION_RESOURCES_UPDATED constant - Implement notifyResourcesUpdated() methods in McpAsyncServer and McpSyncServer - Add client-side support for resource update consumers and handlers - Client automatically reads updated resource content when notification received - Add tests for both async and sync server implementations Signed-off-by: Christian Tzolov --- .../server/AbstractMcpAsyncServerTests.java | 12 ++++++++ .../server/AbstractMcpSyncServerTests.java | 11 +++++++ .../client/McpAsyncClient.java | 30 +++++++++++++++++++ .../client/McpClient.java | 30 ++++++++++++++++--- .../client/McpClientFeatures.java | 18 +++++++++-- .../server/McpAsyncServer.java | 9 ++++++ .../server/McpSyncServer.java | 7 +++++ .../modelcontextprotocol/spec/McpSchema.java | 13 ++++++++ .../server/AbstractMcpAsyncServerTests.java | 12 ++++++++ .../server/AbstractMcpSyncServerTests.java | 11 +++++++ 10 files changed, 146 insertions(+), 7 deletions(-) diff --git a/mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java b/mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java index 025cfea..12827f4 100644 --- a/mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java +++ b/mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java @@ -194,6 +194,18 @@ public abstract class AbstractMcpAsyncServerTests { assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException(); } + @Test + void testNotifyResourcesUpdated() { + var mcpAsyncServer = McpServer.async(createMcpTransportProvider()).serverInfo("test-server", "1.0.0").build(); + + StepVerifier + .create(mcpAsyncServer + .notifyResourcesUpdated(new McpSchema.ResourcesUpdatedNotification(TEST_RESOURCE_URI))) + .verifyComplete(); + + assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException(); + } + @Test void testAddResource() { var mcpAsyncServer = McpServer.async(createMcpTransportProvider()) diff --git a/mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java b/mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java index e313454..eefcdf9 100644 --- a/mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java +++ b/mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java @@ -191,6 +191,17 @@ public abstract class AbstractMcpSyncServerTests { assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException(); } + @Test + void testNotifyResourcesUpdated() { + var mcpSyncServer = McpServer.sync(createMcpTransportProvider()).serverInfo("test-server", "1.0.0").build(); + + assertThatCode(() -> mcpSyncServer + .notifyResourcesUpdated(new McpSchema.ResourcesUpdatedNotification(TEST_RESOURCE_URI))) + .doesNotThrowAnyException(); + + assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException(); + } + @Test void testAddResource() { var mcpSyncServer = McpServer.sync(createMcpTransportProvider()) diff --git a/mcp/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java b/mcp/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java index d6804c8..a7dac4c 100644 --- a/mcp/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java +++ b/mcp/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java @@ -236,6 +236,18 @@ public class McpAsyncClient { notificationHandlers.put(McpSchema.METHOD_NOTIFICATION_RESOURCES_LIST_CHANGED, asyncResourcesChangeNotificationHandler(resourcesChangeConsumersFinal)); + // Resources Update Notification + List, Mono>> resourcesUpdateConsumersFinal = new ArrayList<>(); + resourcesUpdateConsumersFinal + .add((notification) -> Mono.fromRunnable(() -> logger.debug("Resources updated: {}", notification))); + + if (!Utils.isEmpty(features.resourcesUpdateConsumers())) { + resourcesUpdateConsumersFinal.addAll(features.resourcesUpdateConsumers()); + } + + notificationHandlers.put(McpSchema.METHOD_NOTIFICATION_RESOURCES_UPDATED, + asyncResourcesUpdatedNotificationHandler(resourcesUpdateConsumersFinal)); + // Prompts Change Notification List, Mono>> promptsChangeConsumersFinal = new ArrayList<>(); promptsChangeConsumersFinal @@ -867,6 +879,24 @@ public class McpAsyncClient { .then()); } + private NotificationHandler asyncResourcesUpdatedNotificationHandler( + List, Mono>> resourcesUpdateConsumers) { + return params -> { + McpSchema.ResourcesUpdatedNotification resourcesUpdatedNotification = transport.unmarshalFrom(params, + new TypeReference<>() { + }); + + return readResource(new McpSchema.ReadResourceRequest(resourcesUpdatedNotification.uri())) + .flatMap(readResourceResult -> Flux.fromIterable(resourcesUpdateConsumers) + .flatMap(consumer -> consumer.apply(readResourceResult.contents())) + .onErrorResume(error -> { + logger.error("Error handling resource update notification", error); + return Mono.empty(); + }) + .then()); + }; + } + // -------------------------- // Prompts // -------------------------- diff --git a/mcp/src/main/java/io/modelcontextprotocol/client/McpClient.java b/mcp/src/main/java/io/modelcontextprotocol/client/McpClient.java index 280906c..d8925b0 100644 --- a/mcp/src/main/java/io/modelcontextprotocol/client/McpClient.java +++ b/mcp/src/main/java/io/modelcontextprotocol/client/McpClient.java @@ -171,6 +171,8 @@ public interface McpClient { private final List>> resourcesChangeConsumers = new ArrayList<>(); + private final List>> resourcesUpdateConsumers = new ArrayList<>(); + private final List>> promptsChangeConsumers = new ArrayList<>(); private final List> loggingConsumers = new ArrayList<>(); @@ -382,8 +384,8 @@ public interface McpClient { */ public McpSyncClient build() { McpClientFeatures.Sync syncFeatures = new McpClientFeatures.Sync(this.clientInfo, this.capabilities, - this.roots, this.toolsChangeConsumers, this.resourcesChangeConsumers, this.promptsChangeConsumers, - this.loggingConsumers, this.samplingHandler, this.elicitationHandler); + this.roots, this.toolsChangeConsumers, this.resourcesChangeConsumers, this.resourcesUpdateConsumers, + this.promptsChangeConsumers, this.loggingConsumers, this.samplingHandler, this.elicitationHandler); McpClientFeatures.Async asyncFeatures = McpClientFeatures.Async.fromSync(syncFeatures); @@ -427,6 +429,8 @@ public interface McpClient { private final List, Mono>> resourcesChangeConsumers = new ArrayList<>(); + private final List, Mono>> resourcesUpdateConsumers = new ArrayList<>(); + private final List, Mono>> promptsChangeConsumers = new ArrayList<>(); private final List>> loggingConsumers = new ArrayList<>(); @@ -589,6 +593,23 @@ public interface McpClient { return this; } + /** + * Adds a consumer to be notified when a specific resource is updated. This allows + * the client to react to changes in individual resources, such as updates to + * their content or metadata. + * @param resourcesUpdateConsumer A consumer function that processes the updated + * resource and returns a Mono indicating the completion of the processing. Must + * not be null. + * @return This builder instance for method chaining. + * @throws IllegalArgumentException If the resourcesUpdateConsumer is null. + */ + public AsyncSpec resourcesUpdateConsumer( + Function, Mono> resourcesUpdateConsumer) { + Assert.notNull(resourcesUpdateConsumer, "Resources update consumer must not be null"); + this.resourcesUpdateConsumers.add(resourcesUpdateConsumer); + return this; + } + /** * Adds a consumer to be notified when the available prompts change. This allows * the client to react to changes in the server's prompt templates, such as new @@ -641,8 +662,9 @@ public interface McpClient { public McpAsyncClient build() { return new McpAsyncClient(this.transport, this.requestTimeout, this.initializationTimeout, new McpClientFeatures.Async(this.clientInfo, this.capabilities, this.roots, - this.toolsChangeConsumers, this.resourcesChangeConsumers, this.promptsChangeConsumers, - this.loggingConsumers, this.samplingHandler, this.elicitationHandler)); + this.toolsChangeConsumers, this.resourcesChangeConsumers, this.resourcesUpdateConsumers, + this.promptsChangeConsumers, this.loggingConsumers, this.samplingHandler, + this.elicitationHandler)); } } diff --git a/mcp/src/main/java/io/modelcontextprotocol/client/McpClientFeatures.java b/mcp/src/main/java/io/modelcontextprotocol/client/McpClientFeatures.java index 23d7c6a..bd1a098 100644 --- a/mcp/src/main/java/io/modelcontextprotocol/client/McpClientFeatures.java +++ b/mcp/src/main/java/io/modelcontextprotocol/client/McpClientFeatures.java @@ -65,6 +65,7 @@ class McpClientFeatures { record Async(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities clientCapabilities, Map roots, List, Mono>> toolsChangeConsumers, List, Mono>> resourcesChangeConsumers, + List, Mono>> resourcesUpdateConsumers, List, Mono>> promptsChangeConsumers, List>> loggingConsumers, Function> samplingHandler, @@ -85,6 +86,7 @@ class McpClientFeatures { Map roots, List, Mono>> toolsChangeConsumers, List, Mono>> resourcesChangeConsumers, + List, Mono>> resourcesUpdateConsumers, List, Mono>> promptsChangeConsumers, List>> loggingConsumers, Function> samplingHandler, @@ -101,6 +103,7 @@ class McpClientFeatures { this.toolsChangeConsumers = toolsChangeConsumers != null ? toolsChangeConsumers : List.of(); this.resourcesChangeConsumers = resourcesChangeConsumers != null ? resourcesChangeConsumers : List.of(); + this.resourcesUpdateConsumers = resourcesUpdateConsumers != null ? resourcesUpdateConsumers : List.of(); this.promptsChangeConsumers = promptsChangeConsumers != null ? promptsChangeConsumers : List.of(); this.loggingConsumers = loggingConsumers != null ? loggingConsumers : List.of(); this.samplingHandler = samplingHandler; @@ -128,8 +131,13 @@ class McpClientFeatures { .subscribeOn(Schedulers.boundedElastic())); } - List, Mono>> promptsChangeConsumers = new ArrayList<>(); + List, Mono>> resourcesUpdateConsumers = new ArrayList<>(); + for (Consumer> consumer : syncSpec.resourcesUpdateConsumers()) { + resourcesUpdateConsumers.add(r -> Mono.fromRunnable(() -> consumer.accept(r)) + .subscribeOn(Schedulers.boundedElastic())); + } + List, Mono>> promptsChangeConsumers = new ArrayList<>(); for (Consumer> consumer : syncSpec.promptsChangeConsumers()) { promptsChangeConsumers.add(p -> Mono.fromRunnable(() -> consumer.accept(p)) .subscribeOn(Schedulers.boundedElastic())); @@ -150,8 +158,8 @@ class McpClientFeatures { .subscribeOn(Schedulers.boundedElastic()); return new Async(syncSpec.clientInfo(), syncSpec.clientCapabilities(), syncSpec.roots(), - toolsChangeConsumers, resourcesChangeConsumers, promptsChangeConsumers, loggingConsumers, - samplingHandler, elicitationHandler); + toolsChangeConsumers, resourcesChangeConsumers, resourcesUpdateConsumers, promptsChangeConsumers, + loggingConsumers, samplingHandler, elicitationHandler); } } @@ -172,6 +180,7 @@ class McpClientFeatures { public record Sync(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities clientCapabilities, Map roots, List>> toolsChangeConsumers, List>> resourcesChangeConsumers, + List>> resourcesUpdateConsumers, List>> promptsChangeConsumers, List> loggingConsumers, Function samplingHandler, @@ -184,6 +193,7 @@ class McpClientFeatures { * @param roots the roots. * @param toolsChangeConsumers the tools change consumers. * @param resourcesChangeConsumers the resources change consumers. + * @param resourcesUpdateConsumers the resource update consumers. * @param promptsChangeConsumers the prompts change consumers. * @param loggingConsumers the logging consumers. * @param samplingHandler the sampling handler. @@ -192,6 +202,7 @@ class McpClientFeatures { public Sync(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities clientCapabilities, Map roots, List>> toolsChangeConsumers, List>> resourcesChangeConsumers, + List>> resourcesUpdateConsumers, List>> promptsChangeConsumers, List> loggingConsumers, Function samplingHandler, @@ -208,6 +219,7 @@ class McpClientFeatures { this.toolsChangeConsumers = toolsChangeConsumers != null ? toolsChangeConsumers : List.of(); this.resourcesChangeConsumers = resourcesChangeConsumers != null ? resourcesChangeConsumers : List.of(); + this.resourcesUpdateConsumers = resourcesUpdateConsumers != null ? resourcesUpdateConsumers : List.of(); this.promptsChangeConsumers = promptsChangeConsumers != null ? promptsChangeConsumers : List.of(); this.loggingConsumers = loggingConsumers != null ? loggingConsumers : List.of(); this.samplingHandler = samplingHandler; diff --git a/mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java b/mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java index 1efa13d..02ad955 100644 --- a/mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java +++ b/mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java @@ -430,6 +430,15 @@ public class McpAsyncServer { return this.mcpTransportProvider.notifyClients(McpSchema.METHOD_NOTIFICATION_RESOURCES_LIST_CHANGED, null); } + /** + * Notifies clients that the resources have updated. + * @return A Mono that completes when all clients have been notified + */ + public Mono notifyResourcesUpdated(McpSchema.ResourcesUpdatedNotification resourcesUpdatedNotification) { + return this.mcpTransportProvider.notifyClients(McpSchema.METHOD_NOTIFICATION_RESOURCES_UPDATED, + resourcesUpdatedNotification); + } + private McpServerSession.RequestHandler resourcesListRequestHandler() { return (exchange, params) -> { var resourceList = this.resources.values() diff --git a/mcp/src/main/java/io/modelcontextprotocol/server/McpSyncServer.java b/mcp/src/main/java/io/modelcontextprotocol/server/McpSyncServer.java index bf31045..91f8d9e 100644 --- a/mcp/src/main/java/io/modelcontextprotocol/server/McpSyncServer.java +++ b/mcp/src/main/java/io/modelcontextprotocol/server/McpSyncServer.java @@ -141,6 +141,13 @@ public class McpSyncServer { this.asyncServer.notifyResourcesListChanged().block(); } + /** + * Notify clients that the resources have updated. + */ + public void notifyResourcesUpdated(McpSchema.ResourcesUpdatedNotification resourcesUpdatedNotification) { + this.asyncServer.notifyResourcesUpdated(resourcesUpdatedNotification).block(); + } + /** * Notify clients that the list of available prompts has changed. */ diff --git a/mcp/src/main/java/io/modelcontextprotocol/spec/McpSchema.java b/mcp/src/main/java/io/modelcontextprotocol/spec/McpSchema.java index 8a00bf5..5971309 100644 --- a/mcp/src/main/java/io/modelcontextprotocol/spec/McpSchema.java +++ b/mcp/src/main/java/io/modelcontextprotocol/spec/McpSchema.java @@ -71,6 +71,8 @@ public final class McpSchema { public static final String METHOD_NOTIFICATION_RESOURCES_LIST_CHANGED = "notifications/resources/list_changed"; + public static final String METHOD_NOTIFICATION_RESOURCES_UPDATED = "notifications/resources/updated"; + public static final String METHOD_RESOURCES_TEMPLATES_LIST = "resources/templates/list"; public static final String METHOD_RESOURCES_SUBSCRIBE = "resources/subscribe"; @@ -1314,6 +1316,17 @@ public final class McpSchema { @JsonProperty("total") Double total) { }// @formatter:on + /** + * The Model Context Protocol (MCP) provides a standardized way for servers to send + * resources update message to clients. + * + * @param uri The updated resource uri. + */ + @JsonIgnoreProperties(ignoreUnknown = true) + public record ResourcesUpdatedNotification(// @formatter:off + @JsonProperty("uri") String uri) { + }// @formatter:on + /** * The Model Context Protocol (MCP) provides a standardized way for servers to send * structured log messages to clients. Clients can control logging verbosity by diff --git a/mcp/src/test/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java b/mcp/src/test/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java index df0b0c7..dd9f658 100644 --- a/mcp/src/test/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java +++ b/mcp/src/test/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java @@ -193,6 +193,18 @@ public abstract class AbstractMcpAsyncServerTests { assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException(); } + @Test + void testNotifyResourcesUpdated() { + var mcpAsyncServer = McpServer.async(createMcpTransportProvider()).serverInfo("test-server", "1.0.0").build(); + + StepVerifier + .create(mcpAsyncServer + .notifyResourcesUpdated(new McpSchema.ResourcesUpdatedNotification(TEST_RESOURCE_URI))) + .verifyComplete(); + + assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException(); + } + @Test void testAddResource() { var mcpAsyncServer = McpServer.async(createMcpTransportProvider()) diff --git a/mcp/src/test/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java b/mcp/src/test/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java index 0b38da8..6cbb863 100644 --- a/mcp/src/test/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java +++ b/mcp/src/test/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java @@ -190,6 +190,17 @@ public abstract class AbstractMcpSyncServerTests { assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException(); } + @Test + void testNotifyResourcesUpdated() { + var mcpSyncServer = McpServer.sync(createMcpTransportProvider()).serverInfo("test-server", "1.0.0").build(); + + assertThatCode(() -> mcpSyncServer + .notifyResourcesUpdated(new McpSchema.ResourcesUpdatedNotification(TEST_RESOURCE_URI))) + .doesNotThrowAnyException(); + + assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException(); + } + @Test void testAddResource() { var mcpSyncServer = McpServer.sync(createMcpTransportProvider())