From 4a16d692480e07160ee256e83d9ae210b4e48f53 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 7 Nov 2026 16:01:38 -0500 Subject: [PATCH 1/5] Serve MCP 2026-07-28 resources, prompts and listen streams, and over stdio Modern-only resources, resource templates and prompts are served next to the bridged ones, with duplicates across both lists failing the start. subscriptions/listen streams carry the changes published through getChanges(), on their own limit and without time limit; stop() first ends them with their result. Over stdio, a message whose _meta names 2026-07-28 goes to the modern exchange: streamed messages are written as separate lines, and notifications/cancelled aborts the request it names without a response. Co-Authored-By: Claude Opus 5.5 --- changes.md | 12 +- .../restlet/ext/mcp/McpModernResource.java | 35 ++ .../ext/mcp/McpModernResourceTemplate.java | 36 +++ .../java/org/restlet/ext/mcp/McpProtocol.java | 3 + .../org/restlet/ext/mcp/McpServerAdapter.java | 105 +++++- .../restlet/ext/mcp/StdioServerHelper.java | 300 ++++++++++++++++-- .../ext/mcp/internal/McpModernExchange.java | 64 +++- .../ext/mcp/internal/McpModernFeatures.java | 243 ++++++++------ .../restlet/ext/mcp/internal/McpWorkers.java | 54 +++- .../ext/mcp/McpModernServerTestCase.java | 228 +++++++++++++ .../ext/mcp/McpModernStdioTestCase.java | 281 ++++++++++++++++ 11 files changed, 1210 insertions(+), 151 deletions(-) create mode 100644 org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResource.java create mode 100644 org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResourceTemplate.java create mode 100644 org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernStdioTestCase.java diff --git a/changes.md b/changes.md index bf107ca16b..bc90ce64b1 100644 --- a/changes.md +++ b/changes.md @@ -103,8 +103,16 @@ Changes log 2026-07-28 clients only, for `input_required` results and streamed progress. The interceptors wrap each 2026-07-28 request and its final result, the tool filters and the call limit apply to both revisions, and `setRequestStateKey()` shares the key - sealing `input_required` state between instances. Not yet supported for 2026-07-28: - stdio, `subscriptions/listen`, and the MCP client. The SDK's modern API is a preview. + sealing `input_required` state between instances. `getModernResources()`, + `getModernResourceTemplates()` and `getModernPrompts()` add resources, templates and + prompts served to 2026-07-28 clients only; a URI, URI template or prompt name may only + be used once across both revisions' lists. `subscriptions/listen` streams carry the + changes published through `getChanges()`; they have their own limit, + `setMaxListenStreams()`, no time limit, and end when the adapter stops or the client + goes away. Over stdio, a message whose `_meta` names 2026-07-28 goes to the modern + server: streamed messages are written as separate lines as they come, and + `notifications/cancelled` ends the request it names without a response. Not yet + supported for 2026-07-28: the MCP client. The SDK's modern API is a preview. - Added experimental SDK-native gRPC hosting of unary, server-streaming, client-streaming and bidirectional methods on a borrowed Restlet/Netty HTTP/2 listener. Request messages are read as the application requests them, within the queue budget, diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResource.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResource.java new file mode 100644 index 0000000000..b384b95afb --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResource.java @@ -0,0 +1,35 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.mcp; + +import io.modelcontextprotocol.modern.McpSchema; +import io.modelcontextprotocol.modern.server.feature.McpAsyncResource; +import java.util.Objects; + +/** + * A resource served to MCP {@link McpProtocol#MODERN_VERSION} clients only, for {@link + * McpServerAdapter#getModernResources()}: its definition, listed, and its read handler, which may + * answer {@code input_required}. Experimental: the SDK's modern API is a preview. + * + * @param definition The resource, with its URI. + * @param resource The read handler. + */ +public record McpModernResource(McpSchema.Resource definition, McpAsyncResource resource) { + + /** + * Creates a modern resource. + * + * @param definition The resource, with its URI. + * @param resource The read handler. + */ + public McpModernResource { + Objects.requireNonNull(definition, "definition"); + Objects.requireNonNull(resource, "resource"); + } +} diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResourceTemplate.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResourceTemplate.java new file mode 100644 index 0000000000..bc46823fb0 --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResourceTemplate.java @@ -0,0 +1,36 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.mcp; + +import io.modelcontextprotocol.modern.McpSchema; +import io.modelcontextprotocol.modern.server.feature.McpAsyncResource; +import java.util.Objects; + +/** + * A resource template served to MCP {@link McpProtocol#MODERN_VERSION} clients only, for {@link + * McpServerAdapter#getModernResourceTemplates()}: its definition, listed, and the read handler of + * the URIs it matches. Experimental: the SDK's modern API is a preview. + * + * @param definition The template, with its URI template. + * @param resource The read handler of the matching URIs. + */ +public record McpModernResourceTemplate( + McpSchema.ResourceTemplate definition, McpAsyncResource resource) { + + /** + * Creates a modern resource template. + * + * @param definition The template, with its URI template. + * @param resource The read handler of the matching URIs. + */ + public McpModernResourceTemplate { + Objects.requireNonNull(definition, "definition"); + Objects.requireNonNull(resource, "resource"); + } +} diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpProtocol.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpProtocol.java index 4a50a7148f..7faadfe182 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpProtocol.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpProtocol.java @@ -26,5 +26,8 @@ public final class McpProtocol { public static final int DEFAULT_MAX_MESSAGE_BYTES = 1024 * 1024; public static final int DEFAULT_MAX_CONCURRENT_CALLS = 128; + /** The default number of open MCP 2026-07-28 {@code subscriptions/listen} streams. */ + public static final int DEFAULT_MAX_LISTEN_STREAMS = 64; + private McpProtocol() {} } diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpServerAdapter.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpServerAdapter.java index 3e989917e1..b3f979b33f 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpServerAdapter.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpServerAdapter.java @@ -14,7 +14,10 @@ import io.modelcontextprotocol.json.McpJsonMapper; import io.modelcontextprotocol.json.jackson2.JacksonMcpJsonMapper; import io.modelcontextprotocol.modern.server.HmacRequestStateCodec; +import io.modelcontextprotocol.modern.server.feature.McpAsyncPrompt; import io.modelcontextprotocol.modern.server.feature.McpAsyncTool; +import io.modelcontextprotocol.modern.server.feature.McpChangeBroadcaster; +import io.modelcontextprotocol.modern.server.feature.McpChangeFeed; import io.modelcontextprotocol.server.McpAsyncListFilter; import io.modelcontextprotocol.server.McpServer; import io.modelcontextprotocol.server.McpStatelessAsyncServer; @@ -61,6 +64,9 @@ public final class McpServerAdapter extends Restlet { private static final McpJsonMapper MAPPER = initMapper(); + /** How long stopping waits for listen streams to end with their result. */ + private static final Duration LISTEN_DRAIN = Duration.ofSeconds(2); + private static McpJsonMapper initMapper() { McpJsonMapper mapper = McpJsonDefaults.getMapper(); if (mapper instanceof JacksonMcpJsonMapper jacksonMapper) { @@ -82,7 +88,13 @@ private static McpJsonMapper initMapper() { new CopyOnWriteArrayList<>(); private final List interceptors = new CopyOnWriteArrayList<>(); private final List modernTools = new CopyOnWriteArrayList<>(); + private final List modernResources = new CopyOnWriteArrayList<>(); + private final List modernResourceTemplates = + new CopyOnWriteArrayList<>(); + private final List modernPrompts = new CopyOnWriteArrayList<>(); + private final McpChangeFeed changes = new McpChangeFeed(); private byte[] requestStateKey; + private int maxListenStreams = McpProtocol.DEFAULT_MAX_LISTEN_STREAMS; private final Set allowedOrigins = ConcurrentHashMap.newKeySet(); private McpSchema.Implementation serverInfo = McpSchema.Implementation.builder("restlet", "3.0").build(); @@ -94,6 +106,7 @@ private static McpJsonMapper initMapper() { private volatile McpStatelessAsyncServer sdkServer; private volatile io.modelcontextprotocol.modern.server.McpServer modernServer; private volatile McpModernExchange modernExchange; + private volatile McpWorkers listeners; private volatile McpWorkers workers; private volatile boolean accepting; private volatile Sinks.One stopping; @@ -147,6 +160,70 @@ public List getModernTools() { return modernTools; } + /** + * Resources served to MCP {@link McpProtocol#MODERN_VERSION} clients only, whose read may + * answer {@code input_required}. The resources of {@link #getResources()} are served to both + * revisions; a URI may only be used once across both lists. Experimental. + * + * @return The modern resources, modifiable until the adapter starts. + */ + public List getModernResources() { + return modernResources; + } + + /** + * Resource templates served to MCP {@link McpProtocol#MODERN_VERSION} clients only. The + * templates of {@link #getResourceTemplates()} are served to both revisions; a URI template may + * only be used once across both lists. Experimental. + * + * @return The modern resource templates, modifiable until the adapter starts. + */ + public List getModernResourceTemplates() { + return modernResourceTemplates; + } + + /** + * Prompts served to MCP {@link McpProtocol#MODERN_VERSION} clients only, on the SDK's modern + * types. The prompts of {@link #getPrompts()} are served to both revisions; a name may only be + * used once across both lists. Experimental. + * + * @return The modern prompts, modifiable until the adapter starts. + */ + public List getModernPrompts() { + return modernPrompts; + } + + /** + * Publishes changes to the MCP {@link McpProtocol#MODERN_VERSION} clients listening through + * {@code subscriptions/listen}: tools, prompts and resources list changes, and resource + * updates. A change nobody listens to is dropped. Experimental. + * + * @return The change broadcaster, usable at any time. + */ + public McpChangeBroadcaster getChanges() { + return changes; + } + + public synchronized int getMaxListenStreams() { + return maxListenStreams; + } + + /** + * Sets how many {@code subscriptions/listen} streams may be open at once, {@value + * McpProtocol#DEFAULT_MAX_LISTEN_STREAMS} by default. Listen streams have no time limit and + * don't count against {@link #getMaxConcurrentCalls()}; they end when the adapter stops or the + * client goes away, and carry heartbeats. + * + * @param value The number of streams, positive. + */ + public synchronized void setMaxListenStreams(int value) { + requireConfigurable(); + if (value <= 0) { + throw new IllegalArgumentException("maxListenStreams must be positive"); + } + maxListenStreams = value; + } + /** * Sets the key sealing the {@code requestState} of {@code input_required} results, at least 32 * bytes. Without it, each adapter generates its own random key, so a retry must reach the @@ -278,6 +355,7 @@ public List protocolVersions() { .toolFilters(filters -> filters.addAll(toolFilters)) .build(); workers = new McpWorkers(maxConcurrentCalls, requestTimeout); + listeners = new McpWorkers(maxListenStreams, null, requestTimeout); stopping = Sinks.one(); modernServer = modern; modernExchange = @@ -285,6 +363,7 @@ public List protocolVersions() { modernServer, MAPPER, workers, + listeners, requestTimeout, maxMessageBytes, (message, context, terminal) -> invoke(0, message, context, terminal), @@ -316,7 +395,11 @@ private io.modelcontextprotocol.modern.server.McpServer modernServer() { List.copyOf(resourceTemplates), List.copyOf(prompts), List.copyOf(toolFilters), - List.copyOf(modernTools)))); + List.copyOf(modernTools), + List.copyOf(modernResources), + List.copyOf(modernResourceTemplates), + List.copyOf(modernPrompts)))) + .subscriptions(changes); var codec = HmacRequestStateCodec.builder() .principalExtractor(McpServerAdapter::principal) @@ -356,18 +439,26 @@ public synchronized void stop() throws Exception { return; } accepting = false; + // Listen streams end with their result first; what remains is then cut short. + modernServer.closeGracefully(); + try { + listeners.awaitIdle( + requestTimeout.compareTo(LISTEN_DRAIN) < 0 ? requestTimeout : LISTEN_DRAIN); + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + } stopping.tryEmitValue(Boolean.TRUE); try { - workers.close(); + try { + workers.close(); + } finally { + listeners.close(); + } } finally { try { sdkServer.closeGracefully().block(requestTimeout); } finally { - try { - modernServer.closeGracefully(); - } finally { - super.stop(); - } + super.stop(); } } } diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioServerHelper.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioServerHelper.java index fb082047ce..d9011e7169 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioServerHelper.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioServerHelper.java @@ -8,8 +8,15 @@ */ package org.restlet.ext.mcp; +import io.modelcontextprotocol.json.McpJsonDefaults; +import io.modelcontextprotocol.json.McpJsonMapper; +import io.modelcontextprotocol.json.TypeRef; +import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.logging.Level; @@ -21,19 +28,53 @@ import org.restlet.data.Protocol; import org.restlet.engine.connector.ServerHelper; import org.restlet.ext.mcp.internal.McpMessages; +import org.restlet.ext.mcp.internal.McpModernExchange; import org.restlet.ext.mcp.internal.McpWorkers; import org.restlet.representation.ByteArrayRepresentation; +import org.restlet.representation.Representation; /** * Explicit newline-framed MCP STDIO connector. Every message traverses the Server's configured * Restlet chain. The owner must supply a StreamTransport; stdout is exclusively protocol output. + * + *

A message whose {@code _meta} names MCP {@link McpProtocol#MODERN_VERSION} goes to the + * adapter's 2026-07-28 server: the messages it streams, such as progress, are written as separate + * lines as they come, a {@code subscriptions/listen} stream stays open without time limit, and a + * {@code notifications/cancelled} naming an in-flight request ends it without a response. */ public final class StdioServerHelper extends ServerHelper { + private static final String CANCELLED = "notifications/cancelled"; + + /** Request attribute marking a request the client cancelled, which gets no response. */ + private static final String CANCELLED_ATTRIBUTE = "org.restlet.ext.mcp.stdio.cancelled"; + + /** Room for the {@code data: } prefix of a streamed message's line. */ + private static final int EVENT_LINE_OVERHEAD = 16; + + private final McpJsonMapper mapper = McpJsonDefaults.getMapper(); private final Set requests = ConcurrentHashMap.newKeySet(); + + /** The in-flight 2026-07-28 requests, by id, which a client may cancel. */ + private final Map inFlight = new ConcurrentHashMap<>(); + private org.restlet.util.StreamTransport stream; private McpWorkers workers; + + /** + * The workers of 2026-07-28 requests, without time limit of their own: the adapter's request + * timeout applies, and a request it cuts short is answered rather than failing the connector. + */ + private McpWorkers modernWorkers; + + /** The workers of {@code subscriptions/listen} streams, without time limit. */ + private McpWorkers listeners; + private Thread reader; private volatile boolean closed; + + /** Whether the connector was stopped, after which failures are expected. */ + private volatile boolean stopped; + private volatile Throwable failure; public StdioServerHelper(Server server) { @@ -55,6 +96,12 @@ public synchronized void start() throws IOException { stream.setMaxOperationBytes(McpProtocol.DEFAULT_MAX_MESSAGE_BYTES + 1); stream.start(); workers = new McpWorkers(McpProtocol.DEFAULT_MAX_CONCURRENT_CALLS, Duration.ofSeconds(30)); + modernWorkers = + new McpWorkers( + McpProtocol.DEFAULT_MAX_CONCURRENT_CALLS, null, Duration.ofSeconds(30)); + listeners = + new McpWorkers( + McpProtocol.DEFAULT_MAX_LISTEN_STREAMS, null, Duration.ofSeconds(30)); reader = Thread.ofVirtual().name("restlet-mcp-stdio-server").start(this::read); } @@ -66,9 +113,7 @@ private void read() { McpMessages.readFrame( stream, McpProtocol.DEFAULT_MAX_MESSAGE_BYTES)) != null) { - byte[] frame = bytes; - workers.submit(() -> dispatch(frame), () -> {}) - .subscribe(ignored -> {}, this::fail); + accept(bytes); } if (!closed) { fail(new IOException("MCP STDIO input ended")); @@ -80,7 +125,137 @@ private void read() { } } + /** Builds the request of a frame, in reading order, and hands it to a worker. */ + private void accept(byte[] bytes) { + Map envelope = envelope(bytes); + boolean modern = envelope != null && modern(envelope); + String method = envelope != null && envelope.get("method") instanceof String m ? m : null; + // A cancellation of an in-flight 2026-07-28 request, with or without _meta. + if (CANCELLED.equals(method) && (cancel(envelope) || modern)) { + return; + } + Request request = + new Request( + Method.POST, + "stdio://localhost/mcp", + new ByteArrayRepresentation(bytes, MediaType.APPLICATION_JSON)); + request.getHeaders() + .set( + McpProtocol.VERSION_HEADER, + modern ? McpProtocol.MODERN_VERSION : McpProtocol.VERSION); + request.getHeaders().set("Accept", "application/json, text/event-stream"); + if (!modern) { + workers.submit(() -> dispatch(request), () -> {}).subscribe(ignored -> {}, this::fail); + return; + } + if (method != null) { + request.getHeaders().set("Mcp-Method", method); + } + request.getAttributes().put(McpModernExchange.STDIO_ATTRIBUTE, Boolean.TRUE); + Object id = envelope.get("id"); + Object key = id == null ? null : key(id); + if (key != null && inFlight.putIfAbsent(key, request) != null) { + key = null; // A duplicate id: only the first may be cancelled. + } + Object registered = key; + boolean listen = McpModernExchange.isListen(method); + (listen ? listeners : modernWorkers) + .submit( + () -> { + try { + dispatch(request); + } finally { + if (registered != null) { + inFlight.remove(registered, request); + } + } + }, + () -> {}) + .subscribe( + ignored -> {}, + error -> { + if (registered != null) { + inFlight.remove(registered, request); + } + if (!closed && !isCancelled(request)) { + // Refused, too large or broken: answered, the connector stays up. + refuse(id, error); + } + }); + } + + /** The JSON object of a frame, or null when it isn't one; the adapter then rejects it. */ + private Map envelope(byte[] bytes) { + try { + return mapper.readValue(McpMessages.utf8(bytes), new TypeRef() {}) + instanceof Map map + ? map + : null; + } catch (IOException | RuntimeException _) { + return null; + } + } + + /** Whether a message's {@code _meta} names the 2026-07-28 revision. */ + private static boolean modern(Map envelope) { + return envelope.get("params") instanceof Map params + && params.get("_meta") instanceof Map meta + && McpProtocol.MODERN_VERSION.equals( + meta.get( + io.modelcontextprotocol.modern.McpSchema.MetaKeys + .PROTOCOL_VERSION)); + } + + /** A request id comparable across its JSON integer forms, such as 1 and 1.0. */ + private static Object key(Object id) { + return id instanceof Number number && number.doubleValue() == number.longValue() + ? (Object) number.longValue() + : id; + } + + /** + * Ends the in-flight request a {@code notifications/cancelled} names, if any. + * + * @return True if it named one. + */ + private boolean cancel(Map envelope) { + Object id = + envelope.get("params") instanceof Map params ? params.get("requestId") : null; + Request request = id == null ? null : inFlight.remove(key(id)); + if (request == null) { + return false; + } + request.getAttributes().put(CANCELLED_ATTRIBUTE, Boolean.TRUE); + request.abort(); + return true; + } + + private static boolean isCancelled(Request request) { + return Boolean.TRUE.equals(request.getAttributes().get(CANCELLED_ATTRIBUTE)); + } + + /** Answers a 2026-07-28 request refused, cut short or failed, unless it was cancelled. */ + private void refuse(Object id, Throwable cause) { + getLogger().log(Level.FINE, "MCP STDIO request refused or failed", cause); + if (id == null) { + return; + } + try { + McpMessages.writeFrame( + stream, + McpMessages.encode( + mapper, + McpMessages.error(id, -32603, "MCP exchange unavailable"), + McpProtocol.DEFAULT_MAX_MESSAGE_BYTES)); + } catch (IOException | RuntimeException e) { + fail(e); + } + } + private void fail(Throwable cause) { + if (stopped) { + return; // Stopping cuts exchanges short: not a failure. + } failure = cause; closed = true; getLogger().log(Level.WARNING, "MCP STDIO server input/exchange failed", cause); @@ -93,17 +268,10 @@ private void fail(Throwable cause) { } } - private void dispatch(byte[] bytes) throws IOException { + private void dispatch(Request request) throws IOException { if (closed) { throw new IOException("MCP STDIO server is closed", failure); } - Request request = - new Request( - Method.POST, - "stdio://localhost/mcp", - new ByteArrayRepresentation(bytes, MediaType.APPLICATION_JSON)); - request.getHeaders().set(McpProtocol.VERSION_HEADER, McpProtocol.VERSION); - request.getHeaders().set("Accept", "application/json, text/event-stream"); Response response = new Response(request); requests.add(request); try { @@ -112,14 +280,24 @@ private void dispatch(byte[] bytes) throws IOException { throw new IOException("MCP STDIO server is closed", failure); } handle(request, response); - if (response.getEntity() != null) { - if (!MediaType.APPLICATION_JSON.isCompatible(response.getEntity().getMediaType())) { + Representation entity = response.getEntity(); + if (isCancelled(request)) { + return; // A cancelled request gets no response. + } + if (entity != null) { + if (MediaType.TEXT_EVENT_STREAM.isCompatible(entity.getMediaType())) { + try (var frames = new EventFrames(request)) { + entity.write(frames); + } + } else if (MediaType.APPLICATION_JSON.isCompatible(entity.getMediaType())) { + try (var input = entity.getStream()) { + McpMessages.writeFrame( + stream, + McpMessages.read(input, McpProtocol.DEFAULT_MAX_MESSAGE_BYTES)); + } + } else { throw new IOException("MCP STDIO handler returned a non-JSON entity"); } - try (var input = response.getEntity().getStream()) { - McpMessages.writeFrame( - stream, McpMessages.read(input, McpProtocol.DEFAULT_MAX_MESSAGE_BYTES)); - } } else if (!response.getStatus().isSuccess()) { throw new IOException( "MCP STDIO handler rejected exchange: " + response.getStatus()); @@ -132,6 +310,73 @@ private void dispatch(byte[] bytes) throws IOException { } } + /** + * Writes the data of each event of a server-sent event stream as one line, as it ends. + * Comments, such as heartbeats, are dropped, and so is what follows a cancellation. + */ + private final class EventFrames extends OutputStream { + private final Request request; + private final ByteArrayOutputStream line = new ByteArrayOutputStream(); + private final ByteArrayOutputStream data = new ByteArrayOutputStream(); + private boolean hasData; + + /** + * @param request The request whose stream this is, which may be cancelled meanwhile. + */ + private EventFrames(Request request) { + this.request = request; + } + + @Override + public void write(int b) throws IOException { + if (b == '\n') { + endLine(); + } else if (line.size() > McpProtocol.DEFAULT_MAX_MESSAGE_BYTES + EVENT_LINE_OVERHEAD) { + throw new IOException("MCP streamed message exceeds the message size"); + } else { + line.write(b); + } + } + + @Override + public void write(byte[] bytes, int offset, int length) throws IOException { + for (int i = offset; i < offset + length; i++) { + write(bytes[i]); + } + } + + private void endLine() throws IOException { + byte[] bytes = line.toByteArray(); + line.reset(); + int length = bytes.length; + if (length > 0 && bytes[length - 1] == '\r') { + length--; + } + if (length == 0) { + // A cancelled request's stream ends without the error that ends it. + if (hasData && !isCancelled(request)) { + McpMessages.writeFrame(stream, data.toByteArray()); + } + data.reset(); + hasData = false; + return; + } + String text = new String(bytes, 0, length, StandardCharsets.UTF_8); + if (text.startsWith("data:")) { + String value = text.substring(text.startsWith("data: ") ? 6 : 5); + byte[] value8 = value.getBytes(StandardCharsets.UTF_8); + if (data.size() + value8.length + 1 > McpProtocol.DEFAULT_MAX_MESSAGE_BYTES) { + throw new IOException("MCP streamed message exceeds the message size"); + } + if (hasData) { + data.write('\n'); + } + data.writeBytes(value8); + hasData = true; + } + } + } + /** * Terminal input/worker failure, if any; connector owners should monitor this for subprocesses. */ @@ -141,6 +386,7 @@ public Throwable getFailure() { @Override public synchronized void stop() throws Exception { + stopped = true; closed = true; requests.forEach(Request::abort); Exception error = null; @@ -151,15 +397,17 @@ public synchronized void stop() throws Exception { } catch (IOException e) { error = e; } - try { - if (workers != null) { - workers.close(); - } - } catch (IOException e) { - if (error == null) { - error = e; - } else { - error.addSuppressed(e); + for (McpWorkers pool : new McpWorkers[] {workers, modernWorkers, listeners}) { + try { + if (pool != null) { + pool.close(); + } + } catch (IOException e) { + if (error == null) { + error = e; + } else { + error.addSuppressed(e); + } } } if (reader != null && !reader.join(Duration.ofSeconds(30))) { diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernExchange.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernExchange.java index 3bb83ae1e6..1aa4941f92 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernExchange.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernExchange.java @@ -87,6 +87,13 @@ Mono run( McpInterceptor.Chain terminal); } + /** + * Request attribute marking an exchange that came over stdio, which has no headers: the header + * checks are skipped and the {@code Mcp-Param} headers count as not mirrored, as on the SDK's + * own stdio transport. + */ + public static final String STDIO_ATTRIBUTE = "org.restlet.ext.mcp.stdio"; + /** Streamed messages a slow client may leave pending before the SDK is held back. */ private static final int QUEUE_CAPACITY = 64; @@ -106,6 +113,7 @@ Mono run( private final McpServer sdk; private final McpJsonMapper mapper; private final McpWorkers workers; + private final McpWorkers listeners; private final Duration timeout; private final int maxMessageBytes; private final Interceptors interceptors; @@ -117,6 +125,7 @@ Mono run( * @param sdk The SDK's modern server. * @param mapper The JSON mapper. * @param workers The adapter's admission workers. + * @param listeners The workers of {@code subscriptions/listen} streams, without time limit. * @param timeout The request timeout. * @param maxMessageBytes The largest request or response message. * @param interceptors The adapter's interceptor chain. @@ -126,6 +135,7 @@ public McpModernExchange( McpServer sdk, McpJsonMapper mapper, McpWorkers workers, + McpWorkers listeners, Duration timeout, int maxMessageBytes, Interceptors interceptors, @@ -133,6 +143,7 @@ public McpModernExchange( this.sdk = sdk; this.mapper = mapper; this.workers = workers; + this.listeners = listeners; this.timeout = timeout; this.maxMessageBytes = maxMessageBytes; this.interceptors = interceptors; @@ -147,8 +158,13 @@ public McpModernExchange( * @param response The response. */ public void handle(Request request, Response response) { + // Recognised before the body is read, which must then name the same method. + boolean listen = isListen(request.getHeaders().getFirstValue("Mcp-Method", true)); var exchange = - new Exchange(request, new McpRequestContext(request, McpProtocol.MODERN_VERSION)); + new Exchange( + request, + new McpRequestContext(request, McpProtocol.MODERN_VERSION), + listen); Mono stopped = stopping.get() .flatMap(ignored -> Mono.error(new IOException("MCP server stopped"))); @@ -162,7 +178,8 @@ public void handle(Request request, Response response) { Mono.firstWithSignal( stopped, disconnected, - workers.submit(exchange::run, () -> request.getEntity().release())) + (listen ? listeners : workers) + .submit(exchange::run, () -> request.getEntity().release())) .subscribe(null, exchange::fail, exchange::end); Reply reply; try { @@ -180,6 +197,16 @@ public void handle(Request request, Response response) { reply.apply(response, exchange, running, heartbeat()); } + /** + * Whether a method opens a {@code subscriptions/listen} stream, which has no time limit. + * + * @param method The method, or null. + * @return True for {@code subscriptions/listen}. + */ + public static boolean isListen(String method) { + return io.modelcontextprotocol.modern.McpSchema.METHOD_SUBSCRIPTIONS_LISTEN.equals(method); + } + /** Comment heartbeats on a stream, as the SDK's servlet transport sends: at most 30 s apart. */ private Duration heartbeat() { Duration half = timeout.dividedBy(2); @@ -209,9 +236,22 @@ private final class Exchange { /** The id of the request, once read; answers a stream cut short. */ private volatile Object requestId; - private Exchange(Request request, McpRequestContext context) { + /** Whether this is a {@code subscriptions/listen} stream, which has no time limit. */ + private final boolean listen; + + /** Whether this exchange came over stdio, without headers. */ + private final boolean stdio; + + private Exchange(Request request, McpRequestContext context, boolean listen) { this.request = request; this.context = context; + this.listen = listen; + this.stdio = Boolean.TRUE.equals(request.getAttributes().get(STDIO_ATTRIBUTE)); + } + + /** Waits for a chain's response, within the request timeout unless listening. */ + private Optional await(Mono chain) { + return listen ? chain.blockOptional() : chain.blockOptional(timeout); } /** The SDK transport context for the request context an interceptor forwarded. */ @@ -266,19 +306,18 @@ private void notification(JsonRpc.JSONRPCNotification notification) { private void call(JsonRpc.JSONRPCRequest rpc) throws InterruptedException { requestId = rpc.id(); - String mismatch = mismatch(rpc); + String mismatch = stdio ? null : mismatch(rpc); if (mismatch != null) { head.complete(error(rpc.id(), ErrorCodes.HEADER_MISMATCH, mismatch)); return; } Optional last = - interceptors - .run( + await( + interceptors.run( mapper.convertValue(rpc, McpSchema.JSONRPCRequest.class), context, (next, nextContext) -> - Mono.fromCallable(() -> terminal(next, nextContext))) - .blockOptional(timeout); + Mono.fromCallable(() -> terminal(next, nextContext)))); if (last.isEmpty()) { // An interceptor answered nothing: accepted, as on the 2025-11-25 path, unless the // stream already started, which must still answer the request. @@ -314,9 +353,14 @@ private McpSchema.JSONRPCResponse terminal( sdkContext.putAll( Map.of( McpParamHeaders.TRANSPORT_CONTEXT_KEY, - McpHeaderDecoder.paramHeaders(headers()))); + stdio + ? McpParamHeaders.NOT_MIRRORED + : McpHeaderDecoder.paramHeaders(headers()))); } - McpTransportResponse reply = sdk.handleBlocking(sdkContext, rpc).block(timeout); + McpTransportResponse reply = + listen + ? sdk.handle(sdkContext, rpc).block(timeout) + : sdk.handleBlocking(sdkContext, rpc).block(timeout); if (reply instanceof McpTransportResponse.Result result) { return classic(result.response()); } diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernFeatures.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernFeatures.java index 3df8f33269..0ed6612dfb 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernFeatures.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernFeatures.java @@ -8,6 +8,7 @@ */ package org.restlet.ext.mcp.internal; +import io.modelcontextprotocol.common.McpTransportContext; import io.modelcontextprotocol.json.McpJsonMapper; import io.modelcontextprotocol.modern.McpSchema; import io.modelcontextprotocol.modern.server.McpAsyncResponse; @@ -33,6 +34,9 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.function.BiFunction; +import org.restlet.ext.mcp.McpModernResource; +import org.restlet.ext.mcp.McpModernResourceTemplate; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -40,8 +44,8 @@ * The features of the MCP {@code 2026-07-28} server, built from the adapter's registrations. The * {@code 2025-11-25} specifications are bridged: their definitions and results are converted * between the two SDK schemas through JSON, and their handlers receive the same transport context - * as on the stateless server. Modern tools are served as registered. List filters apply to every - * listed tool. + * as on the stateless server. Modern registrations are served as registered. List filters apply to + * every listed tool. A name, URI or URI template may only be registered once across both kinds. */ public final class McpModernFeatures { @@ -56,6 +60,9 @@ private McpModernFeatures() {} * @param prompts The {@code 2025-11-25} prompts. * @param toolFilters The tool list filters. * @param modernTools The {@code 2026-07-28} tools. + * @param modernResources The {@code 2026-07-28} resources. + * @param modernResourceTemplates The {@code 2026-07-28} resource templates. + * @param modernPrompts The {@code 2026-07-28} prompts. */ public record Registrations( List tools, @@ -63,7 +70,10 @@ public record Registrations( List resourceTemplates, List prompts, List> toolFilters, - List modernTools) {} + List modernTools, + List modernResources, + List modernResourceTemplates, + List modernPrompts) {} /** * Builds the features for registrations, one per kind present. @@ -71,18 +81,21 @@ public record Registrations( * @param mapper The JSON mapper converting between the two schemas. * @param registrations The adapter's registrations. * @return The features. - * @throws IllegalArgumentException If two tools share a name. + * @throws IllegalArgumentException If a name, URI or URI template is registered twice. */ public static List of(McpJsonMapper mapper, Registrations registrations) { List features = new ArrayList<>(); if (!registrations.tools().isEmpty() || !registrations.modernTools().isEmpty()) { features.add(ToolsFeature.ofAsync(new Tools(mapper, registrations))); } - if (!registrations.resources().isEmpty() || !registrations.resourceTemplates().isEmpty()) { + if (!registrations.resources().isEmpty() + || !registrations.resourceTemplates().isEmpty() + || !registrations.modernResources().isEmpty() + || !registrations.modernResourceTemplates().isEmpty()) { features.add(ResourcesFeature.ofAsync(new Resources(mapper, registrations))); } - if (!registrations.prompts().isEmpty()) { - features.add(PromptsFeature.ofAsync(new Prompts(mapper, registrations.prompts()))); + if (!registrations.prompts().isEmpty() || !registrations.modernPrompts().isEmpty()) { + features.add(PromptsFeature.ofAsync(new Prompts(mapper, registrations))); } return features; } @@ -154,66 +167,70 @@ public Mono find(McpRequestContext ctx, String name) { } } - /** The bridged resources: exact URIs first, then templates, in registration order. */ + /** A URI template with its matcher and read handler. */ + private record Template( + McpSchema.ResourceTemplate definition, + DefaultMcpUriTemplateManager matcher, + McpAsyncResource resource) {} + + /** + * The resources, bridged and modern: exact URIs first, then templates, in registration order. + * Subscriptions are supported: the adapter publishes resource updates through its change feed. + */ private static final class Resources implements McpAsyncResourceRepository { private final McpJsonMapper mapper; - private final List exact; - private final List templates; + private final List listed = new ArrayList<>(); + private final Map byUri = new LinkedHashMap<>(); + private final List