Merge branch 'main' into bug/protocol_versioning

Author: dishtam

docs/client.md

@@ -408,6 +408,49 @@ Resources represent server-side data sources that clients can access using URI t
         .subscribe();
     ```
 
+### Resource Subscriptions
+
+When the server advertises `resources.subscribe` support, clients can subscribe to individual resources and receive a callback whenever the server pushes a `notifications/resources/updated` notification for that URI. The SDK automatically re-reads the resource on notification and delivers the updated contents to the registered consumer.
+
+Register a consumer on the client builder, then subscribe/unsubscribe at any time:
+
+=== "Sync API"
+
+    ```java
+    McpSyncClient client = McpClient.sync(transport)
+        .resourcesUpdateConsumer(contents -> {
+            // called with the updated resource contents after each notification
+            System.out.println("Resource updated: " + contents);
+        })
+        .build();
+
+    client.initialize();
+
+    // Subscribe to a specific resource URI
+    client.subscribeResource(new McpSchema.SubscribeRequest("custom://resource"));
+
+    // ... later, stop receiving updates
+    client.unsubscribeResource(new McpSchema.UnsubscribeRequest("custom://resource"));
+    ```
+
+=== "Async API"
+
+    ```java
+    McpAsyncClient client = McpClient.async(transport)
+        .resourcesUpdateConsumer(contents -> Mono.fromRunnable(() -> {
+            System.out.println("Resource updated: " + contents);
+        }))
+        .build();
+
+    client.initialize()
+        .then(client.subscribeResource(new McpSchema.SubscribeRequest("custom://resource")))
+        .subscribe();
+
+    // ... later, stop receiving updates
+    client.unsubscribeResource(new McpSchema.UnsubscribeRequest("custom://resource"))
+        .subscribe();
+    ```
+
 ### Prompt System
 
 The prompt system enables interaction with server-side prompt templates. These templates can be discovered and executed with custom parameters, allowing for dynamic text generation based on predefined patterns.

docs/server.md

@@ -33,7 +33,7 @@ The server supports both synchronous and asynchronous APIs, allowing for flexibl
     McpSyncServer syncServer = McpServer.sync(transportProvider)
         .serverInfo("my-server", "1.0.0")
         .capabilities(ServerCapabilities.builder()
-            .resources(false, true)  // Enable resource support with list changes
+            .resources(false, true)  // Resource support: subscribe=false, listChanged=true
             .tools(true)             // Enable tool support with list changes
             .prompts(true)           // Enable prompt support with list changes
             .completions()           // Enable completions support
@@ -57,7 +57,7 @@ The server supports both synchronous and asynchronous APIs, allowing for flexibl
     McpAsyncServer asyncServer = McpServer.async(transportProvider)
         .serverInfo("my-server", "1.0.0")
         .capabilities(ServerCapabilities.builder()
-            .resources(false, true)  // Enable resource support with list changes
+            .resources(false, true)  // Resource support: subscribe=false, listChanged=true
             .tools(true)             // Enable tool support with list changes
             .prompts(true)           // Enable prompt support with list changes
             .completions()           // Enable completions support
@@ -319,7 +319,7 @@ The server can be configured with various capabilities:
 
 ```java
 var capabilities = ServerCapabilities.builder()
-    .resources(false, true)  // Resource support (subscribe, listChanged)
+    .resources(true, true)   // Resource support: subscribe=true, listChanged=true
     .tools(true)             // Tool support with list changes notifications
     .prompts(true)           // Prompt support with list changes notifications
     .completions()           // Enable completions support
