subscription/listen without params or notifications returns error

mcp-core/src/main/java/io/modelcontextprotocol/modern/server/SubscriptionsFeature.java

@@ -71,15 +71,22 @@ public Set<String> methods() {
 
 	@Override
 	public Mono<McpAsyncResponse<Result>> 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)));
 	}
 

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<String, Object> 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<String, Object> 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<String, Object> 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<Object> 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<String, Object> 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);
 	}

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<String> 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<String> response = callTool("quota");

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