diff --git a/mcp/mcp-server/README.md b/mcp/mcp-server/README.md index 62a51350b..557e87acc 100644 --- a/mcp/mcp-server/README.md +++ b/mcp/mcp-server/README.md @@ -176,6 +176,28 @@ support stateless discovery, the engine initializes that remote independently using the proxy server's identity and the highest-priority compatible stateful protocol. +## List-change notifications + +When the tools or prompts of a proxied server change, or a service or remote +client is added to a running stdio server, the server tells its client: + +- A connection that completed `initialize` receives the untagged + `notifications/tools/list_changed` and `notifications/prompts/list_changed` + notifications of the handshake protocol versions. +- On `2026-07-28`, a client opts in with `subscriptions/listen`. The stdio server + advertises `listChanged` for tools and prompts from `server/discover`, + acknowledges a listen request with `notifications/subscriptions/acknowledged`, + and then delivers only the requested types, each tagged with + `_meta["io.modelcontextprotocol/subscriptionId"]`. Resource notifications are + not accepted on subscriptions. The client ends a subscription with + `notifications/cancelled`; a graceful server shutdown answers each open listen + request with a `complete` result. +- A connection that has neither initialized nor subscribed receives no + unsolicited notifications. + +The HTTP handler does not support `subscriptions/listen` yet: it answers with +`-32601` and does not advertise `listChanged`. + ## HTTP parameter headers Annotate a Smithy input member with `smithy.ai#mcpHeader` to mirror that value in diff --git a/mcp/mcp-server/src/it/java/software/amazon/smithy/java/mcp/server/StdioSubscriptionsIntegrationTest.java b/mcp/mcp-server/src/it/java/software/amazon/smithy/java/mcp/server/StdioSubscriptionsIntegrationTest.java new file mode 100644 index 000000000..622f92b3c --- /dev/null +++ b/mcp/mcp-server/src/it/java/software/amazon/smithy/java/mcp/server/StdioSubscriptionsIntegrationTest.java @@ -0,0 +1,213 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +package software.amazon.smithy.java.mcp.server; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.jupiter.api.Test; +import software.amazon.smithy.java.core.serde.document.Document; +import software.amazon.smithy.java.mcp.model.JsonRpcRequest; +import software.amazon.smithy.java.mcp.model.JsonRpcResponse; + +class StdioSubscriptionsIntegrationTest { + @Test + void modernStdioClientLearnsAboutUpstreamToolChanges() throws Exception { + var discovered = new AtomicBoolean(); + var http = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + http.setExecutor(executor); + http.createContext("/mcp", exchange -> { + try (exchange) { + var request = McpJson.CODEC.deserializeShape(exchange.getRequestBody().readAllBytes(), + JsonRpcRequest.builder()); + if (request.getId() == null) { + exchange.sendResponseHeaders(202, -1); + return; + } + Map result; + boolean notify = false; + switch (request.getMethod()) { + case "server/discover" -> { + respond(exchange, + "application/json", + "{\"jsonrpc\":\"2.0\",\"id\":" + + McpJson.CODEC.serializeToString(request.getId()) + + ",\"error\":{\"code\":-32601,\"message\":\"Method not found\"}}"); + return; + } + case "initialize" -> result = Map.of("protocolVersion", + "2025-06-18", + "capabilities", + Map.of("tools", Map.of("listChanged", true)), + "serverInfo", + Map.of("name", "upstream", "version", "1")); + case "tools/list" -> result = Map.of("tools", + discovered.get() ? List.of(tool("discover"), tool("new_tool")) + : List.of(tool("discover"))); + case "prompts/list" -> result = Map.of("prompts", List.of()); + case "tools/call" -> { + notify = request.getParams().getMember("name").asString().equals("discover"); + if (notify) { + discovered.set(true); + } + result = Map.of("content", List.of(Map.of("type", "text", "text", "success"))); + } + default -> throw new IOException("Unexpected " + request.getMethod()); + } + var response = McpJson.CODEC.serializeToString(JsonRpcResponse.builder() + .jsonrpc("2.0") + .id(request.getId()) + .result(Document.ofObject(result)) + .build()); + if (notify) { + respond(exchange, + "text/event-stream", + "data: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/tools/list_changed\"}\n\n" + + "data: " + response + "\n\n"); + } else { + respond(exchange, "application/json", response); + } + } + }); + http.start(); + + var input = new TestInputStream(); + var output = new TestOutputStream(); + var remote = HttpMcpClient.builder() + .endpoint("http://127.0.0.1:" + http.getAddress().getPort() + "/mcp") + .build(); + var server = StdioMcpServer.builder() + .engine(McpEngine.builder().remoteClients(List.of(remote)).build()) + .input(input) + .output(output) + .build(); + server.start(); + try { + send(input, "d", "server/discover", Map.of()); + var caps = read(output).getMember("result").getMember("capabilities"); + assertTrue(caps.getMember("tools").getMember("listChanged").asBoolean()); + + send(input, + "listen:0", + "subscriptions/listen", + Map.of("notifications", + Document.of(Map.of("toolsListChanged", + Document.of(true), + "promptsListChanged", + Document.of(true))))); + var ack = read(output); + assertEquals("notifications/subscriptions/acknowledged", ack.getMember("method").asString()); + + send(input, "l1", "tools/list", Map.of()); + assertEquals(List.of("discover"), names(read(output))); + + // The upstream reports the change inside the call's response stream; the client learns about + // it through its subscription, tagged with the listen request's id, before the call result. + send(input, + "c1", + "tools/call", + Map.of("name", + Document.of("discover"), + "arguments", + Document.of(Map.of()))); + var changed = read(output); + assertEquals("notifications/tools/list_changed", changed.getMember("method").asString()); + assertEquals("listen:0", + changed.getMember("params") + .getMember("_meta") + .getMember(McpWireNames.SUBSCRIPTION_ID) + .asString()); + var callResult = read(output); + assertEquals("c1", callResult.getMember("id").asString()); + + send(input, "l2", "tools/list", Map.of()); + assertEquals(List.of("discover", "new_tool"), names(read(output))); + + send(input, + "c2", + "tools/call", + Map.of("name", + Document.of("new_tool"), + "arguments", + Document.of(Map.of()))); + var called = read(output); + assertNull(called.getMember("error"), called.toString()); + output.assertNoOutput(100); + + // Graceful shutdown ends the open subscription with a complete result. + server.shutdown().join(); + var complete = read(output); + assertEquals("listen:0", complete.getMember("id").asString()); + assertEquals("complete", complete.getMember("result").getMember("resultType").asString()); + } finally { + server.shutdown().join(); + } + } finally { + http.stop(0); + } + } + + private static Map tool(String name) { + return Map.of("name", name, "inputSchema", Map.of("type", "object", "properties", Map.of())); + } + + private static List names(Document response) { + return response.getMember("result") + .getMember("tools") + .asList() + .stream() + .map(t -> t.getMember("name").asString()) + .sorted() + .toList(); + } + + private static void respond(HttpExchange exchange, String contentType, String content) throws IOException { + var bytes = content.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", contentType); + exchange.sendResponseHeaders(200, bytes.length); + exchange.getResponseBody().write(bytes); + } + + private static void send(TestInputStream input, String id, String method, Map members) { + var params = new HashMap<>(members); + params.put("_meta", + Document.of(Map.of( + "io.modelcontextprotocol/protocolVersion", + Document.of("2026-07-28"), + "io.modelcontextprotocol/clientInfo", + Document.of(Map.of("name", + Document.of("c"), + "version", + Document.of("1"))), + "io.modelcontextprotocol/clientCapabilities", + Document.of(Map.of())))); + input.write(McpJson.CODEC.serializeToString(JsonRpcRequest.builder() + .jsonrpc("2.0") + .id(Document.of(id)) + .method(method) + .params(Document.of(params)) + .build()) + "\n"); + } + + private static Document read(TestOutputStream output) { + var line = assertTimeoutPreemptively(Duration.ofSeconds(5), output::read); + return McpJson.CODEC.createDeserializer(line.getBytes(StandardCharsets.UTF_8)).readDocument(); + } +} diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/BuiltInProtocols.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/BuiltInProtocols.java index 0239b7928..e2e74af11 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/BuiltInProtocols.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/BuiltInProtocols.java @@ -35,7 +35,8 @@ final class BuiltInProtocols { McpMethod.Standard.PROMPTS_GET, McpMethod.Standard.COMPLETION_COMPLETE, McpMethod.Standard.TOOLS_LIST, - McpMethod.Standard.TOOLS_CALL); + McpMethod.Standard.TOOLS_CALL, + McpMethod.Standard.SUBSCRIPTIONS_LISTEN); private static final BuiltInProtocol V2024_11_05 = new BuiltInProtocol( KnownProtocolVersion.V2024_11_05, diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpCall.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpCall.java index e111518b3..b215272c4 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpCall.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpCall.java @@ -29,6 +29,7 @@ public sealed interface McpCall permits McpCall.Complete, McpCall.SetLogLevel, McpCall.ReadResource, + McpCall.Listen, McpCall.Notification, McpCall.ExtensionCall, McpCall.UnknownCall { @@ -171,6 +172,25 @@ public McpMethod method() { } } + /** + * Opens a long-lived notification stream ({@code subscriptions/listen}, protocol 2026-07-28). + * + * @param id the JSON-RPC id, which also identifies the subscription + * @param notifications the notification types the client opts in to + * @param metadata request metadata + */ + record Listen(Document id, McpSubscriptionFilter notifications, McpMetadata metadata) implements McpCall { + public Listen { + Objects.requireNonNull(notifications, "notifications"); + metadata = metadata == null ? McpMetadata.EMPTY : metadata; + } + + @Override + public McpMethod method() { + return McpMethod.Standard.SUBSCRIPTIONS_LISTEN; + } + } + record Notification(McpMethod.Standard method, Document params, McpMetadata metadata) implements McpCall { public Notification { Objects.requireNonNull(method, "method"); diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpCatalog.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpCatalog.java index 9fdad47d4..689ede17e 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpCatalog.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpCatalog.java @@ -62,6 +62,7 @@ final class McpCatalog implements McpSources { new CopyOnWriteArrayList<>(); private final CopyOnWriteArrayList> responseWriters = new CopyOnWriteArrayList<>(); + private final CopyOnWriteArrayList listeners = new CopyOnWriteArrayList<>(); private volatile JsonRpcRequest initializeRequest; private volatile McpProtocol initializeProtocol; @@ -206,6 +207,13 @@ public void bindTransport( runOnce(remoteStart, () -> forEachRemoteInParallel("start", McpRemoteClient::start)); } + @Override + public void bindTransport(CatalogListener listener, Consumer responseWriter) { + listeners.addIfAbsent(listener); + responseWriters.addIfAbsent(responseWriter); + runOnce(remoteStart, () -> forEachRemoteInParallel("start", McpRemoteClient::start)); + } + @Override public void initializeRemoteClients(McpProtocol protocol) { var request = initializeRequest(protocol); @@ -409,12 +417,24 @@ private JsonRpcRequest initializeRequest(McpProtocol protocol) { } private void onRemoteNotification(McpRemoteClient client, JsonRpcRequest notification) { - if (McpMethod.Standard.NOTIFICATIONS_TOOLS_LIST_CHANGED.wireName().equals(notification.getMethod())) { + var method = McpMethod.parse(notification.getMethod()); + boolean listChanged = false; + if (method == McpMethod.Standard.NOTIFICATIONS_TOOLS_LIST_CHANGED) { invalidate(client, toolRefreshStates, this::refreshTools); - } else if (McpMethod.Standard.NOTIFICATIONS_PROMPTS_LIST_CHANGED.wireName().equals(notification.getMethod())) { + listChanged = true; + } else if (method == McpMethod.Standard.NOTIFICATIONS_PROMPTS_LIST_CHANGED) { invalidate(client, promptRefreshStates, this::refreshPrompts); + listChanged = true; } + // Invalidate before delivery so a re-list triggered by the notification is fresh (#1377). notificationWriters.forEach(writer -> writer.accept(notification)); + for (var listener : listeners) { + if (listChanged) { + listener.onListChanged((McpMethod.Standard) method, notification); + } else { + listener.onNotification(notification); + } + } } private void writeResponse(JsonRpcResponse response) { diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpDomainOperations.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpDomainOperations.java index 4bf75fe0f..02d2fd541 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpDomainOperations.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpDomainOperations.java @@ -80,7 +80,7 @@ public McpOutcome ping(McpCall.Ping call, McpRequestContext context) { public McpOutcome discover(McpCall.Discover call, McpRequestContext context) { var protocol = protocols.require(context.protocolVersion()); sources.ensureRemoteCatalogLoaded(protocol); - var capabilities = discoverCapabilities(protocol); + var capabilities = discoverCapabilities(protocol, context.transport()); return new McpOutcome.Success( call.id(), Document.of(Map.of( @@ -184,17 +184,47 @@ private Capabilities initializeCapabilities( return builder.build(); } - private Document discoverCapabilities(McpProtocol protocol) { + private Document discoverCapabilities(McpProtocol protocol, McpTransportContext transport) { + var deliverable = subscribable(protocol, transport); var capabilities = new HashMap(); - if (supports(protocol, McpMethod.Standard.TOOLS_LIST)) { + if (deliverable.toolsListChanged()) { + capabilities.put("tools", Document.of(Map.of("listChanged", Document.of(true)))); + } else if (supports(protocol, McpMethod.Standard.TOOLS_LIST)) { capabilities.put("tools", Document.of(Map.of())); } - if (supports(protocol, McpMethod.Standard.PROMPTS_LIST)) { + if (deliverable.promptsListChanged()) { + capabilities.put("prompts", Document.of(Map.of("listChanged", Document.of(true)))); + } else if (supports(protocol, McpMethod.Standard.PROMPTS_LIST)) { capabilities.put("prompts", Document.of(Map.of())); } return Document.of(capabilities); } + /** + * Notification types this server can deliver on a subscription over {@code transport}. + */ + private McpSubscriptionFilter subscribable(McpProtocol protocol, McpTransportContext transport) { + if (!transport.supportsSubscriptions() || !supports(protocol, McpMethod.Standard.SUBSCRIPTIONS_LISTEN)) { + return McpSubscriptionFilter.NONE; + } + return new McpSubscriptionFilter( + supports(protocol, McpMethod.Standard.TOOLS_LIST), + supports(protocol, McpMethod.Standard.PROMPTS_LIST), + false, + List.of()); + } + + @Override + public McpOutcome listen(McpCall.Listen call, McpRequestContext context) { + var protocol = protocols.require(context.protocolVersion()); + if (!context.transport().supportsSubscriptions()) { + throw protocol.unsupported(call.method()); + } + return new McpOutcome.Subscribed( + call.id(), + call.notifications().intersect(subscribable(protocol, context.transport()))); + } + private boolean supports(McpProtocol protocol, McpMethod.Standard method) { return protocol.supportedMethods().contains(method); } diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpEngine.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpEngine.java index 22d978a6a..7803abdc5 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpEngine.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpEngine.java @@ -102,6 +102,10 @@ public McpOutcome execute(McpCall call, McpRequestContext requestContext) { * *

