From 432bf5f832c9c9966dc81828c3393e53acdf3bf3 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 7 Nov 2026 13:52:11 -0500 Subject: [PATCH 1/4] Serve MCP 2026-07-28 from McpServerAdapter, experimentally McpServerAdapter now also serves MCP revision 2026-07-28 on the same endpoint: a request whose MCP-Protocol-Version header names it goes to the MCP SDK's modern stateless server (mcp-core 2.2.0-M2), any other request keeps the 2025-11-25 path unchanged. - The tools, resources, templates and prompts registered on the adapter are served to both revisions, bridged through JSON between the two SDK schemas; getModernTools() adds tools on the modern types, served to 2026-07-28 clients only, for input_required and streamed progress. - The interceptors wrap each 2026-07-28 request and its final result or error; progress streamed before the result passes without them. - The Origin allow-list, message size, call limit, request timeout, stop and peer disconnect apply to both revisions; the tool filters apply to 2026-07-28 listings, modern tools included. - A streamed reply goes over core SSE through a bounded queue, with heartbeats; leaving it early, on a timeout, a disconnect or a stop, cancels the SDK's stream and ends the client's with an error answering the request. - Mcp-Param headers are passed to the SDK, which validates x-mcp-header arguments against them. - input_required state is sealed bound to the authenticated user; setRequestStateKey() shares the key between instances. McpRequestContext reports the revision of its request, and McpProtocol.MODERN_VERSION names 2026-07-28. Not yet for 2026-07-28: stdio, subscriptions/listen, modern resources and prompts, the client. The test-only spike of #1662 is replaced by McpModernServerTestCase. Co-Authored-By: Claude Opus 5.5 --- changes.md | 12 +- .../java/org/restlet/ext/mcp/McpProtocol.java | 7 + .../restlet/ext/mcp/McpRequestContext.java | 15 +- .../org/restlet/ext/mcp/McpServerAdapter.java | 137 ++- .../ext/mcp/internal/McpModernExchange.java | 524 ++++++++++++ .../ext/mcp/internal/McpModernFeatures.java | 299 +++++++ .../ext/mcp/McpModernServerTestCase.java | 782 ++++++++++++++++++ .../ext/mcp/McpModernSpikeTestCase.java | 570 ------------- 8 files changed, 1770 insertions(+), 576 deletions(-) create mode 100644 org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernExchange.java create mode 100644 org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernFeatures.java create mode 100644 org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java delete mode 100644 org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernSpikeTestCase.java diff --git a/changes.md b/changes.md index 227390c6f4..405cc02527 100644 --- a/changes.md +++ b/changes.md @@ -89,12 +89,22 @@ Changes log supporting helper class. Applications with multiple STDIO protocol helpers must select one explicitly. - Enhancements - - Added the MCP extension using the official Java SDK 2.1.0 on shared Restlet HTTP + - Added the MCP extension using the official Java SDK (2.2.0-M2) on shared Restlet HTTP connectors and explicitly configured STDIO helpers. The supported stateless profile uses revision 2025-11-25, native SDK tools/resources/prompts and in-JVM dispatch, bounded messages/calls, one-shot SSE responses, and request cancellation. Sessions, listening GET streams, replay and Ikanos revision compatibility are not supported. Borrowed HTTP connectors are not stopped by the MCP client. + - The MCP server adapter also serves MCP revision 2026-07-28, experimentally, on the same + endpoint: a request whose `MCP-Protocol-Version` header names it goes to the SDK's + modern stateless server, other requests to the 2025-11-25 one. The tools, resources, + templates and prompts registered on the adapter are served to both revisions; + `McpServerAdapter.getModernTools()` adds tools on the SDK's modern types, served to + 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. - 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/McpProtocol.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpProtocol.java index 0facf407aa..4a50a7148f 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 @@ -13,6 +13,13 @@ */ public final class McpProtocol { public static final String VERSION = "2025-11-25"; + + /** + * The stateless revision also served, experimentally, by {@link McpServerAdapter}: requests + * whose {@link #VERSION_HEADER} names it go to the MCP SDK's modern server. + */ + public static final String MODERN_VERSION = "2026-07-28"; + public static final String VERSION_HEADER = "MCP-Protocol-Version"; public static final String CONTEXT_KEY = "org.restlet.ext.mcp.requestContext"; public static final String STREAM_TRANSPORT_ATTRIBUTE = "org.restlet.ext.mcp.streamTransport"; diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpRequestContext.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpRequestContext.java index deff7461c4..adb150c03f 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpRequestContext.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpRequestContext.java @@ -16,10 +16,23 @@ /** MCP-only transport metadata, also available in native SDK callbacks under CONTEXT_KEY. */ public final class McpRequestContext { private final Request request; + private final String protocolVersion; private final Map attributes = new ConcurrentHashMap<>(); public McpRequestContext(Request request) { + this(request, McpProtocol.VERSION); + } + + /** + * Creates the context of a request using a given MCP revision. + * + * @param request The transport request, or null. + * @param protocolVersion The MCP revision of the request, such as {@link + * McpProtocol#MODERN_VERSION}. + */ + public McpRequestContext(Request request, String protocolVersion) { this.request = request; + this.protocolVersion = java.util.Objects.requireNonNull(protocolVersion, "protocolVersion"); } public Optional transportRequest() { @@ -27,7 +40,7 @@ public Optional transportRequest() { } public String protocolVersion() { - return McpProtocol.VERSION; + return protocolVersion; } public Map attributes() { 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 733476da26..f0c9434937 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 @@ -13,6 +13,8 @@ import io.modelcontextprotocol.json.McpJsonDefaults; import io.modelcontextprotocol.json.McpJsonMapper; import io.modelcontextprotocol.json.jackson2.JacksonMcpJsonMapper; +import io.modelcontextprotocol.modern.server.HmacRequestStateCodec; +import io.modelcontextprotocol.modern.server.feature.McpAsyncTool; import io.modelcontextprotocol.server.McpAsyncListFilter; import io.modelcontextprotocol.server.McpServer; import io.modelcontextprotocol.server.McpStatelessAsyncServer; @@ -42,6 +44,8 @@ import org.restlet.data.Status; import org.restlet.engine.header.PreferenceReader; import org.restlet.ext.mcp.internal.McpMessages; +import org.restlet.ext.mcp.internal.McpModernExchange; +import org.restlet.ext.mcp.internal.McpModernFeatures; import org.restlet.ext.mcp.internal.McpWorkers; import org.restlet.representation.ByteArrayRepresentation; import reactor.core.publisher.Mono; @@ -77,6 +81,8 @@ private static McpJsonMapper initMapper() { private final List> toolFilters = new CopyOnWriteArrayList<>(); private final List interceptors = new CopyOnWriteArrayList<>(); + private final List modernTools = new CopyOnWriteArrayList<>(); + private byte[] requestStateKey; private final Set allowedOrigins = ConcurrentHashMap.newKeySet(); private McpSchema.Implementation serverInfo = McpSchema.Implementation.builder("restlet", "3.0").build(); @@ -86,6 +92,8 @@ private static McpJsonMapper initMapper() { private Duration requestTimeout = Duration.ofSeconds(30); private volatile McpStatelessServerHandler handler; private volatile McpStatelessAsyncServer sdkServer; + private volatile io.modelcontextprotocol.modern.server.McpServer modernServer; + private volatile McpModernExchange modernExchange; private volatile McpWorkers workers; private volatile boolean accepting; private volatile Sinks.One stopping; @@ -126,6 +134,35 @@ public List getInterceptors() { return interceptors; } + /** + * Tools served to MCP {@link McpProtocol#MODERN_VERSION} clients only, on the SDK's modern + * types, for what that revision adds: {@code input_required} results and streamed progress. The + * tools of {@link #getTools()} are served to both revisions. A name may only be used once + * across both lists. The tool filters apply to them too, on their {@code 2025-11-25} + * projection. Experimental: the SDK's modern API is a preview. + * + * @return The modern tools, modifiable until the adapter starts. + */ + public List getModernTools() { + return modernTools; + } + + /** + * 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 + * instance that answered; instances behind a load balancer need the same key. The state is also + * bound to the authenticated user of the request, so another user can't replay it. + * + * @param value The key, copied. + */ + public synchronized void setRequestStateKey(byte[] value) { + requireConfigurable(); + if (value == null || value.length < 32) { + throw new IllegalArgumentException("requestStateKey must have at least 32 bytes"); + } + requestStateKey = value.clone(); + } + /** Allowed exact Origin header values. Requests with any other Origin are forbidden. */ public Set getAllowedOrigins() { return allowedOrigins; @@ -198,6 +235,8 @@ public synchronized void start() throws Exception { return; } requireConfigurable(); + // Built first: it rejects duplicate tool names before anything needs releasing. + var modern = modernServer(); var caps = McpSchema.ServerCapabilities.builder(); if (!tools.isEmpty()) { caps.tools(false); @@ -240,6 +279,16 @@ public List protocolVersions() { .build(); workers = new McpWorkers(maxConcurrentCalls, requestTimeout); stopping = Sinks.one(); + modernServer = modern; + modernExchange = + new McpModernExchange( + modernServer, + MAPPER, + workers, + requestTimeout, + maxMessageBytes, + (message, context, terminal) -> invoke(0, message, context, terminal), + () -> stopping.asMono()); runningInterceptors = List.copyOf(interceptors); runningOrigins = Set.copyOf(allowedOrigins); configured = true; @@ -247,6 +296,60 @@ public List protocolVersions() { accepting = true; } + /** Builds the SDK's modern server from the registrations, bridged and modern. */ + private io.modelcontextprotocol.modern.server.McpServer modernServer() { + var builder = + io.modelcontextprotocol.modern.server.McpServer.builder() + .serverInfo( + MAPPER.convertValue( + serverInfo, + io.modelcontextprotocol.modern.McpSchema.Implementation + .class)) + .instructions(instructions) + .jsonMapper(MAPPER) + .features( + McpModernFeatures.of( + MAPPER, + new McpModernFeatures.Registrations( + List.copyOf(tools), + List.copyOf(resources), + List.copyOf(resourceTemplates), + List.copyOf(prompts), + List.copyOf(toolFilters), + List.copyOf(modernTools)))); + var codec = + HmacRequestStateCodec.builder() + .principalExtractor(McpServerAdapter::principal) + .jsonMapper(MAPPER); + if (requestStateKey != null) { + codec.key(requestStateKey); + } + return builder.requestStateCodec(codec.build()).build(); + } + + /** + * The authenticated user of a request, binding its {@code input_required} state, or null. + * + * @param transport The SDK transport context. + * @return The user identifier, else the first principal name, or null. + */ + private static String principal(McpTransportContext transport) { + if (!(transport.get(McpProtocol.CONTEXT_KEY) instanceof McpRequestContext context)) { + return null; + } + return context.transportRequest() + .map(Request::getClientInfo) + .map( + info -> + info.getUser() != null + ? info.getUser().getIdentifier() + : info.getPrincipals().stream() + .map(java.security.Principal::getName) + .findFirst() + .orElse(null)) + .orElse(null); + } + @Override public synchronized void stop() throws Exception { if (!isStarted()) { @@ -260,7 +363,11 @@ public synchronized void stop() throws Exception { try { sdkServer.closeGracefully().block(requestTimeout); } finally { - super.stop(); + try { + modernServer.closeGracefully(); + } finally { + super.stop(); + } } } } @@ -315,14 +422,29 @@ private Optional admit( private Mono invoke( int index, McpSchema.JSONRPCMessage message, McpRequestContext context) { + return invoke(index, message, context, this::dispatchStateless); + } + + /** Runs the interceptors from {@code index}, then {@code terminal}. */ + private Mono invoke( + int index, + McpSchema.JSONRPCMessage message, + McpRequestContext context, + McpInterceptor.Chain terminal) { if (index < runningInterceptors.size()) { return runningInterceptors .get(index) .intercept( message, context, - (next, nextContext) -> invoke(index + 1, next, nextContext)); + (next, nextContext) -> invoke(index + 1, next, nextContext, terminal)); } + return terminal.proceed(message, context); + } + + /** Dispatches a 2025-11-25 message to the SDK's stateless server. */ + private Mono dispatchStateless( + McpSchema.JSONRPCMessage message, McpRequestContext context) { McpTransportContext sdkContext = McpTransportContext.create(Map.of(McpProtocol.CONTEXT_KEY, context)); if (message instanceof McpSchema.JSONRPCRequest request) { @@ -367,8 +489,11 @@ public void handle(Request request, Response response) { return; } String version = request.getHeaders().getFirstValue(McpProtocol.VERSION_HEADER, true); - if (version != null && !McpProtocol.VERSION.equals(version) - || request.getHeaders().getFirstValue("Mcp-Session-Id", true) != null) { + // A 2026-07-28 request goes to the modern server; it ignores a stale Mcp-Session-Id. + boolean modern = McpProtocol.MODERN_VERSION.equals(version); + if (!modern + && (version != null && !McpProtocol.VERSION.equals(version) + || request.getHeaders().getFirstValue("Mcp-Session-Id", true) != null)) { response.setStatus(Status.CLIENT_ERROR_BAD_REQUEST); return; } @@ -407,6 +532,10 @@ public void handle(Request request, Response response) { response.setEntity((org.restlet.representation.Representation) null); return; } + if (modern) { + modernExchange.handle(request, response); + return; + } var context = new McpRequestContext(request); var parsed = new AtomicReference(); Reply reply; 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 new file mode 100644 index 0000000000..5cf8a384c9 --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernExchange.java @@ -0,0 +1,524 @@ +/** + * 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.internal; + +import io.modelcontextprotocol.common.McpTransportContext; +import io.modelcontextprotocol.json.McpJsonMapper; +import io.modelcontextprotocol.modern.JsonRpc; +import io.modelcontextprotocol.modern.McpSchema.ErrorCodes; +import io.modelcontextprotocol.modern.McpSchema.MetaKeys; +import io.modelcontextprotocol.modern.server.McpParamHeaders; +import io.modelcontextprotocol.modern.server.McpServer; +import io.modelcontextprotocol.modern.server.McpTransportResponse; +import io.modelcontextprotocol.modern.server.transport.McpHeaderDecoder; +import io.modelcontextprotocol.server.transport.HeaderAccessor; +import io.modelcontextprotocol.spec.McpSchema; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.function.Supplier; +import java.util.stream.Stream; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.data.MediaType; +import org.restlet.data.ServerSentEvent; +import org.restlet.data.Status; +import org.restlet.ext.mcp.McpInterceptor; +import org.restlet.ext.mcp.McpProtocol; +import org.restlet.ext.mcp.McpRequestContext; +import org.restlet.representation.ByteArrayRepresentation; +import org.restlet.representation.EventStreamRepresentation; +import org.restlet.representation.Representation; +import reactor.core.Disposable; +import reactor.core.publisher.Mono; + +/** + * One MCP {@code 2026-07-28} HTTP exchange, as the SDK's servlet transport handles it: POST only, + * the {@code Mcp-Method}, {@code MCP-Protocol-Version} and {@code Mcp-Name} headers checked against + * the body, 202 for a notification, one JSON reply with an error mapped to its HTTP status, or an + * SSE stream when the SDK streams notifications before the result. + * + *

The exchange runs in an admitted worker, within the adapter's call limit and timeout. The + * adapter's interceptors wrap the request and the final result or error, on the {@code 2025-11-25} + * message types. Notifications streamed before the result pass without them, and so does the + * in-band error ending a stream cut short by a timeout, a disconnect or a stop. Streamed messages + * reach the response through a bounded queue, so a slow client holds back the SDK, and leaving the + * stream early cancels the SDK's. + */ +public final class McpModernExchange { + + /** Runs the adapter's interceptors around a terminal step. */ + @FunctionalInterface + public interface Interceptors { + + /** + * Runs the interceptor chain. + * + * @param message The message, on the {@code 2025-11-25} types. + * @param context The request context. + * @param terminal The step after the last interceptor. + * @return The final response, empty for a notification. + */ + Mono run( + McpSchema.JSONRPCMessage message, + McpRequestContext context, + McpInterceptor.Chain terminal); + } + + /** Streamed messages a slow client may leave pending before the SDK is held back. */ + private static final int QUEUE_CAPACITY = 64; + + /** Marks the end of a stream in the queue. */ + private static final byte[] END = new byte[0]; + + private static final Map ERROR_STATUSES = + Map.of( + ErrorCodes.PARSE_ERROR, Status.CLIENT_ERROR_BAD_REQUEST, + ErrorCodes.INVALID_REQUEST, Status.CLIENT_ERROR_BAD_REQUEST, + ErrorCodes.INVALID_PARAMS, Status.CLIENT_ERROR_BAD_REQUEST, + ErrorCodes.HEADER_MISMATCH, Status.CLIENT_ERROR_BAD_REQUEST, + ErrorCodes.MISSING_REQUIRED_CLIENT_CAPABILITY, Status.CLIENT_ERROR_BAD_REQUEST, + ErrorCodes.UNSUPPORTED_PROTOCOL_VERSION, Status.CLIENT_ERROR_BAD_REQUEST, + ErrorCodes.METHOD_NOT_FOUND, Status.CLIENT_ERROR_NOT_FOUND); + + private final McpServer sdk; + private final McpJsonMapper mapper; + private final McpWorkers workers; + private final Duration timeout; + private final int maxMessageBytes; + private final Interceptors interceptors; + private final Supplier> stopping; + + /** + * Creates the exchange handler of an adapter. + * + * @param sdk The SDK's modern server. + * @param mapper The JSON mapper. + * @param workers The adapter's admission workers. + * @param timeout The request timeout. + * @param maxMessageBytes The largest request or response message. + * @param interceptors The adapter's interceptor chain. + * @param stopping Emits when the adapter stops. + */ + public McpModernExchange( + McpServer sdk, + McpJsonMapper mapper, + McpWorkers workers, + Duration timeout, + int maxMessageBytes, + Interceptors interceptors, + Supplier> stopping) { + this.sdk = sdk; + this.mapper = mapper; + this.workers = workers; + this.timeout = timeout; + this.maxMessageBytes = maxMessageBytes; + this.interceptors = interceptors; + this.stopping = stopping; + } + + /** + * Handles a {@code 2026-07-28} POST whose method, origin, media types and size the adapter + * already checked. + * + * @param request The request. + * @param response The response. + */ + public void handle(Request request, Response response) { + var exchange = + new Exchange(request, new McpRequestContext(request, McpProtocol.MODERN_VERSION)); + Mono stopped = + stopping.get() + .flatMap(ignored -> Mono.error(new IOException("MCP server stopped"))); + Mono disconnected = + Mono.fromCompletionStage(request.getDisconnectSignal()) + .flatMap( + cause -> + Mono.error( + new IOException("MCP peer disconnected", cause))); + Disposable running = + Mono.firstWithSignal( + stopped, + disconnected, + workers.submit(exchange::run, () -> request.getEntity().release())) + .subscribe(null, exchange::fail, exchange::end); + Reply reply; + try { + reply = + exchange.head.get( + timeout.toNanos() + TimeUnit.SECONDS.toNanos(1), TimeUnit.NANOSECONDS); + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + running.dispose(); + reply = unavailable(); + } catch (ExecutionException | TimeoutException _) { + running.dispose(); + reply = unavailable(); + } + reply.apply(response, exchange, running, heartbeat()); + } + + /** 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); + Duration max = Duration.ofSeconds(30); + return half.compareTo(max) < 0 ? half : max; + } + + private Reply unavailable() { + return json( + Status.SERVER_ERROR_SERVICE_UNAVAILABLE, + McpMessages.error(null, ErrorCodes.INTERNAL_ERROR, "MCP exchange unavailable")); + } + + /** The state of one exchange, shared by its worker and the thread writing the response. */ + private final class Exchange { + private final Request request; + private final McpRequestContext context; + private final McpTransportContext transport; + private final CompletableFuture head = new CompletableFuture<>(); + private final BlockingQueue events = new LinkedBlockingQueue<>(QUEUE_CAPACITY); + private volatile boolean finished; + + /** The id of the request, once read; answers a stream cut short. */ + private volatile Object requestId; + + private Exchange(Request request, McpRequestContext context) { + this.request = request; + this.context = context; + this.transport = McpTransportContext.create(Map.of(McpProtocol.CONTEXT_KEY, context)); + } + + /** Runs in the admitted worker: reads, checks and dispatches the message. */ + private void run() throws IOException, InterruptedException { + JsonRpc.JSONRPCMessage message; + try (InputStream input = request.getEntity().getStream()) { + byte[] body = McpMessages.read(input, maxMessageBytes); + message = JsonRpc.deserializeMessage(mapper, McpMessages.utf8(body)); + } catch (McpMessages.MessageLimitException _) { + head.complete(new Reply(Status.CLIENT_ERROR_REQUEST_ENTITY_TOO_LARGE, null, false)); + return; + } catch (JsonRpc.InvalidMessageException e) { + head.complete( + error(e.id(), ErrorCodes.INVALID_REQUEST, "Invalid JSON-RPC message")); + return; + } catch (IOException _) { + head.complete(error(null, ErrorCodes.PARSE_ERROR, "Parse error")); + return; + } + if (message instanceof JsonRpc.JSONRPCNotification notification) { + notification(notification); + } else if (message instanceof JsonRpc.JSONRPCRequest rpc) { + call(rpc); + } else { + head.complete( + error(null, ErrorCodes.INVALID_REQUEST, "Requests or notifications only")); + } + } + + private void notification(JsonRpc.JSONRPCNotification notification) { + interceptors + .run( + mapper.convertValue(notification, McpSchema.JSONRPCNotification.class), + context, + (next, nextContext) -> + sdk.handleNotification( + transport, + mapper.convertValue( + next, + JsonRpc.JSONRPCNotification.class)) + .then(Mono.empty())) + .blockOptional(timeout); + head.complete(new Reply(Status.SUCCESS_ACCEPTED, null, false)); + } + + private void call(JsonRpc.JSONRPCRequest rpc) throws InterruptedException { + requestId = rpc.id(); + String mismatch = mismatch(rpc); + if (mismatch != null) { + head.complete(error(rpc.id(), ErrorCodes.HEADER_MISMATCH, mismatch)); + return; + } + Optional last = + interceptors + .run( + mapper.convertValue(rpc, McpSchema.JSONRPCRequest.class), + context, + (next, nextContext) -> Mono.fromCallable(() -> terminal(next))) + .blockOptional(timeout); + if (last.isEmpty()) { + // An interceptor answered nothing: accepted, as on the 2025-11-25 path. + head.complete(new Reply(Status.SUCCESS_ACCEPTED, null, false)); + return; + } + byte[] encoded; + try { + encoded = McpMessages.encode(mapper, last.get(), maxMessageBytes); + } catch (IOException _) { + // Too large to send, as on the 2025-11-25 path. + if (!head.complete(new Reply(Status.SERVER_ERROR_INTERNAL, null, false))) { + events.put(failure("MCP response too large")); + } + return; + } + if (!head.complete(json(status(last.get()), encoded))) { + events.put(encoded); + } + } + + /** Dispatches to the SDK after the interceptors, streaming what comes before the result. */ + private McpSchema.JSONRPCResponse terminal(McpSchema.JSONRPCMessage next) + throws IOException, InterruptedException { + JsonRpc.JSONRPCRequest rpc = mapper.convertValue(next, JsonRpc.JSONRPCRequest.class); + McpTransportContext sdkContext = transport; + if (McpSchema.METHOD_TOOLS_CALL.equals(rpc.method())) { + // The tools feature validates x-mcp-header arguments against these. + sdkContext = + transport.putAll( + Map.of( + McpParamHeaders.TRANSPORT_CONTEXT_KEY, + McpHeaderDecoder.paramHeaders(headers()))); + } + McpTransportResponse reply = sdk.handleBlocking(sdkContext, rpc).block(timeout); + if (reply instanceof McpTransportResponse.Result result) { + return classic(result.response()); + } + if (reply instanceof McpTransportResponse.Error failure) { + return classic(failure.response()); + } + if (!(reply instanceof McpTransportResponse.Streaming streaming)) { + throw new IOException("MCP request got no response"); + } + head.complete(new Reply(Status.SUCCESS_OK, null, true)); + JsonRpc.JSONRPCResponse last = null; + // Closing the stream cancels the SDK's, should the loop end early. + try (Stream messages = streaming.messages().toStream()) { + for (Iterator it = messages.iterator(); it.hasNext(); ) { + JsonRpc.JSONRPCMessage each = it.next(); + if (each instanceof JsonRpc.JSONRPCResponse response) { + last = response; + } else { + events.put(McpMessages.encode(mapper, each, maxMessageBytes)); + } + } + } + if (last == null) { + throw new IOException("MCP stream ended without a response"); + } + return classic(last); + } + + private McpSchema.JSONRPCResponse classic(JsonRpc.JSONRPCResponse response) { + return mapper.convertValue(response, McpSchema.JSONRPCResponse.class); + } + + /** The servlet transport's header checks, on Restlet headers. */ + private String mismatch(JsonRpc.JSONRPCRequest rpc) { + String method = header("Mcp-Method"); + if (method == null) { + return "Missing required header: Mcp-Method"; + } + if (!method.equals(rpc.method())) { + return "Mcp-Method header does not match request method"; + } + Map params = rpc.params() instanceof Map map ? map : Map.of(); + if (params.get("_meta") instanceof Map meta + && meta.get(MetaKeys.PROTOCOL_VERSION) instanceof String version + && !version.equals(header(McpProtocol.VERSION_HEADER))) { + return "MCP-Protocol-Version header does not match _meta protocol version"; + } + String key = + switch (rpc.method()) { + case "tools/call", "prompts/get" -> "name"; + case "resources/read" -> "uri"; + default -> null; + }; + if (key != null && params.get(key) instanceof String expected) { + String name = header("Mcp-Name"); + if (name == null) { + return "Missing required header: Mcp-Name"; + } + String decoded = McpHeaderDecoder.decodeValue(name); + if (decoded == null) { + return "Mcp-Name header has a malformed value"; + } + if (!expected.equals(decoded)) { + return "Mcp-Name header does not match request name/uri"; + } + } + return null; + } + + private String header(String name) { + return request.getHeaders().getFirstValue(name, true); + } + + /** The request headers, as the SDK's header decoder reads them. */ + private HeaderAccessor headers() { + return new HeaderAccessor() { + @Override + public List getHeader(String name) { + return Arrays.asList(request.getHeaders().getValuesArray(name, true)); + } + + @Override + public List getHeaderNames() { + return new ArrayList<>(request.getHeaders().getNames()); + } + }; + } + + /** An in-band error answering the request, for a stream that can't end with its result. */ + private byte[] failure(String message) { + try { + return McpMessages.encode( + mapper, + McpMessages.error(requestId, ErrorCodes.INTERNAL_ERROR, message), + maxMessageBytes); + } catch (IOException e) { + throw new java.io.UncheckedIOException(e); + } + } + + /** + * Ends the exchange after a failure: unavailable when nothing was answered yet, or an + * in-band error once streaming. The stream always ends, whatever happens here. + */ + private void fail(Throwable cause) { + try { + if (head.isDone()) { + byte[] error = failure("MCP exchange failed"); + if (!events.offer(error)) { + // A stalled client: what it hasn't read yet gives way to the answer. + events.clear(); + events.offer(error); + } + } else { + head.complete(unavailable()); + } + } catch (RuntimeException _) { + // The stream ends below anyway. + } finally { + finish(); + } + } + + /** Ends the exchange once the worker is done; it answered, unless it was cut short. */ + private void end() { + try { + if (!head.isDone()) { + head.complete(unavailable()); + } + } finally { + finish(); + } + } + + /** Marks the end of the stream, for the writer to stop once it has written the rest. */ + private void finish() { + finished = true; + events.offer(END); + } + + /** + * Takes the next streamed message, or null at the end of the stream: after its end marker, + * or once the worker finished and nothing is left, should the marker not fit the queue. + */ + private byte[] next() throws InterruptedException { + for (; ; ) { + byte[] event = events.poll(100, TimeUnit.MILLISECONDS); + if (event == END || (event == null && finished && events.isEmpty())) { + return null; + } + if (event != null) { + return event; + } + } + } + } + + /** A JSON-RPC error reply, built as a map since the id may be null. */ + private Reply error(Object id, int code, String message) { + return json( + ERROR_STATUSES.getOrDefault(code, Status.SUCCESS_OK), + McpMessages.error(id, code, message)); + } + + private Reply json(Status status, Object message) { + try { + return json(status, McpMessages.encode(mapper, message, maxMessageBytes)); + } catch (IOException _) { + return new Reply(Status.SERVER_ERROR_INTERNAL, null, false); + } + } + + private static Reply json(Status status, byte[] body) { + return new Reply(status, body, false); + } + + private static Status status(McpSchema.JSONRPCResponse response) { + return response.error() == null + ? Status.SUCCESS_OK + : ERROR_STATUSES.getOrDefault(response.error().code(), Status.SUCCESS_OK); + } + + /** The reply to write: a status and an optional JSON body, or an SSE stream. */ + private record Reply(Status status, byte[] body, boolean streaming) { + + private void apply( + Response response, Exchange exchange, Disposable running, Duration heartbeat) { + response.setStatus(status); + if (streaming) { + response.setEntity(stream(exchange, running, heartbeat)); + } else if (body != null) { + response.setEntity(new ByteArrayRepresentation(body, MediaType.APPLICATION_JSON)); + } else { + response.setEntity((Representation) null); + } + } + + private static Representation stream( + Exchange exchange, Disposable running, Duration heartbeat) { + return new EventStreamRepresentation( + sink -> { + try { + for (byte[] event = exchange.next(); + event != null; + event = exchange.next()) { + sink.send( + ServerSentEvent.builder() + .event("message") + .data(new String(event, StandardCharsets.UTF_8)) + .build()); + } + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + running.dispose(); + } catch (IOException e) { + running.dispose(); + throw e; + } + }, + heartbeat); + } + } +} 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 new file mode 100644 index 0000000000..0bde065ca1 --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernFeatures.java @@ -0,0 +1,299 @@ +/** + * 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.internal; + +import io.modelcontextprotocol.json.McpJsonMapper; +import io.modelcontextprotocol.modern.McpSchema; +import io.modelcontextprotocol.modern.server.McpAsyncResponse; +import io.modelcontextprotocol.modern.server.McpFeature; +import io.modelcontextprotocol.modern.server.McpRequestContext; +import io.modelcontextprotocol.modern.server.feature.McpAsyncPrompt; +import io.modelcontextprotocol.modern.server.feature.McpAsyncPromptRepository; +import io.modelcontextprotocol.modern.server.feature.McpAsyncResource; +import io.modelcontextprotocol.modern.server.feature.McpAsyncResourceRepository; +import io.modelcontextprotocol.modern.server.feature.McpAsyncTool; +import io.modelcontextprotocol.modern.server.feature.McpAsyncToolRepository; +import io.modelcontextprotocol.modern.server.feature.PromptsFeature; +import io.modelcontextprotocol.modern.server.feature.PromptsPage; +import io.modelcontextprotocol.modern.server.feature.ResourceTemplatesPage; +import io.modelcontextprotocol.modern.server.feature.ResourcesFeature; +import io.modelcontextprotocol.modern.server.feature.ResourcesPage; +import io.modelcontextprotocol.modern.server.feature.ToolsFeature; +import io.modelcontextprotocol.modern.server.feature.ToolsPage; +import io.modelcontextprotocol.server.McpAsyncListFilter; +import io.modelcontextprotocol.server.McpStatelessServerFeatures; +import io.modelcontextprotocol.util.DefaultMcpUriTemplateManager; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * 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. + */ +public final class McpModernFeatures { + + private McpModernFeatures() {} + + /** + * The registrations of an adapter. + * + * @param tools The {@code 2025-11-25} tools. + * @param resources The {@code 2025-11-25} resources. + * @param resourceTemplates The {@code 2025-11-25} resource templates. + * @param prompts The {@code 2025-11-25} prompts. + * @param toolFilters The tool list filters. + * @param modernTools The {@code 2026-07-28} tools. + */ + public record Registrations( + List tools, + List resources, + List resourceTemplates, + List prompts, + List> toolFilters, + List modernTools) {} + + /** + * Builds the features for registrations, one per kind present. + * + * @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. + */ + 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()) { + features.add(ResourcesFeature.ofAsync(new Resources(mapper, registrations))); + } + if (!registrations.prompts().isEmpty()) { + features.add(PromptsFeature.ofAsync(new Prompts(mapper, registrations.prompts()))); + } + return features; + } + + /** The tools, bridged and modern, indexed by name. */ + private static final class Tools implements McpAsyncToolRepository { + private final McpJsonMapper mapper; + private final Map byName = new LinkedHashMap<>(); + private final McpAsyncListFilter filter; + + private Tools(McpJsonMapper mapper, Registrations registrations) { + this.mapper = mapper; + this.filter = McpAsyncListFilter.and(List.copyOf(registrations.toolFilters())); + for (McpStatelessServerFeatures.AsyncToolSpecification spec : registrations.tools()) { + add(bridge(spec)); + } + for (McpAsyncTool tool : registrations.modernTools()) { + add(tool); + } + } + + private void add(McpAsyncTool tool) { + if (byName.putIfAbsent(tool.definition().name(), tool) != null) { + throw new IllegalArgumentException( + "Duplicate MCP tool name: " + tool.definition().name()); + } + } + + private McpAsyncTool bridge(McpStatelessServerFeatures.AsyncToolSpecification spec) { + return McpAsyncTool.of( + mapper.convertValue(spec.tool(), McpSchema.Tool.class), + (ctx, request) -> + spec.callHandler() + .apply( + ctx.transportContext(), + mapper.convertValue( + request, + io.modelcontextprotocol.spec.McpSchema + .CallToolRequest.class)) + .map( + result -> + McpAsyncResponse + .result( + mapper.convertValue( + result, + McpSchema.CallToolResult + .class)))); + } + + @Override + public Mono list(McpRequestContext ctx, String cursor) { + return Flux.fromIterable(byName.values()) + .map(McpAsyncTool::definition) + .filterWhen( + tool -> + filter.isVisible( + ctx.transportContext(), + mapper.convertValue( + tool, + io.modelcontextprotocol.spec.McpSchema.Tool + .class))) + .collectList() + .map(ToolsPage::of); + } + + @Override + public Mono find(McpRequestContext ctx, String name) { + return Mono.justOrEmpty(byName.get(name)); + } + } + + /** The bridged resources: exact URIs first, then templates, in registration order. */ + private static final class Resources implements McpAsyncResourceRepository { + private final McpJsonMapper mapper; + private final List resources; + private final List templates; + + private Resources(McpJsonMapper mapper, Registrations registrations) { + this.mapper = mapper; + this.resources = List.copyOf(registrations.resources()); + this.templates = List.copyOf(registrations.resourceTemplates()); + } + + @Override + public Mono list(McpRequestContext ctx, String cursor) { + return Mono.fromSupplier( + () -> + ResourcesPage.of( + resources.stream() + .map( + spec -> + mapper.convertValue( + spec.resource(), + McpSchema.Resource.class)) + .toList())); + } + + @Override + public Mono listTemplates(McpRequestContext ctx, String cursor) { + return Mono.fromSupplier( + () -> + ResourceTemplatesPage.of( + templates.stream() + .map( + spec -> + mapper.convertValue( + spec.resourceTemplate(), + McpSchema.ResourceTemplate + .class)) + .toList())); + } + + @Override + public Mono find(McpRequestContext ctx, String uri) { + for (McpStatelessServerFeatures.AsyncResourceSpecification spec : resources) { + if (spec.resource().uri().equals(uri)) { + return Mono.just(read(spec.readHandler())); + } + } + for (McpStatelessServerFeatures.AsyncResourceTemplateSpecification spec : templates) { + if (new DefaultMcpUriTemplateManager(spec.resourceTemplate().uriTemplate()) + .matches(uri)) { + return Mono.just(read(spec.readHandler())); + } + } + return Mono.empty(); + } + + private McpAsyncResource read( + java.util.function.BiFunction< + io.modelcontextprotocol.common.McpTransportContext, + io.modelcontextprotocol.spec.McpSchema.ReadResourceRequest, + Mono> + handler) { + return McpAsyncResource.of( + (ctx, request) -> + handler.apply( + ctx.transportContext(), + mapper.convertValue( + request, + io.modelcontextprotocol.spec.McpSchema + .ReadResourceRequest.class)) + .map( + result -> + McpAsyncResponse + .result( + mapper.convertValue( + result, + McpSchema + .ReadResourceResult + .class)))); + } + } + + /** The bridged prompts, indexed by name. */ + private static final class Prompts implements McpAsyncPromptRepository { + private final McpJsonMapper mapper; + private final Map byName = + new LinkedHashMap<>(); + + private Prompts( + McpJsonMapper mapper, + List prompts) { + this.mapper = mapper; + for (McpStatelessServerFeatures.AsyncPromptSpecification spec : prompts) { + byName.put(spec.prompt().name(), spec); + } + } + + @Override + public Mono list(McpRequestContext ctx, String cursor) { + return Mono.fromSupplier( + () -> + PromptsPage.of( + byName.values().stream() + .map( + spec -> + mapper.convertValue( + spec.prompt(), + McpSchema.Prompt.class)) + .toList())); + } + + @Override + public Mono find(McpRequestContext ctx, String name) { + McpStatelessServerFeatures.AsyncPromptSpecification spec = byName.get(name); + if (spec == null) { + return Mono.empty(); + } + return Mono.just( + McpAsyncPrompt.of( + mapper.convertValue(spec.prompt(), McpSchema.Prompt.class), + (ctx2, request) -> + spec.promptHandler() + .apply( + ctx2.transportContext(), + mapper.convertValue( + request, + io.modelcontextprotocol.spec.McpSchema + .GetPromptRequest.class)) + .map( + result -> + McpAsyncResponse + . + result( + mapper + .convertValue( + result, + McpSchema + .GetPromptResult + .class))))); + } + } +} diff --git a/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java new file mode 100644 index 0000000000..3ed97c3f52 --- /dev/null +++ b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java @@ -0,0 +1,782 @@ +/** + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.modelcontextprotocol.json.McpJsonDefaults; +import io.modelcontextprotocol.json.McpJsonMapper; +import io.modelcontextprotocol.json.TypeRef; +import io.modelcontextprotocol.modern.McpSchema.CallToolOutcome; +import io.modelcontextprotocol.modern.McpSchema.CallToolResult; +import io.modelcontextprotocol.modern.McpSchema.ElicitFormRequest; +import io.modelcontextprotocol.modern.McpSchema.InputRequiredResult; +import io.modelcontextprotocol.modern.McpSchema.MetaKeys; +import io.modelcontextprotocol.modern.McpSchema.TextContent; +import io.modelcontextprotocol.modern.McpSchema.Tool; +import io.modelcontextprotocol.modern.server.McpAsyncResponse; +import io.modelcontextprotocol.modern.server.feature.McpAsyncTool; +import io.modelcontextprotocol.server.McpStatelessServerFeatures; +import io.modelcontextprotocol.spec.McpSchema; +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.data.MediaType; +import org.restlet.data.Method; +import org.restlet.representation.StringRepresentation; +import reactor.core.publisher.Mono; + +/** + * MCP {@code 2026-07-28} served by {@link McpServerAdapter} next to {@code 2025-11-25}: bridged + * registrations, modern tools, interceptors, filters, limits and protocol errors. + */ +class McpModernServerTestCase { + private static final McpJsonMapper MAPPER = McpJsonDefaults.getMapper(); + private static final String MODERN = McpProtocol.MODERN_VERSION; + + private final List adapters = new ArrayList<>(); + + @AfterEach + void cleanup() throws Exception { + for (McpServerAdapter adapter : adapters) { + adapter.stop(); + } + } + + /** The 2025-11-25 registrations are served to 2026-07-28 clients, converted. */ + @Test + void bridgedRegistrationsAnswerModernClients() throws Exception { + McpServerAdapter adapter = adapter(); + addResourcesAndPrompts(adapter); + adapter.start(); + + Map discover = result(call(adapter, "server/discover", Map.of(), null)); + assertEquals(List.of(MODERN), discover.get("supportedVersions")); + Map capabilities = (Map) discover.get("capabilities"); + for (String kind : List.of("tools", "resources", "prompts")) { + assertTrue(capabilities.containsKey(kind), kind + " in " + capabilities); + } + + List tools = (List) result(call(adapter, "tools/list", Map.of(), null)).get("tools"); + assertEquals( + List.of("echo", "progress", "approve"), + tools.stream().map(tool -> ((Map) tool).get("name")).toList()); + assertEquals( + "echo:hi", + firstText( + result( + call( + adapter, + "tools/call", + Map.of("name", "echo", "arguments", Map.of("text", "hi")), + "echo")))); + for (String uri : List.of("test://fixed", "test://items/7")) { + Map read = + result(call(adapter, "resources/read", Map.of("uri", uri), uri)); + assertEquals(uri, ((Map) ((List) read.get("contents")).get(0)).get("uri")); + } + Map prompt = + result(call(adapter, "prompts/get", Map.of("name", "greet"), "greet")); + assertEquals("greeting", prompt.get("description")); + } + + /** A modern tool streams its progress over SSE, then the result answering the request. */ + @Test + void aModernToolStreamsItsProgressBeforeItsResult() throws Exception { + McpServerAdapter adapter = adapter(); + adapter.start(); + Map meta = meta(); + meta.put(MetaKeys.PROGRESS_TOKEN, "p1"); + + Response response = + adapter.handle( + modern( + "tools/call", + Map.of("name", "progress", "arguments", Map.of()), + "progress", + meta)); + + assertEquals(200, response.getStatus().getCode()); + assertEquals(MediaType.TEXT_EVENT_STREAM, response.getEntity().getMediaType()); + List> events = events(body(response)); + assertEquals(2, events.size(), events.toString()); + assertEquals("notifications/progress", events.get(0).get("method")); + assertEquals("half", ((Map) events.get(0).get("params")).get("message")); + assertEquals(1, ((Number) events.get(1).get("id")).intValue()); + @SuppressWarnings("unchecked") + Map result = (Map) events.get(1).get("result"); + assertEquals("done", firstText(result)); + } + + /** + * An input_required result seals its state with the request-state key: the retry reaches + * another instance with the same key, but not one with another key. + */ + @Test + void anInputRequiredRetryReachesAnInstanceSharingTheKey() throws Exception { + byte[] key = "0123456789abcdef0123456789abcdef".getBytes(StandardCharsets.US_ASCII); + McpServerAdapter first = adapter(); + first.setRequestStateKey(key); + first.start(); + McpServerAdapter second = adapter(); + second.setRequestStateKey(key); + second.start(); + McpServerAdapter other = adapter(); + other.setRequestStateKey( + "fedcba9876543210fedcba9876543210".getBytes(StandardCharsets.US_ASCII)); + other.start(); + Map meta = meta(); + meta.put(MetaKeys.CLIENT_CAPABILITIES, Map.of("elicitation", Map.of("form", Map.of()))); + + Map asked = + result( + first.handle( + modern( + "tools/call", + Map.of("name", "approve", "arguments", Map.of()), + "approve", + meta))); + assertEquals("input_required", asked.get("resultType"), asked.toString()); + String state = (String) asked.get("requestState"); + assertNotNull(state); + assertFalse(state.contains("s1"), "the state must be sealed, not sent in clear"); + + Map retry = new LinkedHashMap<>(); + retry.put("name", "approve"); + retry.put("arguments", Map.of()); + retry.put("requestState", state); + retry.put( + "inputResponses", + Map.of("confirm", Map.of("action", "accept", "content", Map.of()))); + Map answered = + result(second.handle(modern("tools/call", retry, "approve", meta))); + assertEquals("approved:s1", firstText(answered)); + + Response refused = other.handle(modern("tools/call", retry, "approve", meta)); + assertNotNull(envelope(refused).get("error"), body(refused)); + } + + /** Interceptors see 2026-07-28 requests, may refuse them, and see their final result. */ + @Test + void interceptorsWrapModernRequests() throws Exception { + McpServerAdapter adapter = adapter(); + List seen = new CopyOnWriteArrayList<>(); + adapter.getInterceptors() + .add( + (message, context, chain) -> { + seen.add(context.protocolVersion()); + if (message instanceof McpSchema.JSONRPCRequest request + && request.params() instanceof Map params + && "approve".equals(params.get("name"))) { + return Mono.just( + new McpSchema.JSONRPCResponse( + "2.0", + request.id(), + null, + new McpSchema.JSONRPCResponse.JSONRPCError( + -32001, "Not allowed", null))); + } + return chain.proceed(message, context) + .doOnNext(response -> seen.add("final:" + response.id())); + }); + adapter.start(); + + Response refused = + adapter.handle( + modern( + "tools/call", + Map.of("name", "approve", "arguments", Map.of()), + "approve", + meta())); + assertEquals( + -32001, + ((Number) ((Map) envelope(refused).get("error")).get("code")).intValue()); + assertEquals( + "echo:x", + firstText( + result( + call( + adapter, + "tools/call", + Map.of("name", "echo", "arguments", Map.of("text", "x")), + "echo")))); + assertEquals(List.of(MODERN, MODERN, "final:1"), seen); + } + + /** The adapter's tool filters apply to 2026-07-28 listings, modern tools included. */ + @Test + void toolFiltersApplyToModernListings() throws Exception { + McpServerAdapter adapter = adapter(); + adapter.getToolFilters().add((context, tool) -> Mono.just(!tool.name().equals("approve"))); + adapter.start(); + + List tools = (List) result(call(adapter, "tools/list", Map.of(), null)).get("tools"); + + assertEquals( + List.of("echo", "progress"), + tools.stream().map(tool -> ((Map) tool).get("name")).toList()); + } + + /** The same adapter still serves 2025-11-25 clients. */ + @Test + void legacyClientsAreStillServed() throws Exception { + McpServerAdapter adapter = adapter(); + adapter.start(); + + Response response = + adapter.handle( + McpServerAdapterTestCase.request( + "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":" + + "{\"name\":\"echo\",\"arguments\":{\"text\":\"old\"}}}")); + + Map envelope = McpServerAdapterTestCase.envelope(response); + @SuppressWarnings("unchecked") + Map result = (Map) envelope.get("result"); + assertEquals("echo:old", firstText(result)); + assertFalse(result.containsKey("resultType"), "2025-11-25 results have no resultType"); + } + + /** The call limit counts 2026-07-28 requests: a call over it is unavailable. */ + @Test + void theCallLimitAppliesToModernRequests() throws Exception { + McpServerAdapter adapter = adapter(); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + adapter.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder("hold", Map.of("type", "object")).build(), + (ctx, request) -> + Mono.fromCallable( + () -> { + entered.countDown(); + release.await(10, TimeUnit.SECONDS); + return McpAsyncResponse.result( + text("held")); + }))); + adapter.setMaxConcurrentCalls(1); + adapter.start(); + Map hold = Map.of("name", "hold", "arguments", Map.of()); + + CompletableFuture held = + CompletableFuture.supplyAsync( + () -> adapter.handle(modern("tools/call", hold, "hold", meta()))); + assertTrue(entered.await(10, TimeUnit.SECONDS)); + Response refused = adapter.handle(modern("tools/list", Map.of(), null, meta())); + release.countDown(); + + assertEquals(503, refused.getStatus().getCode(), body(refused)); + assertEquals("held", firstText(result(held.get(10, TimeUnit.SECONDS)))); + } + + /** Protocol errors use the 2026-07-28 codes and HTTP statuses; a notification gets 202. */ + @Test + void protocolErrorsAndNotifications() throws Exception { + McpServerAdapter adapter = adapter(); + adapter.setMaxMessageBytes(4096); + adapter.start(); + + Request notification = + post( + "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\",\"params\":" + + "{\"requestId\":1,\"_meta\":{\"" + + MetaKeys.PROTOCOL_VERSION + + "\":\"" + + MODERN + + "\"}}}", + "notifications/cancelled", + null); + Response accepted = adapter.handle(notification); + assertEquals(202, accepted.getStatus().getCode()); + assertTrue(accepted.getEntity() == null || accepted.getEntity().getSize() <= 0); + + Response mismatch = + adapter.handle( + modern( + "tools/call", + Map.of("name", "echo", "arguments", Map.of()), + "other", + meta())); + assertEquals(400, mismatch.getStatus().getCode()); + assertEquals(-32020, code(mismatch)); + + Response unknown = adapter.handle(modern("nothing/here", Map.of(), null, meta())); + assertEquals(404, unknown.getStatus().getCode()); + assertEquals(-32601, code(unknown)); + + Response oversized = + adapter.handle( + modern( + "tools/call", + Map.of( + "name", + "echo", + "arguments", + Map.of("text", "x".repeat(8192))), + "echo", + meta())); + assertEquals(413, oversized.getStatus().getCode()); + } + + /** A tool's x-mcp-header arguments are checked against their Mcp-Param headers. */ + @Test + void paramHeadersAreValidatedForAnnotatedTools() throws Exception { + McpServerAdapter adapter = adapter(); + adapter.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder( + "region", + Map.of( + "type", + "object", + "properties", + Map.of( + "region", + Map.of( + "type", + "string", + "x-mcp-header", + "Region")))) + .build(), + (ctx, request) -> + Mono.just( + McpAsyncResponse.result( + text("ok"))))); + adapter.start(); + Map params = Map.of("name", "region", "arguments", Map.of("region", "eu")); + + Request matching = modern("tools/call", params, "region", meta()); + matching.getHeaders().set("Mcp-Param-Region", "eu"); + assertEquals("ok", firstText(result(adapter.handle(matching)))); + + Request differing = modern("tools/call", params, "region", meta()); + differing.getHeaders().set("Mcp-Param-Region", "us"); + assertHeaderMismatch(adapter.handle(differing)); + assertHeaderMismatch(adapter.handle(modern("tools/call", params, "region", meta()))); + } + + /** Checks a 400 header-mismatch error, reading the body once. */ + private static void assertHeaderMismatch(Response response) throws Exception { + String body = body(response); + assertEquals(400, response.getStatus().getCode(), body); + Map envelope = + MAPPER.readValue(body, new TypeRef>() {}); + assertEquals(-32020, ((Number) ((Map) envelope.get("error")).get("code")).intValue()); + } + + /** A client gone mid-stream cancels the tool's stream and frees the call slot. */ + @Test + void aClientGoneMidStreamCancelsTheTool() throws Exception { + McpServerAdapter adapter = adapter(); + CountDownLatch cancelled = new CountDownLatch(1); + adapter.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder("endless", Map.of("type", "object")).build(), + (ctx, request) -> + Mono.just( + McpAsyncResponse.streaming( + notifier -> + notifier.progress( + 1, null, "started") + .then( + Mono + . + never() + .doOnCancel( + cancelled + ::countDown)))))); + adapter.setMaxConcurrentCalls(1); + adapter.start(); + Map meta = meta(); + meta.put(MetaKeys.PROGRESS_TOKEN, "p1"); + + Response response = + adapter.handle( + modern( + "tools/call", + Map.of("name", "endless", "arguments", Map.of()), + "endless", + meta)); + java.io.OutputStream gone = + new java.io.OutputStream() { + @Override + public void write(int b) throws java.io.IOException { + throw new java.io.IOException("client gone"); + } + + @Override + public void write(byte[] b, int off, int len) throws java.io.IOException { + throw new java.io.IOException("client gone"); + } + }; + try { + response.getEntity().write(gone); + } catch (java.io.IOException expected) { + // The writer reports the broken client. + } + + assertTrue(cancelled.await(10, TimeUnit.SECONDS), "the tool's stream was not cancelled"); + assertEquals( + "echo:freed", + firstText( + result( + call( + adapter, + "tools/call", + Map.of( + "name", + "echo", + "arguments", + Map.of("text", "freed")), + "echo")))); + } + + /** A stream cut short by the timeout ends with an error answering the request. */ + @Test + void aTimeoutDuringAStreamEndsItWithAnError() throws Exception { + McpServerAdapter adapter = adapter(); + adapter.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder("stuck", Map.of("type", "object")).build(), + (ctx, request) -> + Mono.just( + McpAsyncResponse.streaming( + notifier -> + notifier.progress( + 1, null, "started") + .then( + Mono + . + never()))))); + adapter.setRequestTimeout(java.time.Duration.ofMillis(500)); + adapter.start(); + Map meta = meta(); + meta.put(MetaKeys.PROGRESS_TOKEN, "p1"); + + Response response = + adapter.handle( + modern( + "tools/call", + Map.of("name", "stuck", "arguments", Map.of()), + "stuck", + meta)); + List> events = + CompletableFuture.supplyAsync( + () -> { + try { + return events(body(response)); + } catch (Exception e) { + throw new IllegalStateException(e); + } + }) + .get(10, TimeUnit.SECONDS); + + assertEquals(2, events.size(), events.toString()); + assertEquals("notifications/progress", events.get(0).get("method")); + assertEquals(1, ((Number) events.get(1).get("id")).intValue(), events.toString()); + assertNotNull(events.get(1).get("error"), events.toString()); + } + + /** The input_required state is bound to the user: another user can't replay it. */ + @Test + void inputRequiredStateIsBoundToTheUser() throws Exception { + McpServerAdapter adapter = adapter(); + adapter.start(); + Map meta = meta(); + meta.put(MetaKeys.CLIENT_CAPABILITIES, Map.of("elicitation", Map.of("form", Map.of()))); + Request ask = + modern( + "tools/call", + Map.of("name", "approve", "arguments", Map.of()), + "approve", + meta); + ask.getClientInfo().setUser(new org.restlet.security.User("alice")); + String state = (String) result(adapter.handle(ask)).get("requestState"); + Map retry = new LinkedHashMap<>(); + retry.put("name", "approve"); + retry.put("arguments", Map.of()); + retry.put("requestState", state); + retry.put( + "inputResponses", + Map.of("confirm", Map.of("action", "accept", "content", Map.of()))); + + Request asBob = modern("tools/call", retry, "approve", meta); + asBob.getClientInfo().setUser(new org.restlet.security.User("bob")); + Response refused = adapter.handle(asBob); + assertNotNull(envelope(refused).get("error"), body(refused)); + + Request asAlice = modern("tools/call", retry, "approve", meta); + asAlice.getClientInfo().setUser(new org.restlet.security.User("alice")); + assertEquals("approved:s1", firstText(result(adapter.handle(asAlice)))); + } + + /** A 2026-07-28 request from an origin that isn't allowed is forbidden. */ + @Test + void anUnallowedOriginIsForbidden() throws Exception { + McpServerAdapter adapter = adapter(); + adapter.getAllowedOrigins().add("https://allowed.example"); + adapter.start(); + + Request request = modern("tools/list", Map.of(), null, meta()); + request.getHeaders().set("Origin", "https://evil.example"); + + assertEquals(403, adapter.handle(request).getStatus().getCode()); + } + + /** A tool name can't be used by both a 2025-11-25 tool and a modern one. */ + @Test + void aDuplicateToolNameFailsTheStart() { + McpServerAdapter adapter = adapter(); + adapter.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder("echo", Map.of("type", "object")).build(), + (ctx, request) -> + Mono.just( + McpAsyncResponse.result( + text("modern"))))); + + IllegalArgumentException error = + org.junit.jupiter.api.Assertions.assertThrows( + IllegalArgumentException.class, adapter::start); + assertTrue(error.getMessage().contains("echo"), error.getMessage()); + } + + /** An interceptor sees and may change the final result of a streamed call. */ + @Test + void anInterceptorChangesTheResultOfAStreamedCall() throws Exception { + McpServerAdapter adapter = adapter(); + adapter.getInterceptors() + .add( + (message, context, chain) -> + chain.proceed(message, context) + .map( + response -> + new McpSchema.JSONRPCResponse( + response.jsonrpc(), + response.id(), + Map.of( + "content", + List.of( + Map.of( + "type", + "text", + "text", + "intercepted")), + "resultType", + "complete"), + null))); + adapter.start(); + Map meta = meta(); + meta.put(MetaKeys.PROGRESS_TOKEN, "p1"); + + List> events = + events( + body( + adapter.handle( + modern( + "tools/call", + Map.of("name", "progress", "arguments", Map.of()), + "progress", + meta)))); + + assertEquals(2, events.size(), events.toString()); + @SuppressWarnings("unchecked") + Map result = (Map) events.get(1).get("result"); + assertEquals("intercepted", firstText(result)); + } + + private McpServerAdapter adapter() { + McpServerAdapter adapter = new McpServerAdapter(); + McpServerAdapterTestCase.addEcho(adapter); + adapter.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder("progress", Map.of("type", "object")).build(), + (ctx, request) -> + Mono.just( + McpAsyncResponse.streaming( + notifier -> + notifier.progress(1, 2.0, "half") + .then( + Mono.just( + text( + "done"))))))); + adapter.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder("approve", Map.of("type", "object")).build(), + (ctx, request) -> + Mono.just( + request.requestState() == null + ? McpAsyncResponse.result( + InputRequiredResult.builder() + .elicit( + "confirm", + ElicitFormRequest + .builder( + "Proceed?", + Map + .of( + "type", + "object", + "properties", + Map + .of())) + .build()) + .requestState("s1") + .build()) + : McpAsyncResponse.result( + text( + "approved:" + + request + .requestState()))))); + adapters.add(adapter); + return adapter; + } + + private static void addResourcesAndPrompts(McpServerAdapter adapter) { + adapter.getResources() + .add( + new McpStatelessServerFeatures.AsyncResourceSpecification( + McpSchema.Resource.builder("test://fixed", "fixed").build(), + (context, request) -> Mono.just(contents(request.uri())))); + adapter.getResourceTemplates() + .add( + new McpStatelessServerFeatures.AsyncResourceTemplateSpecification( + McpSchema.ResourceTemplate.builder("test://items/{id}", "items") + .build(), + (context, request) -> Mono.just(contents(request.uri())))); + adapter.getPrompts() + .add( + new McpStatelessServerFeatures.AsyncPromptSpecification( + McpSchema.Prompt.builder("greet").build(), + (context, request) -> + Mono.just( + new McpSchema.GetPromptResult( + "greeting", + List.of( + new McpSchema.PromptMessage( + McpSchema.Role.USER, + McpSchema.TextContent + .builder("hello") + .build())), + null)))); + } + + private static McpSchema.ReadResourceResult contents(String uri) { + return new McpSchema.ReadResourceResult( + List.of(new McpSchema.TextResourceContents(uri, "text/plain", "text", null)), null); + } + + private static CallToolResult text(String text) { + return CallToolResult.builder().addContent(TextContent.builder(text).build()).build(); + } + + private static Map meta() { + Map meta = new LinkedHashMap<>(); + meta.put(MetaKeys.PROTOCOL_VERSION, MODERN); + meta.put(MetaKeys.CLIENT_CAPABILITIES, Map.of()); + return meta; + } + + private static Response call( + McpServerAdapter adapter, String method, Map params, String name) { + return adapter.handle(modern(method, params, name, meta())); + } + + /** A 2026-07-28 request with its _meta and headers. */ + private static Request modern( + String method, Map params, String name, Map meta) { + Map withMeta = new LinkedHashMap<>(params); + withMeta.put("_meta", meta); + Map envelope = new LinkedHashMap<>(); + envelope.put("jsonrpc", "2.0"); + envelope.put("id", 1); + envelope.put("method", method); + envelope.put("params", withMeta); + try { + return post(MAPPER.writeValueAsString(envelope), method, name); + } catch (java.io.IOException e) { + throw new IllegalStateException(e); + } + } + + private static Request post(String body, String method, String name) { + Request request = + new Request( + Method.POST, + "http://localhost/mcp", + new StringRepresentation(body, MediaType.APPLICATION_JSON)); + request.getHeaders().set(McpProtocol.VERSION_HEADER, MODERN); + request.getHeaders().set("Accept", "application/json, text/event-stream"); + request.getHeaders().set("Mcp-Method", method); + if (name != null) { + request.getHeaders().set("Mcp-Name", name); + } + return request; + } + + private static String body(Response response) throws Exception { + if (response.getEntity() == null) { + return ""; + } + var bytes = new ByteArrayOutputStream(); + response.getEntity().write(bytes); + return bytes.toString(StandardCharsets.UTF_8); + } + + private static Map envelope(Response response) throws Exception { + return MAPPER.readValue(body(response), new TypeRef>() {}); + } + + @SuppressWarnings("unchecked") + private static Map result(Response response) throws Exception { + Map envelope = envelope(response); + assertTrue(envelope.containsKey("result"), envelope.toString()); + return (Map) envelope.get("result"); + } + + private static int code(Response response) throws Exception { + return ((Number) ((Map) envelope(response).get("error")).get("code")).intValue(); + } + + private static String firstText(Map result) { + return (String) ((Map) ((List) result.get("content")).get(0)).get("text"); + } + + private static List> events(String body) throws Exception { + List> events = new ArrayList<>(); + for (String line : body.split("\n")) { + if (line.startsWith("data:")) { + events.add( + MAPPER.readValue( + line.substring(5).trim(), new TypeRef>() {})); + } + } + return events; + } +} diff --git a/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernSpikeTestCase.java b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernSpikeTestCase.java deleted file mode 100644 index 0e41253838..0000000000 --- a/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernSpikeTestCase.java +++ /dev/null @@ -1,570 +0,0 @@ -/** - * 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 static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertTrue; - -import io.modelcontextprotocol.common.McpTransportContext; -import io.modelcontextprotocol.json.McpJsonDefaults; -import io.modelcontextprotocol.json.McpJsonMapper; -import io.modelcontextprotocol.json.TypeRef; -import io.modelcontextprotocol.modern.JsonRpc; -import io.modelcontextprotocol.modern.JsonRpc.JSONRPCMessage; -import io.modelcontextprotocol.modern.JsonRpc.JSONRPCNotification; -import io.modelcontextprotocol.modern.JsonRpc.JSONRPCRequest; -import io.modelcontextprotocol.modern.JsonRpc.JSONRPCResponse; -import io.modelcontextprotocol.modern.JsonRpc.JSONRPCResponse.JSONRPCError; -import io.modelcontextprotocol.modern.McpSchema; -import io.modelcontextprotocol.modern.McpSchema.CallToolOutcome; -import io.modelcontextprotocol.modern.McpSchema.CallToolResult; -import io.modelcontextprotocol.modern.McpSchema.ErrorCodes; -import io.modelcontextprotocol.modern.McpSchema.Implementation; -import io.modelcontextprotocol.modern.McpSchema.MetaKeys; -import io.modelcontextprotocol.modern.McpSchema.TextContent; -import io.modelcontextprotocol.modern.McpSchema.Tool; -import io.modelcontextprotocol.modern.server.McpRequestContext; -import io.modelcontextprotocol.modern.server.McpServer; -import io.modelcontextprotocol.modern.server.McpSyncResponse; -import io.modelcontextprotocol.modern.server.McpTransportResponse; -import io.modelcontextprotocol.modern.server.feature.McpSyncTool; -import io.modelcontextprotocol.modern.server.feature.McpSyncToolRepository; -import io.modelcontextprotocol.modern.server.feature.ToolsFeature; -import io.modelcontextprotocol.modern.server.feature.ToolsPage; -import io.modelcontextprotocol.modern.server.transport.McpHeaderDecoder; -import java.io.IOException; -import java.net.URI; -import java.net.http.HttpClient; -import java.net.http.HttpRequest; -import java.net.http.HttpResponse; -import java.nio.charset.StandardCharsets; -import java.time.Duration; -import java.util.Base64; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.restlet.Context; -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.data.ServerSentEvent; -import org.restlet.data.Status; -import org.restlet.representation.EventStreamRepresentation; -import org.restlet.representation.StringRepresentation; - -/** - * R6b gate 0 for MCP 2026-07-28: the MCP SDK's modern stateless server (mcp-core 2.2.0-M2, a - * preview) behind a test-only Restlet transport, mirroring the SDK's own servlet transport, and - * called as the Ikanos client calls a server. Test-only: no production API is implied. - */ -class McpModernSpikeTestCase { - private static final Duration TIMEOUT = Duration.ofSeconds(10); - private static final McpJsonMapper MAPPER = McpJsonDefaults.getMapper(); - private static final String VERSION = McpSchema.LATEST_PROTOCOL_VERSION; - - private static final Tool ECHO = - Tool.builder( - "echo", - Map.of( - "type", - "object", - "properties", - Map.of("text", Map.of("type", "string")))) - .description("Echoes text") - .build(); - - private static final Tool PROGRESS = - Tool.builder("progress", Map.of("type", "object")) - .description("Reports progress, then answers") - .build(); - - private McpServer sdkServer; - private Server server; - private HttpClient http; - private String target; - - @BeforeEach - void start() throws Exception { - McpSyncToolRepository tools = - new McpSyncToolRepository() { - @Override - public ToolsPage list(McpRequestContext ctx, String cursor) { - return ToolsPage.of(List.of(ECHO, PROGRESS)); - } - - @Override - public McpSyncTool find(McpRequestContext ctx, String name) { - if (ECHO.name().equals(name)) { - return McpSyncTool.of( - ECHO, - (context, request) -> - McpSyncResponse.result( - text( - String.valueOf( - request.arguments() - .get("text"))))); - } - if (PROGRESS.name().equals(name)) { - // A streaming reply: a progress notification, then the result. - return McpSyncTool.of( - PROGRESS, - (context, request) -> - McpSyncResponse.streaming( - notifier -> { - notifier.progress(1, 2.0, "half"); - return text("done"); - })); - } - return null; - } - }; - sdkServer = - McpServer.builder() - .serverInfo(Implementation.builder("restlet-modern-spike", "1").build()) - .feature(ToolsFeature.ofSync(tools)) - .build(); - server = new Server(new Context(), Protocol.HTTP, 0, new ModernHost(sdkServer)); - server.start(); - target = "http://localhost:" + server.getActualPort() + "/mcp"; - http = HttpClient.newHttpClient(); - } - - @AfterEach - void stop() throws Exception { - try { - if (http != null) { - http.close(); - } - } finally { - try { - if (sdkServer != null) { - sdkServer.closeGracefully(); - } - } finally { - if (server != null) { - server.stop(); - } - } - } - } - - /** Discovery replaces initialize: versions, capabilities and the server identity. */ - @Test - void serverDiscoverAdvertisesTheRevision() throws Exception { - Reply reply = call("server/discover", Map.of(), null); - assertEquals(200, reply.status, reply.body); - Map result = reply.result(); - assertEquals(List.of(VERSION), result.get("supportedVersions")); - assertEquals("complete", result.get("resultType")); - assertTrue(((Map) result.get("capabilities")).containsKey("tools"), reply.body); - assertEquals( - "restlet-modern-spike", - ((Map) ((Map) result.get("_meta")).get(MetaKeys.SERVER_INFO)) - .get("name")); - } - - /** A tool is listed and called as Ikanos calls one, with a plain or base64 Mcp-Name. */ - @Test - void toolsAreListedAndCalledAsIkanosCallsThem() throws Exception { - Reply list = call("tools/list", Map.of(), null); - assertEquals(200, list.status, list.body); - assertEquals( - "echo", ((Map) ((List) list.result().get("tools")).get(0)).get("name")); - assertNotNull(list.result().get("ttlMs"), list.body); - assertNotNull(list.result().get("cacheScope"), list.body); - - Map params = Map.of("name", "echo", "arguments", Map.of("text", "hello")); - Reply plain = call("tools/call", params, "echo"); - assertEquals(200, plain.status, plain.body); - assertEquals("hello", firstText(plain.result())); - assertEquals("complete", plain.result().get("resultType")); - - String encoded = - "=?base64?" - + Base64.getEncoder() - .encodeToString("echo".getBytes(StandardCharsets.UTF_8)) - + "?="; - Reply base64 = call("tools/call", params, encoded); - assertEquals(200, base64.status, base64.body); - assertEquals("hello", firstText(base64.result())); - } - - /** - * A streaming reply comes as an SSE stream over core SSE: the progress notification, then the - * result answering the request, then the end of the stream. - */ - @Test - void aStreamingReplyCarriesProgressThenTheResult() throws Exception { - String request = - body( - "tools/call", - Map.of("name", "progress", "arguments", Map.of()), - Map.of( - MetaKeys.PROTOCOL_VERSION, - VERSION, - MetaKeys.CLIENT_CAPABILITIES, - Map.of(), - MetaKeys.PROGRESS_TOKEN, - "p1")); - HttpRequest.Builder post = - HttpRequest.newBuilder(URI.create(target)) - .timeout(TIMEOUT) - .header("Content-Type", "application/json") - .POST(HttpRequest.BodyPublishers.ofString(request)); - headers("tools/call", VERSION, "progress").forEach(post::header); - // Returns once the stream has ended. - HttpResponse response = - http.send(post.build(), HttpResponse.BodyHandlers.ofString()); - - assertEquals(200, response.statusCode(), response.body()); - assertTrue( - response.headers() - .firstValue("Content-Type") - .orElse("") - .startsWith("text/event-stream"), - response.headers().toString()); - List> events = events(response.body()); - assertEquals(2, events.size(), response.body()); - - Map progress = events.get(0); - assertEquals("notifications/progress", progress.get("method")); - Map params = (Map) progress.get("params"); - assertEquals("p1", params.get("progressToken")); - assertEquals("half", params.get("message")); - - Map answer = events.get(1); - assertEquals(1, ((Number) answer.get("id")).intValue()); - @SuppressWarnings("unchecked") - Map result = (Map) answer.get("result"); - assertEquals("done", firstText(result)); - assertEquals("complete", result.get("resultType")); - } - - /** Header and version errors use the 2026-07-28 codes and the HTTP statuses Ikanos uses. */ - @Test - void protocolErrorsMatchTheRevision() throws Exception { - Map params = Map.of("name", "echo", "arguments", Map.of("text", "x")); - - Reply mismatch = call("tools/call", params, "other"); - assertEquals(400, mismatch.status, mismatch.body); - assertEquals(ErrorCodes.HEADER_MISMATCH, mismatch.errorCode()); - - Reply noMethod = send(body("tools/list", Map.of(), VERSION), headers(null, VERSION, null)); - assertEquals(400, noMethod.status, noMethod.body); - assertEquals(ErrorCodes.HEADER_MISMATCH, noMethod.errorCode()); - - Reply oldVersion = - send( - body("tools/list", Map.of(), "2025-11-25"), - headers("tools/list", "2025-11-25", null)); - assertEquals(400, oldVersion.status, oldVersion.body); - assertEquals(ErrorCodes.UNSUPPORTED_PROTOCOL_VERSION, oldVersion.errorCode()); - assertEquals( - List.of(VERSION), ((Map) oldVersion.error().get("data")).get("supported")); - - Reply unknown = call("nothing/here", Map.of(), null); - assertEquals(404, unknown.status, unknown.body); - assertEquals(ErrorCodes.METHOD_NOT_FOUND, unknown.errorCode()); - } - - /** - * Legacy traffic: GET is refused with 405, a notification gets 202, and a 2025-11-25 initialize - * is refused, which a dual-era adapter would route to the stateless server instead. - */ - @Test - void legacyTrafficIsRefused() throws Exception { - HttpResponse get = - http.send( - HttpRequest.newBuilder(URI.create(target)).timeout(TIMEOUT).GET().build(), - HttpResponse.BodyHandlers.ofString()); - assertEquals(405, get.statusCode()); - - String notification = - "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\",\"params\":{\"requestId\":7}}"; - Reply accepted = send(notification, headers("notifications/cancelled", VERSION, null)); - assertEquals(202, accepted.status, accepted.body); - - String initialize = - "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":" - + "{\"protocolVersion\":\"2025-11-25\",\"capabilities\":{}," - + "\"clientInfo\":{\"name\":\"legacy\",\"version\":\"1\"}}}"; - Reply legacy = send(initialize, headers("initialize", null, null)); - // The SDK refuses it as invalid params; Ikanos answers -32022 with the supported version. - assertEquals(400, legacy.status, legacy.body); - assertEquals(ErrorCodes.INVALID_PARAMS, legacy.errorCode()); - } - - /** - * Every 2026-07-28 request carries its client capabilities in _meta. Ikanos' client sends only - * the protocol version, which the SDK refuses: the client must declare its capabilities, even - * empty, to call an SDK-based server. - */ - @Test - void aRequestWithoutClientCapabilitiesIsRefused() throws Exception { - String request = body("tools/list", Map.of(), Map.of(MetaKeys.PROTOCOL_VERSION, VERSION)); - Reply reply = send(request, headers("tools/list", VERSION, null)); - assertEquals(400, reply.status, reply.body); - assertEquals(ErrorCodes.INVALID_PARAMS, reply.errorCode()); - assertTrue( - ((String) reply.error().get("message")).contains(MetaKeys.CLIENT_CAPABILITIES), - reply.body); - } - - private Reply call(String method, Map params, String name) throws Exception { - return send(body(method, params, VERSION), headers(method, VERSION, name)); - } - - /** A request as the 2026-07-28 revision requires: version and client capabilities in _meta. */ - private static String body(String method, Map params, String version) - throws IOException { - return body( - method, - params, - Map.of(MetaKeys.PROTOCOL_VERSION, version, MetaKeys.CLIENT_CAPABILITIES, Map.of())); - } - - /** A request with the given _meta. */ - private static String body(String method, Map params, Map meta) - throws IOException { - Map withMeta = new LinkedHashMap<>(params); - withMeta.put("_meta", meta); - Map request = new LinkedHashMap<>(); - request.put("jsonrpc", "2.0"); - request.put("id", 1); - request.put("method", method); - request.put("params", withMeta); - return MAPPER.writeValueAsString(request); - } - - private static Map headers(String method, String version, String name) { - Map headers = new LinkedHashMap<>(); - headers.put("Accept", "application/json, text/event-stream"); - if (version != null) { - headers.put("MCP-Protocol-Version", version); - } - if (method != null) { - headers.put("Mcp-Method", method); - } - if (name != null) { - headers.put("Mcp-Name", name); - } - return headers; - } - - private Reply send(String body, Map headers) throws Exception { - HttpRequest.Builder request = - HttpRequest.newBuilder(URI.create(target)) - .timeout(TIMEOUT) - .header("Content-Type", "application/json") - .POST(HttpRequest.BodyPublishers.ofString(body)); - headers.forEach(request::header); - HttpResponse response = - http.send(request.build(), HttpResponse.BodyHandlers.ofString()); - return new Reply(response.statusCode(), response.body()); - } - - private static CallToolResult text(String text) { - return CallToolResult.builder().addContent(TextContent.builder(text).build()).build(); - } - - /** Parses the data lines of an SSE body, one JSON-RPC message per event. */ - private static List> events(String body) throws IOException { - List> events = new java.util.ArrayList<>(); - for (String line : body.split("\n")) { - if (line.startsWith("data:")) { - events.add( - MAPPER.readValue( - line.substring(5).trim(), new TypeRef>() {})); - } - } - return events; - } - - private static String firstText(Map result) { - return (String) ((Map) ((List) result.get("content")).get(0)).get("text"); - } - - private record Reply(int status, String body) { - Map json() throws IOException { - return MAPPER.readValue(body, new TypeRef>() {}); - } - - @SuppressWarnings("unchecked") - Map result() throws IOException { - return (Map) json().get("result"); - } - - @SuppressWarnings("unchecked") - Map error() throws IOException { - return (Map) json().get("error"); - } - - int errorCode() throws IOException { - return ((Number) error().get("code")).intValue(); - } - } - - /** - * The test-only Restlet transport: what the SDK's HttpServletMcpTransport does, on the Restlet - * API. POST only; header checks against the body; one JSON reply, an error mapped to its HTTP - * status, or an SSE stream for a streaming reply. - */ - private static final class ModernHost extends Restlet { - private static final Map ERROR_STATUSES = - Map.of( - ErrorCodes.PARSE_ERROR, Status.CLIENT_ERROR_BAD_REQUEST, - ErrorCodes.INVALID_REQUEST, Status.CLIENT_ERROR_BAD_REQUEST, - ErrorCodes.INVALID_PARAMS, Status.CLIENT_ERROR_BAD_REQUEST, - ErrorCodes.HEADER_MISMATCH, Status.CLIENT_ERROR_BAD_REQUEST, - ErrorCodes.MISSING_REQUIRED_CLIENT_CAPABILITY, - Status.CLIENT_ERROR_BAD_REQUEST, - ErrorCodes.UNSUPPORTED_PROTOCOL_VERSION, Status.CLIENT_ERROR_BAD_REQUEST, - ErrorCodes.METHOD_NOT_FOUND, Status.CLIENT_ERROR_NOT_FOUND); - - private final McpServer sdk; - - private ModernHost(McpServer sdk) { - this.sdk = sdk; - } - - @Override - public void handle(Request request, Response response) { - if (!Method.POST.equals(request.getMethod())) { - response.setStatus(Status.CLIENT_ERROR_METHOD_NOT_ALLOWED); - response.getAllowedMethods().add(Method.POST); - return; - } - if (request.getEntity() == null - || !MediaType.APPLICATION_JSON.isCompatible( - request.getEntity().getMediaType())) { - response.setStatus(Status.CLIENT_ERROR_UNSUPPORTED_MEDIA_TYPE); - return; - } - JSONRPCMessage message; - try { - message = JsonRpc.deserializeMessage(MAPPER, request.getEntity().getText()); - } catch (JsonRpc.InvalidMessageException e) { - error(response, e.id(), ErrorCodes.INVALID_REQUEST, "Invalid JSON-RPC message"); - return; - } catch (IOException _) { - error(response, null, ErrorCodes.PARSE_ERROR, "Parse error"); - return; - } - if (message instanceof JSONRPCNotification notification) { - sdk.handleNotification(McpTransportContext.EMPTY, notification).block(TIMEOUT); - response.setStatus(Status.SUCCESS_ACCEPTED); - return; - } - if (!(message instanceof JSONRPCRequest rpc)) { - error(response, null, ErrorCodes.INVALID_REQUEST, "Requests or notifications only"); - return; - } - String mismatch = mismatch(request, rpc); - if (mismatch != null) { - error(response, rpc.id(), ErrorCodes.HEADER_MISMATCH, mismatch); - return; - } - McpTransportResponse reply = - sdk.handleBlocking(McpTransportContext.EMPTY, rpc).block(TIMEOUT); - if (reply instanceof McpTransportResponse.Result result) { - json(response, Status.SUCCESS_OK, result.response()); - } else if (reply instanceof McpTransportResponse.Error failure) { - json( - response, - ERROR_STATUSES.getOrDefault( - failure.response().error().code(), Status.SUCCESS_OK), - failure.response()); - } else if (reply instanceof McpTransportResponse.Streaming streaming) { - response.setEntity( - new EventStreamRepresentation( - sink -> - streaming - .messages() - .doOnNext( - each -> { - try { - sink.send( - ServerSentEvent.builder() - .event("message") - .data( - MAPPER - .writeValueAsString( - each)) - .build()); - } catch (IOException e) { - throw new java.io - .UncheckedIOException(e); - } - }) - .blockLast(TIMEOUT))); - } - } - - /** The servlet transport's header checks, on Restlet headers. */ - private static String mismatch(Request request, JSONRPCRequest rpc) { - String method = header(request, "Mcp-Method"); - if (method == null) { - return "Missing required header: Mcp-Method"; - } - if (!method.equals(rpc.method())) { - return "Mcp-Method header does not match request method"; - } - Map params = rpc.params() instanceof Map map ? map : Map.of(); - if (params.get("_meta") instanceof Map meta - && meta.get(MetaKeys.PROTOCOL_VERSION) instanceof String version - && !version.equals(header(request, "MCP-Protocol-Version"))) { - return "MCP-Protocol-Version header does not match _meta protocol version"; - } - String key = - switch (rpc.method()) { - case "tools/call", "prompts/get" -> "name"; - case "resources/read" -> "uri"; - default -> null; - }; - if (key != null && params.get(key) instanceof String expected) { - String name = header(request, "Mcp-Name"); - String decoded = name == null ? null : McpHeaderDecoder.decodeValue(name); - if (!expected.equals(decoded)) { - return "Mcp-Name header does not match request name/uri"; - } - } - return null; - } - - private static String header(Request request, String name) { - return request.getHeaders().getFirstValue(name, true); - } - - private static void error(Response response, Object id, int code, String message) { - json( - response, - ERROR_STATUSES.getOrDefault(code, Status.SUCCESS_OK), - JSONRPCResponse.error(id, new JSONRPCError(code, message))); - } - - private static void json(Response response, Status status, JSONRPCResponse message) { - try { - response.setStatus(status); - response.setEntity( - new StringRepresentation( - MAPPER.writeValueAsString(message), MediaType.APPLICATION_JSON)); - } catch (IOException _) { - response.setStatus(Status.SERVER_ERROR_INTERNAL); - } - } - } -} From e4b9604666c1daa72fa1a4c39f928935a4fd08ce Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 7 Nov 2026 14:13:18 -0500 Subject: [PATCH 2/4] Address SonarCloud findings on the MCP 2026-07-28 server Split McpServerAdapter.handle's checks into helpers, in the same order; end a stream on the finished flag rather than a queued marker, so no queue offer can be lost; dispatch the message type with a switch; turn the reply record holding an array into a class; move the error reply into the exchange; rename a field named like its class. Co-Authored-By: Claude Opus 5.5 --- .../org/restlet/ext/mcp/McpServerAdapter.java | 116 ++++++++++-------- .../ext/mcp/internal/McpModernExchange.java | 63 ++++++---- .../ext/mcp/internal/McpModernFeatures.java | 8 +- .../ext/mcp/McpModernServerTestCase.java | 2 +- 4 files changed, 112 insertions(+), 77 deletions(-) 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 f0c9434937..3e989917e1 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 @@ -477,59 +477,21 @@ public void handle(Request request, Response response) { response.getAllowedMethods().add(Method.POST); return; } - String origin = request.getHeaders().getFirstValue("Origin", true); - if (request.getHeaders().getValuesArray("Origin", true).length > 1 - || request.getHeaders().getValuesArray(McpProtocol.VERSION_HEADER, true).length - > 1) { - response.setStatus(Status.CLIENT_ERROR_BAD_REQUEST); - return; - } - if (origin != null && !runningOrigins.contains(origin)) { - response.setStatus(Status.CLIENT_ERROR_FORBIDDEN); - return; - } String version = request.getHeaders().getFirstValue(McpProtocol.VERSION_HEADER, true); // A 2026-07-28 request goes to the modern server; it ignores a stale Mcp-Session-Id. boolean modern = McpProtocol.MODERN_VERSION.equals(version); - if (!modern - && (version != null && !McpProtocol.VERSION.equals(version) - || request.getHeaders().getFirstValue("Mcp-Session-Id", true) != null)) { - response.setStatus(Status.CLIENT_ERROR_BAD_REQUEST); - return; + Status rejected = headerRejection(request, version, modern); + if (rejected == null && !acceptsJsonAndEventStream(request)) { + rejected = Status.CLIENT_ERROR_NOT_ACCEPTABLE; } - ClientInfo accepted = request.getClientInfo(); - String accept = request.getHeaders().getValues("Accept"); - if (accept != null) { - accepted = new ClientInfo(); - PreferenceReader.addMediaTypes(accept, accepted); - } - boolean jsonAccepted = - accepted.getAcceptedMediaTypes().stream() - .anyMatch( - p -> - p.getQuality() > 0 - && MediaType.APPLICATION_JSON.isCompatible( - p.getMetadata())); - boolean sseAccepted = - accepted.getAcceptedMediaTypes().stream() - .anyMatch( - p -> - p.getQuality() > 0 - && MediaType.TEXT_EVENT_STREAM.isCompatible( - p.getMetadata())); - if (!jsonAccepted || !sseAccepted) { - response.setStatus(Status.CLIENT_ERROR_NOT_ACCEPTABLE); - return; - } - if (request.getEntity() == null - || request.getEntity().getMediaType() == null - || !MediaType.APPLICATION_JSON.isCompatible(request.getEntity().getMediaType())) { - response.setStatus(Status.CLIENT_ERROR_UNSUPPORTED_MEDIA_TYPE); - return; + if (rejected == null) { + rejected = entityRejection(request); } - if (request.getEntity().getSize() > maxMessageBytes) { - response.setStatus(Status.CLIENT_ERROR_REQUEST_ENTITY_TOO_LARGE); - response.setEntity((org.restlet.representation.Representation) null); + if (rejected != null) { + response.setStatus(rejected); + if (Status.CLIENT_ERROR_REQUEST_ENTITY_TOO_LARGE.equals(rejected)) { + response.setEntity((org.restlet.representation.Representation) null); + } return; } if (modern) { @@ -559,6 +521,64 @@ public void handle(Request request, Response response) { } } + /** + * Checks the headers: one Origin and one version at most, an allowed Origin, and for a + * 2025-11-25 request a supported version and no session. + * + * @return The status refusing the request, or null. + */ + private Status headerRejection(Request request, String version, boolean modern) { + if (request.getHeaders().getValuesArray("Origin", true).length > 1 + || request.getHeaders().getValuesArray(McpProtocol.VERSION_HEADER, true).length + > 1) { + return Status.CLIENT_ERROR_BAD_REQUEST; + } + String origin = request.getHeaders().getFirstValue("Origin", true); + if (origin != null && !runningOrigins.contains(origin)) { + return Status.CLIENT_ERROR_FORBIDDEN; + } + if (!modern + && (version != null && !McpProtocol.VERSION.equals(version) + || request.getHeaders().getFirstValue("Mcp-Session-Id", true) != null)) { + return Status.CLIENT_ERROR_BAD_REQUEST; + } + return null; + } + + /** Whether the request accepts both JSON and an event stream, as MCP requires. */ + private static boolean acceptsJsonAndEventStream(Request request) { + ClientInfo accepted = request.getClientInfo(); + String accept = request.getHeaders().getValues("Accept"); + if (accept != null) { + accepted = new ClientInfo(); + PreferenceReader.addMediaTypes(accept, accepted); + } + return accepts(accepted, MediaType.APPLICATION_JSON) + && accepts(accepted, MediaType.TEXT_EVENT_STREAM); + } + + private static boolean accepts(ClientInfo accepted, MediaType type) { + return accepted.getAcceptedMediaTypes().stream() + .anyMatch(p -> p.getQuality() > 0 && type.isCompatible(p.getMetadata())); + } + + /** + * Checks the request entity: JSON, within the message size. + * + * @return The status refusing the request, or null. + */ + private Status entityRejection(Request request) { + if (request.getEntity() == null + || request.getEntity().getMediaType() == null + || !MediaType.APPLICATION_JSON.isCompatible(request.getEntity().getMediaType())) { + return Status.CLIENT_ERROR_UNSUPPORTED_MEDIA_TYPE; + } + if (request.getEntity().getSize() > maxMessageBytes) { + return Status.CLIENT_ERROR_REQUEST_ENTITY_TOO_LARGE; + } + return null; + } + /** Reads, decodes and dispatches one HTTP message in an admitted worker. */ private Reply exchange( Request request, 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 5cf8a384c9..108c536da6 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,7 +87,7 @@ Mono run( /** Streamed messages a slow client may leave pending before the SDK is held back. */ private static final int QUEUE_CAPACITY = 64; - /** Marks the end of a stream in the queue. */ + /** Returned by the writer's next message once the stream ended; never a real message. */ private static final byte[] END = new byte[0]; private static final Map ERROR_STATUSES = @@ -225,13 +225,15 @@ private void run() throws IOException, InterruptedException { head.complete(error(null, ErrorCodes.PARSE_ERROR, "Parse error")); return; } - if (message instanceof JsonRpc.JSONRPCNotification notification) { - notification(notification); - } else if (message instanceof JsonRpc.JSONRPCRequest rpc) { - call(rpc); - } else { - head.complete( - error(null, ErrorCodes.INVALID_REQUEST, "Requests or notifications only")); + switch (message) { + case JsonRpc.JSONRPCNotification notification -> notification(notification); + case JsonRpc.JSONRPCRequest rpc -> call(rpc); + default -> + head.complete( + error( + null, + ErrorCodes.INVALID_REQUEST, + "Requests or notifications only")); } } @@ -410,7 +412,7 @@ private void fail(Throwable cause) { if (!events.offer(error)) { // A stalled client: what it hasn't read yet gives way to the answer. events.clear(); - events.offer(error); + events.add(error); } } else { head.complete(unavailable()); @@ -433,34 +435,38 @@ private void end() { } } - /** Marks the end of the stream, for the writer to stop once it has written the rest. */ + /** + * Marks the end of the stream, after the last message was queued: the writer stops once it + * has written the rest. + */ private void finish() { finished = true; - events.offer(END); } /** - * Takes the next streamed message, or null at the end of the stream: after its end marker, - * or once the worker finished and nothing is left, should the marker not fit the queue. + * Takes the next streamed message, or {@link #END} once the worker finished and nothing is + * left. */ private byte[] next() throws InterruptedException { for (; ; ) { + // Read before polling, so that a message queued just before the end is not lost. + boolean ended = finished; byte[] event = events.poll(100, TimeUnit.MILLISECONDS); - if (event == END || (event == null && finished && events.isEmpty())) { - return null; - } if (event != null) { return event; } + if (ended) { + return END; + } } } - } - /** A JSON-RPC error reply, built as a map since the id may be null. */ - private Reply error(Object id, int code, String message) { - return json( - ERROR_STATUSES.getOrDefault(code, Status.SUCCESS_OK), - McpMessages.error(id, code, message)); + /** A JSON-RPC error reply, built as a map since the id may be null. */ + private Reply error(Object id, int code, String message) { + return json( + ERROR_STATUSES.getOrDefault(code, Status.SUCCESS_OK), + McpMessages.error(id, code, message)); + } } private Reply json(Status status, Object message) { @@ -482,7 +488,16 @@ private static Status status(McpSchema.JSONRPCResponse response) { } /** The reply to write: a status and an optional JSON body, or an SSE stream. */ - private record Reply(Status status, byte[] body, boolean streaming) { + private static final class Reply { + private final Status status; + private final byte[] body; + private final boolean streaming; + + private Reply(Status status, byte[] body, boolean streaming) { + this.status = status; + this.body = body; + this.streaming = streaming; + } private void apply( Response response, Exchange exchange, Disposable running, Duration heartbeat) { @@ -502,7 +517,7 @@ private static Representation stream( sink -> { try { for (byte[] event = exchange.next(); - event != null; + event != END; event = exchange.next()) { sink.send( ServerSentEvent.builder() 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 0bde065ca1..3df8f33269 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 @@ -157,12 +157,12 @@ public Mono find(McpRequestContext ctx, String name) { /** The bridged resources: exact URIs first, then templates, in registration order. */ private static final class Resources implements McpAsyncResourceRepository { private final McpJsonMapper mapper; - private final List resources; + private final List exact; private final List templates; private Resources(McpJsonMapper mapper, Registrations registrations) { this.mapper = mapper; - this.resources = List.copyOf(registrations.resources()); + this.exact = List.copyOf(registrations.resources()); this.templates = List.copyOf(registrations.resourceTemplates()); } @@ -171,7 +171,7 @@ public Mono list(McpRequestContext ctx, String cursor) { return Mono.fromSupplier( () -> ResourcesPage.of( - resources.stream() + exact.stream() .map( spec -> mapper.convertValue( @@ -197,7 +197,7 @@ public Mono listTemplates(McpRequestContext ctx, String c @Override public Mono find(McpRequestContext ctx, String uri) { - for (McpStatelessServerFeatures.AsyncResourceSpecification spec : resources) { + for (McpStatelessServerFeatures.AsyncResourceSpecification spec : exact) { if (spec.resource().uri().equals(uri)) { return Mono.just(read(spec.readHandler())); } diff --git a/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java index 3ed97c3f52..ce20c1ee8f 100644 --- a/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java +++ b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java @@ -433,7 +433,7 @@ public void write(byte[] b, int off, int len) throws java.io.IOException { }; try { response.getEntity().write(gone); - } catch (java.io.IOException expected) { + } catch (java.io.IOException _) { // The writer reports the broken client. } From fb39da1df9fe791d762e71e6201f49990527c8d9 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 7 Nov 2026 14:38:39 -0500 Subject: [PATCH 3/4] Answer every MCP 2026-07-28 call, forward the interceptors' context, release on a failed flush An exchange cut short before its answer names its request in the unavailable error; one losing the race with a starting stream, or whose streamed result an interceptor suppresses, ends the stream with an error answering the request. The context an interceptor forwards is the one the SDK and the handlers get, for requests and notifications. The event stream releases its exchange however its write ends, including when the flush of the response head fails before any event. Co-Authored-By: Claude Opus 5.5 --- .../ext/mcp/internal/McpModernExchange.java | 122 ++++++++----- .../ext/mcp/McpModernServerTestCase.java | 167 ++++++++++++++++++ 2 files changed, 249 insertions(+), 40 deletions(-) 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 108c536da6..2251b9a33d 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 @@ -21,6 +21,7 @@ import io.modelcontextprotocol.spec.McpSchema; import java.io.IOException; import java.io.InputStream; +import java.io.OutputStream; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; @@ -39,6 +40,7 @@ import java.util.stream.Stream; import org.restlet.Request; import org.restlet.Response; +import org.restlet.data.CharacterSet; import org.restlet.data.MediaType; import org.restlet.data.ServerSentEvent; import org.restlet.data.Status; @@ -47,6 +49,7 @@ import org.restlet.ext.mcp.McpRequestContext; import org.restlet.representation.ByteArrayRepresentation; import org.restlet.representation.EventStreamRepresentation; +import org.restlet.representation.OutputRepresentation; import org.restlet.representation.Representation; import reactor.core.Disposable; import reactor.core.publisher.Mono; @@ -169,10 +172,10 @@ public void handle(Request request, Response response) { } catch (InterruptedException _) { Thread.currentThread().interrupt(); running.dispose(); - reply = unavailable(); + reply = unavailable(exchange.requestId); } catch (ExecutionException | TimeoutException _) { running.dispose(); - reply = unavailable(); + reply = unavailable(exchange.requestId); } reply.apply(response, exchange, running, heartbeat()); } @@ -184,17 +187,21 @@ private Duration heartbeat() { return half.compareTo(max) < 0 ? half : max; } - private Reply unavailable() { + /** + * The reply to an exchange cut short before it answered. + * + * @param id The id of the request, or null when it wasn't read yet. + */ + private Reply unavailable(Object id) { return json( Status.SERVER_ERROR_SERVICE_UNAVAILABLE, - McpMessages.error(null, ErrorCodes.INTERNAL_ERROR, "MCP exchange unavailable")); + McpMessages.error(id, ErrorCodes.INTERNAL_ERROR, "MCP exchange unavailable")); } /** The state of one exchange, shared by its worker and the thread writing the response. */ private final class Exchange { private final Request request; private final McpRequestContext context; - private final McpTransportContext transport; private final CompletableFuture head = new CompletableFuture<>(); private final BlockingQueue events = new LinkedBlockingQueue<>(QUEUE_CAPACITY); private volatile boolean finished; @@ -205,7 +212,11 @@ private final class Exchange { private Exchange(Request request, McpRequestContext context) { this.request = request; this.context = context; - this.transport = McpTransportContext.create(Map.of(McpProtocol.CONTEXT_KEY, context)); + } + + /** The SDK transport context for the request context an interceptor forwarded. */ + private static McpTransportContext transport(McpRequestContext forwarded) { + return McpTransportContext.create(Map.of(McpProtocol.CONTEXT_KEY, forwarded)); } /** Runs in the admitted worker: reads, checks and dispatches the message. */ @@ -244,7 +255,7 @@ private void notification(JsonRpc.JSONRPCNotification notification) { context, (next, nextContext) -> sdk.handleNotification( - transport, + transport(nextContext), mapper.convertValue( next, JsonRpc.JSONRPCNotification.class)) @@ -265,11 +276,15 @@ private void call(JsonRpc.JSONRPCRequest rpc) throws InterruptedException { .run( mapper.convertValue(rpc, McpSchema.JSONRPCRequest.class), context, - (next, nextContext) -> Mono.fromCallable(() -> terminal(next))) + (next, nextContext) -> + Mono.fromCallable(() -> terminal(next, nextContext))) .blockOptional(timeout); if (last.isEmpty()) { - // An interceptor answered nothing: accepted, as on the 2025-11-25 path. - head.complete(new Reply(Status.SUCCESS_ACCEPTED, null, false)); + // An interceptor answered nothing: accepted, as on the 2025-11-25 path, unless the + // stream already started, which must still answer the request. + if (!head.complete(new Reply(Status.SUCCESS_ACCEPTED, null, false))) { + events.put(failure("MCP result suppressed")); + } return; } byte[] encoded; @@ -288,14 +303,15 @@ private void call(JsonRpc.JSONRPCRequest rpc) throws InterruptedException { } /** Dispatches to the SDK after the interceptors, streaming what comes before the result. */ - private McpSchema.JSONRPCResponse terminal(McpSchema.JSONRPCMessage next) + private McpSchema.JSONRPCResponse terminal( + McpSchema.JSONRPCMessage next, McpRequestContext forwarded) throws IOException, InterruptedException { JsonRpc.JSONRPCRequest rpc = mapper.convertValue(next, JsonRpc.JSONRPCRequest.class); - McpTransportContext sdkContext = transport; + McpTransportContext sdkContext = transport(forwarded); if (McpSchema.METHOD_TOOLS_CALL.equals(rpc.method())) { // The tools feature validates x-mcp-header arguments against these. sdkContext = - transport.putAll( + sdkContext.putAll( Map.of( McpParamHeaders.TRANSPORT_CONTEXT_KEY, McpHeaderDecoder.paramHeaders(headers()))); @@ -407,15 +423,15 @@ private byte[] failure(String message) { */ private void fail(Throwable cause) { try { - if (head.isDone()) { + // Completing the head first, atomically: losing means the worker started a + // stream, which must then end with an error answering the request. + if (!head.complete(unavailable(requestId))) { byte[] error = failure("MCP exchange failed"); if (!events.offer(error)) { // A stalled client: what it hasn't read yet gives way to the answer. events.clear(); events.add(error); } - } else { - head.complete(unavailable()); } } catch (RuntimeException _) { // The stream ends below anyway. @@ -427,9 +443,7 @@ private void fail(Throwable cause) { /** Ends the exchange once the worker is done; it answered, unless it was cut short. */ private void end() { try { - if (!head.isDone()) { - head.complete(unavailable()); - } + head.complete(unavailable(requestId)); } finally { finish(); } @@ -513,27 +527,55 @@ private void apply( private static Representation stream( Exchange exchange, Disposable running, Duration heartbeat) { - return new EventStreamRepresentation( - sink -> { - try { - for (byte[] event = exchange.next(); - event != END; - event = exchange.next()) { - sink.send( - ServerSentEvent.builder() - .event("message") - .data(new String(event, StandardCharsets.UTF_8)) - .build()); - } - } catch (InterruptedException _) { - Thread.currentThread().interrupt(); - running.dispose(); - } catch (IOException e) { - running.dispose(); - throw e; - } - }, - heartbeat); + return new Releasing( + new EventStreamRepresentation( + sink -> { + try { + for (byte[] event = exchange.next(); + event != END; + event = exchange.next()) { + sink.send( + ServerSentEvent.builder() + .event("message") + .data( + new String( + event, + StandardCharsets.UTF_8)) + .build()); + } + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + } + }, + heartbeat), + running); + } + } + + /** + * An event stream that releases its exchange however its write ends, including when the flush + * of the response head fails, before any event: the SDK's stream is cancelled and the call slot + * freed. Releasing a finished exchange does nothing. + */ + private static final class Releasing extends OutputRepresentation { + private final Representation stream; + private final Disposable running; + + private Releasing(Representation stream, Disposable running) { + super(MediaType.TEXT_EVENT_STREAM); + setCharacterSet(CharacterSet.UTF_8); + setTransient(true); + this.stream = stream; + this.running = running; + } + + @Override + public void write(OutputStream output) throws IOException { + try { + stream.write(output); + } finally { + running.dispose(); + } } } } diff --git a/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java index ce20c1ee8f..f4eaffc3a0 100644 --- a/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java +++ b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java @@ -609,6 +609,173 @@ void anInterceptorChangesTheResultOfAStreamedCall() throws Exception { assertEquals("intercepted", firstText(result)); } + /** A call timing out before its answer is unavailable, with an error naming the request. */ + @Test + void aTimedOutCallNamesTheRequest() throws Exception { + McpServerAdapter adapter = adapter(); + adapter.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder("silent", Map.of("type", "object")).build(), + (ctx, request) -> Mono.never())); + adapter.setRequestTimeout(java.time.Duration.ofMillis(500)); + adapter.start(); + + Response response = + adapter.handle( + modern( + "tools/call", + Map.of("name", "silent", "arguments", Map.of()), + "silent", + meta())); + String body = body(response); + + assertEquals(503, response.getStatus().getCode(), body); + Map envelope = + MAPPER.readValue(body, new TypeRef>() {}); + assertEquals(1, ((Number) envelope.get("id")).intValue(), body); + } + + /** The context an interceptor forwards is the one the handlers get. */ + @Test + void handlersGetTheContextAnInterceptorForwards() throws Exception { + McpServerAdapter adapter = new McpServerAdapter(); + adapters.add(adapter); + adapter.getTools() + .add( + new McpStatelessServerFeatures.AsyncToolSpecification( + McpServerAdapterTestCase.echoTool(), + (transport, call) -> + Mono.just( + McpSchema.CallToolResult.builder() + .addTextContent( + String.valueOf( + ((McpRequestContext) + transport + .get( + McpProtocol + .CONTEXT_KEY)) + .attributes() + .get("tenant"))) + .build()))); + adapter.getInterceptors() + .add( + (message, context, chain) -> { + McpRequestContext forwarded = + new McpRequestContext( + context.transportRequest().orElse(null), + context.protocolVersion()); + forwarded.attributes().put("tenant", "acme"); + return chain.proceed(message, forwarded); + }); + adapter.start(); + + assertEquals( + "acme", + firstText( + result( + call( + adapter, + "tools/call", + Map.of("name", "echo", "arguments", Map.of("text", "x")), + "echo")))); + } + + /** A streamed result an interceptor suppresses still ends with an error answering the call. */ + @Test + void aSuppressedStreamedResultStillAnswersTheCall() throws Exception { + McpServerAdapter adapter = adapter(); + adapter.getInterceptors() + .add( + (message, context, chain) -> + chain.proceed(message, context).then(Mono.empty())); + adapter.start(); + Map meta = meta(); + meta.put(MetaKeys.PROGRESS_TOKEN, "p1"); + + List> events = + events( + body( + adapter.handle( + modern( + "tools/call", + Map.of("name", "progress", "arguments", Map.of()), + "progress", + meta)))); + + assertEquals(2, events.size(), events.toString()); + assertEquals(1, ((Number) events.get(1).get("id")).intValue(), events.toString()); + assertNotNull(events.get(1).get("error"), events.toString()); + } + + /** A client gone before the first event, at the head's flush, also frees the call slot. */ + @Test + void aClientGoneAtTheFirstFlushCancelsTheTool() throws Exception { + McpServerAdapter adapter = adapter(); + CountDownLatch cancelled = new CountDownLatch(1); + adapter.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder("endless", Map.of("type", "object")).build(), + (ctx, request) -> + Mono.just( + McpAsyncResponse.streaming( + notifier -> + notifier.progress( + 1, null, "started") + .then( + Mono + . + never() + .doOnCancel( + cancelled + ::countDown)))))); + adapter.setMaxConcurrentCalls(1); + adapter.start(); + Map meta = meta(); + meta.put(MetaKeys.PROGRESS_TOKEN, "p1"); + + Response response = + adapter.handle( + modern( + "tools/call", + Map.of("name", "endless", "arguments", Map.of()), + "endless", + meta)); + java.io.OutputStream unflushable = + new java.io.OutputStream() { + @Override + public void write(int b) { + // Never reached: the flush fails first. + } + + @Override + public void flush() throws java.io.IOException { + throw new java.io.IOException("client gone"); + } + }; + try { + response.getEntity().write(unflushable); + } catch (java.io.IOException _) { + // The write reports the broken client. + } + + assertTrue(cancelled.await(10, TimeUnit.SECONDS), "the tool's stream was not cancelled"); + assertEquals( + "echo:freed", + firstText( + result( + call( + adapter, + "tools/call", + Map.of( + "name", + "echo", + "arguments", + Map.of("text", "freed")), + "echo")))); + } + private McpServerAdapter adapter() { McpServerAdapter adapter = new McpServerAdapter(); McpServerAdapterTestCase.addEcho(adapter); From e9826515639067535f63dd57c72b22aa690d761f Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sat, 7 Nov 2026 14:52:35 -0500 Subject: [PATCH 4/4] Release an unwritten MCP 2026-07-28 stream A streamed response released without being written now cancels its exchange, freeing the call slot and the SDK's stream, and marks the stream finished for any reader; the wrapper compares by identity. Co-Authored-By: Claude Opus 5.5 --- .../ext/mcp/internal/McpModernExchange.java | 39 +++++++++++--- .../ext/mcp/McpModernServerTestCase.java | 52 +++++++++++++++++++ 2 files changed, 84 insertions(+), 7 deletions(-) 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 2251b9a33d..3bb83ae1e6 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 @@ -548,25 +548,29 @@ private static Representation stream( } }, heartbeat), - running); + () -> { + running.dispose(); + exchange.finish(); + }); } } /** * An event stream that releases its exchange however its write ends, including when the flush - * of the response head fails, before any event: the SDK's stream is cancelled and the call slot - * freed. Releasing a finished exchange does nothing. + * of the response head fails, before any event, and when the response is released without being + * written: the SDK's stream is cancelled, the call slot freed, and the stream marked finished + * for any reader. Releasing a finished exchange does nothing. */ private static final class Releasing extends OutputRepresentation { private final Representation stream; - private final Disposable running; + private final Runnable close; - private Releasing(Representation stream, Disposable running) { + private Releasing(Representation stream, Runnable close) { super(MediaType.TEXT_EVENT_STREAM); setCharacterSet(CharacterSet.UTF_8); setTransient(true); this.stream = stream; - this.running = running; + this.close = close; } @Override @@ -574,8 +578,29 @@ public void write(OutputStream output) throws IOException { try { stream.write(output); } finally { - running.dispose(); + close.run(); } } + + @Override + public void release() { + try { + close.run(); + stream.release(); + } finally { + super.release(); + } + } + + /** Identity: two streams are never the same response. */ + @Override + public boolean equals(Object other) { + return this == other; + } + + @Override + public int hashCode() { + return System.identityHashCode(this); + } } } diff --git a/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java index f4eaffc3a0..673241154a 100644 --- a/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java +++ b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java @@ -776,6 +776,58 @@ public void flush() throws java.io.IOException { "echo")))); } + /** A streamed response released without being written cancels the tool and frees its slot. */ + @Test + void aReleasedUnwrittenStreamCancelsTheTool() throws Exception { + McpServerAdapter adapter = adapter(); + CountDownLatch cancelled = new CountDownLatch(1); + adapter.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder("endless", Map.of("type", "object")).build(), + (ctx, request) -> + Mono.just( + McpAsyncResponse.streaming( + notifier -> + notifier.progress( + 1, null, "started") + .then( + Mono + . + never() + .doOnCancel( + cancelled + ::countDown)))))); + adapter.setMaxConcurrentCalls(1); + adapter.start(); + Map meta = meta(); + meta.put(MetaKeys.PROGRESS_TOKEN, "p1"); + + Response response = + adapter.handle( + modern( + "tools/call", + Map.of("name", "endless", "arguments", Map.of()), + "endless", + meta)); + response.getEntity().release(); + + assertTrue(cancelled.await(10, TimeUnit.SECONDS), "the tool's stream was not cancelled"); + assertEquals( + "echo:freed", + firstText( + result( + call( + adapter, + "tools/call", + Map.of( + "name", + "echo", + "arguments", + Map.of("text", "freed")), + "echo")))); + } + private McpServerAdapter adapter() { McpServerAdapter adapter = new McpServerAdapter(); McpServerAdapterTestCase.addEcho(adapter);