+ * The restriction can only narrow the default honoured subscriptions. + */ + public Builder restrictSubscriptions(SubscriptionRestriction restriction) { + Assert.notNull(restriction, "restriction must not be null"); + this.subscriptionRestriction = restriction; + return this; + } + /** * The codec used to seal/open MRTR {@code requestState}. Defaults to * {@link HmacRequestStateCodec#builder()}{@code .build()}. @@ -479,7 +495,7 @@ public McpServer build() { SubscriptionsFeature subscriptionsFeature = null; if (this.changePublisher != null) { subscriptionsFeature = new SubscriptionsFeature(this.changePublisher, mapper, allFeatures, - capabilitiesBuilder.hasTools(), capabilitiesBuilder.hasPrompts(), + this.subscriptionRestriction, capabilitiesBuilder.hasTools(), capabilitiesBuilder.hasPrompts(), capabilitiesBuilder.hasResources(), capabilitiesBuilder.hasResourcesSubscribe()); subscriptionsFeature.capabilities(capabilitiesBuilder); allFeatures.add(subscriptionsFeature); diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/modern/server/SubscriptionRestriction.java b/mcp-core/src/main/java/io/modelcontextprotocol/modern/server/SubscriptionRestriction.java new file mode 100644 index 000000000..b85e442eb --- /dev/null +++ b/mcp-core/src/main/java/io/modelcontextprotocol/modern/server/SubscriptionRestriction.java @@ -0,0 +1,78 @@ +/* + * Copyright 2026-2026 the original author or authors. + */ + +package io.modelcontextprotocol.modern.server; + +import io.modelcontextprotocol.modern.McpException; +import io.modelcontextprotocol.modern.McpSchema.SubscriptionFilter; +import reactor.core.publisher.Mono; + +/** + * Narrows, per {@code subscriptions/listen} request, the notifications the server agrees + * to honour, for example to keep a caller from watching resources that belong to another + * principal. Registered through {@link McpServer.Builder#restrictSubscriptions}. + *
+ * Change notifications are server-wide: a {@code notifications/resources/updated} for a + * URI reaches every stream whose acknowledged filter contains that URI, whoever opened + * it. Without a restriction, any caller may subscribe to any URI and learn when that + * resource changes, though not its content. + *
+ * Return the part of {@code honoured} this caller may receive, typically {@code honoured} + * with some fields removed or some list elements left out. The restriction can only + * narrow: the server keeps only the part of the returned filter that it honours, so it + * cannot grant notifications the client didn't request or the server cannot deliver. + * Anything returned beyond {@code honoured} is ignored and logged as a warning. An empty + * {@code Mono} honours nothing. + *
+ * The restriction runs once, when the stream opens, and the filter it leaves is + * acknowledged to the client and applies for the whole life of the stream. A change in + * what the caller may receive, such as a revoked permission, isn't applied to streams + * that are already open: it takes effect on the caller's next + * {@code subscriptions/listen}. When the server honours nothing of the request, the + * restriction isn't called: the stream opens with an empty acknowledgement and delivers + * nothing, so there is nothing to narrow or refuse. + *
+ * To refuse the request altogether, return {@link Mono#error} or throw an + * {@link McpException}; it is answered with its error, before any stream opens. The spec + * defines no permission error code, so use for example + * {@code new McpException(ErrorCodes.INVALID_REQUEST, "Not permitted")}, with the same + * message whichever subscription is denied, so that the error does not tell the caller + * which URI it may not access. Any other exception is answered as an internal error. + *
+ * The restriction runs while the request is being handled, so it must not block the + * calling thread. Run blocking work, such as a permission lookup, on a scheduler that + * allows it, except where the request is already handled on a thread that may block: + *
{@code
+ * (ctx, requested, honoured) -> {
+ * Mono restricted = Mono.fromCallable(() -> {
+ * Principal principal = (Principal) ctx.transportContext().get("principal");
+ * List allowed = honoured.resourceSubscriptionsOrEmpty()
+ * .stream()
+ * .filter(uri -> permissions.canWatch(principal, uri)) // blocking
+ * .toList();
+ * return honoured.with(SubscriptionFilter.RESOURCE_SUBSCRIPTIONS,
+ * allowed.isEmpty() ? null : allowed);
+ * });
+ * return ctx.isBlocking() ? restricted : restricted.subscribeOn(Schedulers.boundedElastic());
+ * }
+ * }
+ *
+ * @author Daniel Garnier-Moiroux
+ */
+@FunctionalInterface
+public interface SubscriptionRestriction {
+
+ /**
+ * The subscriptions this caller may receive.
+ * @param ctx the {@code subscriptions/listen} request, including the transport
+ * context that identifies the caller
+ * @param requested the filter the client sent
+ * @param honoured the part of {@code requested} the server and its features agree to
+ * honour
+ * @return the subscriptions this caller may receive, of which the server keeps the
+ * part it honours
+ */
+ Mono+ * Changes are typically consumed in an + * {@link io.modelcontextprotocol.modern.server.McpServer}, and broadcast to clients. They + * are filtered by client request and further restricted through + * {@link io.modelcontextprotocol.modern.server.SubscriptionRestriction}. * * @author Dariusz Jędrzejczyk */ diff --git a/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/SubscriptionsFeatureTests.java b/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/SubscriptionsFeatureTests.java index c8434597b..6048033a9 100644 --- a/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/SubscriptionsFeatureTests.java +++ b/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/SubscriptionsFeatureTests.java @@ -5,6 +5,7 @@ package io.modelcontextprotocol.modern.server; import java.time.Duration; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -49,6 +50,7 @@ import static io.modelcontextprotocol.modern.server.ModernTestFixtures.respond; import static org.assertj.core.api.Assertions.as; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.InstanceOfAssertFactories.MAP; import static org.assertj.core.api.InstanceOfAssertFactories.STRING; import static org.assertj.core.api.InstanceOfAssertFactories.type; @@ -339,7 +341,177 @@ void changeDeliveredBeforeAckIsSentFollowsTheAck() { .verify(Duration.ofSeconds(5)); } + @Test + void restrictionNarrowsResourceSubscriptionsAndOnlyThoseAreForwarded() { + McpChangeFeed feed = new McpChangeFeed(); + McpServer server = resourcesServerBuilder(true).subscriptions(feed) + .restrictSubscriptions((ctx, requested, honoured) -> Mono + .just(honoured.with(SubscriptionFilter.RESOURCE_SUBSCRIPTIONS, List.of("test://mine")))) + .build(); + + var listen = (McpTransportResponse.Streaming) listenWith(server, + Map.of("resourceSubscriptions", List.of("test://mine", "test://theirs"))); + + StepVerifier.create(listen.messages()) + .assertNext(msg -> assertThat( + acknowledgedNotifications((JSONRPCNotification) msg).resourceSubscriptionsOrEmpty()) + .containsExactly("test://mine")) + .then(() -> { + feed.broadcast(new ServerChange.ResourceUpdated("test://theirs")); + feed.broadcast(new ServerChange.ResourceUpdated("test://mine")); + }) + .assertNext(msg -> assertThat(((JSONRPCNotification) msg).params()).asInstanceOf(MAP) + .containsEntry("uri", "test://mine")) + .then(server::closeGracefully) + .expectNextMatches(JSONRPCResponse.class::isInstance) + .verifyComplete(); + } + + @Test + void restrictionSeesTheCallerTheRequestedFilterAndWhatTheServerHonours() { + List