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..552cb3f47a 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,32 +8,84 @@ */ 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.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.logging.Level; import org.restlet.Request; import org.restlet.Response; +import org.restlet.Restlet; import org.restlet.Server; import org.restlet.data.MediaType; import org.restlet.data.Method; 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; +import org.restlet.routing.Filter; /** * 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<>(); + + /** + * Held while a frame is written, and while a request is marked cancelled: once a cancellation + * is marked, no frame of that request follows. + */ + private final Object output = new Object(); + private org.restlet.util.StreamTransport stream; private McpWorkers workers; + + /** + * The workers of 2026-07-28 requests, without limits of their own: the adapter's call and + * listen limits and request timeout apply, and what it refuses or cuts short is answered rather + * than failing the connector. + */ + private McpWorkers modernWorkers; + + /** The pending slots of 2026-07-28 requests, which bound their workers and output. */ + private Semaphore pending; + 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 +107,8 @@ 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(Integer.MAX_VALUE, null, Duration.ofSeconds(30)); + pending = new Semaphore(pendingLimit()); reader = Thread.ofVirtual().name("restlet-mcp-stdio-server").start(this::read); } @@ -66,9 +120,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 +132,212 @@ 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 = modern(envelope); + String method = envelope.get("method") instanceof String m ? m : null; + // A cancellation of an in-flight 2026-07-28 request, with or without _meta. A request + // naming this method, or a malformed envelope, goes on to be answered. + if (isCancellation(envelope, method) && (cancel(envelope) || modern)) { + return; + } + Request request = request(bytes, modern); + if (modern) { + submitModern(request, envelope, method); + } else { + workers.submit(() -> dispatch(request, null), () -> {}) + .subscribe(ignored -> {}, this::fail); + } + } + + /** Whether a message is a {@code notifications/cancelled} notification. */ + private static boolean isCancellation(Map envelope, String method) { + return CANCELLED.equals(method) + && "2.0".equals(envelope.get("jsonrpc")) + && !envelope.containsKey("id"); + } + + /** The request of a frame, with the headers it would have over HTTP. */ + private static Request request(byte[] bytes, boolean modern) { + 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"); + return request; + } + + /** + * Hands a 2026-07-28 request to a worker if one of the pending slots is free, else refuses it + * at once: the reader keeps reading, cancellations included, and a client that stops reading + * the output stops being read in turn, when the refusal cannot be written. + */ + private void submitModern(Request request, Map envelope, String method) { + if (method != null) { + request.getHeaders().set("Mcp-Method", method); + } + request.getAttributes().put(McpModernExchange.STDIO_ATTRIBUTE, Boolean.TRUE); + boolean answered = envelope.containsKey("id"); + Object id = validId(envelope.get("id")); + if (!pending.tryAcquire()) { + refuse(request, answered, id, new IOException("Too many pending MCP STDIO requests")); + return; + } + 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; + AtomicBoolean held = new AtomicBoolean(true); + // Freed once the response, or the error answering a failure, is written. + Runnable done = + () -> { + if (registered != null) { + inFlight.remove(registered, request); + } + if (held.compareAndSet(true, false)) { + pending.release(); + } + }; + modernWorkers + .submit(() -> dispatch(request, id), () -> {}) + .subscribe( + null, + error -> { + try { + if (!closed && !isCancelled(request)) { + // Refused, too large or broken: answered, the connector stays. + refuse(request, answered, id, error); + } + } finally { + done.run(); + } + }, + done); + } + + /** A request id as JSON-RPC allows it, a string or a number; else null. */ + private static Object validId(Object id) { + return id instanceof String || id instanceof Number ? id : null; + } + + /** + * How many 2026-07-28 requests may be pending at once, writing or waiting to write: the open + * listen streams the adapter allows, and as many other requests as the default call limit. + */ + private int pendingLimit() { + Restlet next = getHelped() == null ? null : getHelped().getNext(); + while (next instanceof Filter filter) { + next = filter.getNext(); + } + int listens = + next instanceof McpServerAdapter adapter + ? adapter.getMaxListenStreams() + : McpProtocol.DEFAULT_MAX_LISTEN_STREAMS; + return listens + McpProtocol.DEFAULT_MAX_CONCURRENT_CALLS; + } + + /** The JSON object of a frame, or an empty map when it isn't one; the adapter rejects it. */ + private Map envelope(byte[] bytes) { + try { + return mapper.readValue(McpMessages.utf8(bytes), new TypeRef() {}) + instanceof Map map + ? map + : Map.of(); + } catch (IOException | RuntimeException _) { + return Map.of(); + } + } + + /** 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; + } + synchronized (output) { + 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)); + } + + /** + * Writes a frame of a request, unless the client cancelled it. + * + * @param request The request answered, or null for none. + * @param frame The frame, without its newline. + */ + private void emit(Request request, byte[] frame) throws IOException { + synchronized (output) { + if (request == null || !isCancelled(request)) { + McpMessages.writeFrame(stream, frame); + } + } + } + + /** + * Answers a 2026-07-28 request refused, cut short or failed, unless it was cancelled. + * + * @param request The request. + * @param answered Whether it is a request, which gets an answer, rather than a notification. + * @param id Its valid id, or null to answer with a null id. + * @param cause Why it is refused. + */ + private void refuse(Request request, boolean answered, Object id, Throwable cause) { + getLogger().log(Level.FINE, "MCP STDIO request refused or failed", cause); + if (!answered) { + return; + } + try { + emit( + request, + 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 +350,17 @@ private void fail(Throwable cause) { } } - private void dispatch(byte[] bytes) throws IOException { + /** + * Handles a request and writes its response. + * + * @param request The request. + * @param id The id of a 2026-07-28 request, given to an error the adapter answered without it, + * having refused the request unread; or null. + */ + private void dispatch(Request request, Object id) 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 +369,27 @@ 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()) { + emit( + request, + withId( + McpMessages.read( + input, McpProtocol.DEFAULT_MAX_MESSAGE_BYTES), + id)); + } + } 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 +402,99 @@ private void dispatch(byte[] bytes) throws IOException { } } + /** A reply, given the request id when it is an error without one. */ + private byte[] withId(byte[] reply, Object id) throws IOException { + if (id == null) { + return reply; + } + Map envelope = envelope(reply); + if (envelope.get("id") != null || !(envelope.get("error") instanceof Map error)) { + return reply; + } + Map answered = + McpMessages.error( + id, + error.get("code") instanceof Number code ? code.intValue() : -32603, + String.valueOf(error.get("message"))); + return McpMessages.encode(mapper, answered, McpProtocol.DEFAULT_MAX_MESSAGE_BYTES); + } + + /** + * 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) { + endEvent(); + return; + } + String text = new String(bytes, 0, length, StandardCharsets.UTF_8); + if (text.startsWith("data:")) { + addData(text.substring(text.startsWith("data: ") ? 6 : 5)); + } + } + + /** Writes the event that ends, if it has data; a cancelled request's is dropped. */ + private void endEvent() throws IOException { + if (hasData) { + emit(request, data.toByteArray()); + } + data.reset(); + hasData = false; + } + + /** Adds a data line to the event, joined by a newline, within the message size. */ + private void addData(String value) throws IOException { + byte[] bytes = value.getBytes(StandardCharsets.UTF_8); + int separator = hasData ? 1 : 0; + if (data.size() + separator + bytes.length > McpProtocol.DEFAULT_MAX_MESSAGE_BYTES) { + throw new IOException("MCP streamed message exceeds the message size"); + } + if (hasData) { + data.write('\n'); + } + data.writeBytes(bytes); + hasData = true; + } + } + /** * Terminal input/worker failure, if any; connector owners should monitor this for subprocesses. */ @@ -141,38 +504,65 @@ public Throwable getFailure() { @Override public synchronized void stop() throws Exception { + stopped = true; closed = true; requests.forEach(Request::abort); - Exception error = null; - try { - if (stream != null) { - stream.stop(); - } - } catch (IOException e) { - error = e; + Exception error = + attempt( + null, + () -> { + if (stream != null) { + stream.stop(); + } + }); + for (McpWorkers pool : new McpWorkers[] {workers, modernWorkers}) { + error = + attempt( + error, + () -> { + if (pool != null) { + pool.close(); + } + }); + } + boolean joined = reader == null || reader.join(Duration.ofSeconds(30)); + error = + attempt( + error, + () -> { + if (!joined) { + throw new IOException("MCP STDIO server reader shutdown timed out"); + } + }); + super.stop(); + if (error != null) { + throw error; } + } + + /** A shutdown step. */ + @FunctionalInterface + private interface Step { + void run() throws IOException; + } + + /** + * Runs a shutdown step, keeping the first error and suppressing the next ones in it. + * + * @param error The error so far, or null. + * @param step The step. + * @return The error so far, after the step. + */ + private static Exception attempt(Exception error, Step step) { try { - if (workers != null) { - workers.close(); - } + step.run(); + return error; } catch (IOException e) { if (error == null) { - error = e; - } else { - error.addSuppressed(e); + return e; } - } - if (reader != null && !reader.join(Duration.ofSeconds(30))) { - IOException join = new IOException("MCP STDIO server reader shutdown timed out"); - if (error == null) { - error = join; - } else { - error.addSuppressed(join); - } - } - super.stop(); - if (error != null) { - throw error; + error.addSuppressed(e); + return error; } } } 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