Support subscriptions/listen on the stdio MCP server

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

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<String, ?> 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<String, Object> tool(String name) {
+        return Map.of("name", name, "inputSchema", Map.of("type", "object", "properties", Map.of()));
+    }
+
+    private static List<String> 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<String, Document> 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();
+    }
+}

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,

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");

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<Consumer<JsonRpcResponse>> responseWriters =
             new CopyOnWriteArrayList<>();
+    private final CopyOnWriteArrayList<CatalogListener> 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<JsonRpcResponse> 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) {

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