diff --git a/changes.md b/changes.md index bc90ce64b1..8983de4443 100644 --- a/changes.md +++ b/changes.md @@ -111,8 +111,16 @@ Changes log `setMaxListenStreams()`, no time limit, and end when the adapter stops or the client goes away. Over stdio, a message whose `_meta` names 2026-07-28 goes to the modern server: streamed messages are written as separate lines as they come, and - `notifications/cancelled` ends the request it names without a response. Not yet - supported for 2026-07-28: the MCP client. The SDK's modern API is a preview. + `notifications/cancelled` ends the request it names without a response. + `McpClientAdapter.setProtocolVersion(McpProtocol.MODERN_VERSION)` makes the client + speak 2026-07-28 over HTTP: `start()` runs `server/discover`, and + `getModernClient()` returns an `McpModernClient`, on the SDK's modern types, for + listings, tool calls, resource reads and prompts, with streamed progress, + `subscriptions/listen`, and cancellation by disposing the request. An + `input_required` result is answered by the `McpInputHandler` set with + `setInputHandler()` and the request retried, within `setMaxInputRounds()`; + without a handler, it is returned. Not yet supported for 2026-07-28: the client over + stdio. 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/McpClientAdapter.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpClientAdapter.java index edde0d6ad9..56dc2f248b 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpClientAdapter.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpClientAdapter.java @@ -10,9 +10,13 @@ import io.modelcontextprotocol.client.McpAsyncClient; import io.modelcontextprotocol.client.McpClient; +import io.modelcontextprotocol.json.McpJsonDefaults; +import io.modelcontextprotocol.json.McpJsonMapper; import io.modelcontextprotocol.json.TypeRef; import io.modelcontextprotocol.spec.McpClientTransport; import io.modelcontextprotocol.spec.McpSchema; +import java.io.IOException; +import java.io.UncheckedIOException; import java.time.Duration; import java.util.List; import java.util.Map; @@ -21,11 +25,22 @@ import java.util.function.Function; import org.restlet.Client; import org.restlet.Request; +import org.restlet.ext.mcp.internal.McpModernHttpTransport; import org.restlet.ext.mcp.internal.StdioMcpClientTransport; import reactor.core.publisher.Mono; -/** Lifecycle facade over the native SDK. Borrowed Restlet connectors are never stopped. */ +/** + * Lifecycle facade over the native SDK. Borrowed Restlet connectors are never stopped. + * + *

By default it speaks MCP {@link McpProtocol#VERSION} through the SDK's client, {@link + * #getSdkClient()}. With {@link #setProtocolVersion(String)} set to {@link + * McpProtocol#MODERN_VERSION}, it speaks {@code 2026-07-28} over HTTP instead, experimentally: + * {@link #start()} discovers the server, and {@link #getModernClient()} makes the requests. + */ public final class McpClientAdapter implements AutoCloseable { + /** The default number of {@code input_required} rounds answered for one request. */ + public static final int DEFAULT_MAX_INPUT_ROUNDS = 8; + private final Client client; private final String target; private final org.restlet.util.StreamTransport stream; @@ -33,6 +48,12 @@ public final class McpClientAdapter implements AutoCloseable { private int maxMessageBytes = McpProtocol.DEFAULT_MAX_MESSAGE_BYTES; private int maxConcurrentCalls = McpProtocol.DEFAULT_MAX_CONCURRENT_CALLS; private Consumer requestCustomizer = request -> {}; + private String protocolVersion = McpProtocol.VERSION; + private io.modelcontextprotocol.modern.McpSchema.ClientCapabilities modernCapabilities = + io.modelcontextprotocol.modern.McpSchema.ClientCapabilities.NONE; + private McpInputHandler inputHandler; + private int maxInputRounds = DEFAULT_MAX_INPUT_ROUNDS; + private McpModernClient modern; private McpAsyncClient sdk; private McpClientTransport transport; private boolean started; @@ -102,12 +123,85 @@ public synchronized Consumer getRequestCustomizer() { return requestCustomizer; } + /** + * Sets the MCP revision spoken: {@link McpProtocol#VERSION}, the default, through the SDK's + * client, or {@link McpProtocol#MODERN_VERSION} through {@link #getModernClient()}, over HTTP + * only for now. Experimental for {@code 2026-07-28}. + * + * @param value The revision. + */ + public synchronized void setProtocolVersion(String value) { + requireConfigurable(); + if (!McpProtocol.VERSION.equals(value) && !McpProtocol.MODERN_VERSION.equals(value)) { + throw new IllegalArgumentException("Unsupported MCP revision: " + value); + } + protocolVersion = value; + } + + public synchronized String getProtocolVersion() { + return protocolVersion; + } + + /** + * Sets the client capabilities sent with each {@link McpProtocol#MODERN_VERSION} request, none + * by default. A server asks only for the input they declare, such as elicitation. + * + * @param value The capabilities. + */ + public synchronized void setModernCapabilities( + io.modelcontextprotocol.modern.McpSchema.ClientCapabilities value) { + requireConfigurable(); + modernCapabilities = Objects.requireNonNull(value, "value"); + } + + public synchronized io.modelcontextprotocol.modern.McpSchema.ClientCapabilities + getModernCapabilities() { + return modernCapabilities; + } + + /** + * Sets what answers the {@code input_required} results of {@link McpProtocol#MODERN_VERSION} + * requests, after which the request is retried; without one, the result is returned. + * + * @param value The handler, or null for none. + */ + public synchronized void setInputHandler(McpInputHandler value) { + requireConfigurable(); + inputHandler = value; + } + + public synchronized McpInputHandler getInputHandler() { + return inputHandler; + } + + /** + * Sets how many {@code input_required} results the input handler answers for one request, + * {@value #DEFAULT_MAX_INPUT_ROUNDS} by default; the next one is returned. + * + * @param value The number of rounds, zero or more. + */ + public synchronized void setMaxInputRounds(int value) { + requireConfigurable(); + if (value < 0) { + throw new IllegalArgumentException("Input rounds must not be negative"); + } + maxInputRounds = value; + } + + public synchronized int getMaxInputRounds() { + return maxInputRounds; + } + public synchronized void start() { if (started) { return; } requireConfigurable(); configured = true; + if (McpProtocol.MODERN_VERSION.equals(protocolVersion)) { + startModern(); + return; + } transport = createTransport(); try { sdk = @@ -134,6 +228,69 @@ public synchronized void start() { } } + /** Starts a {@code 2026-07-28} client, and discovers the server's support of it. */ + private void startModern() { + if (stream != null || target.startsWith("stdio:")) { + throw new UnsupportedOperationException( + "The MCP 2026-07-28 client supports HTTP only for now"); + } + if (!client.isStarted() + || client.getContext() == null + || !"false" + .equalsIgnoreCase( + client.getContext() + .getParameters() + .getFirstValue("followRedirects", "true"))) { + throw new IllegalStateException( + "Borrowed MCP client must be started with followRedirects=false"); + } + McpJsonMapper mapper = McpJsonDefaults.getMapper(); + McpModernClient candidate = + new McpModernClient( + new McpModernHttpTransport( + client, + target, + mapper, + maxMessageBytes, + maxConcurrentCalls, + requestTimeout, + requestCustomizer), + mapper, + modernCapabilities, + inputHandler, + maxInputRounds); + try { + io.modelcontextprotocol.modern.McpSchema.DiscoverResult discovered = + candidate.discover().block(requestTimeout); + if (discovered == null + || discovered.supportedVersions() == null + || !discovered.supportedVersions().contains(McpProtocol.MODERN_VERSION)) { + throw new IllegalStateException("Peer does not support MCP 2026-07-28"); + } + } catch (RuntimeException e) { + try { + candidate.close(); + } catch (IOException cleanup) { + e.addSuppressed(cleanup); + } + throw e; + } + modern = candidate; + started = true; + } + + /** + * The {@link McpProtocol#MODERN_VERSION} client, when that is the revision set. + * + * @return The client. + */ + public synchronized McpModernClient getModernClient() { + if (!started || modern == null) { + throw new IllegalStateException("MCP 2026-07-28 client is not started"); + } + return modern; + } + private McpClientTransport createTransport() { if (stream != null) { return new StdioMcpClientTransport( @@ -163,6 +320,10 @@ private McpClientTransport createTransport() { /** Exposes native MCP operations, errors and call options without a parallel invocation API. */ public synchronized McpAsyncClient getSdkClient() { + if (modern != null) { + throw new IllegalStateException( + "This MCP client speaks 2026-07-28: use getModernClient()"); + } if (!started) { throw new IllegalStateException("MCP client is not started"); } @@ -184,6 +345,16 @@ public McpSchema.CallToolResult callTool(String name, Map argume } public synchronized void stop() { + if (modern != null) { + started = false; + try { + modern.close(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } finally { + modern = null; + } + } if (sdk != null) { started = false; sdk.closeGracefully().block(requestTimeout); diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpInputHandler.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpInputHandler.java new file mode 100644 index 0000000000..09d36d067f --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpInputHandler.java @@ -0,0 +1,33 @@ +/** + * Copyright 2005-2026 Qlik + *

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

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.mcp; + +import io.modelcontextprotocol.modern.McpSchema; +import java.util.Map; +import reactor.core.publisher.Mono; + +/** + * Answers the input an MCP {@link McpProtocol#MODERN_VERSION} server asks for, through an {@code + * input_required} result, such as an elicitation: {@link McpModernClient} then retries the request + * with the responses and the result's request state. Experimental: the SDK's modern API is a + * preview. + */ +@FunctionalInterface +public interface McpInputHandler { + + /** + * Answers the input requests of a result. + * + * @param required The result, with its input requests by key. + * @return The responses, by the keys of the input requests, such as {@code {"action": "accept", + * "content": {...}}} for an elicitation; or an empty {@code Mono} to return the result to + * the caller instead of retrying. + */ + Mono> respond(McpSchema.InputRequiredResult required); +} diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernClient.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernClient.java new file mode 100644 index 0000000000..9901d36b72 --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernClient.java @@ -0,0 +1,488 @@ +/** + * Copyright 2005-2026 Qlik + *

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

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.mcp; + +import io.modelcontextprotocol.json.McpJsonMapper; +import io.modelcontextprotocol.json.TypeRef; +import io.modelcontextprotocol.modern.JsonRpc; +import io.modelcontextprotocol.modern.McpException; +import io.modelcontextprotocol.modern.McpSchema; +import io.modelcontextprotocol.modern.ToolParamHeaderAnnotations; +import java.io.IOException; +import java.math.BigDecimal; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.BiFunction; +import java.util.function.Consumer; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.restlet.ext.mcp.internal.McpModernHttpTransport; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * A client of MCP revision {@link McpProtocol#MODERN_VERSION}, on the SDK's modern types, over + * HTTP: discovery, listings, tool calls, resource reads and prompts, with streamed progress, {@code + * input_required} retries and {@code subscriptions/listen}. Each request carries the protocol + * version and the client capabilities in its {@code _meta}, as the revision requires. Cancelling a + * returned {@code Mono} or {@code Flux} aborts its HTTP exchange, which the server sees as a + * cancellation. + * + *

Obtained from {@link McpClientAdapter#getModernClient()}, which owns its lifecycle. + * Experimental: the SDK's modern API is a preview. + */ +public final class McpModernClient { + private static final Logger LOGGER = Logger.getLogger(McpModernClient.class.getName()); + private static final TypeRef> MAP = new TypeRef<>() {}; + private static final String META = "_meta"; + private static final String REQUEST = "request"; + + private final McpModernHttpTransport transport; + private final McpJsonMapper mapper; + private final Map capabilities; + private final McpInputHandler inputHandler; + private final int maxInputRounds; + private final AtomicLong ids = new AtomicLong(); + + /** The tools last listed, by name, whose {@code x-mcp-header} arguments calls mirror. */ + private final Map tools = new ConcurrentHashMap<>(); + + McpModernClient( + McpModernHttpTransport transport, + McpJsonMapper mapper, + McpSchema.ClientCapabilities capabilities, + McpInputHandler inputHandler, + int maxInputRounds) { + this.transport = transport; + this.mapper = mapper; + this.capabilities = mapper.convertValue(capabilities, MAP); + this.inputHandler = inputHandler; + this.maxInputRounds = maxInputRounds; + } + + /** + * Discovers the server: its supported versions, capabilities and instructions. + * + * @return The discovery result. + */ + public Mono discover() { + return result(McpSchema.METHOD_SERVER_DISCOVER, new LinkedHashMap<>(), null, Map.of(), null) + .map(result -> mapper.convertValue(result, McpSchema.DiscoverResult.class)); + } + + /** + * Lists tools, and remembers their definitions: a later call mirrors the arguments their {@code + * x-mcp-header} annotations name into {@code Mcp-Param-*} headers, as the server requires. A + * call to an annotated tool this client has not listed is sent without them, and the server + * refuses it with a header mismatch. Listing the first page forgets the tools listed before. + * + * @param cursor The cursor of the page, or null for the first. + * @return The page of tools. + */ + public Mono listTools(String cursor) { + return page(McpSchema.METHOD_TOOLS_LIST, cursor, McpSchema.ListToolsResult.class) + .doOnNext( + page -> { + if (cursor == null) { + tools.clear(); + } + page.tools().forEach(tool -> tools.put(tool.name(), tool)); + }); + } + + /** + * Calls a tool; see {@link #callTool(McpSchema.CallToolRequest, Consumer)}. + * + * @param request The call. + * @return The tool's result, or an {@code input_required} result. + */ + public Mono callTool(McpSchema.CallToolRequest request) { + return callTool(request, null); + } + + /** + * Calls a tool. An {@code input_required} result is answered by the input handler, and the call + * retried, within the round limit; without a handler, or past the limit, it is returned. + * + * @param request The call. + * @param progress Receives the progress notifications of the call, or null for none. + * @return The tool's result, or an {@code input_required} result. + */ + public Mono callTool( + McpSchema.CallToolRequest request, Consumer progress) { + Objects.requireNonNull(request, REQUEST); + // Worked out when subscribed, from the tools listed by then. + return Mono.defer( + () -> { + Map headers = paramHeaders(request); + return withInput( + request.inputResponses(), + request.requestState(), + (responses, state) -> + outcome( + McpSchema.METHOD_TOOLS_CALL, + new McpSchema.CallToolRequest( + request.name(), + request.arguments(), + responses, + state, + request.meta()), + request.name(), + headers, + progress, + McpSchema.CallToolResult.class, + McpSchema.CallToolOutcome.class), + 0); + }); + } + + /** + * Lists resources. + * + * @param cursor The cursor of the page, or null for the first. + * @return The page of resources. + */ + public Mono listResources(String cursor) { + return page(McpSchema.METHOD_RESOURCES_LIST, cursor, McpSchema.ListResourcesResult.class); + } + + /** + * Lists resource templates. + * + * @param cursor The cursor of the page, or null for the first. + * @return The page of resource templates. + */ + public Mono listResourceTemplates(String cursor) { + return page( + McpSchema.METHOD_RESOURCES_TEMPLATES_LIST, + cursor, + McpSchema.ListResourceTemplatesResult.class); + } + + /** + * Reads a resource, answering {@code input_required} results as {@link + * #callTool(McpSchema.CallToolRequest, Consumer)} does. + * + * @param request The read. + * @param progress Receives the progress notifications of the read, or null for none. + * @return The resource's contents, or an {@code input_required} result. + */ + public Mono readResource( + McpSchema.ReadResourceRequest request, Consumer progress) { + Objects.requireNonNull(request, REQUEST); + return withInput( + request.inputResponses(), + request.requestState(), + (responses, state) -> + outcome( + McpSchema.METHOD_RESOURCES_READ, + new McpSchema.ReadResourceRequest( + request.uri(), responses, state, request.meta()), + request.uri(), + Map.of(), + progress, + McpSchema.ReadResourceResult.class, + McpSchema.ReadResourceOutcome.class), + 0); + } + + /** + * Lists prompts. + * + * @param cursor The cursor of the page, or null for the first. + * @return The page of prompts. + */ + public Mono listPrompts(String cursor) { + return page(McpSchema.METHOD_PROMPTS_LIST, cursor, McpSchema.ListPromptsResult.class); + } + + /** + * Gets a prompt, answering {@code input_required} results as {@link + * #callTool(McpSchema.CallToolRequest, Consumer)} does. + * + * @param request The prompt and its arguments. + * @param progress Receives the progress notifications of the request, or null for none. + * @return The prompt, or an {@code input_required} result. + */ + public Mono getPrompt( + McpSchema.GetPromptRequest request, Consumer progress) { + Objects.requireNonNull(request, REQUEST); + return withInput( + request.inputResponses(), + request.requestState(), + (responses, state) -> + outcome( + McpSchema.METHOD_PROMPTS_GET, + new McpSchema.GetPromptRequest( + request.name(), + request.arguments(), + responses, + state, + request.meta()), + request.name(), + Map.of(), + progress, + McpSchema.GetPromptResult.class, + McpSchema.GetPromptOutcome.class), + 0); + } + + /** + * Listens to the changes a filter selects: tools, prompts and resources list changes, and the + * updates of the resources named. The stream emits each change notification as it comes, the + * acknowledgement excepted, and completes when the server ends it with its response, such as + * when it stops. Cancelling it ends the listen stream. It has no time limit. A slow subscriber + * holds back the stream: only a bounded number of notifications wait for it. + * + * @param filter The changes to listen to. + * @return The change notifications. + */ + public Flux listen(McpSchema.SubscriptionFilter filter) { + return Flux.defer(() -> listening(filter)); + } + + private Flux listening(McpSchema.SubscriptionFilter filter) { + Map params = params(new McpSchema.SubscriptionsListenRequest(filter, null)); + JsonRpc.JSONRPCRequest rpc = + new JsonRpc.JSONRPCRequest( + McpSchema.METHOD_SUBSCRIPTIONS_LISTEN, + ids.incrementAndGet(), + stamp(params)); + return transport + .send(rpc, null, Map.of(), true) + .handle( + (message, sink) -> { + if (message instanceof JsonRpc.JSONRPCNotification notification + && !McpSchema.METHOD_NOTIFICATION_SUBSCRIPTIONS_ACKNOWLEDGED + .equals(notification.method())) { + sink.next(notification); + } else if (message instanceof JsonRpc.JSONRPCResponse response + && response.error() != null) { + sink.error(new McpException(response.error())); + } + }); + } + + /** Retries a request with the input its handler gives, within the round limit. */ + private Mono withInput( + Map responses, + String state, + BiFunction, String, Mono> attempt, + int round) { + return attempt.apply(responses, state) + .flatMap( + outcome -> { + if (!(outcome instanceof McpSchema.InputRequiredResult required) + || inputHandler == null + || round >= maxInputRounds) { + return Mono.just(outcome); + } + return inputHandler + .respond(required) + .map(Optional::of) + .defaultIfEmpty(Optional.empty()) + .flatMap( + next -> + next.isPresent() + ? withInput( + next.get(), + required.requestState(), + attempt, + round + 1) + : Mono.just(outcome)); + }); + } + + /** A request whose result is complete, or input required. */ + private Mono outcome( + String method, + Object request, + String name, + Map headers, + Consumer progress, + Class complete, + Class type) { + return result(method, params(request), name, headers, progress) + .map( + result -> + type.cast( + McpSchema.ResultType.INPUT_REQUIRED.equals( + result.get("resultType")) + ? mapper.convertValue( + result, McpSchema.InputRequiredResult.class) + : mapper.convertValue(result, complete))); + } + + private Mono page(String method, String cursor, Class type) { + return result( + method, + params(new McpSchema.PaginatedRequest(cursor, null)), + null, + Map.of(), + null) + .map(result -> mapper.convertValue(result, type)); + } + + /** + * Sends a request, passing its progress on, and returns its result; an error response fails + * with an {@link McpException}. + */ + private Mono> result( + String method, + Map params, + String name, + Map headers, + Consumer progress) { + return Mono.defer(() -> exchange(method, params, name, headers, progress)); + } + + private Mono> exchange( + String method, + Map given, + String name, + Map headers, + Consumer progress) { + // Each subscription sends its own request, with its own id. + Map params = new LinkedHashMap<>(given); + params.put(META, new LinkedHashMap<>(meta(given))); + long id = ids.incrementAndGet(); + Object token = progress == null ? null : "restlet-" + id; + if (token != null) { + meta(params).put(McpSchema.MetaKeys.PROGRESS_TOKEN, token); + } + stamp(params); + // The stream completes after the response: taking the last one doesn't cancel it. + return transport + .send(new JsonRpc.JSONRPCRequest(method, id, params), name, headers, false) + .doOnNext(message -> progress(message, token, progress)) + .ofType(JsonRpc.JSONRPCResponse.class) + .filter(response -> McpModernHttpTransport.sameId(response.id(), id)) + .last() + .onErrorMap( + java.util.NoSuchElementException.class, + e -> new IOException("MCP request got no response", e)) + .switchIfEmpty(Mono.error(() -> new IOException("MCP request got no response"))) + .map( + response -> { + if (response.error() != null) { + throw new McpException(response.error()); + } + return mapper.convertValue(response.result(), MAP); + }); + } + + private void progress( + JsonRpc.JSONRPCMessage message, + Object token, + Consumer progress) { + if (token != null + && message instanceof JsonRpc.JSONRPCNotification notification + && McpSchema.METHOD_NOTIFICATION_PROGRESS.equals(notification.method())) { + try { + McpSchema.ProgressParams params = + mapper.convertValue(notification.params(), McpSchema.ProgressParams.class); + if (token.equals(params.progressToken())) { + progress.accept(params); + } + } catch (RuntimeException e) { + // A malformed notification, or a failing consumer, doesn't fail the call. + LOGGER.log(Level.FINE, "MCP progress notification dropped", e); + } + } + } + + /** The params of a request, with a modifiable {@code _meta} merging what it had. */ + private Map params(Object request) { + Map params = new LinkedHashMap<>(mapper.convertValue(request, MAP)); + Map meta = new LinkedHashMap<>(); + if (params.get(META) instanceof Map given) { + given.forEach((key, value) -> meta.put(String.valueOf(key), value)); + } + params.put(META, meta); + return params; + } + + /** + * Adds what the revision requires in each request's {@code _meta}: the protocol version and the + * client capabilities, unless the request gives them. + */ + private Map stamp(Map params) { + meta(params).putIfAbsent(McpSchema.MetaKeys.PROTOCOL_VERSION, McpProtocol.MODERN_VERSION); + meta(params).putIfAbsent(McpSchema.MetaKeys.CLIENT_CAPABILITIES, capabilities); + return params; + } + + @SuppressWarnings("unchecked") + private static Map meta(Map params) { + return (Map) + params.computeIfAbsent(META, ignored -> new LinkedHashMap()); + } + + /** + * The {@code Mcp-Param-*} headers of a call: the arguments the tool's valid {@code + * x-mcp-header} annotations name, when the tool was listed. + */ + private Map paramHeaders(McpSchema.CallToolRequest request) { + McpSchema.Tool tool = tools.get(request.name()); + if (tool == null || request.arguments() == null) { + return Map.of(); + } + // The arguments as the server sees them, in JSON: a UUID or an enum is a string. + Map arguments = mapper.convertValue(request.arguments(), MAP); + Map headers = new LinkedHashMap<>(); + for (ToolParamHeaderAnnotations.Annotation annotation : + ToolParamHeaderAnnotations.parse(tool).valid()) { + String value = headerValue(argument(arguments, annotation.path())); + if (value != null) { + headers.put(annotation.name(), value); + } + } + return headers; + } + + /** + * An argument as a header value: a number in plain decimal notation, such as 10000000 rather + * than 1.0E7, as the server expects; a string or boolean as is; else null. + */ + private static String headerValue(Object value) { + return switch (value) { + case String text -> text; + case Boolean flag -> flag.toString(); + case Number number -> { + try { + yield new BigDecimal(number.toString()).toPlainString(); + } catch (NumberFormatException _) { + yield null; // NaN or infinite: not a JSON number. + } + } + case null, default -> null; + }; + } + + private static Object argument(Map arguments, List path) { + Object value = arguments; + for (String step : path) { + if (!(value instanceof Map map)) { + return null; + } + value = map.get(step); + } + return value; + } + + /** Ends the exchanges in flight and listen streams. */ + void close() throws IOException { + transport.close(); + } +} diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioServerHelper.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioServerHelper.java index 9b74f584f1..62e91a4abb 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioServerHelper.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioServerHelper.java @@ -14,8 +14,6 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.OutputStream; -import java.math.BigDecimal; -import java.math.BigInteger; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.Map; @@ -191,7 +189,7 @@ private void submitModern(Request request, Map envelope, String method) { refuse(request, answered, id, new IOException("Too many pending MCP STDIO requests")); return; } - Object key = id == null ? null : key(id); + Object key = id == null ? null : McpMessages.idKey(id); if (key != null && inFlight.putIfAbsent(key, request) != null) { key = null; // A duplicate id: only the first may be cancelled. } @@ -267,43 +265,6 @@ private static boolean modern(Map envelope) { .PROTOCOL_VERSION)); } - /** - * A request id comparable across its JSON number forms, such as 1 and 1.0: a long when the - * number is an integer in its range, exactly, else the number itself, normalised. - */ - private static Object key(Object id) { - if (!(id instanceof Number number)) { - return id; - } - BigDecimal exact = exact(number); - if (exact == null) { - return id; // NaN or infinite: compared as given. - } - try { - return exact.longValueExact(); - } catch (ArithmeticException _) { - return exact.stripTrailingZeros(); - } - } - - /** - * The exact value of a number: a floating-point one by its binary value, not its shortest - * decimal form, which may round an integer such as 1000000000000000128.0. - * - * @return The value, or null for NaN or an infinity. - */ - private static BigDecimal exact(Number number) { - return switch (number) { - case BigDecimal decimal -> decimal; - case BigInteger integer -> new BigDecimal(integer); - case Double _, Float _ -> - Double.isFinite(number.doubleValue()) - ? new BigDecimal(number.doubleValue()) - : null; - default -> BigDecimal.valueOf(number.longValue()); - }; - } - /** * Ends the in-flight request a {@code notifications/cancelled} names, if any. * @@ -312,7 +273,7 @@ private static BigDecimal exact(Number number) { private boolean cancel(Map envelope) { Object id = envelope.get("params") instanceof Map params ? params.get("requestId") : null; - Request request = id == null ? null : inFlight.remove(key(id)); + Request request = id == null ? null : inFlight.remove(McpMessages.idKey(id)); if (request == null) { return false; } diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpMessages.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpMessages.java index 982f64aba9..471061f841 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpMessages.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpMessages.java @@ -14,6 +14,8 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; +import java.math.BigDecimal; +import java.math.BigInteger; import java.nio.ByteBuffer; import java.nio.charset.CharacterCodingException; import java.nio.charset.CodingErrorAction; @@ -122,4 +124,44 @@ public static byte[] readFrame(org.restlet.util.StreamTransport transport, int l } return null; } + + /** + * A request id comparable across its JSON number forms, such as 1 and 1.0: a long when the + * number is an integer in its range, exactly, else the number itself, normalised. + * + * @param id The id. + * @return Its key. + */ + public static Object idKey(Object id) { + if (!(id instanceof Number number)) { + return id; + } + BigDecimal exact = exact(number); + if (exact == null) { + return id; // NaN or infinite: compared as given. + } + try { + return exact.longValueExact(); + } catch (ArithmeticException _) { + return exact.stripTrailingZeros(); + } + } + + /** + * The exact value of a number: a floating-point one by its binary value, not its shortest + * decimal form, which may round an integer such as 1000000000000000128.0. + * + * @return The value, or null for NaN or an infinity. + */ + private static BigDecimal exact(Number number) { + return switch (number) { + case BigDecimal decimal -> decimal; + case BigInteger integer -> new BigDecimal(integer); + case Double _, Float _ -> + Double.isFinite(number.doubleValue()) + ? new BigDecimal(number.doubleValue()) + : null; + default -> BigDecimal.valueOf(number.longValue()); + }; + } } diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernHttpTransport.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernHttpTransport.java new file mode 100644 index 0000000000..cba5bb0169 --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernHttpTransport.java @@ -0,0 +1,358 @@ +/** + * 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.JsonRpc; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.InterruptedIOException; +import java.nio.charset.CodingErrorAction; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Base64; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +import org.restlet.Client; +import org.restlet.Request; +import org.restlet.Response; +import org.restlet.data.MediaType; +import org.restlet.data.Method; +import org.restlet.data.Preference; +import org.restlet.data.ServerSentEvent; +import org.restlet.ext.mcp.McpProtocol; +import org.restlet.representation.ByteArrayRepresentation; +import org.restlet.representation.EventStreamReader; +import org.restlet.representation.Representation; +import reactor.core.Disposable; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Operators; +import reactor.core.scheduler.Schedulers; + +/** + * Sends MCP {@code 2026-07-28} requests over a borrowed, started Restlet HTTP connector, one POST + * each, and streams back what answers them: the notifications the server sends first, then the + * response. Calls are admitted within the call limit and timeout; listen streams within their own + * limit, without timeout. Messages are read as they are demanded, so a slow subscriber holds back + * the server rather than filling memory. Cancelling a request aborts its HTTP exchange, which the + * server sees as a cancellation. It never retries or reconnects. + */ +public final class McpModernHttpTransport { + private static final String BASE64_PREFIX = "=?base64?"; + private static final String BASE64_SUFFIX = "?="; + + /** + * The characters an event may have beyond its message: field names, comments and the other + * fields. Each message is limited in bytes on its own. + */ + private static final int MAX_FRAMING_CHARS = 8192; + + private final Client client; + private final String target; + private final McpJsonMapper mapper; + private final int maxMessageBytes; + private final Consumer customizer; + private final McpWorkers calls; + private final McpWorkers listeners; + private final Set requests = ConcurrentHashMap.newKeySet(); + private volatile boolean closed; + + /** + * @param client The borrowed, started HTTP connector, not following redirects. + * @param target The MCP endpoint URI. + * @param mapper The JSON mapper. + * @param maxMessageBytes The size limit of each message, sent or received. + * @param maxConcurrentCalls The number of calls in flight at once. + * @param timeout The time limit of each call, listen streams excepted. + * @param customizer Adds to each request, such as credentials. + */ + public McpModernHttpTransport( + Client client, + String target, + McpJsonMapper mapper, + int maxMessageBytes, + int maxConcurrentCalls, + Duration timeout, + Consumer customizer) { + this.client = Objects.requireNonNull(client, "client"); + this.target = Objects.requireNonNull(target, "target"); + this.mapper = Objects.requireNonNull(mapper, "mapper"); + this.maxMessageBytes = maxMessageBytes; + this.customizer = Objects.requireNonNull(customizer, "customizer"); + calls = new McpWorkers(maxConcurrentCalls, timeout); + listeners = new McpWorkers(McpProtocol.DEFAULT_MAX_LISTEN_STREAMS, null, timeout); + } + + /** + * Sends a request, and streams back the messages answering it: notifications, then its + * response, after which the stream completes. Responses to other requests are dropped, and the + * stream fails if it ends without its response. Cancelling the stream aborts the exchange. + * + * @param rpc The request. + * @param name The value of the {@code Mcp-Name} header, or null. + * @param params The values of the {@code Mcp-Param-*} headers, by name. + * @param listen Whether this is a {@code subscriptions/listen} stream, without time limit. + * @return The messages answering the request. + */ + public Flux send( + JsonRpc.JSONRPCRequest rpc, String name, Map params, boolean listen) { + return Flux.create( + sink -> { + AtomicReference active = new AtomicReference<>(); + Demand demand = new Demand(); + sink.onRequest(demand::add); + Emitter emitter = + message -> { + demand.take(); + sink.next(message); + }; + Disposable running = + (listen ? listeners : calls) + .submit( + () -> + exchange( + rpc, name, params, active, + emitter), + () -> abort(active.get())) + .subscribe(null, sink::error, sink::complete); + sink.onCancel(running); + }) + // What follows runs off the worker, which then frees its slot without waiting. The + // worker reads only what this bounded queue has room for. + .publishOn(Schedulers.boundedElastic()); + } + + private void exchange( + JsonRpc.JSONRPCRequest rpc, + String name, + Map params, + AtomicReference active, + Emitter sink) + throws IOException { + if (closed) { + throw new IOException("MCP client is closed"); + } + Request request = request(rpc, name, params); + active.set(request); + requests.add(request); + Response response = null; + try { + if (closed || Thread.currentThread().isInterrupted()) { + request.abort(); + throw new IOException("MCP exchange cancelled"); + } + response = client.handle(request); + Representation entity = response.getEntity(); + MediaType type = entity == null ? null : entity.getMediaType(); + if (type != null && MediaType.TEXT_EVENT_STREAM.isCompatible(type)) { + stream(entity, rpc.id(), sink); + } else if (type != null && MediaType.APPLICATION_JSON.isCompatible(type)) { + sink.emit(single(entity, response, rpc.id())); + } else { + throw new IOException( + "MCP HTTP failure: " + response.getStatus(), + response.getStatus().getThrowable()); + } + } finally { + if (response != null && response.getEntity() != null) { + response.getEntity().release(); + } + requests.remove(request); + } + } + + private Request request(JsonRpc.JSONRPCRequest rpc, String name, Map params) + throws IOException { + Request request = + new Request( + Method.POST, + target, + new ByteArrayRepresentation( + McpMessages.encode(mapper, rpc, maxMessageBytes), + MediaType.APPLICATION_JSON)); + customizer.accept(request); + request.getHeaders().set(McpProtocol.VERSION_HEADER, McpProtocol.MODERN_VERSION); + request.getHeaders().set("Mcp-Method", rpc.method()); + if (name != null) { + request.getHeaders().set("Mcp-Name", headerValue(name)); + } + params.forEach( + (header, value) -> + request.getHeaders().set("Mcp-Param-" + header, headerValue(value))); + request.getClientInfo() + .setAcceptedMediaTypes( + List.of( + new Preference<>(MediaType.APPLICATION_JSON), + new Preference<>(MediaType.TEXT_EVENT_STREAM))); + return request; + } + + /** A JSON reply: the response, success or error, whatever the HTTP status. */ + private JsonRpc.JSONRPCMessage single(Representation entity, Response response, Object id) + throws IOException { + byte[] body; + try (InputStream input = entity.getStream()) { + body = McpMessages.read(input, maxMessageBytes); + } + JsonRpc.JSONRPCMessage message; + try { + message = JsonRpc.deserializeMessage(mapper, McpMessages.utf8(body)); + } catch (IOException | RuntimeException e) { + throw new IOException("MCP HTTP failure: " + response.getStatus(), e); + } + if (!(message instanceof JsonRpc.JSONRPCResponse reply) || !sameId(reply.id(), id)) { + throw new IOException("MCP JSON reply must be the response to its request"); + } + return message; + } + + /** + * An event stream: each message as it comes, until the response to the request. Responses to + * other requests are dropped. The stream must be valid UTF-8, and each message within the + * message size limit, in bytes. + */ + private void stream(Representation entity, Object id, Emitter sink) throws IOException { + try (InputStream input = entity.getStream(); + EventStreamReader reader = + new EventStreamReader( + new InputStreamReader( + input, + StandardCharsets.UTF_8 + .newDecoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter( + CodingErrorAction.REPORT)))) { + // Characters bound the event as read, its framing included; bytes bound its message. + reader.setMaxEventSize( + (int) Math.min(Integer.MAX_VALUE, (long) maxMessageBytes + MAX_FRAMING_CHARS)); + for (ServerSentEvent event = reader.readEvent(); + event != null; + event = reader.readEvent()) { + String data = event.getData(); + if (data == null || data.isEmpty()) { + continue; // A heartbeat. + } + if (utf8Length(data) > maxMessageBytes) { + throw new McpMessages.MessageLimitException( + "MCP message exceeds " + maxMessageBytes + " bytes"); + } + JsonRpc.JSONRPCMessage message = JsonRpc.deserializeMessage(mapper, data); + boolean response = message instanceof JsonRpc.JSONRPCResponse; + boolean ours = response && sameId(((JsonRpc.JSONRPCResponse) message).id(), id); + if (!response || ours) { + sink.emit(message); // Another request's response is dropped. + } + if (ours) { + return; + } + } + } + throw new IOException("MCP event stream ended without a response"); + } + + /** The number of UTF-8 bytes of a well-formed string. */ + private static long utf8Length(String value) { + long bytes = 0; + for (int i = 0; i < value.length(); i++) { + char c = value.charAt(i); + if (c < 0x80) { + bytes += 1; + } else if (c < 0x800 || Character.isSurrogate(c)) { + bytes += 2; // Each surrogate: 2 of its supplementary character's 4 bytes. + } else { + bytes += 3; + } + } + return bytes; + } + + /** + * Whether two JSON-RPC ids are the same, across their JSON number forms, exactly: 1 and 1.0 + * are, 1 and 1.5 aren't. + * + * @param a An id. + * @param b Another id. + * @return True if they are the same. + */ + public static boolean sameId(Object a, Object b) { + return Objects.equals(McpMessages.idKey(a), McpMessages.idKey(b)); + } + + /** + * A header value as MCP sends it: as is when it is printable ASCII without surrounding spaces, + * else UTF-8 in base64 between {@code =?base64?} and {@code ?=}. + * + * @param value The value. + * @return The header value. + */ + public static String headerValue(String value) { + boolean plain = + !value.startsWith(BASE64_PREFIX) + && value.strip().equals(value) + && value.chars().allMatch(c -> c >= 0x20 && c <= 0x7e); + return plain + ? value + : BASE64_PREFIX + + Base64.getEncoder().encodeToString(value.getBytes(StandardCharsets.UTF_8)) + + BASE64_SUFFIX; + } + + private static void abort(Request request) { + if (request != null) { + request.abort(); + } + } + + /** Hands a message on, once it is demanded. */ + @FunctionalInterface + private interface Emitter { + void emit(JsonRpc.JSONRPCMessage message) throws IOException; + } + + /** The messages demanded and not handed on yet. Cancelling interrupts a worker waiting. */ + private static final class Demand { + private long requested; + + synchronized void add(long count) { + requested = Operators.addCap(requested, count); + notifyAll(); + } + + synchronized void take() throws InterruptedIOException { + try { + while (requested == 0) { + wait(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("MCP exchange cancelled"); + } + if (requested != Long.MAX_VALUE) { + requested--; + } + } + } + + /** Aborts the exchanges in flight and refuses new ones. */ + public void close() throws IOException { + closed = true; + requests.forEach(Request::abort); + try { + calls.close(); + } finally { + listeners.close(); + } + } +} diff --git a/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernClientTestCase.java b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernClientTestCase.java new file mode 100644 index 0000000000..5ca08ef5ee --- /dev/null +++ b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernClientTestCase.java @@ -0,0 +1,548 @@ +/** + * 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.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.modelcontextprotocol.modern.JsonRpc; +import io.modelcontextprotocol.modern.McpException; +import io.modelcontextprotocol.modern.McpSchema; +import io.modelcontextprotocol.modern.McpSchema.CallToolOutcome; +import io.modelcontextprotocol.modern.McpSchema.CallToolRequest; +import io.modelcontextprotocol.modern.McpSchema.CallToolResult; +import io.modelcontextprotocol.modern.McpSchema.ElicitFormRequest; +import io.modelcontextprotocol.modern.McpSchema.InputRequiredResult; +import io.modelcontextprotocol.modern.McpSchema.TextContent; +import io.modelcontextprotocol.modern.McpSchema.Tool; +import io.modelcontextprotocol.modern.server.McpAsyncResponse; +import io.modelcontextprotocol.modern.server.feature.McpAsyncPrompt; +import io.modelcontextprotocol.modern.server.feature.McpAsyncResource; +import io.modelcontextprotocol.modern.server.feature.McpAsyncTool; +import io.modelcontextprotocol.modern.server.feature.ServerChange; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.restlet.Client; +import org.restlet.Context; +import org.restlet.Server; +import org.restlet.data.Protocol; +import reactor.core.Disposable; +import reactor.core.publisher.Mono; + +/** + * {@link McpClientAdapter} speaking MCP {@code 2026-07-28} to {@link McpServerAdapter} over real + * HTTP connectors: discovery, listings, calls, streamed progress, {@code input_required} retries, + * {@code subscriptions/listen} and cancellation. + */ +class McpModernClientTestCase { + private static final Duration WAIT = Duration.ofSeconds(10); + + private final CountDownLatch cancelled = new CountDownLatch(1); + private McpServerAdapter adapter; + private Server server; + private Client connector; + private McpClientAdapter client; + + @AfterEach + void cleanup() throws Exception { + try { + if (client != null) { + client.stop(); + } + } finally { + try { + if (connector != null) { + connector.stop(); + } + } finally { + try { + if (server != null) { + server.stop(); + } + } finally { + if (adapter != null) { + adapter.stop(); + } + } + } + } + } + + @Test + void discoversAndCallsOverHttp1() throws Exception { + discoversAndCalls("HTTP1_1"); + } + + @Test + void discoversAndCallsOverHttp2() throws Exception { + discoversAndCalls("HTTP2"); + } + + /** Discovery, then the tools, resources and prompts of both revisions' registrations. */ + private void discoversAndCalls(String mode) throws Exception { + McpModernClient modern = start(mode, null); + + McpSchema.DiscoverResult discovered = modern.discover().block(WAIT); + assertEquals(List.of(McpProtocol.MODERN_VERSION), discovered.supportedVersions()); + assertEquals( + List.of("echo", "progress", "approve", "endless", "region", "level"), + modern.listTools(null).block(WAIT).tools().stream().map(Tool::name).toList()); + assertEquals( + "echo:café", + text( + modern.callTool( + new CallToolRequest( + "echo", Map.of("text", "café"), null, null, null)) + .block(WAIT))); + assertEquals( + List.of("test://modern"), + modern.listResources(null).block(WAIT).resources().stream() + .map(McpSchema.Resource::uri) + .toList()); + McpSchema.ReadResourceResult read = + (McpSchema.ReadResourceResult) + modern.readResource( + new McpSchema.ReadResourceRequest( + "test://modern", null, null, null), + null) + .block(WAIT); + assertEquals( + "modern", ((McpSchema.TextResourceContents) read.contents().getFirst()).text()); + assertEquals( + List.of("welcome"), + modern.listPrompts(null).block(WAIT).prompts().stream() + .map(McpSchema.Prompt::name) + .toList()); + McpSchema.GetPromptResult prompt = + (McpSchema.GetPromptResult) + modern.getPrompt( + new McpSchema.GetPromptRequest( + "welcome", null, null, null, null), + null) + .block(WAIT); + assertEquals("modern welcome", prompt.description()); + } + + /** A call's progress notifications reach its consumer before its result. */ + @Test + void streamsProgress() throws Exception { + McpModernClient modern = start("HTTP2", null); + List progress = new CopyOnWriteArrayList<>(); + + CallToolOutcome outcome = + modern.callTool( + new CallToolRequest("progress", Map.of(), null, null, null), + update -> progress.add(update.message())) + .block(WAIT); + + assertEquals("done", text(outcome)); + assertEquals(List.of("half"), progress); + } + + /** Without a handler, an input_required result is returned; with one, the call is retried. */ + @Test + void answersInputRequired() throws Exception { + McpModernClient modern = start("HTTP2", null); + CallToolRequest approve = new CallToolRequest("approve", Map.of(), null, null, null); + + InputRequiredResult required = + assertInstanceOf(InputRequiredResult.class, modern.callTool(approve).block(WAIT)); + assertNotNull(required.requestState()); // Sealed by the server. + assertTrue(required.inputRequests().containsKey("confirm")); + assertEquals( + "approved:s1", + text( + modern.callTool( + new CallToolRequest( + "approve", + Map.of(), + Map.of("confirm", accepted()), + required.requestState(), + null)) + .block(WAIT))); + + client.stop(); + McpModernClient handled = + start("HTTP2", result -> Mono.just(Map.of("confirm", accepted()))); + assertEquals("approved:s1", text(handled.callTool(approve).block(WAIT))); + } + + /** A listen stream carries the changes published, and completes when the server stops. */ + @Test + void listensToChanges() throws Exception { + McpModernClient modern = start("HTTP2", null); + BlockingQueue changes = new LinkedBlockingQueue<>(); + CountDownLatch completed = new CountDownLatch(1); + List errors = new CopyOnWriteArrayList<>(); + Disposable listening = + modern.listen(new McpSchema.SubscriptionFilter(true, null, null, null)) + .doOnComplete(completed::countDown) + .subscribe(changes::add, errors::add); + try { + // The listen stream opens asynchronously: publish until the change arrives. + JsonRpc.JSONRPCNotification change = null; + for (int i = 0; i < 200 && change == null; i++) { + adapter.getChanges().broadcast(new ServerChange.ToolsListChanged()); + change = changes.poll(50, TimeUnit.MILLISECONDS); + } + assertEquals( + McpSchema.METHOD_NOTIFICATION_TOOLS_LIST_CHANGED, + change == null ? null : change.method(), + errors.toString()); + + adapter.stop(); + assertTrue(completed.await(10, TimeUnit.SECONDS), "the listen stream did not end"); + assertEquals(List.of(), errors); + } finally { + listening.dispose(); + } + } + + /** Cancelling a call aborts its exchange, which cancels the tool on the server. */ + @Test + void cancellingACallCancelsTheTool() throws Exception { + McpModernClient modern = start("HTTP2", null); + CountDownLatch started = new CountDownLatch(1); + + Disposable call = + modern.callTool( + new CallToolRequest("endless", Map.of(), null, null, null), + update -> started.countDown()) + .subscribe(); + assertTrue(started.await(10, TimeUnit.SECONDS), "the tool did not start"); + call.dispose(); + + assertTrue(cancelled.await(10, TimeUnit.SECONDS), "the tool was not cancelled"); + assertEquals( + "echo:after", + text( + modern.callTool( + new CallToolRequest( + "echo", Map.of("text", "after"), null, null, null)) + .block(WAIT))); + } + + /** A listed tool's x-mcp-header arguments are mirrored into Mcp-Param headers. */ + @Test + void mirrorsAnnotatedArguments() throws Exception { + McpModernClient modern = start("HTTP2", null); + CallToolRequest region = + new CallToolRequest("region", Map.of("region", "eu"), null, null, null); + + Mono unlistedCall = modern.callTool(region); + McpException unlisted = assertThrows(McpException.class, () -> unlistedCall.block(WAIT)); + assertEquals(McpSchema.ErrorCodes.HEADER_MISMATCH, unlisted.error().code().intValue()); + + modern.listTools(null).block(WAIT); + assertEquals("region:eu", text(modern.callTool(region).block(WAIT))); + } + + /** + * With a single call slot, an input handler may make a blocking call: the first call no longer + * holds its slot, nor runs the handler on its interrupted worker. + */ + @Test + void anInputHandlerMayCallWithOneSlot() throws Exception { + AtomicReference holder = new AtomicReference<>(); + List nested = new CopyOnWriteArrayList<>(); + McpModernClient modern = + start( + "HTTP2", + required -> + Mono.fromCallable( + () -> { + nested.add( + text( + holder.get() + .callTool( + new CallToolRequest( + "echo", + Map.of( + "text", + "nested"), + null, + null, + null)) + .block(WAIT))); + return Map.of("confirm", accepted()); + }), + configured -> configured.setMaxConcurrentCalls(1)); + holder.set(modern); + + assertEquals( + "approved:s1", + text( + modern.callTool(new CallToolRequest("approve", Map.of(), null, null, null)) + .block(WAIT))); + assertEquals(List.of("echo:nested"), nested); + } + + /** A call's headers are worked out when it is sent, after a listing chained before it. */ + @Test + void headersFollowAChainedListing() throws Exception { + McpModernClient modern = start("HTTP2", null); + CallToolRequest region = + new CallToolRequest("region", Map.of("region", "eu"), null, null, null); + + assertEquals( + "region:eu", + text(modern.listTools(null).then(modern.callTool(region)).block(WAIT))); + } + + /** A number argument is mirrored in plain decimal notation. */ + @Test + void mirrorsNumbersInPlainNotation() throws Exception { + McpModernClient modern = start("HTTP2", null); + modern.listTools(null).block(WAIT); + + assertEquals( + "level:1.0E7", + text( + modern.callTool( + new CallToolRequest( + "level", Map.of("level", 1.0e7), null, null, null)) + .block(WAIT))); + } + + /** An error response fails the request with its JSON-RPC error. */ + @Test + void anErrorResponseFails() throws Exception { + McpModernClient modern = start("HTTP2", null); + + Mono missing = + modern.readResource( + new McpSchema.ReadResourceRequest("test://missing", null, null, null), + null); + McpException error = assertThrows(McpException.class, () -> missing.block(WAIT)); + assertEquals(McpSchema.ErrorCodes.INVALID_PARAMS, error.error().code().intValue()); + } + + /** The modern client is reached through getModernClient() only, over HTTP for now. */ + @Test + void modernModeLimits() throws Exception { + start("HTTP2", null); + assertThrows(IllegalStateException.class, client::getSdkClient); + client.stop(); + assertFalse(client.isStarted()); + + McpClientAdapter stdio = new McpClientAdapter(connector, "stdio://localhost/mcp"); + stdio.setProtocolVersion(McpProtocol.MODERN_VERSION); + assertThrows(UnsupportedOperationException.class, stdio::start); + McpClientAdapter unknown = new McpClientAdapter(connector, "http://localhost/mcp"); + assertThrows( + IllegalArgumentException.class, () -> unknown.setProtocolVersion("2024-11-05")); + } + + /** Starts the server, if needed, and a modern client with an optional input handler. */ + private McpModernClient start(String mode, McpInputHandler inputHandler) throws Exception { + return start(mode, inputHandler, ignored -> {}); + } + + private McpModernClient start( + String mode, McpInputHandler inputHandler, Consumer configure) + throws Exception { + if (adapter == null) { + adapter = adapter(); + Context sc = new Context(); + sc.getParameters().set("http.transport.protocol", mode); + server = new Server(sc, Protocol.HTTP, 0, adapter); + server.start(); + Context cc = new Context(); + cc.getParameters().set("httpClientTransportMode", mode); + cc.getParameters().set("followRedirects", "false"); + connector = new Client(cc, Protocol.HTTP); + connector.start(); + } + client = + new McpClientAdapter( + connector, "http://localhost:" + server.getActualPort() + "/mcp"); + client.setProtocolVersion(McpProtocol.MODERN_VERSION); + client.setModernCapabilities( + McpSchema.ClientCapabilities.builder() + .elicitation( + new McpSchema.ClientCapabilities.Elicitation( + new McpSchema.ClientCapabilities.Elicitation.Form(), null)) + .build()); + client.setInputHandler(inputHandler); + configure.accept(client); + client.start(); + return client.getModernClient(); + } + + private McpServerAdapter adapter() throws Exception { + McpServerAdapter served = new McpServerAdapter(); + McpServerAdapterTestCase.addEcho(served); + served.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( + result( + "done"))))))); + served.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder("approve", Map.of("type", "object")).build(), + (ctx, request) -> { + CallToolOutcome outcome = + request.requestState() == null + ? InputRequiredResult.builder() + .elicit( + "confirm", + ElicitFormRequest.builder( + "Proceed?", + Map.of( + "type", + "object", + "properties", + Map + .of())) + .build()) + .requestState("s1") + .build() + : result("approved:" + request.requestState()); + return Mono.just(McpAsyncResponse.result(outcome)); + })); + served.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)))))); + served.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( + result( + "region:" + + request.arguments() + .get("region")))))); + served.getModernTools() + .add( + McpAsyncTool.of( + Tool.builder( + "level", + Map.of( + "type", + "object", + "properties", + Map.of( + "level", + Map.of( + "type", + "number", + "x-mcp-header", + "Level")))) + .build(), + (ctx, request) -> + Mono.just( + McpAsyncResponse.result( + result( + "level:" + + request.arguments() + .get("level")))))); + served.getModernResources() + .add( + new McpModernResource( + McpSchema.Resource.builder("test://modern", "modern").build(), + McpAsyncResource.of( + (ctx, request) -> { + McpSchema.ReadResourceOutcome outcome = + McpSchema.ReadResourceResult.builder( + List.of( + new McpSchema + .TextResourceContents( + request.uri(), + "text/plain", + "modern", + null))) + .build(); + return Mono.just(McpAsyncResponse.result(outcome)); + }))); + served.getModernPrompts() + .add( + McpAsyncPrompt.of( + McpSchema.Prompt.builder("welcome").build(), + (ctx, request) -> { + McpSchema.GetPromptOutcome outcome = + new McpSchema.GetPromptResult( + "modern welcome", + List.of( + new McpSchema.PromptMessage( + McpSchema.Role.USER, + TextContent.builder("hi") + .build())), + null, + null); + return Mono.just(McpAsyncResponse.result(outcome)); + })); + served.start(); + return served; + } + + private static Map accepted() { + return Map.of("action", "accept", "content", Map.of()); + } + + private static CallToolResult result(String text) { + return CallToolResult.builder().addContent(TextContent.builder(text).build()).build(); + } + + private static String text(Object outcome) { + CallToolResult result = assertInstanceOf(CallToolResult.class, outcome); + return ((TextContent) result.content().getFirst()).text(); + } +} diff --git a/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernHttpTransportTestCase.java b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernHttpTransportTestCase.java new file mode 100644 index 0000000000..9f5c63869b --- /dev/null +++ b/org.restlet.ext.mcp/src/test/java/org/restlet/ext/mcp/McpModernHttpTransportTestCase.java @@ -0,0 +1,265 @@ +/** + * 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.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.modelcontextprotocol.json.McpJsonDefaults; +import io.modelcontextprotocol.modern.JsonRpc; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.reactivestreams.Subscription; +import org.restlet.Client; +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.Protocol; +import org.restlet.ext.mcp.internal.McpMessages; +import org.restlet.ext.mcp.internal.McpModernHttpTransport; +import org.restlet.representation.ByteArrayRepresentation; +import reactor.core.publisher.BaseSubscriber; +import reactor.core.publisher.Flux; +import reactor.util.retry.Retry; + +/** + * {@link McpModernHttpTransport} against a server sending crafted replies: correlation by exact id, + * strict UTF-8, per-message byte limits and demand. + */ +class McpModernHttpTransportTestCase { + private static final Duration WAIT = Duration.ofSeconds(10); + private static final String NOTIFICATION = + "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\",\"params\":{}}"; + + private volatile byte[] reply = new byte[0]; + private volatile MediaType type = MediaType.TEXT_EVENT_STREAM; + private Server server; + private Client connector; + private McpModernHttpTransport transport; + + @AfterEach + void cleanup() throws Exception { + try { + if (transport != null) { + transport.close(); + } + } finally { + try { + if (connector != null) { + connector.stop(); + } + } finally { + if (server != null) { + server.stop(); + } + } + } + } + + @Test + void comparesIdsExactly() { + assertTrue(McpModernHttpTransport.sameId(1L, 1.0)); + assertTrue(McpModernHttpTransport.sameId("a", "a")); + assertFalse(McpModernHttpTransport.sameId(1L, 1.5)); + assertFalse(McpModernHttpTransport.sameId(9007199254740993L, 9007199254740992.0)); + assertFalse(McpModernHttpTransport.sameId(1L, "1")); + } + + /** Other requests' responses are dropped; the stream ends with its own. */ + @Test + void streamsUntilItsOwnResponse() throws Exception { + start(1024, 1); + reply = events(NOTIFICATION, response(99), response(1), NOTIFICATION); + + List messages = send(1).collectList().block(WAIT); + + assertEquals(2, messages.size()); + assertInstanceOf(JsonRpc.JSONRPCNotification.class, messages.get(0)); + assertEquals(1, ((Number) ((JsonRpc.JSONRPCResponse) messages.get(1)).id()).intValue()); + } + + @Test + void aStreamWithoutItsResponseFails() throws Exception { + start(1024, 1); + reply = + events( + NOTIFICATION, + response(99), + "{\"jsonrpc\":\"2.0\",\"id\":\"1\",\"result\":{}}"); + + Flux sent = send(1); + IOException failure = assertThrows(IOException.class, () -> block(sent)); + assertTrue(failure.getMessage().contains("without a response"), failure.getMessage()); + } + + @Test + void aJsonReplyMustBeItsResponse() throws Exception { + start(1024, 1); + type = MediaType.APPLICATION_JSON; + for (String body : List.of(NOTIFICATION, response(99))) { + reply = body.getBytes(StandardCharsets.UTF_8); + Flux sent = send(1); + assertThrows(IOException.class, () -> block(sent)); + } + reply = response(1).getBytes(StandardCharsets.UTF_8); + assertEquals(1, send(1).collectList().block(WAIT).size()); + } + + @Test + void aStreamWithMalformedUtf8Fails() throws Exception { + start(1024, 1); + ByteArrayOutputStream body = new ByteArrayOutputStream(); + body.writeBytes( + "data: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"text\":\"" + .getBytes(StandardCharsets.UTF_8)); + body.write(0xFF); + body.writeBytes("\"}}\n\n".getBytes(StandardCharsets.UTF_8)); + reply = body.toByteArray(); + + Flux sent = send(1); + assertThrows(CharacterCodingException.class, () -> block(sent)); + } + + /** The limit is on each message's bytes: framing doesn't count, non-ASCII characters do. */ + @Test + void limitsEachMessageInBytes() throws Exception { + // Padded past the size of the request sent. + String exact = response(1).replace("{}", "{\"pad\":\"" + "x".repeat(200) + "\"}"); + start(exact.getBytes(StandardCharsets.UTF_8).length, 1); + reply = + (": a comment\nevent: message\nid: 7\ndata: " + exact + "\n\n") + .getBytes(StandardCharsets.UTF_8); + assertEquals(1, send(1).collectList().block(WAIT).size()); + + // As many characters, but "é" takes 2 bytes. + String wide = exact.replace("xx\"", "xé\""); + transport.close(); + start(exact.length(), 1); + reply = events(wide); + Flux sent = send(1); + assertThrows(McpMessages.MessageLimitException.class, () -> block(sent)); + } + + /** + * A subscriber demanding nothing holds back the stream, which it can still cancel: the worker + * waiting for demand ends, and frees the only call slot. + */ + @Test + void aSlowSubscriberHoldsBackTheStream() throws Exception { + start(1024, 1); + String[] many = new String[5000]; + java.util.Arrays.fill(many, NOTIFICATION); + reply = events(many); + CountDownLatch first = new CountDownLatch(1); + BaseSubscriber slow = + new BaseSubscriber<>() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + request(1); + } + + @Override + protected void hookOnNext(JsonRpc.JSONRPCMessage message) { + first.countDown(); + } + }; + send(1).subscribe(slow); + assertTrue(first.await(WAIT.toSeconds(), TimeUnit.SECONDS)); + + // Read ahead without demand, the stream would have ended and freed its slot by now. + Thread.sleep(500); + reply = events(response(2)); + Flux refused = send(2); + IOException busy = assertThrows(IOException.class, () -> block(refused)); + assertTrue(busy.getMessage().contains("call limit"), busy.getMessage()); + + slow.cancel(); + assertEquals( + 1, + send(2).retryWhen(Retry.fixedDelay(50, Duration.ofMillis(100))) + .collectList() + .block(WAIT) + .size()); + } + + private void start(int maxMessageBytes, int maxCalls) throws Exception { + if (server == null) { + server = + new Server( + new Context(), + Protocol.HTTP, + 0, + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity(new ByteArrayRepresentation(reply, type)); + } + }); + server.start(); + Context cc = new Context(); + cc.getParameters().set("followRedirects", "false"); + connector = new Client(cc, Protocol.HTTP); + connector.start(); + } + transport = + new McpModernHttpTransport( + connector, + "http://localhost:" + server.getActualPort() + "/mcp", + McpJsonDefaults.getMapper(), + maxMessageBytes, + maxCalls, + WAIT, + ignored -> {}); + } + + private Flux send(long id) { + return transport.send( + new JsonRpc.JSONRPCRequest("tools/call", id, Map.of()), null, Map.of(), false); + } + + /** Blocks on a stream, rethrowing the checked exception it failed with. */ + private static void block(Flux sent) throws Exception { + try { + sent.blockLast(WAIT); + } catch (RuntimeException e) { + Throwable cause = reactor.core.Exceptions.unwrap(e); + if (cause instanceof Exception checked) { + throw checked; + } + throw e; + } + } + + private static String response(int id) { + return "{\"jsonrpc\":\"2.0\",\"id\":" + id + ",\"result\":{}}"; + } + + private static byte[] events(String... messages) { + StringBuilder body = new StringBuilder(); + for (String message : messages) { + body.append("data: ").append(message).append("\n\n"); + } + return body.toString().getBytes(StandardCharsets.UTF_8); + } +}