modelcontextprotocol/java-sdk
Watch
1
0
Fork
You've already forked java-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: implement resource update notifications (#264)

Browse source 
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 <christian.tzolov@broadcom.com>
This commit is contained in:
gongwn1 2025-05-26 12:08:37 +08:00 • committed by Christian Tzolov
commit 9ebff0c5d9
10 changed files with 146 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

12
mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java
View file

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

11
mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java
View file

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

30
mcp/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java
View file

@ -236,6 +236,18 @@ public class McpAsyncClient {
notificationHandlers.put(McpSchema.METHOD_NOTIFICATION_RESOURCES_LIST_CHANGED,
asyncResourcesChangeNotificationHandler(resourcesChangeConsumersFinal));
// Resources Update Notification
List<Function<List<McpSchema.ResourceContents>, Mono<Void>>> 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<Function<List<McpSchema.Prompt>, Mono<Void>>> promptsChangeConsumersFinal = new ArrayList<>();
promptsChangeConsumersFinal
@ -867,6 +879,24 @@ public class McpAsyncClient {
.then());
}
private NotificationHandler asyncResourcesUpdatedNotificationHandler(
List<Function<List<McpSchema.ResourceContents>, Mono<Void>>> 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
// --------------------------

30
mcp/src/main/java/io/modelcontextprotocol/client/McpClient.java
View file

@ -171,6 +171,8 @@ public interface McpClient {
private final List<Consumer<List<McpSchema.Resource>>> resourcesChangeConsumers = new ArrayList<>();
private final List<Consumer<List<McpSchema.ResourceContents>>> resourcesUpdateConsumers = new ArrayList<>();
private final List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers = new ArrayList<>();
private final List<Consumer<McpSchema.LoggingMessageNotification>> 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<Function<List<McpSchema.Resource>, Mono<Void>>> resourcesChangeConsumers = new ArrayList<>();
private final List<Function<List<McpSchema.ResourceContents>, Mono<Void>>> resourcesUpdateConsumers = new ArrayList<>();
private final List<Function<List<McpSchema.Prompt>, Mono<Void>>> promptsChangeConsumers = new ArrayList<>();
private final List<Function<McpSchema.LoggingMessageNotification, Mono<Void>>> 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<List<McpSchema.ResourceContents>, Mono<Void>> 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));
}
}

18
mcp/src/main/java/io/modelcontextprotocol/client/McpClientFeatures.java
View file

@ -65,6 +65,7 @@ class McpClientFeatures {
record Async(McpSchema.Implementation clientInfo, McpSchema.ClientCapabilities clientCapabilities,
Map<String, McpSchema.Root> roots, List<Function<List<McpSchema.Tool>, Mono<Void>>> toolsChangeConsumers,
List<Function<List<McpSchema.Resource>, Mono<Void>>> resourcesChangeConsumers,
List<Function<List<McpSchema.ResourceContents>, Mono<Void>>> resourcesUpdateConsumers,
List<Function<List<McpSchema.Prompt>, Mono<Void>>> promptsChangeConsumers,
List<Function<McpSchema.LoggingMessageNotification, Mono<Void>>> loggingConsumers,
Function<McpSchema.CreateMessageRequest, Mono<McpSchema.CreateMessageResult>> samplingHandler,
@ -85,6 +86,7 @@ class McpClientFeatures {
Map<String, McpSchema.Root> roots,
List<Function<List<McpSchema.Tool>, Mono<Void>>> toolsChangeConsumers,
List<Function<List<McpSchema.Resource>, Mono<Void>>> resourcesChangeConsumers,
List<Function<List<McpSchema.ResourceContents>, Mono<Void>>> resourcesUpdateConsumers,
List<Function<List<McpSchema.Prompt>, Mono<Void>>> promptsChangeConsumers,
List<Function<McpSchema.LoggingMessageNotification, Mono<Void>>> loggingConsumers,
Function<McpSchema.CreateMessageRequest, Mono<McpSchema.CreateMessageResult>> 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<Function<List<McpSchema.Prompt>, Mono<Void>>> promptsChangeConsumers = new ArrayList<>();
List<Function<List<McpSchema.ResourceContents>, Mono<Void>>> resourcesUpdateConsumers = new ArrayList<>();
for (Consumer<List<McpSchema.ResourceContents>> consumer : syncSpec.resourcesUpdateConsumers()) {
resourcesUpdateConsumers.add(r -> Mono.<Void>fromRunnable(() -> consumer.accept(r))
.subscribeOn(Schedulers.boundedElastic()));
}
List<Function<List<McpSchema.Prompt>, Mono<Void>>> promptsChangeConsumers = new ArrayList<>();
for (Consumer<List<McpSchema.Prompt>> consumer : syncSpec.promptsChangeConsumers()) {
promptsChangeConsumers.add(p -> Mono.<Void>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<String, McpSchema.Root> roots, List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers,
List<Consumer<List<McpSchema.Resource>>> resourcesChangeConsumers,
List<Consumer<List<McpSchema.ResourceContents>>> resourcesUpdateConsumers,
List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers,
List<Consumer<McpSchema.LoggingMessageNotification>> loggingConsumers,
Function<McpSchema.CreateMessageRequest, McpSchema.CreateMessageResult> 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<String, McpSchema.Root> roots, List<Consumer<List<McpSchema.Tool>>> toolsChangeConsumers,
List<Consumer<List<McpSchema.Resource>>> resourcesChangeConsumers,
List<Consumer<List<McpSchema.ResourceContents>>> resourcesUpdateConsumers,
List<Consumer<List<McpSchema.Prompt>>> promptsChangeConsumers,
List<Consumer<McpSchema.LoggingMessageNotification>> loggingConsumers,
Function<McpSchema.CreateMessageRequest, McpSchema.CreateMessageResult> 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;

9
mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java
View file

@ -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<Void> notifyResourcesUpdated(McpSchema.ResourcesUpdatedNotification resourcesUpdatedNotification) {
return this.mcpTransportProvider.notifyClients(McpSchema.METHOD_NOTIFICATION_RESOURCES_UPDATED,
resourcesUpdatedNotification);
}
private McpServerSession.RequestHandler<McpSchema.ListResourcesResult> resourcesListRequestHandler() {
return (exchange, params) -> {
var resourceList = this.resources.values()

7
mcp/src/main/java/io/modelcontextprotocol/server/McpSyncServer.java
View file

@ -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.
*/

13
mcp/src/main/java/io/modelcontextprotocol/spec/McpSchema.java
View file

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

12
mcp/src/test/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java
View file

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

11
mcp/src/test/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java
View file

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