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..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 @@ -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) { @@ -355,56 +477,25 @@ 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); - if (version != null && !McpProtocol.VERSION.equals(version) - || request.getHeaders().getFirstValue("Mcp-Session-Id", true) != null) { - response.setStatus(Status.CLIENT_ERROR_BAD_REQUEST); - return; - } - 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); + // A 2026-07-28 request goes to the modern server; it ignores a stale Mcp-Session-Id. + boolean modern = McpProtocol.MODERN_VERSION.equals(version); + Status rejected = headerRejection(request, version, modern); + if (rejected == null && !acceptsJsonAndEventStream(request)) { + rejected = Status.CLIENT_ERROR_NOT_ACCEPTABLE; + } + if (rejected == null) { + rejected = entityRejection(request); + } + 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 (request.getEntity().getSize() > maxMessageBytes) { - response.setStatus(Status.CLIENT_ERROR_REQUEST_ENTITY_TOO_LARGE); - response.setEntity((org.restlet.representation.Representation) null); + if (modern) { + modernExchange.handle(request, response); return; } var context = new McpRequestContext(request); @@ -430,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 new file mode 100644 index 0000000000..3bb83ae1e6 --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernExchange.java @@ -0,0 +1,606 @@ +/** + * 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.io.OutputStream; +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.CharacterSet; +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.OutputRepresentation; +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; + + /** 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 = + 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(exchange.requestId); + } catch (ExecutionException | TimeoutException _) { + running.dispose(); + reply = unavailable(exchange.requestId); + } + 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; + } + + /** + * 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(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 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; + } + + /** 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. */ + 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; + } + 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")); + } + } + + private void notification(JsonRpc.JSONRPCNotification notification) { + interceptors + .run( + mapper.convertValue(notification, McpSchema.JSONRPCNotification.class), + context, + (next, nextContext) -> + sdk.handleNotification( + transport(nextContext), + 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, nextContext))) + .blockOptional(timeout); + if (last.isEmpty()) { + // An interceptor answered nothing: accepted, as on the 2025-11-25 path, unless the + // stream already started, which must still answer the request. + if (!head.complete(new Reply(Status.SUCCESS_ACCEPTED, null, false))) { + events.put(failure("MCP result suppressed")); + } + 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, McpRequestContext forwarded) + throws IOException, InterruptedException { + JsonRpc.JSONRPCRequest rpc = mapper.convertValue(next, JsonRpc.JSONRPCRequest.class); + McpTransportContext sdkContext = transport(forwarded); + if (McpSchema.METHOD_TOOLS_CALL.equals(rpc.method())) { + // The tools feature validates x-mcp-header arguments against these. + sdkContext = + sdkContext.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 { + // 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); + } + } + } 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 { + head.complete(unavailable(requestId)); + } finally { + finish(); + } + } + + /** + * 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; + } + + /** + * 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 != 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)); + } + } + + 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 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) { + 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 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.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, 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 Runnable close; + + private Releasing(Representation stream, Runnable close) { + super(MediaType.TEXT_EVENT_STREAM); + setCharacterSet(CharacterSet.UTF_8); + setTransient(true); + this.stream = stream; + this.close = close; + } + + @Override + public void write(OutputStream output) throws IOException { + try { + stream.write(output); + } finally { + 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/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..3df8f33269 --- /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 exact; + private final List templates; + + private Resources(McpJsonMapper mapper, Registrations registrations) { + this.mapper = mapper; + this.exact = List.copyOf(registrations.resources()); + this.templates = List.copyOf(registrations.resourceTemplates()); + } + + @Override + public Mono list(McpRequestContext ctx, String cursor) { + return Mono.fromSupplier( + () -> + ResourcesPage.of( + exact.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 : exact) { + 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..673241154a --- /dev/null +++ b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernServerTestCase.java @@ -0,0 +1,1001 @@ +/** + * 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 _) { + // 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)); + } + + /** 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")))); + } + + /** 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); + 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); - } - } - } -}