Skip to content

Commit 4f50bfe

Browse files
KehrlannKehrlann
committed
subscription/listen without params or notifications returns error
Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>
1 parent 97edf22 commit 4f50bfe

4 files changed

Lines changed: 116 additions & 3 deletions

File tree

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

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -71,15 +71,20 @@ public Set<String> methods() {
7171

7272
@Override
7373
public Mono<McpAsyncResponse<Result>> handle(McpRequestContext ctx, Object params) {
74+
// The spec marks `notifications` required on the listen request. Checked on the
75+
// raw params because SubscriptionsListenRequest deserializes a missing or null
76+
// value to an empty filter, which would be acknowledged as a valid subscription.
77+
if (params == null || (params instanceof Map<?, ?> paramsMap && paramsMap.get("notifications") == null)) {
78+
return Mono.error(McpException.invalidParams("Invalid params: 'notifications' is required"));
79+
}
7480
SubscriptionsListenRequest request;
7581
try {
76-
request = params == null ? null : this.jsonMapper.convertValue(params, SubscriptionsListenRequest.class);
82+
request = this.jsonMapper.convertValue(params, SubscriptionsListenRequest.class);
7783
}
7884
catch (RuntimeException ex) {
7985
return Mono.error(McpException.invalidParams("Malformed SubscriptionsListenRequest"));
8086
}
81-
SubscriptionFilter requested = request == null || request.notifications() == null ? SubscriptionFilter.EMPTY
82-
: request.notifications();
87+
SubscriptionFilter requested = request.notifications();
8388
return Mono.just(McpAsyncResponse.streaming(notifier -> listen(ctx, intersect(requested), notifier)));
8489
}
8590

‎mcp-test/src/test/java/io/modelcontextprotocol/modern/server/SubscriptionsFeatureTests.java‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,10 @@
1313
import io.modelcontextprotocol.modern.JsonRpc.JSONRPCNotification;
1414
import io.modelcontextprotocol.modern.JsonRpc.JSONRPCRequest;
1515
import io.modelcontextprotocol.modern.JsonRpc.JSONRPCResponse;
16+
import io.modelcontextprotocol.modern.JsonRpc.JSONRPCResponse.JSONRPCError;
17+
import io.modelcontextprotocol.modern.McpException;
1618
import io.modelcontextprotocol.modern.McpSchema.CacheScope;
19+
import io.modelcontextprotocol.modern.McpSchema.ErrorCodes;
1720
import io.modelcontextprotocol.modern.McpSchema.MetaKeys;
1821
import io.modelcontextprotocol.modern.McpSchema.SubscriptionFilter;
1922
import io.modelcontextprotocol.modern.McpSchema.SubscriptionsAcknowledgedParams;
@@ -35,7 +38,10 @@
3538
import static io.modelcontextprotocol.modern.server.ModernTestFixtures.emptyTools;
3639
import static io.modelcontextprotocol.modern.server.ModernTestFixtures.meta;
3740
import static io.modelcontextprotocol.modern.server.ModernTestFixtures.respond;
41+
import static org.assertj.core.api.Assertions.as;
3842
import static org.assertj.core.api.Assertions.assertThat;
43+
import static org.assertj.core.api.InstanceOfAssertFactories.STRING;
44+
import static org.assertj.core.api.InstanceOfAssertFactories.type;
3945

4046
class SubscriptionsFeatureTests {
4147

@@ -171,6 +177,66 @@ void listenAfterEarlierListenEndedReceivesChanges() {
171177
.verifyComplete();
172178
}
173179

180+
@Test
181+
void listenWithoutNotificationsIsRejected() {
182+
McpServer server = toolsServer(new McpChangeFeed());
183+
184+
Map<String, Object> params = new HashMap<>();
185+
params.put("_meta", meta());
186+
McpTransportResponse response = server
187+
.handle(McpTransportContext.EMPTY, new JSONRPCRequest("subscriptions/listen", 1, params))
188+
.block();
189+
190+
assertNotificationsRequired(response);
191+
}
192+
193+
@Test
194+
void listenWithNullNotificationsIsRejected() {
195+
McpServer server = toolsServer(new McpChangeFeed());
196+
197+
Map<String, Object> params = new HashMap<>();
198+
params.put("_meta", meta());
199+
params.put("notifications", null);
200+
McpTransportResponse response = server
201+
.handle(McpTransportContext.EMPTY, new JSONRPCRequest("subscriptions/listen", 1, params))
202+
.block();
203+
204+
assertNotificationsRequired(response);
205+
}
206+
207+
@Test
208+
void listenWithoutParamsIsRejected() {
209+
SubscriptionsFeature feature = new SubscriptionsFeature(new McpChangeFeed(), McpJsonDefaults.getMapper(), true,
210+
false, false, false);
211+
212+
StepVerifier.create(feature.handle(null, null)).expectErrorSatisfies(error -> {
213+
JSONRPCError jsonRpcError = ((McpException) error).error();
214+
assertThat(jsonRpcError.code()).isEqualTo(ErrorCodes.INVALID_PARAMS);
215+
assertThat(jsonRpcError.message()).contains("notifications").contains("required");
216+
}).verify();
217+
}
218+
219+
@Test
220+
void emptyNotificationsIsAcknowledgedAndStaysOpen() {
221+
McpChangeFeed feed = new McpChangeFeed();
222+
McpServer server = toolsServer(feed);
223+
224+
Map<String, Object> params = new HashMap<>();
225+
params.put("_meta", meta());
226+
params.put("notifications", Map.of());
227+
var invocation = (McpTransportResponse.Streaming) server
228+
.handle(McpTransportContext.EMPTY, new JSONRPCRequest("subscriptions/listen", 1, params))
229+
.block();
230+
231+
StepVerifier.create(invocation.messages())
232+
.assertNext(msg -> assertThat(acknowledgedNotifications((JSONRPCNotification) msg))
233+
.isEqualTo(SubscriptionFilter.EMPTY))
234+
.then(() -> feed.broadcast(new ServerChange.ToolsListChanged()))
235+
.then(server::closeGracefully)
236+
.expectNextMatches(msg -> msg instanceof JSONRPCResponse)
237+
.verifyComplete();
238+
}
239+
174240
@Test
175241
void resourceSubscribeIsAdvertisedAndHonouredWhenRepositorySupportsIt() {
176242
McpServer server = resourcesServer(true, new McpChangeFeed());
@@ -268,6 +334,15 @@ private static McpTransportResponse.Streaming listen(McpServer server, int id) {
268334
return (McpTransportResponse.Streaming) server.handle(McpTransportContext.EMPTY, request).block();
269335
}
270336

337+
private static void assertNotificationsRequired(McpTransportResponse response) {
338+
// An error response, not a stream: no acknowledgement is ever sent.
339+
assertThat(response).asInstanceOf(type(McpTransportResponse.Error.class))
340+
.extracting(error -> error.response().error())
341+
.returns(ErrorCodes.INVALID_PARAMS, JSONRPCError::code)
342+
.extracting(JSONRPCError::message, as(STRING))
343+
.contains("notifications", "required");
344+
}
345+
271346
private static Object subscriptionIdOf(JSONRPCNotification notification) {
272347
return ((SubscriptionsAcknowledgedParams) notification.params()).meta().get(MetaKeys.SUBSCRIPTION_ID);
273348
}

‎mcp-test/src/test/java/io/modelcontextprotocol/modern/server/transport/HttpServletMcpTransportIntegrationTests.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -296,6 +296,20 @@ void errorIsAnsweredAsJsonWithTheStatusForItsCode() throws Exception {
296296
assertThat(errorCode(response)).isEqualTo(ErrorCodes.INVALID_PARAMS);
297297
}
298298

299+
@Test
300+
void listenWithoutNotificationsIsAnsweredAsJsonError() throws Exception {
301+
HttpRequest request = post("subscriptions/listen", Map.of("_meta", meta())).build();
302+
// Bounded: an accepted listen would open an event stream that never ends.
303+
HttpResponse<String> response = HttpClient.newHttpClient()
304+
.sendAsync(request, HttpResponse.BodyHandlers.ofString())
305+
.get(5, TimeUnit.SECONDS);
306+
307+
assertThat(response.statusCode()).isEqualTo(400);
308+
assertThat(response.headers().firstValue("Content-Type"))
309+
.hasValueSatisfying(v -> assertThat(v).contains("application/json"));
310+
assertThat(errorCode(response)).isEqualTo(ErrorCodes.INVALID_PARAMS);
311+
}
312+
299313
@Test
300314
void customErrorCodeUsesConfiguredStatus() throws Exception {
301315
HttpResponse<String> response = callTool("quota");

‎mcp-test/src/test/java/io/modelcontextprotocol/modern/server/transport/StdioMcpTransportTests.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@
5050
import io.modelcontextprotocol.modern.server.McpSyncResponse;
5151
import io.modelcontextprotocol.modern.server.McpTransportResponse;
5252
import io.modelcontextprotocol.modern.server.feature.McpAsyncTool;
53+
import io.modelcontextprotocol.modern.server.feature.McpChangeFeed;
5354
import io.modelcontextprotocol.modern.server.feature.McpChangePublisher;
5455
import io.modelcontextprotocol.modern.server.feature.McpSyncTool;
5556
import io.modelcontextprotocol.modern.server.feature.McpSyncToolRepository;
@@ -623,6 +624,24 @@ void closeGracefullyCancelsListenStreams() throws Exception {
623624
await().atMost(Duration.ofSeconds(1)).untilTrue(publisherCancelled);
624625
}
625626

627+
@Test
628+
void listenWithoutNotificationsIsAnsweredWithAnError() throws Exception {
629+
McpServer server = McpServer.builder()
630+
.serverInfo(SERVER_INFO)
631+
.jsonMapper(this.jsonMapper)
632+
.feature(ToolsFeature.ofAsync(emptyTools(), this.jsonMapper, PERMISSIVE_VALIDATOR, 0L, CacheScope.PRIVATE))
633+
.subscriptions(new McpChangeFeed())
634+
.build();
635+
start(server);
636+
637+
send("subscriptions/listen", 3, Map.of("_meta", meta()));
638+
639+
// The error is the first line: no acknowledgement precedes it.
640+
String line = readLineWithTimeout();
641+
assertThatJson(line).inPath("$.id").isEqualTo(3);
642+
assertThatJson(line).inPath("$.error.code").isEqualTo(ErrorCodes.INVALID_PARAMS);
643+
}
644+
626645
@Test
627646
void closeGracefullyCompletesTheStartMono() throws Exception {
628647
McpRequestManager manager = managerOf((transportContext, request) -> Mono

0 commit comments

Comments
 (0)