Support resource subscriptions (#839)

Clients can now subscribe to specific resources and receive targeted
notifications when those resources change. Previously, calling
`notifyResourceUpdated` on the server would broadcast the notification
to every connected client regardless of interest — now only sessions
that have explicitly subscribed to a given resource URI receive the
update, making resource change propagation both correct and efficient.
The subscription lifecycle is fully handled: clients can subscribe and
unsubscribe at any time, and the server cleans up subscription state
when a session closes.

Supersedes #838.
Resolves #837, #776.

---------

Signed-off-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>

Author: chemicL

conformance-tests/VALIDATION_RESULTS.md

@@ -2,27 +2,22 @@
 
 ## Summary
 
-**Server Tests:** 37/40 passed (92.5%)
+**Server Tests:** 40/40 passed (100%)
 **Client Tests:** 3/4 scenarios passed (9/10 checks passed)
 **Auth Tests:** 12/14 scenarios fully passing (178 passed, 1 failed, 1 warning, 85.7% scenarios, 98.9% checks)
 
 ## Server Test Results
 
-### Passing (37/40)
+### Passing (40/40)
 
 - **Lifecycle & Utilities (4/4):** initialize, ping, logging-set-level, completion-complete
 - **Tools (11/11):** All scenarios including progress notifications ✨
 - **Elicitation (10/10):** SEP-1034 defaults (5 checks), SEP-1330 enums (5 checks)
-- **Resources (4/6):** list, read-text, read-binary, templates-read
+- **Resources (6/6):** list, read-text, read-binary, templates-read, subscribe, unsubscribe
 - **Prompts (4/4):** list, simple, with-args, embedded-resource, with-image
 - **SSE Transport (2/2):** Multiple streams
 - **Security (2/2):** Localhost validation passes, DNS rebinding protection
 
-### Failing (3/40)
-
-1. **resources-subscribe** - Not implemented in SDK
-2. **resources-unsubscribe** - Not implemented in SDK
-
 ## Client Test Results
 
 ### Passing (3/4 scenarios, 9/10 checks)
@@ -68,10 +63,9 @@ Uses the `client-spring-http-client` module with Spring Security OAuth2 and the
 
 ## Known Limitations
 
-1. **Resource Subscriptions:** SDK doesn't implement `resources/subscribe` and `resources/unsubscribe` handlers
-2. **Client SSE Retry:** Client doesn't parse or respect the `retry:` field, reconnects immediately, and doesn't send Last-Event-ID header
-3. **Auth Scope Step-Up:** Client does not fully handle scope step-up challenges where the server requests additional scopes after initial authorization
-4. **Auth Basic CIMD:** Minor conformance warning in the basic Client-Initiated Metadata Discovery flow
+1. **Client SSE Retry:** Client doesn't parse or respect the `retry:` field, reconnects immediately, and doesn't send Last-Event-ID header
+2. **Auth Scope Step-Up:** Client does not fully handle scope step-up challenges where the server requests additional scopes after initial authorization
+3. **Auth Basic CIMD:** Minor conformance warning in the basic Client-Initiated Metadata Discovery flow
 
 ## Running Tests
 
@@ -118,6 +112,5 @@ npx @modelcontextprotocol/conformance@0.1.15 client \
 
 ### High Priority
 1. Fix client SSE retry field handling in `HttpClientStreamableHttpTransport`
-2. Implement resource subscription handlers in `McpStatelessAsyncServer`
-3. Implement CIMD
-4. Implement scope step up
+2. Implement CIMD
+3. Implement scope step up

conformance-tests/conformance-baseline.yml

@@ -2,11 +2,6 @@
 # This file lists known failing scenarios that are expected to fail until fixed.
 # See: https://github.com/modelcontextprotocol/conformance/blob/main/SDK_INTEGRATION.md
 
-server:
-  # Resource subscription not implemented in SDK
-  - resources-subscribe
-  - resources-unsubscribe
-
 client:
   # SSE retry field handling not implemented
   # - Client does not parse or respect retry: field timing

conformance-tests/server-servlet/README.md

@@ -4,7 +4,7 @@ This module contains a comprehensive MCP (Model Context Protocol) server impleme
 
 ## Conformance Test Results
 
-**Status: 37 out of 40 tests passing (92.5%)**
+**Status: 40 out of 40 tests passing (100%)**
 
 The server has been validated against the official [MCP conformance test suite](https://github.com/modelcontextprotocol/conformance). See [VALIDATION_RESULTS.md](../VALIDATION_RESULTS.md) for detailed results.
 
@@ -22,9 +22,8 @@ The server has been validated against the official [MCP conformance test suite](
 - SEP-1034: Default values for all primitive types
 - SEP-1330: All enum schema variants
 
-✅ **Resources** (4/6)
-- List, read text/binary, templates
-- ⚠️ Subscribe/unsubscribe (SDK limitation)
+✅ **Resources** (6/6)
+- List, read text/binary, templates, subscribe, unsubscribe
 
 ✅ **Prompts** (4/4)
 - Simple, parameterized, embedded resources, images
@@ -191,12 +190,7 @@ curl -X POST http://localhost:8080/mcp \
 
 ## Known Limitations
 
-See [VALIDATION_RESULTS.md](../VALIDATION_RESULTS.md) for details on:
-
-1. **Resource Subscriptions** - Not implemented in Java SDK
-2. **DNS Rebinding Protection** - Missing Host/Origin validation
-
-These are SDK-level limitations that require fixes in the core framework.
+See [VALIDATION_RESULTS.md](../VALIDATION_RESULTS.md) for details on remaining client-side limitations.
 
 ## References
 

mcp-core/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java

@@ -5,10 +5,12 @@
 package io.modelcontextprotocol.server;
 
 import java.time.Duration;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -25,7 +27,6 @@
 import io.modelcontextprotocol.spec.McpSchema.CompleteResult.CompleteCompletion;
 import io.modelcontextprotocol.spec.McpSchema.ErrorCodes;
 import io.modelcontextprotocol.spec.McpSchema.LoggingLevel;
-import io.modelcontextprotocol.spec.McpSchema.LoggingMessageNotification;
 import io.modelcontextprotocol.spec.McpSchema.PromptReference;
 import io.modelcontextprotocol.spec.McpSchema.ResourceReference;
 import io.modelcontextprotocol.spec.McpSchema.SetLevelRequest;
@@ -111,12 +112,10 @@ public class McpAsyncServer {
 
 	private final ConcurrentHashMap<String, McpServerFeatures.AsyncPromptSpecification> prompts = new ConcurrentHashMap<>();
 
-	// FIXME: this field is deprecated and should be remvoed together with the
-	// broadcasting loggingNotification.
-	private LoggingLevel minLoggingLevel = LoggingLevel.DEBUG;
-
 	private final ConcurrentHashMap<McpSchema.CompleteReference, McpServerFeatures.AsyncCompletionSpecification> completions = new ConcurrentHashMap<>();
 
+	private final ConcurrentHashMap<String, Set<String>> resourceSubscriptions = new ConcurrentHashMap<>();
+
 	private List<String> protocolVersions;
 
 	private McpUriTemplateManagerFactory uriTemplateManagerFactory = new DefaultMcpUriTemplateManagerFactory();
@@ -149,8 +148,11 @@ public class McpAsyncServer {
 
 		this.protocolVersions = mcpTransportProvider.protocolVersions();
 
-		mcpTransportProvider.setSessionFactory(transport -> new McpServerSession(UUID.randomUUID().toString(),
-				requestTimeout, transport, this::asyncInitializeRequestHandler, requestHandlers, notificationHandlers));
+		mcpTransportProvider.setSessionFactory(transport -> {
+			String sessionId = UUID.randomUUID().toString();
+			return new McpServerSession(sessionId, requestTimeout, transport, this::asyncInitializeRequestHandler,
+					requestHandlers, notificationHandlers, () -> this.cleanupForSession(sessionId));
+		});
 	}
 
 	McpAsyncServer(McpStreamableServerTransportProvider mcpTransportProvider, McpJsonMapper jsonMapper,
@@ -174,8 +176,9 @@ public class McpAsyncServer {
 
 		this.protocolVersions = mcpTransportProvider.protocolVersions();
 
-		mcpTransportProvider.setSessionFactory(new DefaultMcpStreamableServerSessionFactory(requestTimeout,
-				this::asyncInitializeRequestHandler, requestHandlers, notificationHandlers));
+		mcpTransportProvider.setSessionFactory(
+				new DefaultMcpStreamableServerSessionFactory(requestTimeout, this::asyncInitializeRequestHandler,
+						requestHandlers, notificationHandlers, sessionId -> this.cleanupForSession(sessionId)));
 	}
 
 	private Map<String, McpNotificationHandler> prepareNotificationHandlers(McpServerFeatures.Async features) {
@@ -215,6 +218,10 @@ private Map<String, McpRequestHandler<?>> prepareRequestHandlers() {
 			requestHandlers.put(McpSchema.METHOD_RESOURCES_LIST, resourcesListRequestHandler());
 			requestHandlers.put(McpSchema.METHOD_RESOURCES_READ, resourcesReadRequestHandler());
 			requestHandlers.put(McpSchema.METHOD_RESOURCES_TEMPLATES_LIST, resourceTemplateListRequestHandler());
+			if (Boolean.TRUE.equals(this.serverCapabilities.resources().subscribe())) {
+				requestHandlers.put(McpSchema.METHOD_RESOURCES_SUBSCRIBE, resourcesSubscribeRequestHandler());
+				requestHandlers.put(McpSchema.METHOD_RESOURCES_UNSUBSCRIBE, resourcesUnsubscribeRequestHandler());
+			}
 		}
 
 		// Add prompts API handlers if provider exists
@@ -685,12 +692,73 @@ public Mono<Void> notifyResourcesListChanged() {
 	}
 
 	/**
-	 * Notifies clients that the resources have updated.
-	 * @return A Mono that completes when all clients have been notified
+	 * Notifies only the sessions that have subscribed to the updated resource URI.
+	 * @param resourcesUpdatedNotification the notification containing the updated
+	 * resource URI
+	 * @return A Mono that completes when all subscribed sessions have been notified
 	 */
 	public Mono<Void> notifyResourcesUpdated(McpSchema.ResourcesUpdatedNotification resourcesUpdatedNotification) {
-		return this.mcpTransportProvider.notifyClients(McpSchema.METHOD_NOTIFICATION_RESOURCES_UPDATED,
-				resourcesUpdatedNotification);
+		return Mono.defer(() -> {
+			String uri = resourcesUpdatedNotification.uri();
+			Set<String> subscribedSessions = this.resourceSubscriptions.get(uri);
+			if (subscribedSessions == null || subscribedSessions.isEmpty()) {
+				logger.debug("No sessions subscribed to resource URI: {}", uri);
+				return Mono.empty();
+			}
+			return Flux.fromIterable(subscribedSessions)
+				.flatMap(sessionId -> this.mcpTransportProvider
+					.notifyClient(sessionId, McpSchema.METHOD_NOTIFICATION_RESOURCES_UPDATED,
+							resourcesUpdatedNotification)
+					.doOnError(e -> logger.error("Failed to notify session {} of resource update for {}", sessionId,
+							uri, e))
+					.onErrorComplete())
+				.then();
+		});
+	}
+
+	private Mono<Void> cleanupForSession(String sessionId) {
+		return Mono.fromRunnable(() -> {
+			removeSessionSubscriptions(sessionId);
+		});
+	}
+
+	private void removeSessionSubscriptions(String sessionId) {
+		this.resourceSubscriptions.forEach((uri, sessions) -> sessions.remove(sessionId));
+		this.resourceSubscriptions.entrySet().removeIf(entry -> entry.getValue().isEmpty());
+	}
+
+	private McpRequestHandler<Object> resourcesSubscribeRequestHandler() {
+		return (exchange, params) -> Mono.defer(() -> {
+			McpSchema.SubscribeRequest subscribeRequest = jsonMapper.convertValue(params,
+					new TypeRef<McpSchema.SubscribeRequest>() {
+					});
+			String uri = subscribeRequest.uri();
+			String sessionId = exchange.sessionId();
+			this.resourceSubscriptions.computeIfAbsent(uri, k -> Collections.newSetFromMap(new ConcurrentHashMap<>()))
+				.add(sessionId);
+			logger.debug("Session {} subscribed to resource URI: {}", sessionId, uri);
+
+			return Mono.just(Map.of());
+		});
+	}
+
+	private McpRequestHandler<Object> resourcesUnsubscribeRequestHandler() {
+		return (exchange, params) -> Mono.defer(() -> {
+			McpSchema.UnsubscribeRequest unsubscribeRequest = jsonMapper.convertValue(params,
+					new TypeRef<McpSchema.UnsubscribeRequest>() {
+					});
+			String uri = unsubscribeRequest.uri();
+			String sessionId = exchange.sessionId();
+			Set<String> sessions = this.resourceSubscriptions.get(uri);
+			if (sessions != null) {
+				sessions.remove(sessionId);
+				if (sessions.isEmpty()) {
+					this.resourceSubscriptions.remove(uri, sessions);
+				}
+			}
+			logger.debug("Session {} unsubscribed from resource URI: {}", sessionId, uri);
+			return Mono.just(Map.of());
+		});
 	}
 
 	private McpRequestHandler<McpSchema.ListResourcesResult> resourcesListRequestHandler() {
@@ -878,10 +946,6 @@ private McpRequestHandler<Object> setLoggerRequestHandler() {
 
 				exchange.setMinLoggingLevel(newMinLoggingLevel.level());
 
-				// FIXME: this field is deprecated and should be removed together
-				// with the broadcasting loggingNotification.
-				this.minLoggingLevel = newMinLoggingLevel.level();
-
 				return Mono.just(Map.of());
 			});
 		};

mcp-core/src/main/java/io/modelcontextprotocol/server/transport/HttpServletSseServerTransportProvider.java

@@ -228,6 +228,25 @@ public Mono<Void> notifyClients(String method, Object params) {
 			.then();
 	}
 
+	@Override
+	public Mono<Void> notifyClient(String sessionId, String method, Object params) {
+		return Mono.defer(() -> {
+			// Need to iterate in O(n) because the transport session id
+			// is different from the server-logical session id (in streamable http this
+			// design issue was solved)
+			McpServerSession session = sessions.values()
+				.stream()
+				.filter(s -> sessionId.equals(s.getId()))
+				.findFirst()
+				.orElse(null);
+			if (session == null) {
+				logger.debug("Session {} not found", sessionId);
+				return Mono.empty();
+			}
+			return session.sendNotification(method, params);
+		});
+	}
+
 	/**
 	 * Handles GET requests to establish SSE connections.
 	 * <p>

mcp-core/src/main/java/io/modelcontextprotocol/server/transport/HttpServletStreamableServerTransportProvider.java

@@ -206,6 +206,18 @@ public Mono<Void> notifyClients(String method, Object params) {
 		});
 	}
 
+	@Override
+	public Mono<Void> notifyClient(String sessionId, String method, Object params) {
+		return Mono.defer(() -> {
+			McpStreamableServerSession session = this.sessions.get(sessionId);
+			if (session == null) {
+				logger.debug("Session {} not found", sessionId);
+				return Mono.empty();
+			}
+			return session.sendNotification(method, params);
+		});
+	}
+
 	/**
 	 * Initiates a graceful shutdown of the transport.
 	 * @return A Mono that completes when all cleanup operations are finished

mcp-core/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java

@@ -98,12 +98,26 @@ public void setSessionFactory(McpServerSession.Factory sessionFactory) {
 	@Override
 	public Mono<Void> notifyClients(String method, Object params) {
 		if (this.session == null) {
-			return Mono.error(new IllegalStateException("No session to close"));
+			return Mono.error(new IllegalStateException("No session to notify"));
 		}
 		return this.session.sendNotification(method, params)
 			.doOnError(e -> logger.error("Failed to send notification: {}", e.getMessage()));
 	}
 
+	@Override
+	public Mono<Void> notifyClient(String sessionId, String method, Object params) {
+		return Mono.defer(() -> {
+			if (this.session == null) {
+				return Mono.error(new IllegalStateException("No session to notify"));
+			}
+			if (!this.session.getId().equals(sessionId)) {
+				return Mono.error(new IllegalStateException("Existing session id " + this.session.getId()
+						+ " doesn't match the notification target: " + sessionId));
+			}
+			return this.session.sendNotification(method, params);
+		});
+	}
+
 	@Override
 	public Mono<Void> closeGracefully() {
 		if (this.session == null) {