diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/modern/server/SubscriptionsFeature.java b/mcp-core/src/main/java/io/modelcontextprotocol/modern/server/SubscriptionsFeature.java index 3e815da23..fa50d9344 100644 --- a/mcp-core/src/main/java/io/modelcontextprotocol/modern/server/SubscriptionsFeature.java +++ b/mcp-core/src/main/java/io/modelcontextprotocol/modern/server/SubscriptionsFeature.java @@ -71,15 +71,22 @@ public Set methods() { @Override public Mono> handle(McpRequestContext ctx, Object params) { + // The spec requires `notifications` to be a SubscriptionFilter object. Checked on + // the raw params because SubscriptionsListenRequest deserializes a missing or + // null value to an empty filter, which would be acknowledged as a valid + // subscription. Requiring an object (not just non-null) also stops a lenient + // mapper from coercing "" or [] to null, and therefore to that same empty filter. + if (!(params instanceof Map map) || !(map.get("notifications") instanceof Map)) { + return Mono.error(McpException.invalidParams("params.notifications is required")); + } SubscriptionsListenRequest request; try { - request = params == null ? null : this.jsonMapper.convertValue(params, SubscriptionsListenRequest.class); + request = this.jsonMapper.convertValue(params, SubscriptionsListenRequest.class); } catch (RuntimeException ex) { return Mono.error(McpException.invalidParams("Malformed SubscriptionsListenRequest")); } - SubscriptionFilter requested = request == null || request.notifications() == null ? SubscriptionFilter.EMPTY - : request.notifications(); + SubscriptionFilter requested = request.notifications(); return Mono.just(McpAsyncResponse.streaming(notifier -> listen(ctx, intersect(requested), notifier))); } 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 b5172d1a4..115a7d8e5 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 @@ -7,13 +7,16 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.stream.Stream; import io.modelcontextprotocol.common.McpTransportContext; import io.modelcontextprotocol.json.McpJsonDefaults; import io.modelcontextprotocol.modern.JsonRpc.JSONRPCNotification; import io.modelcontextprotocol.modern.JsonRpc.JSONRPCRequest; import io.modelcontextprotocol.modern.JsonRpc.JSONRPCResponse; +import io.modelcontextprotocol.modern.JsonRpc.JSONRPCResponse.JSONRPCError; import io.modelcontextprotocol.modern.McpSchema.CacheScope; +import io.modelcontextprotocol.modern.McpSchema.ErrorCodes; import io.modelcontextprotocol.modern.McpSchema.MetaKeys; import io.modelcontextprotocol.modern.McpSchema.SubscriptionFilter; import io.modelcontextprotocol.modern.McpSchema.SubscriptionsAcknowledgedParams; @@ -27,6 +30,8 @@ import io.modelcontextprotocol.modern.server.feature.ServerChange; import io.modelcontextprotocol.modern.server.feature.ToolsFeature; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; @@ -35,7 +40,10 @@ import static io.modelcontextprotocol.modern.server.ModernTestFixtures.emptyTools; import static io.modelcontextprotocol.modern.server.ModernTestFixtures.meta; 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.STRING; +import static org.assertj.core.api.InstanceOfAssertFactories.type; class SubscriptionsFeatureTests { @@ -171,6 +179,77 @@ void listenAfterEarlierListenEndedReceivesChanges() { .verifyComplete(); } + @Test + void listenWithoutNotificationsIsRejected() { + McpServer server = toolsServer(new McpChangeFeed()); + + Map params = new HashMap<>(); + params.put("_meta", meta()); + McpTransportResponse response = server + .handle(McpTransportContext.EMPTY, new JSONRPCRequest("subscriptions/listen", 1, params)) + .block(); + + assertNotificationsRequired(response); + } + + @Test + void listenWithNullNotificationsIsRejected() { + McpServer server = toolsServer(new McpChangeFeed()); + + Map params = new HashMap<>(); + params.put("_meta", meta()); + params.put("notifications", null); + McpTransportResponse response = server + .handle(McpTransportContext.EMPTY, new JSONRPCRequest("subscriptions/listen", 1, params)) + .block(); + + assertNotificationsRequired(response); + } + + @ParameterizedTest + @MethodSource("nonObjectNotifications") + void listenWithNonObjectNotificationsIsRejected(Object notifications) { + McpServer server = toolsServer(new McpChangeFeed()); + + Map params = new HashMap<>(); + params.put("_meta", meta()); + params.put("notifications", notifications); + McpTransportResponse response = server + .handle(McpTransportContext.EMPTY, new JSONRPCRequest("subscriptions/listen", 1, params)) + .block(); + + assertNotificationsRequired(response); + } + + static Stream nonObjectNotifications() { + // "" and [] matter in particular: a mapper configured with + // ACCEPT_EMPTY_STRING_AS_NULL_OBJECT / ACCEPT_EMPTY_ARRAY_AS_NULL_OBJECT would + // coerce them to null, which SubscriptionsListenRequest turns into an empty + // filter. + return Stream.of("", List.of(), true, 1); + } + + @Test + void emptyNotificationsIsAcknowledgedAndStaysOpen() { + McpChangeFeed feed = new McpChangeFeed(); + McpServer server = toolsServer(feed); + + Map params = new HashMap<>(); + params.put("_meta", meta()); + params.put("notifications", Map.of()); + var invocation = (McpTransportResponse.Streaming) server + .handle(McpTransportContext.EMPTY, new JSONRPCRequest("subscriptions/listen", 1, params)) + .block(); + + StepVerifier.create(invocation.messages()) + .assertNext(msg -> assertThat(acknowledgedNotifications((JSONRPCNotification) msg)) + .isEqualTo(SubscriptionFilter.EMPTY)) + .then(() -> feed.broadcast(new ServerChange.ToolsListChanged())) + .then(server::closeGracefully) + .expectNextMatches(msg -> msg instanceof JSONRPCResponse) + .verifyComplete(); + } + @Test void resourceSubscribeIsAdvertisedAndHonouredWhenRepositorySupportsIt() { McpServer server = resourcesServer(true, new McpChangeFeed()); @@ -268,6 +347,15 @@ private static McpTransportResponse.Streaming listen(McpServer server, int id) { return (McpTransportResponse.Streaming) server.handle(McpTransportContext.EMPTY, request).block(); } + private static void assertNotificationsRequired(McpTransportResponse response) { + // An error response, not a stream: no acknowledgement is ever sent. + assertThat(response).asInstanceOf(type(McpTransportResponse.Error.class)) + .extracting(error -> error.response().error()) + .returns(ErrorCodes.INVALID_PARAMS, JSONRPCError::code) + .extracting(JSONRPCError::message, as(STRING)) + .contains("notifications", "required"); + } + private static Object subscriptionIdOf(JSONRPCNotification notification) { return ((SubscriptionsAcknowledgedParams) notification.params()).meta().get(MetaKeys.SUBSCRIPTION_ID); } diff --git a/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/transport/HttpServletMcpTransportIntegrationTests.java b/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/transport/HttpServletMcpTransportIntegrationTests.java index 68c56aefe..eeb85f237 100644 --- a/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/transport/HttpServletMcpTransportIntegrationTests.java +++ b/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/transport/HttpServletMcpTransportIntegrationTests.java @@ -296,6 +296,20 @@ void errorIsAnsweredAsJsonWithTheStatusForItsCode() throws Exception { assertThat(errorCode(response)).isEqualTo(ErrorCodes.INVALID_PARAMS); } + @Test + void listenWithoutNotificationsIsAnsweredAsJsonError() throws Exception { + HttpRequest request = post("subscriptions/listen", Map.of("_meta", meta())).build(); + // Bounded: an accepted listen would open an event stream that never ends. + HttpResponse response = HttpClient.newHttpClient() + .sendAsync(request, HttpResponse.BodyHandlers.ofString()) + .get(5, TimeUnit.SECONDS); + + assertThat(response.statusCode()).isEqualTo(400); + assertThat(response.headers().firstValue("Content-Type")) + .hasValueSatisfying(v -> assertThat(v).contains("application/json")); + assertThat(errorCode(response)).isEqualTo(ErrorCodes.INVALID_PARAMS); + } + @Test void customErrorCodeUsesConfiguredStatus() throws Exception { HttpResponse response = callTool("quota"); diff --git a/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/transport/StdioMcpTransportTests.java b/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/transport/StdioMcpTransportTests.java index 39ffa5ebf..f7303ffcf 100644 --- a/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/transport/StdioMcpTransportTests.java +++ b/mcp-test/src/test/java/io/modelcontextprotocol/modern/server/transport/StdioMcpTransportTests.java @@ -50,6 +50,7 @@ import io.modelcontextprotocol.modern.server.McpSyncResponse; import io.modelcontextprotocol.modern.server.McpTransportResponse; import io.modelcontextprotocol.modern.server.feature.McpAsyncTool; +import io.modelcontextprotocol.modern.server.feature.McpChangeFeed; import io.modelcontextprotocol.modern.server.feature.McpChangePublisher; import io.modelcontextprotocol.modern.server.feature.McpSyncTool; import io.modelcontextprotocol.modern.server.feature.McpSyncToolRepository; @@ -623,6 +624,24 @@ void closeGracefullyCancelsListenStreams() throws Exception { await().atMost(Duration.ofSeconds(1)).untilTrue(publisherCancelled); } + @Test + void listenWithoutNotificationsIsAnsweredWithAnError() throws Exception { + McpServer server = McpServer.builder() + .serverInfo(SERVER_INFO) + .jsonMapper(this.jsonMapper) + .feature(ToolsFeature.ofAsync(emptyTools(), this.jsonMapper, PERMISSIVE_VALIDATOR, 0L, CacheScope.PRIVATE)) + .subscriptions(new McpChangeFeed()) + .build(); + start(server); + + send("subscriptions/listen", 3, Map.of("_meta", meta())); + + // The error is the first line: no acknowledgement precedes it. + String line = readLineWithTimeout(); + assertThatJson(line).inPath("$.id").isEqualTo(3); + assertThatJson(line).inPath("$.error.code").isEqualTo(ErrorCodes.INVALID_PARAMS); + } + @Test void closeGracefullyCompletesTheStartMono() throws Exception { McpRequestManager manager = managerOf((transportContext, request) -> Mono