@@ -438,6 +438,42 @@ Resources provide context to AI models by exposing data such as: File contents,
     );
     ```
 
+### Resource Subscriptions
+
+When the `subscribe` capability is enabled, clients can subscribe to specific resources and receive targeted `notifications/resources/updated` notifications when those resources change. Only sessions that have explicitly subscribed to a given URI receive the notification — not every connected client.
+
+Enable subscription support in the server capabilities:
+
+```java
+McpSyncServer server = McpServer.sync(transportProvider)
+    .serverInfo("my-server", "1.0.0")
+    .capabilities(ServerCapabilities.builder()
+        .resources(true, false)  // subscribe=true, listChanged=false
+        .build())
+    .resources(myResourceSpec)
+    .build();
+```
+
+When a subscribed resource changes, notify only the interested sessions:
+
+=== "Sync"
+
+    ```java
+    server.notifyResourcesUpdated(
+        new McpSchema.ResourcesUpdatedNotification("custom://resource")
+    );
+    ```
+
+=== "Async"
+
+    ```java
+    server.notifyResourcesUpdated(
+        new McpSchema.ResourcesUpdatedNotification("custom://resource")
+    ).subscribe();
+    ```
+
+If no sessions are subscribed to the given URI the call completes immediately without sending any messages. Subscription state is automatically cleaned up when a client session closes.
+
 ### Resource Template Specification
 
 Resource templates allow servers to expose parameterized resources using URI templates:

mcp-test/src/main/java/io/modelcontextprotocol/AbstractMcpClientServerIntegrationTests.java

@@ -53,7 +53,6 @@
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
 
 import static io.modelcontextprotocol.util.ToolsUtils.EMPTY_JSON_SCHEMA;
 import static net.javacrumbs.jsonunit.assertj.JsonAssertions.assertThatJson;
@@ -404,6 +403,8 @@ void testCreateElicitationSuccess(String clientType) {
 			.addContent(new McpSchema.TextContent("CALL RESPONSE"))
 			.build();
 
+		AtomicReference<McpSchema.ElicitResult> elicitResultRef = new AtomicReference<>();
+
 		McpServerFeatures.AsyncToolSpecification tool = McpServerFeatures.AsyncToolSpecification.builder()
 			.tool(Tool.builder().name("tool1").description("tool1 description").inputSchema(EMPTY_JSON_SCHEMA).build())
 			.callHandler((exchange, request) -> {
@@ -414,13 +415,9 @@ void testCreateElicitationSuccess(String clientType) {
 							Map.of("type", "object", "properties", Map.of("message", Map.of("type", "string"))))
 					.build();
 
-				StepVerifier.create(exchange.createElicitation(elicitationRequest)).consumeNextWith(result -> {
-					assertThat(result).isNotNull();
-					assertThat(result.action()).isEqualTo(McpSchema.ElicitResult.Action.ACCEPT);
-					assertThat(result.content().get("message")).isEqualTo("Test message");
-				}).verifyComplete();
-
-				return Mono.just(callResponse);
+				return exchange.createElicitation(elicitationRequest)
+					.doOnNext(elicitResultRef::set)
+					.thenReturn(callResponse);
 			})
 			.build();
 
@@ -438,6 +435,11 @@ void testCreateElicitationSuccess(String clientType) {
 
 			assertThat(response).isNotNull();
 			assertThat(response).isEqualTo(callResponse);
+			assertWith(elicitResultRef.get(), result -> {
+				assertThat(result).isNotNull();
+				assertThat(result.action()).isEqualTo(McpSchema.ElicitResult.Action.ACCEPT);
+				assertThat(result.content().get("message")).isEqualTo("Test message");
+			});
 		}
 		finally {
 			mcpServer.closeGracefully().block();

mcp-test/src/main/java/io/modelcontextprotocol/client/AbstractMcpAsyncClientTests.java

@@ -610,22 +610,17 @@ void testListAllResourceTemplatesReturnsImmutableList() {
 		});
 	}
 
-	// @Test
+	@Test
 	void testResourceSubscription() {
 		withClient(createMcpTransport(), mcpAsyncClient -> {
-			StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> {
-				if (!resources.resources().isEmpty()) {
-					Resource firstResource = resources.resources().get(0);
-
-					// Test subscribe
-					StepVerifier.create(mcpAsyncClient.subscribeResource(new SubscribeRequest(firstResource.uri())))
-						.verifyComplete();
-
-					// Test unsubscribe
-					StepVerifier.create(mcpAsyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())))
-						.verifyComplete();
+			StepVerifier.create(mcpAsyncClient.listResources().flatMap(resources -> {
+				if (resources.resources().isEmpty()) {
+					return Mono.empty();
 				}
-			}).verifyComplete();
+				Resource firstResource = resources.resources().get(0);
+				return mcpAsyncClient.subscribeResource(new SubscribeRequest(firstResource.uri()))
+					.then(mcpAsyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())));
+			})).verifyComplete();
 		});
 	}