This is primarily useful for transport adapters. Application code should * prefer the typed-call overload. + * + *

An accepted {@code subscriptions/listen} returns {@code null}, like a notification: the + * request stays open, and delivering its acknowledgment and notifications is up to the transport + * (see {@link McpOutcome.Subscribed}). */ public JsonRpcResponse execute(JsonRpcRequest request, ProtocolVersion protocolVersion) { var session = newSession(); @@ -131,7 +135,11 @@ McpOutcome execute( } catch (RuntimeException e) { return errorOutcome(call, e); } - return execute(call, new McpRequestContext(version, transportContext, Context.create())); + var outcome = execute(call, new McpRequestContext(version, transportContext, Context.create())); + if (call instanceof McpCall.Initialize && outcome instanceof McpOutcome.Success) { + session.markHandshake(); + } + return outcome; } JsonRpcRequest encode(McpCall call) { @@ -169,6 +177,10 @@ void bindTransport( sources.bindTransport(notificationWriter, responseWriter); } + void bindTransport(McpSources.CatalogListener listener, Consumer responseWriter) { + sources.bindTransport(listener, responseWriter); + } + void addService(String id, Service service) { sources.addService(id, service); } diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpMethod.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpMethod.java index d7c74d89f..a476b0dd1 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpMethod.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpMethod.java @@ -45,7 +45,10 @@ enum Standard implements McpMethod { TOOLS_LIST("tools/list"), TOOLS_CALL("tools/call"), RESOURCES_READ("resources/read"), + SUBSCRIPTIONS_LISTEN("subscriptions/listen"), NOTIFICATIONS_INITIALIZED("notifications/initialized"), + NOTIFICATIONS_CANCELLED("notifications/cancelled"), + NOTIFICATIONS_SUBSCRIPTIONS_ACKNOWLEDGED("notifications/subscriptions/acknowledged"), NOTIFICATIONS_PROMPTS_LIST_CHANGED("notifications/prompts/list_changed"), NOTIFICATIONS_TOOLS_LIST_CHANGED("notifications/tools/list_changed"); diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpOperations.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpOperations.java index 60c13aec1..a11f434bd 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpOperations.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpOperations.java @@ -37,6 +37,14 @@ default McpOutcome setLogLevel(McpCall.SetLogLevel call, McpRequestContext conte throw new UnsupportedOperationException("logging/setLevel is not implemented"); } + /** + * Handles {@code subscriptions/listen}, typically by returning {@link McpOutcome.Subscribed} with the + * notification types the transport can deliver. + */ + default McpOutcome listen(McpCall.Listen call, McpRequestContext context) { + throw new UnsupportedOperationException("subscriptions/listen is not implemented"); + } + default McpOutcome readResource(McpCall.ReadResource call, McpRequestContext context) { throw new UnsupportedOperationException("resources/read is not implemented"); } diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpOutcome.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpOutcome.java index 27bdf91ad..4c79fd5c7 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpOutcome.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpOutcome.java @@ -13,7 +13,8 @@ * The result of blocking MCP execution. */ @SmithyUnstableApi -public sealed interface McpOutcome permits McpOutcome.Success, McpOutcome.Failure, McpOutcome.NoResponse { +public sealed interface McpOutcome + permits McpOutcome.Success, McpOutcome.Failure, McpOutcome.Subscribed, McpOutcome.NoResponse { record Success(Document id, Document result) implements McpOutcome { public Success { @@ -27,6 +28,24 @@ record Failure(Document id, McpError error) implements McpOutcome { } } + /** + * A {@code subscriptions/listen} request was accepted. + * + *

No JSON-RPC response is written now. The transport must first send + * {@code notifications/subscriptions/acknowledged} with {@code accepted}, then deliver only those + * notification types tagged with the subscription id, and keep the request open until the client + * cancels it or the transport ends it. + * + * @param id the JSON-RPC id of the listen request, which is also the subscription id + * @param accepted the requested notification types this server will deliver + */ + record Subscribed(Document id, McpSubscriptionFilter accepted) implements McpOutcome { + public Subscribed { + Objects.requireNonNull(id, "id"); + Objects.requireNonNull(accepted, "accepted"); + } + } + enum NoResponse implements McpOutcome { INSTANCE } diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpProtocol.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpProtocol.java index b2869e541..06c8bf231 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpProtocol.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpProtocol.java @@ -102,6 +102,7 @@ default McpOutcome dispatch(McpCall call, McpOperations operations, McpRequestCo yield operations.setLogLevel(setLogLevel, context); } case McpCall.ReadResource readResource -> operations.readResource(readResource, context); + case McpCall.Listen listen -> operations.listen(listen, context); case McpCall.Notification ignored -> McpOutcome.NoResponse.INSTANCE; case McpCall.ExtensionCall ignored -> throw new IllegalStateException("Extension calls are dispatched before standard calls"); diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpRequestDecoder.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpRequestDecoder.java index cddad15b5..3398c239a 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpRequestDecoder.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpRequestDecoder.java @@ -51,6 +51,10 @@ McpCall decode(JsonRpcRequest request) { new McpCall.SetLogLevel(request.getId(), optionalString(params, "level"), metadata); case McpMethod.Standard.RESOURCES_READ -> new McpCall.ReadResource(request.getId(), requiredString(params, "uri"), metadata); + case McpMethod.Standard.SUBSCRIPTIONS_LISTEN -> new McpCall.Listen( + request.getId(), + McpSubscriptionFilter.decode(member(params, "notifications")), + metadata); case McpMethod.Standard standard when standard.wireName().startsWith("notifications/") -> new McpCall.Notification(standard, params, metadata); case McpMethod.Standard standard -> diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSession.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSession.java index cfe0b6e71..ab314c161 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSession.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSession.java @@ -11,12 +11,22 @@ final class McpSession { private final ReentrantLock negotiationLock = new ReentrantLock(); private final McpProtocolRegistry protocols; private ProtocolVersion version; + // Era of the connection for unsolicited delivery; independent of the per-request version above. + private volatile boolean handshake; McpSession(McpProtocolRegistry protocols) { this.protocols = protocols; version = protocols.defaultProtocol().protocolVersion(); } + void markHandshake() { + handshake = true; + } + + boolean handshake() { + return handshake; + } + ProtocolVersion negotiate(McpCall call, ProtocolVersion transportClaim) { negotiationLock.lock(); try { diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSources.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSources.java index fb79f1318..02bad4d24 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSources.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSources.java @@ -34,6 +34,24 @@ void bindTransport( Consumer responseWriter ); + /** + * Binds a transport that applies its own delivery policy to catalog change events. + */ + void bindTransport(CatalogListener listener, Consumer responseWriter); + + /** + * Receives catalog changes and other notifications relayed from remote peers. + */ + interface CatalogListener { + /** + * A tools or prompts list changed. {@code original} is the remote notification that + * reported it, or {@code null} for a local change. + */ + void onListChanged(McpMethod.Standard method, JsonRpcRequest original); + + void onNotification(JsonRpcRequest notification); + } + void initializeRemoteClients(McpProtocol protocol); default void ensureRemoteCatalogLoaded() { diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSubscriptionFilter.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSubscriptionFilter.java new file mode 100644 index 000000000..4b9ecc91c --- /dev/null +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSubscriptionFilter.java @@ -0,0 +1,101 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +package software.amazon.smithy.java.mcp.server; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import software.amazon.smithy.java.core.serde.document.Document; +import software.amazon.smithy.model.shapes.ShapeType; +import software.amazon.smithy.utils.SmithyUnstableApi; + +/** + * The notification types requested by, or acknowledged for, a {@code subscriptions/listen} stream. + * + * @param toolsListChanged whether {@code notifications/tools/list_changed} is included + * @param promptsListChanged whether {@code notifications/prompts/list_changed} is included + * @param resourcesListChanged whether {@code notifications/resources/list_changed} is included + * @param resourceSubscriptions resource URIs whose {@code notifications/resources/updated} are included + */ +@SmithyUnstableApi +public record McpSubscriptionFilter( + boolean toolsListChanged, + boolean promptsListChanged, + boolean resourcesListChanged, + List resourceSubscriptions) { + /** + * A filter that includes no notification types. + */ + public static final McpSubscriptionFilter NONE = new McpSubscriptionFilter(false, false, false, List.of()); + + public McpSubscriptionFilter { + resourceSubscriptions = resourceSubscriptions == null ? List.of() : List.copyOf(resourceSubscriptions); + } + + static McpSubscriptionFilter decode(Document document) { + if (document == null || !(document.isType(ShapeType.MAP) || document.isType(ShapeType.STRUCTURE))) { + throw new McpProtocolException(-32602, "Missing or invalid notifications filter"); + } + var uris = document.getMember("resourceSubscriptions"); + List resources = List.of(); + if (uris != null) { + if (!uris.isType(ShapeType.LIST)) { + throw new McpProtocolException(-32602, "resourceSubscriptions must be an array of strings"); + } + resources = uris.asList().stream().map(value -> { + if (!value.isType(ShapeType.STRING)) { + throw new McpProtocolException(-32602, "resourceSubscriptions must be an array of strings"); + } + return value.asString(); + }).toList(); + } + return new McpSubscriptionFilter( + flag(document, "toolsListChanged"), + flag(document, "promptsListChanged"), + flag(document, "resourcesListChanged"), + resources); + } + + private static boolean flag(Document document, String name) { + var value = document.getMember(name); + if (value == null) { + return false; + } + if (!value.isType(ShapeType.BOOLEAN)) { + throw new McpProtocolException(-32602, name + " must be a boolean"); + } + return value.asBoolean(); + } + + /** + * Returns the subset of this filter that is also enabled in {@code supported}. + */ + public McpSubscriptionFilter intersect(McpSubscriptionFilter supported) { + return new McpSubscriptionFilter( + toolsListChanged && supported.toolsListChanged, + promptsListChanged && supported.promptsListChanged, + resourcesListChanged && supported.resourcesListChanged, + resourceSubscriptions.stream().filter(supported.resourceSubscriptions::contains).toList()); + } + + Document toDocument() { + var values = new HashMap(); + if (toolsListChanged) { + values.put("toolsListChanged", Document.of(true)); + } + if (promptsListChanged) { + values.put("promptsListChanged", Document.of(true)); + } + if (resourcesListChanged) { + values.put("resourcesListChanged", Document.of(true)); + } + if (!resourceSubscriptions.isEmpty()) { + values.put("resourceSubscriptions", + Document.of(resourceSubscriptions.stream().map(Document::of).toList())); + } + return Document.of(Map.copyOf(values)); + } +} diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSubscriptionRegistry.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSubscriptionRegistry.java new file mode 100644 index 000000000..ee94e532a --- /dev/null +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpSubscriptionRegistry.java @@ -0,0 +1,148 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +package software.amazon.smithy.java.mcp.server; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.function.Consumer; +import software.amazon.smithy.java.core.schema.SerializableStruct; +import software.amazon.smithy.java.core.serde.document.Document; +import software.amazon.smithy.java.mcp.model.JsonRpcErrorResponse; +import software.amazon.smithy.java.mcp.model.JsonRpcRequest; +import software.amazon.smithy.java.mcp.model.JsonRpcResponse; + +/** + * Connection-scoped {@code subscriptions/listen} streams sharing one ordered output channel. + * + *

Every mutation and every write happens while holding {@code lock}, the same monitor the + * transport uses for all frame writes. Registering a subscription and writing its + * acknowledgment is therefore atomic with respect to event delivery: no notification tagged + * with a subscription id can precede that subscription's acknowledgment, and no message is + * written for a subscription after it is cancelled. + */ +final class McpSubscriptionRegistry { + private final Object lock; + private final Consumer writer; + private final McpServerIdentity identity; + private final Map active = new LinkedHashMap<>(); + + McpSubscriptionRegistry(Object lock, Consumer writer, McpServerIdentity identity) { + this.lock = lock; + this.writer = writer; + this.identity = identity; + } + + void open(McpOutcome.Subscribed subscribed) { + var key = StdioMcpClient.requestKey(subscribed.id()); + synchronized (lock) { + if (active.containsKey(key)) { + writer.accept(JsonRpcResponse.builder() + .jsonrpc("2.0") + .id(subscribed.id()) + .error(JsonRpcErrorResponse.builder() + .code(-32600) + .message("A subscription with this id is already active") + .build()) + .build()); + return; + } + writer.accept(notification( + McpMethod.Standard.NOTIFICATIONS_SUBSCRIPTIONS_ACKNOWLEDGED, + subscribed.id(), + subscribed.accepted().toDocument())); + active.put(key, new Subscription(subscribed.id(), subscribed.accepted())); + } + } + + void cancel(Document requestId) { + if (requestId == null) { + return; + } + final String key; + try { + key = StdioMcpClient.requestKey(requestId); + } catch (IllegalStateException e) { + return; // Malformed cancellation: ignore per the cancellation rules. + } + synchronized (lock) { + active.remove(key); + } + } + + /** + * Writes a tagged list-changed notification to every subscription that requested it. + * Callers must hold {@code lock} when combining this with other writes. + */ + void deliver(McpMethod.Standard method) { + synchronized (lock) { + for (var subscription : active.values()) { + if (subscription.accepts(method)) { + writer.accept(notification(method, subscription.id(), null)); + } + } + } + } + + /** + * Ends every subscription gracefully with a {@code complete} result. + */ + void completeAll() { + synchronized (lock) { + for (var subscription : active.values()) { + writer.accept(JsonRpcResponse.builder() + .jsonrpc("2.0") + .id(subscription.id()) + .result(Document.of(Map.of( + "resultType", + Document.of("complete"), + "_meta", + Document.of(Map.of( + McpWireNames.SUBSCRIPTION_ID, + subscription.id(), + McpWireNames.SERVER_INFO, + Document.of(Map.of( + "name", + Document.of(identity.name()), + "version", + Document.of(identity.version())))))))) + .build()); + } + active.clear(); + } + } + + /** + * Drops every subscription without writing: the transport closed underneath it. + */ + void clear() { + synchronized (lock) { + active.clear(); + } + } + + private static JsonRpcRequest notification(McpMethod.Standard method, Document id, Document filter) { + var params = new LinkedHashMap(); + params.put("_meta", Document.of(Map.of(McpWireNames.SUBSCRIPTION_ID, id))); + if (filter != null) { + params.put("notifications", filter); + } + return JsonRpcRequest.builder() + .jsonrpc("2.0") + .method(method.wireName()) + .params(Document.of(params)) + .build(); + } + + private record Subscription(Document id, McpSubscriptionFilter filter) { + boolean accepts(McpMethod.Standard method) { + return switch (method) { + case NOTIFICATIONS_TOOLS_LIST_CHANGED -> filter.toolsListChanged(); + case NOTIFICATIONS_PROMPTS_LIST_CHANGED -> filter.promptsListChanged(); + default -> false; + }; + } + } +} diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpTransportContext.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpTransportContext.java index 4af58ba46..3a35b7417 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpTransportContext.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpTransportContext.java @@ -23,11 +23,23 @@ default boolean supportsServerNotifications() { return false; } + /** + * Returns whether this transport can host long-lived {@code subscriptions/listen} streams. + */ + default boolean supportsSubscriptions() { + return false; + } + record Stdio() implements McpTransportContext { @Override public boolean supportsServerNotifications() { return true; } + + @Override + public boolean supportsSubscriptions() { + return true; + } } record Http(Map> headers, boolean loopbackOnly) implements McpTransportContext { diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpWireCodec.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpWireCodec.java index 403d9277a..969e63b00 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpWireCodec.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpWireCodec.java @@ -46,6 +46,7 @@ JsonRpcRequest encode(McpCall call) { case McpCall.Complete c -> completionParams(c); case McpCall.SetLogLevel c -> Document.of(Map.of("level", Document.of(c.level()))); case McpCall.ReadResource c -> Document.of(Map.of("uri", Document.of(c.uri()))); + case McpCall.Listen c -> Document.of(Map.of("notifications", c.notifications().toDocument())); case McpCall.Notification c -> c.params(); case McpCall.ExtensionCall extension -> encodeExtension(extension); case McpCall.UnknownCall c -> c.params(); @@ -79,6 +80,8 @@ JsonRpcResponse encode(McpOutcome outcome) { .error(error.build()) .build(); } + // The transport acknowledges the subscription and answers the request when it ends. + case McpOutcome.Subscribed ignored -> null; case McpOutcome.NoResponse ignored -> null; }; } diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpWireNames.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpWireNames.java index c30b2c10d..b70d7cd62 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpWireNames.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/McpWireNames.java @@ -10,6 +10,7 @@ final class McpWireNames { static final String CLIENT_INFO = "io.modelcontextprotocol/clientInfo"; static final String CLIENT_CAPABILITIES = "io.modelcontextprotocol/clientCapabilities"; static final String SERVER_INFO = "io.modelcontextprotocol/serverInfo"; + static final String SUBSCRIPTION_ID = "io.modelcontextprotocol/subscriptionId"; private McpWireNames() {} } diff --git a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/StdioMcpServer.java b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/StdioMcpServer.java index 3028a41cf..f93cd48ba 100644 --- a/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/StdioMcpServer.java +++ b/mcp/mcp-server/src/main/java/software/amazon/smithy/java/mcp/server/StdioMcpServer.java @@ -11,6 +11,7 @@ import java.io.InputStreamReader; import java.io.OutputStream; import java.nio.charset.StandardCharsets; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; @@ -37,6 +38,9 @@ @SmithyUnstableApi public final class StdioMcpServer implements Server { private static final InternalLogger LOG = InternalLogger.getLogger(StdioMcpServer.class); + private static final Set SUBSCRIPTION_ONLY_NOTIFICATIONS = Set.of( + "notifications/resources/list_changed", + "notifications/resources/updated"); private static final byte[] TOOLS_CHANGED = """ {"jsonrpc":"2.0","method":"notifications/tools/list_changed"} """.getBytes(StandardCharsets.UTF_8); @@ -49,12 +53,14 @@ public final class StdioMcpServer implements Server { private final ExecutorService requests = Executors.newVirtualThreadPerTaskExecutor(); private final CountDownLatch done = new CountDownLatch(1); private final AtomicBoolean shuttingDown = new AtomicBoolean(); + private final McpSubscriptionRegistry subscriptions; StdioMcpServer(StdioMcpServerBuilder builder) { engine = builder.engine; session = engine.newSession(); input = builder.input; output = builder.output; + subscriptions = new McpSubscriptionRegistry(output, this::write, engine.identity()); listener = Thread.ofPlatform() .name("stdio-dispatcher") .daemon() @@ -82,8 +88,15 @@ private void listen() { continue; } + var method = McpMethod.parse(request.getMethod()); + if (method == McpMethod.Standard.NOTIFICATIONS_CANCELLED && request.getId() == null) { + // Handled in arrival order on the reader thread, after any listen it references. + var params = request.getParams(); + subscriptions.cancel(McpHttpBinding.isObject(params) ? params.getMember("requestId") : null); + continue; + } var task = requests.submit(() -> handleRequest(request)); - if (McpMethod.Standard.INITIALIZE.wireName().equals(request.getMethod())) { + if (method == McpMethod.Standard.INITIALIZE || method == McpMethod.Standard.SUBSCRIPTIONS_LISTEN) { try { task.get(); } catch (InterruptedException e) { @@ -99,19 +112,68 @@ private void listen() { LOG.error("Error reading MCP input", e); } } finally { + // EOF or read failure: the transport is gone, so subscriptions end without a response. + subscriptions.clear(); requests.shutdown(); } } private void handleRequest(JsonRpcRequest request) { var outcome = engine.execute(request, session, null, McpTransportContext.STDIO); + if (outcome instanceof McpOutcome.Subscribed subscribed) { + subscriptions.open(subscribed); + return; + } var response = engine.encode(outcome); if (response != null) { write(response); } } + /** + * Tells the client that the tool list changed. + * + *

A connection that completed {@code initialize} receives an untagged + * {@code notifications/tools/list_changed}; {@code subscriptions/listen} streams that requested tool + * changes receive a copy tagged with their subscription id. Nothing is sent to a connection that + * has neither. + */ public void refreshTools() { + publish(McpMethod.Standard.NOTIFICATIONS_TOOLS_LIST_CHANGED, null); + } + + /** + * Writes the untagged copy for a handshake connection and tagged copies for matching subscriptions, + * under the output monitor so they cannot interleave with an acknowledgment. + * + * @param original the remote notification that reported the change, written unchanged to a + * handshake connection; {@code null} for a local change + */ + private void publish(McpMethod.Standard method, JsonRpcRequest original) { + synchronized (output) { + if (session.handshake()) { + if (original != null) { + write(original); + } else if (method == McpMethod.Standard.NOTIFICATIONS_TOOLS_LIST_CHANGED) { + writeToolsChanged(); + } + } + subscriptions.deliver(method); + } + } + + /** + * A local change can affect both tools and prompts. Handshake connections keep receiving the + * single untagged tools notification they always have. + */ + private void publishLocalChange() { + synchronized (output) { + publish(McpMethod.Standard.NOTIFICATIONS_TOOLS_LIST_CHANGED, null); + subscriptions.deliver(McpMethod.Standard.NOTIFICATIONS_PROMPTS_LIST_CHANGED); + } + } + + private void writeToolsChanged() { try { synchronized (output) { output.write(TOOLS_CHANGED); @@ -122,14 +184,34 @@ public void refreshTools() { } } + /** + * Applies this connection's delivery policy to catalog events. + */ + private final class CatalogEvents implements McpSources.CatalogListener { + @Override + public void onListChanged(McpMethod.Standard method, JsonRpcRequest original) { + publish(method, original); + } + + @Override + public void onNotification(JsonRpcRequest notification) { + // 2026-07-28 delivers resource change notifications only on subscriptions, and resource + // subscriptions are not accepted, so only handshake connections receive them. + if (SUBSCRIPTION_ONLY_NOTIFICATIONS.contains(notification.getMethod()) && !session.handshake()) { + return; + } + write(notification); + } + } + public void addService(String id, Service service) { engine.addService(id, service); - refreshTools(); + publishLocalChange(); } public void addRemoteClient(McpRemoteClient client) { engine.addRemoteClient(client); - refreshTools(); + publishLocalChange(); } public boolean containsServer(String id) { @@ -165,13 +247,14 @@ private JsonRpcResponse parseError() { @Override public void start() { - engine.bindTransport(this::write, this::write); + engine.bindTransport(new CatalogEvents(), this::write); listener.start(); } @Override public CompletableFuture shutdown() { if (shuttingDown.compareAndSet(false, true)) { + subscriptions.completeAll(); requests.shutdownNow(); engine.close(); try { diff --git a/mcp/mcp-server/src/test/java/software/amazon/smithy/java/mcp/server/StdioMcpServerSubscriptionsTest.java b/mcp/mcp-server/src/test/java/software/amazon/smithy/java/mcp/server/StdioMcpServerSubscriptionsTest.java new file mode 100644 index 000000000..e8124dc00 --- /dev/null +++ b/mcp/mcp-server/src/test/java/software/amazon/smithy/java/mcp/server/StdioMcpServerSubscriptionsTest.java @@ -0,0 +1,642 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +package software.amazon.smithy.java.mcp.server; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.PipedInputStream; +import java.io.PipedOutputStream; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import software.amazon.smithy.java.core.serde.document.Document; +import software.amazon.smithy.java.mcp.model.JsonRpcErrorResponse; +import software.amazon.smithy.java.mcp.model.JsonRpcRequest; +import software.amazon.smithy.java.mcp.model.JsonRpcResponse; +import software.amazon.smithy.java.mcp.model.PromptInfo; +import software.amazon.smithy.java.mcp.model.ToolInfo; + +class StdioMcpServerSubscriptionsTest { + private static final String TOOLS_CHANGED = "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/tools/list_changed\"}"; + + private TestInputStream input; + private TestOutputStream output; + private StdioMcpServer server; + private FakeRemote remote; + + @BeforeEach + void setUp() { + input = new TestInputStream(); + output = new TestOutputStream(); + remote = new FakeRemote("fake"); + } + + @AfterEach + void tearDown() { + if (server != null) { + server.shutdown().join(); + } + } + + @Test + void discoverAdvertisesListChangedOnStdio() { + start(); + + send("d", "server/discover", modern(Map.of())); + var capabilities = read().getMember("result").getMember("capabilities"); + + assertTrue(capabilities.getMember("tools").getMember("listChanged").asBoolean()); + assertTrue(capabilities.getMember("prompts").getMember("listChanged").asBoolean()); + } + + @Test + void listenAcknowledgesTheSupportedSubsetAndTagsChanges() { + start(); + + send("listen:0", + "subscriptions/listen", + listen(Map.of( + "toolsListChanged", + Document.of(true), + "resourcesListChanged", + Document.of(true), + "somethingNew", + Document.of(true)))); + var ack = read(); + + assertEquals("notifications/subscriptions/acknowledged", ack.getMember("method").asString()); + assertNull(ack.getMember("id")); + assertEquals("listen:0", subscriptionId(ack).asString()); + var accepted = ack.getMember("params").getMember("notifications"); + assertTrue(accepted.getMember("toolsListChanged").asBoolean()); + assertNull(accepted.getMember("promptsListChanged")); + assertNull(accepted.getMember("resourcesListChanged")); + assertNull(accepted.getMember("somethingNew")); + + server.refreshTools(); + var changed = read(); + assertEquals("notifications/tools/list_changed", changed.getMember("method").asString()); + assertEquals("listen:0", subscriptionId(changed).asString()); + // No untagged copy: this connection never completed initialize. + output.assertNoOutput(100); + } + + @Test + void subscriptionIdsKeepTheirJsonType() { + start(); + + send("7", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + assertEquals("\"7\"", rawSubscriptionId(output.read())); + send(7, "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + assertEquals("7", rawSubscriptionId(output.read())); + + server.refreshTools(); + var ids = new ArrayList(); + for (int i = 0; i < 2; i++) { + ids.add(rawSubscriptionId(output.read())); + } + assertEquals(List.of("\"7\"", "7"), ids.stream().sorted().toList()); + output.assertNoOutput(100); + } + + @Test + void invalidFiltersAreRejected() { + start(); + + send("missing", "subscriptions/listen", modern(Map.of())); + assertEquals(-32602, errorCode(read())); + send("not-boolean", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of("yes")))); + assertEquals(-32602, errorCode(read())); + send("not-array", + "subscriptions/listen", + listen(Map.of("resourceSubscriptions", Document.of("file:///a")))); + assertEquals(-32602, errorCode(read())); + send("not-strings", + "subscriptions/listen", + listen(Map.of("resourceSubscriptions", Document.of(List.of(Document.of(1)))))); + assertEquals(-32602, errorCode(read())); + + send("ok", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + assertEquals("ok", subscriptionId(read()).asString()); + } + + @Test + void emptyAcceptedFilterIsStillAcknowledged() { + start(); + + send("resources-only", "subscriptions/listen", listen(Map.of("resourcesListChanged", Document.of(true)))); + var ack = read(); + + assertEquals("notifications/subscriptions/acknowledged", ack.getMember("method").asString()); + assertTrue(ack.getMember("params").getMember("notifications").asStringMap().isEmpty()); + server.refreshTools(); + output.assertNoOutput(100); + } + + @Test + void handshakeEraListenIsMethodNotFound() { + start(); + initialize(); + + send(2, + "subscriptions/listen", + Document.of(Map.of( + "notifications", + Document.of(Map.of("toolsListChanged", Document.of(true)))))); + + assertEquals(-32601, errorCode(read())); + } + + @Test + void duplicateActiveSubscriptionIdIsRejected() { + start(); + + send("s", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + read(); + send("s", "subscriptions/listen", listen(Map.of("promptsListChanged", Document.of(true)))); + var duplicate = read(); + + assertEquals("s", duplicate.getMember("id").asString()); + assertEquals(-32600, errorCode(duplicate)); + server.refreshTools(); + assertEquals("s", subscriptionId(read()).asString()); + output.assertNoOutput(100); + } + + @Test + void cancellationStopsDeliveryAndMatchesTheIdType() { + start(); + send("7", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + read(); + send(7, "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + read(); + + // Unknown and malformed cancellations are ignored without a response. + cancel(Document.of("unknown")); + writeLine(McpJson.CODEC.serializeToString(JsonRpcRequest.builder() + .jsonrpc("2.0") + .method("notifications/cancelled") + .params(Document.of(Map.of("reason", Document.of("no id")))) + .build())); + // The string id cancels only the string subscription. + cancel(Document.of("7")); + barrier(); + + server.refreshTools(); + assertEquals("7", rawSubscriptionId(output.read())); + output.assertNoOutput(100); + + cancel(Document.of(7)); + barrier(); + server.refreshTools(); + output.assertNoOutput(100); + + // A cancelled subscription is never answered, even on graceful shutdown. + server.shutdown().join(); + server = null; + output.assertNoOutput(100); + } + + @Test + void shutdownCompletesOpenSubscriptions() { + start(); + send("s", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + read(); + + server.shutdown().join(); + server = null; + var complete = read(); + + assertEquals("s", complete.getMember("id").asString()); + assertNull(complete.getMember("error")); + var result = complete.getMember("result"); + assertEquals("complete", result.getMember("resultType").asString()); + assertEquals("s", result.getMember("_meta").getMember(McpWireNames.SUBSCRIPTION_ID).asString()); + assertEquals("test-server", + result.getMember("_meta").getMember(McpWireNames.SERVER_INFO).getMember("name").asString()); + } + + @Test + void endOfInputDropsSubscriptionsWithoutAResponse() throws Exception { + var clientEnd = new PipedOutputStream(); + var serverInput = new PipedInputStream(clientEnd, 64 * 1024); + server = StdioMcpServer.builder() + .engine(engine(McpInterceptor.NOOP)) + .input(serverInput) + .output(output) + .build(); + server.start(); + + clientEnd.write((McpJson.CODEC.serializeToString(request( + "s", + "subscriptions/listen", + listen(Map.of("toolsListChanged", Document.of(true))))) + "\n") + .getBytes(StandardCharsets.UTF_8)); + clientEnd.flush(); + read(); + clientEnd.close(); + + // Once input ends, the stream is gone: neither changes nor a completion are written. + assertTimeoutPreemptively(Duration.ofSeconds(2), () -> { + while (true) { + server.refreshTools(); + if (!output.hasOutput()) { + Thread.sleep(50); + if (!output.hasOutput()) { + return; + } + } + output.read(); + } + }); + server.shutdown().join(); + server = null; + output.assertNoOutput(100); + } + + @Test + void untaggedChangesAreSentOnlyAfterInitialize() { + start(); + + server.refreshTools(); + output.assertNoOutput(100); + + initialize(); + server.refreshTools(); + assertEquals(TOOLS_CHANGED, output.read().strip()); + remote.emit("notifications/tools/list_changed", null); + assertEquals(TOOLS_CHANGED, output.read().strip()); + output.assertNoOutput(100); + } + + @Test + void remoteChangesReachOnlyMatchingSubscriptionsWithTheirOwnTag() { + start(); + send("d", "server/discover", modern(Map.of())); + read(); + send("tools", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + read(); + send("prompts", "subscriptions/listen", listen(Map.of("promptsListChanged", Document.of(true)))); + read(); + + // An upstream tag must never be echoed downstream. + remote.emit("notifications/prompts/list_changed", + Document.of(Map.of("_meta", + Document.of(Map.of(McpWireNames.SUBSCRIPTION_ID, Document.of("upstream-1")))))); + var prompts = read(); + assertEquals("notifications/prompts/list_changed", prompts.getMember("method").asString()); + assertEquals("prompts", subscriptionId(prompts).asString()); + output.assertNoOutput(100); + + remote.emit("notifications/tools/list_changed", null); + assertEquals("tools", subscriptionId(read()).asString()); + output.assertNoOutput(100); + } + + @Test + void resourceChangesAreOnlyRelayedToHandshakeConnections() { + start(); + send("d", "server/discover", modern(Map.of())); + read(); + + remote.emit("notifications/resources/list_changed", null); + remote.emit("notifications/resources/updated", Document.of(Map.of("uri", Document.of("file:///a")))); + output.assertNoOutput(100); + + // Request-scoped notifications keep flowing. + remote.emit("notifications/progress", + Document.of(Map.of("progressToken", Document.of("t"), "progress", Document.of(1)))); + assertEquals("notifications/progress", read().getMember("method").asString()); + + initialize(); + remote.emit("notifications/resources/list_changed", null); + assertEquals("notifications/resources/list_changed", read().getMember("method").asString()); + } + + @Test + void localChangesNotifyToolAndPromptSubscribers() { + start(); + send("tools", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + read(); + send("prompts", "subscriptions/listen", listen(Map.of("promptsListChanged", Document.of(true)))); + read(); + + server.addRemoteClient(new FakeRemote("added")); + + var methods = new ArrayList(); + for (int i = 0; i < 2; i++) { + var message = read(); + methods.add(message.getMember("method").asString() + "@" + subscriptionId(message).asString()); + } + assertEquals( + List.of( + "notifications/prompts/list_changed@prompts", + "notifications/tools/list_changed@tools"), + methods.stream().sorted().toList()); + output.assertNoOutput(100); + } + + @Test + void handshakeConnectionsKeepTheSingleToolsNotificationForLocalChanges() { + start(); + initialize(); + + server.addRemoteClient(new FakeRemote("added")); + + assertEquals(TOOLS_CHANGED, output.read().strip()); + output.assertNoOutput(100); + } + + @Test + void connectionsUsingBothEraReceiveBothCopies() { + start(); + initialize(); + send("s", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + read(); + + server.refreshTools(); + + assertEquals(TOOLS_CHANGED, output.read().strip()); + assertEquals("s", subscriptionId(read()).asString()); + output.assertNoOutput(100); + } + + @Test + void acknowledgmentPrecedesChangesThatRaceActivation() throws Exception { + var inAfterHook = new CountDownLatch(1); + var release = new CountDownLatch(1); + start(new McpInterceptor() { + @Override + public void readAfterExecution(McpExecutionContext context, McpOutcome outcome, RuntimeException error) { + if (context.call() instanceof McpCall.Listen) { + inAfterHook.countDown(); + try { + release.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + }); + + send("s", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + assertTrue(inAfterHook.await(2, TimeUnit.SECONDS)); + // The subscription is not active yet, so this change is not delivered to it. + server.refreshTools(); + output.assertNoOutput(100); + release.countDown(); + + assertEquals("notifications/subscriptions/acknowledged", read().getMember("method").asString()); + server.refreshTools(); + assertEquals("notifications/tools/list_changed", read().getMember("method").asString()); + } + + @Test + void interceptorsCanRejectASubscription() { + start(new McpInterceptor() { + @Override + public McpOutcome modifyAfterExecution( + McpExecutionContext context, + McpOutcome outcome, + RuntimeException error + ) { + if (context.call() instanceof McpCall.Listen listen) { + return new McpOutcome.Failure(listen.id(), new McpError(-32603, "No subscriptions", null)); + } + return McpInterceptor.super.modifyAfterExecution(context, outcome, error); + } + }); + + send("s", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))); + + assertEquals(-32603, errorCode(read())); + server.refreshTools(); + output.assertNoOutput(100); + } + + @Test + void httpNeitherAdvertisesNorAcceptsListen() { + try (var engine = engine(McpInterceptor.NOOP)) { + var handler = new McpHttpHandler(engine); + + var discover = handler.handle( + request("d", "server/discover", modern(Map.of())), + Map.of( + "MCP-Protocol-Version", + List.of("2026-07-28"), + "Mcp-Method", + List.of("server/discover"))); + assertNull(discover.body() + .getResult() + .getMember("capabilities") + .getMember("tools") + .getMember("listChanged")); + + var listen = handler.handle( + request("l", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))), + Map.of( + "MCP-Protocol-Version", + List.of("2026-07-28"), + "Mcp-Method", + List.of("subscriptions/listen"))); + assertEquals(404, listen.statusCode()); + assertEquals(-32601, listen.body().getError().getCode()); + } + } + + @Test + void engineExecuteReturnsNoResponseForAnAcceptedListen() { + try (var engine = engine(McpInterceptor.NOOP)) { + assertNull(engine.execute( + request("s", "subscriptions/listen", listen(Map.of("toolsListChanged", Document.of(true)))), + null)); + } + } + + private void start() { + start(McpInterceptor.NOOP); + } + + private void start(McpInterceptor interceptor) { + server = StdioMcpServer.builder() + .engine(engine(interceptor)) + .input(input) + .output(output) + .build(); + server.start(); + } + + private McpEngine engine(McpInterceptor interceptor) { + return McpEngine.builder() + .name("test-server") + .remoteClients(List.of(remote)) + .interceptor(interceptor) + .build(); + } + + private void initialize() { + send(1, + "initialize", + Document.of(Map.of( + "protocolVersion", + Document.of("2025-11-25"), + "capabilities", + Document.of(Map.of()), + "clientInfo", + Document.of(Map.of("name", Document.of("client"), "version", Document.of("1")))))); + assertNull(read().getMember("error")); + } + + /** + * Cancellations are applied in arrival order on the reader thread; once a later request is answered, + * every earlier cancellation has taken effect. + */ + private void barrier() { + send("barrier", "server/discover", modern(Map.of())); + assertEquals("barrier", read().getMember("id").asString()); + } + + private void cancel(Document requestId) { + writeLine(McpJson.CODEC.serializeToString(JsonRpcRequest.builder() + .jsonrpc("2.0") + .method("notifications/cancelled") + .params(Document.of(Map.of("requestId", requestId))) + .build())); + } + + private void send(Object id, String method, Document params) { + writeLine(McpJson.CODEC.serializeToString(request(id, method, params))); + } + + private void writeLine(String line) { + input.write(line + "\n"); + } + + private Document read() { + var line = assertTimeoutPreemptively(Duration.ofSeconds(2), output::read); + return McpJson.CODEC.createDeserializer(line.getBytes(StandardCharsets.UTF_8)).readDocument(); + } + + private static JsonRpcRequest request(Object id, String method, Document params) { + return JsonRpcRequest.builder() + .jsonrpc("2.0") + .id(id instanceof Integer number ? Document.of(number) : Document.of((String) id)) + .method(method) + .params(params) + .build(); + } + + private static Document listen(Map filter) { + return modern(Map.of("notifications", Document.of(filter))); + } + + private static Document modern(Map members) { + var params = new HashMap<>(members); + params.put("_meta", + Document.of(Map.of( + "io.modelcontextprotocol/protocolVersion", + Document.of("2026-07-28"), + "io.modelcontextprotocol/clientCapabilities", + Document.of(Map.of())))); + return Document.of(params); + } + + private static Document subscriptionId(Document message) { + return message.getMember("params").getMember("_meta").getMember(McpWireNames.SUBSCRIPTION_ID); + } + + /** + * Returns the subscription id exactly as written, so string and number ids stay distinguishable. + */ + private static String rawSubscriptionId(String line) { + var key = "\"" + McpWireNames.SUBSCRIPTION_ID + "\":"; + var start = line.indexOf(key) + key.length(); + var end = line.indexOf('}', start); + return line.substring(start, end).strip(); + } + + private static int errorCode(Document response) { + return response.getMember("error").getMember("code").asInteger(); + } + + /** + * A session-based remote: it rejects {@code server/discover}, so the catalog initializes it, which + * installs the notification consumer. + */ + private static final class FakeRemote extends McpRemoteClient { + private final String name; + + FakeRemote(String name) { + this.name = name; + } + + @Override + public McpPage listTools() { + return McpPage.last(List.of()); + } + + @Override + public McpPage listPrompts() { + return McpPage.last(List.of()); + } + + @Override + protected JsonRpcResponse exchange(JsonRpcRequest request) { + if (request.getId() == null) { + return null; + } + if ("server/discover".equals(request.getMethod())) { + return JsonRpcResponse.builder() + .jsonrpc("2.0") + .id(request.getId()) + .error(JsonRpcErrorResponse.builder().code(-32601).message("Method not found").build()) + .build(); + } + if ("initialize".equals(request.getMethod())) { + return JsonRpcResponse.builder() + .jsonrpc("2.0") + .id(request.getId()) + .result(Document.of(Map.of("protocolVersion", Document.of("2025-06-18")))) + .build(); + } + return JsonRpcResponse.builder() + .jsonrpc("2.0") + .id(request.getId()) + .result(Document.of(Map.of())) + .build(); + } + + void emit(String method, Document params) { + var notification = JsonRpcRequest.builder().jsonrpc("2.0").method(method); + if (params != null) { + notification.params(params); + } + notify(notification.build()); + } + + @Override + public void start() {} + + @Override + public void close() {} + + @Override + public String name() { + return name; + } + } +} diff --git a/mcp/mcp-server/src/testFixtures/java/software/amazon/smithy/java/mcp/server/TestOutputStream.java b/mcp/mcp-server/src/testFixtures/java/software/amazon/smithy/java/mcp/server/TestOutputStream.java index 08beccec6..3a67650b4 100644 --- a/mcp/mcp-server/src/testFixtures/java/software/amazon/smithy/java/mcp/server/TestOutputStream.java +++ b/mcp/mcp-server/src/testFixtures/java/software/amazon/smithy/java/mcp/server/TestOutputStream.java @@ -26,21 +26,18 @@ public void write(int b) { @Override public void write(byte[] b, int off, int len) { - int rem = len; + int end = off + len; int pos = off; - while (rem > 0) { - int nl = find(b, pos, pos + rem, (byte) '\n'); + while (pos < end) { + int nl = find(b, pos, end, (byte) '\n'); if (nl == -1) { - baos.write(b, off, len); + baos.write(b, pos, end - pos); return; - } else { - int toWrite = nl - off; - baos.write(b, off, toWrite); - lines.add(baos.toString(StandardCharsets.UTF_8)); - baos.reset(); - rem -= toWrite; - pos += toWrite; } + baos.write(b, pos, nl - pos); + lines.add(baos.toString(StandardCharsets.UTF_8)); + baos.reset(); + pos = nl + 1; } }