diff --git a/changes.md b/changes.md index f0f79c5b01..6ef52cc402 100644 --- a/changes.md +++ b/changes.md @@ -91,37 +91,53 @@ Changes log select one explicitly. - Enhancements - 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. `getModernResources()`, - `getModernResourceTemplates()` and `getModernPrompts()` add resources, templates and - prompts served to 2026-07-28 clients only; a URI, URI template or prompt name may only - be used once across both revisions' lists. `subscriptions/listen` streams carry the - changes published through `getChanges()`; they have their own limit, - `setMaxListenStreams()`, no time limit, and end when the adapter stops or the client - goes away. Over stdio, a message whose `_meta` names 2026-07-28 goes to the modern - server: streamed messages are written as separate lines as they come, and - `notifications/cancelled` ends the request it names without a response. - `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 + connectors and explicitly configured STDIO helpers, built on the SDK's stateless MCP + 2026-07-28 model, experimentally: the SDK's modern API is a preview. Messages and calls + are bounded, responses may stream over SSE, and requests are cancelled when the client + goes away. Borrowed HTTP connectors are not stopped by the MCP client. + - `McpServerAdapter` registers tools, resources, resource templates and prompts on the + SDK's modern types (`getTools()`, `getResources()` with `McpResource`, + `getResourceTemplates()` with `McpResourceTemplate`, `getPrompts()`); a name, URI or URI + template may only be used once. Tool filters (`McpToolFilter`) and interceptors + (`McpInterceptor`) also use the modern types, and apply once to each request of either + era. 2026-07-28 requests get `input_required` results, sealed with a key that + `setRequestStateKey()` shares between instances, and streamed progress; + `subscriptions/listen` streams carry the changes published through `getChanges()`, + within `setMaxListenStreams()`. Over stdio, a message whose `_meta` names 2026-07-28 is + answered line by line as it streams, and `notifications/cancelled` ends the request it + names. + - MCP 2025-11-25 clients are served by a legacy edge translating their requests into the + 2026-07-28 model, so it can be dropped when the deprecation window closes. It exposes + nothing 2026-07-28 deprecates or removes: `initialize` is answered from + `server/discover` with tools, resources and prompts only, `ping` gets an empty result, + notifications are dropped, `logging/setLevel` and resource subscriptions are not found, + no session is issued, `GET` gets a `405`, and nothing is requested from the client. A + legacy client gets the final result of a streaming request, and an error instead of an + `input_required` result. + - `McpClientAdapter` speaks MCP 2025-11-25 by default, through the SDK's client behind the + same legacy edge. `setProtocolVersion(McpProtocol.MODERN_VERSION)` makes it speak + 2026-07-28 over HTTP: `start()` runs `server/discover`, and `getModernClient()` returns + an `McpModernClient` 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. + `setInputHandler()` and the request retried, within `setMaxInputRounds()`. + `setProtocolVersion(McpProtocol.AUTO)` speaks whichever era the server does, as the + 2026-07-28 revision specifies for a dual-era client: over HTTP, `start()` sends + `server/discover`; a `4xx` without a JSON-RPC response, or a JSON-RPC error other than + an unsupported version (`-32022`), marks a legacy server, which gets an `initialize` + handshake offering `McpProtocol.LEGACY_VERSIONS` (2025-11-25 back to 2024-11-05). The era + found is remembered per endpoint URI for the life of the JVM, and probed again when it + proves wrong; over stdio, `auto` speaks 2025-11-25. `getNegotiatedVersion()` tells the + revision in use, and `listTools()`, `callTool()`, `readResource()` and `getPrompt()` work + in both eras on the modern types. With a legacy server, the client declares no roots, + sampling or elicitation capability, never sets a logging level, refuses the requests the + server sends, and opens no listening `GET` stream. + - Legacy MCP sessions are opt-in, since 2026-07-28 removed them: + `McpClientAdapter.setLegacySessions(true)` keeps the `Mcp-Session-Id` a legacy server + returns on `initialize`, sends it with later requests, initializes again after it + expires, and deletes it on `stop()`; `setLegacySessionId()` sends a session id the caller + owns, which is never replaced or deleted; `getLegacySessionId()` tells the one in use. + They have no effect on modern servers, and go away with the legacy edge. - 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 56dc2f248b..897359b297 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 @@ -8,39 +8,66 @@ */ package org.restlet.ext.mcp; -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 io.modelcontextprotocol.modern.McpSchema; import java.io.IOException; import java.io.UncheckedIOException; import java.time.Duration; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.function.Consumer; -import java.util.function.Function; import org.restlet.Client; import org.restlet.Request; +import org.restlet.ext.mcp.internal.McpEras; +import org.restlet.ext.mcp.internal.McpHttpStatusException; +import org.restlet.ext.mcp.internal.McpLegacyEdge; import org.restlet.ext.mcp.internal.McpModernHttpTransport; -import org.restlet.ext.mcp.internal.StdioMcpClientTransport; -import reactor.core.publisher.Mono; +import reactor.core.Exceptions; /** - * Lifecycle facade over the native SDK. Borrowed Restlet connectors are never stopped. + * Lifecycle facade of an MCP client, on the SDK's {@code 2026-07-28} types. 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. + *

By default it speaks MCP {@link McpProtocol#VERSION} through the SDK's client, behind a legacy + * edge. 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. + * + *

With {@link McpProtocol#AUTO}, also experimental, it speaks whichever era the server does, as + * the 2026-07-28 revision specifies for a dual-era client. Over HTTP, {@link #start()} sends {@code + * server/discover}: a discovery result naming {@code 2026-07-28} selects the modern client; a + * {@code 4xx} without a JSON-RPC response, or a JSON-RPC error other than an unsupported version, + * marks a legacy server, which gets an {@code initialize} handshake offering {@link + * McpProtocol#LEGACY_VERSIONS}. Other failures, such as a {@code 5xx} or a timeout, are thrown + * without falling back. The era found is remembered for the endpoint URI, for the life of the JVM: + * a later start skips the probe, and probes again only if the server contradicts that era, with a + * {@code 4xx} or a JSON-RPC error that the other era would answer with, not with a timeout or a + * {@code 5xx}. A started client keeps its era: requests are never replayed on the other one, so + * when a server changes era, stop the client and start a new one. Over stdio, it speaks {@link + * McpProtocol#VERSION}, the modern client being HTTP only. + * + *

With a legacy server, the client exposes nothing 2026-07-28 deprecates or removes: it declares + * no roots, sampling or elicitation capability, never sets a logging level, refuses the requests a + * server sends, and opens no listening {@code GET} stream. It keeps no session unless asked to, + * through {@link #setLegacySessions(boolean)} or {@link #setLegacySessionId(String)}. + * + *

{@link #getNegotiatedVersion()} tells the revision in use. {@link #listTools(String)}, {@link + * #callTool(String, Map)}, {@link #readResource(String)} and {@link #getPrompt(String, Map)} work + * in both eras, on the modern types. */ 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; + /** JSON-RPC errors only a {@code 2026-07-28} server answers {@code server/discover} with. */ + private static final Set MODERN_ERRORS = + Set.of( + McpSchema.ErrorCodes.HEADER_MISMATCH, + McpSchema.ErrorCodes.MISSING_REQUIRED_CLIENT_CAPABILITY); + private final Client client; private final String target; private final org.restlet.util.StreamTransport stream; @@ -49,13 +76,14 @@ public final class McpClientAdapter implements AutoCloseable { 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 McpSchema.ClientCapabilities modernCapabilities = McpSchema.ClientCapabilities.NONE; private McpInputHandler inputHandler; private int maxInputRounds = DEFAULT_MAX_INPUT_ROUNDS; + private boolean legacySessions; + private String legacySessionId; private McpModernClient modern; - private McpAsyncClient sdk; - private McpClientTransport transport; + private McpLegacyEdge.Client legacy; + private String negotiatedVersion; private boolean started; private boolean configured; @@ -125,19 +153,36 @@ public synchronized Consumer getRequestCustomizer() { /** * 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}. + * client; {@link McpProtocol#MODERN_VERSION} through {@link #getModernClient()}, over HTTP only + * for now; or {@link McpProtocol#AUTO}, whichever the server speaks. Experimental for {@code + * 2026-07-28} and {@code auto}. * - * @param value The revision. + * @param value The revision, or {@link McpProtocol#AUTO}. */ public synchronized void setProtocolVersion(String value) { requireConfigurable(); - if (!McpProtocol.VERSION.equals(value) && !McpProtocol.MODERN_VERSION.equals(value)) { + if (!McpProtocol.VERSION.equals(value) + && !McpProtocol.MODERN_VERSION.equals(value) + && !McpProtocol.AUTO.equals(value)) { throw new IllegalArgumentException("Unsupported MCP revision: " + value); } protocolVersion = value; } + /** + * The revision spoken with the server, once started: {@link McpProtocol#MODERN_VERSION}, or the + * one negotiated by the last {@code initialize}, which a handshake after an expired legacy + * session may change. + * + * @return The revision, or null before {@link #start()}. + */ + public synchronized String getNegotiatedVersion() { + if (!started) { + return null; + } + return legacy != null ? legacy.negotiatedVersion() : negotiatedVersion; + } + public synchronized String getProtocolVersion() { return protocolVersion; } @@ -148,20 +193,19 @@ public synchronized String getProtocolVersion() { * * @param value The capabilities. */ - public synchronized void setModernCapabilities( - io.modelcontextprotocol.modern.McpSchema.ClientCapabilities value) { + public synchronized void setModernCapabilities(McpSchema.ClientCapabilities value) { requireConfigurable(); modernCapabilities = Objects.requireNonNull(value, "value"); } - public synchronized io.modelcontextprotocol.modern.McpSchema.ClientCapabilities - getModernCapabilities() { + public synchronized 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. + * requests, after which the request is retried; without one, the result is returned. Legacy + * servers ask for no input: the legacy edge declares no elicitation capability. * * @param value The handler, or null for none. */ @@ -192,6 +236,62 @@ public synchronized int getMaxInputRounds() { return maxInputRounds; } + /** + * Keeps the {@code Mcp-Session-Id} a legacy server returns on {@code initialize}, off by + * default: the id is then sent with each later request, the client initializes once more when + * the session expires, and it deletes the session when it stops. Without it, a legacy server + * opening a session fails the handshake. + * + *

Legacy only. MCP 2026-07-28 removed sessions: this property exists so the client + * can reach legacy servers that require one, and goes away with the legacy edge. It has no + * effect on modern servers. Even with a session, the client opens no listening {@code GET} + * stream, does not resume with {@code Last-Event-ID}, and refuses the requests the server + * sends, such as elicitation, sampling or roots, so servers depending on those will not work. + * + * @param value True to keep a legacy server's session. + */ + public synchronized void setLegacySessions(boolean value) { + requireConfigurable(); + legacySessions = value; + } + + /** + * Whether the session of a legacy server is kept. + * + * @return True if kept. + * @see #setLegacySessions(boolean) + */ + public synchronized boolean isLegacySessions() { + return legacySessions; + } + + /** + * Sets a legacy server's {@code Mcp-Session-Id} the caller owns, such as one a proxy relays: it + * is sent with each legacy request, from {@code initialize} on, but never replaced, never + * deleted, and its expiry fails the request. + * + *

Legacy only, with the same limits as {@link #setLegacySessions(boolean)}: MCP + * 2026-07-28 removed sessions, and this property goes away with the legacy edge. It has no + * effect on modern servers. + * + * @param value The session id, or null for none. + */ + public synchronized void setLegacySessionId(String value) { + requireConfigurable(); + legacySessionId = value; + } + + /** + * The {@code Mcp-Session-Id} sent to a legacy server: once started, the one in use, either the + * caller's or the server's; before, the one the caller set. Legacy only: see {@link + * #setLegacySessions(boolean)}. + * + * @return The session id, or null. + */ + public synchronized String getLegacySessionId() { + return legacy != null ? legacy.sessionId() : legacySessionId; + } + public synchronized void start() { if (started) { return; @@ -199,37 +299,169 @@ public synchronized void start() { requireConfigurable(); configured = true; if (McpProtocol.MODERN_VERSION.equals(protocolVersion)) { - startModern(); + startModern(createModern()); return; } - transport = createTransport(); - try { - sdk = - McpClient.async(transport) - .requestTimeout(requestTimeout) - .initializationTimeout(requestTimeout) - .build(); - McpSchema.InitializeResult initialized = sdk.initialize().block(requestTimeout); - if (initialized == null || !McpProtocol.VERSION.equals(initialized.protocolVersion())) { - throw new IllegalStateException("Peer did not negotiate the supported MCP version"); + if (McpProtocol.AUTO.equals(protocolVersion)) { + startAuto(); + return; + } + startLegacy(List.of(McpProtocol.VERSION)); + } + + /** Starts the legacy client: an {@code initialize} handshake offering these revisions. */ + private void startLegacy(List versions) { + legacy = createLegacy(versions); + negotiatedVersion = legacy.negotiatedVersion(); + started = true; + } + + /** + * Starts the client of the era the server speaks: the one remembered for the endpoint, unless + * the server contradicts it, else the one a {@code server/discover} probe tells. + */ + private void startAuto() { + if (stream != null || target.startsWith("stdio:")) { + startLegacy(List.of(McpProtocol.VERSION)); + return; + } + String cached = McpEras.get(target); + if (cached != null) { + try { + if (McpEras.MODERN.equals(cached)) { + startModern(createModern()); + } else { + startLegacy(McpProtocol.LEGACY_VERSIONS); + } + return; + } catch (RuntimeException e) { + // A timeout or a server error says nothing of the era: probing again would only + // add requests, and might repeat a handshake that opened a session. + if (!contradicts(cached, e)) { + throw e; + } + McpEras.remove(target); } + } + McpModernClient candidate = createModern(); + if (probe(candidate)) { + modern = candidate; + negotiatedVersion = McpProtocol.MODERN_VERSION; started = true; + McpEras.put(target, McpEras.MODERN); + } else { + startLegacy(McpProtocol.LEGACY_VERSIONS); + McpEras.put(target, McpEras.LEGACY); + } + } + + /** + * Whether a client's failed start contradicts the era remembered for the endpoint. + * + * @param era The era remembered. + * @param failure The failure. + * @return True if the endpoint should be probed again. + */ + private static boolean contradicts(String era, RuntimeException failure) { + if (McpEras.LEGACY.equals(era)) { + return McpLegacyEdge.wrongEra(failure); + } + if (failure instanceof NotModern) { + return true; + } + try { + return isLegacy(Exceptions.unwrap(failure)); + } catch (IllegalStateException noSharedRevision) { + return true; + } + } + + /** A discovery result not naming {@code 2026-07-28}. */ + private static final class NotModern extends IllegalStateException { + private static final long serialVersionUID = 1L; + + private NotModern() { + super("Peer does not support MCP 2026-07-28"); + } + } + + /** + * Probes the server with {@code server/discover}. + * + * @return True for a modern server, which the candidate then serves; false for a legacy one, + * for which the candidate is closed. + */ + private boolean probe(McpModernClient candidate) { + boolean kept = false; + try { + McpSchema.DiscoverResult discovered = candidate.discover().block(requestTimeout); + List supported = + discovered == null || discovered.supportedVersions() == null + ? List.of() + : discovered.supportedVersions(); + kept = supported.contains(McpProtocol.MODERN_VERSION); + if (!kept && !offersLegacy(supported)) { + throw new IllegalStateException( + "No MCP revision supported by both the client and the peer: " + supported); + } + return kept; } catch (RuntimeException e) { - try { - if (sdk != null) { - sdk.closeGracefully().block(requestTimeout); - } else { - transport.closeGracefully().block(requestTimeout); + if (!isLegacy(Exceptions.unwrap(e))) { + throw e; + } + return false; + } finally { + if (!kept) { + try { + candidate.close(); + } catch (IOException cleanup) { + // The probe exchange is over: nothing is left to release. } - } catch (RuntimeException cleanup) { - e.addSuppressed(cleanup); } - throw e; } } - /** Starts a {@code 2026-07-28} client, and discovers the server's support of it. */ - private void startModern() { + /** + * Whether a failed {@code server/discover} marks a legacy server: a JSON-RPC error that isn't a + * recognized modern one, or a {@code 4xx} without a response. A header mismatch or a missing + * client capability error comes from a modern server, whose failure is kept. An unsupported + * version error comes from a modern server too, which is legacy-compatible only if it offers a + * legacy revision. + */ + private static boolean isLegacy(Throwable failure) { + if (failure instanceof io.modelcontextprotocol.modern.McpException mcp) { + var error = mcp.error(); + if (error != null && error.code() != null && MODERN_ERRORS.contains(error.code())) { + return false; + } + if (error != null + && error.code() != null + && error.code() == McpProtocol.UNSUPPORTED_PROTOCOL_VERSION) { + List supported = supportedVersions(error.data()); + if (offersLegacy(supported)) { + return true; + } + throw new IllegalStateException( + "No MCP revision supported by both the client and the peer: " + supported, + mcp); + } + return true; + } + return failure instanceof McpHttpStatusException status && status.isClientError(); + } + + private static List supportedVersions(Object data) { + if (data instanceof Map map && map.get("supported") instanceof List list) { + return list.stream().map(String::valueOf).toList(); + } + return List.of(); + } + + private static boolean offersLegacy(List supported) { + return supported.stream().anyMatch(McpProtocol.LEGACY_VERSIONS::contains); + } + + private McpModernClient createModern() { if (stream != null || target.startsWith("stdio:")) { throw new UnsupportedOperationException( "The MCP 2026-07-28 client supports HTTP only for now"); @@ -245,27 +477,29 @@ private void startModern() { "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), + return new McpModernClient( + new McpModernHttpTransport( + client, + target, mapper, - modernCapabilities, - inputHandler, - maxInputRounds); + maxMessageBytes, + maxConcurrentCalls, + requestTimeout, + requestCustomizer), + mapper, + modernCapabilities, + inputHandler, + maxInputRounds); + } + + /** Starts a {@code 2026-07-28} client, and discovers the server's support of it. */ + private void startModern(McpModernClient candidate) { try { - io.modelcontextprotocol.modern.McpSchema.DiscoverResult discovered = - candidate.discover().block(requestTimeout); + 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"); + throw new NotModern(); } } catch (RuntimeException e) { try { @@ -276,6 +510,7 @@ private void startModern() { throw e; } modern = candidate; + negotiatedVersion = McpProtocol.MODERN_VERSION; started = true; } @@ -291,10 +526,10 @@ public synchronized McpModernClient getModernClient() { return modern; } - private McpClientTransport createTransport() { + /** Starts a legacy client, over the stream, the stdio connector or HTTP. */ + private McpLegacyEdge.Client createLegacy(List versions) { if (stream != null) { - return new StdioMcpClientTransport( - stream, maxMessageBytes, maxConcurrentCalls, requestTimeout); + return McpLegacyEdge.stdio(stream, maxMessageBytes, maxConcurrentCalls, requestTimeout); } if (target.startsWith("stdio:")) { if (!client.isStarted() @@ -306,42 +541,121 @@ private McpClientTransport createTransport() { throw new IllegalStateException( "A started explicitly selected MCP STDIO helper is required"); } - return new BorrowedTransport( - helper.bind(maxMessageBytes, maxConcurrentCalls, requestTimeout)); + return helper.bind(maxMessageBytes, maxConcurrentCalls, requestTimeout); } - return new RestletMcpClientTransport( - client, - target, - maxMessageBytes, - maxConcurrentCalls, - requestTimeout, - requestCustomizer); + return McpLegacyEdge.http( + new McpLegacyEdge.HttpOptions( + client, + target, + maxMessageBytes, + maxConcurrentCalls, + requestTimeout, + requestCustomizer, + versions, + legacySessions, + legacySessionId)); } - /** 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()"); - } + public synchronized boolean isStarted() { + return started; + } + + private synchronized McpModernClient modernOrNull() { if (!started) { throw new IllegalStateException("MCP client is not started"); } - return sdk; - } - - public synchronized boolean isStarted() { - return started; + return modern; } + /** + * Lists tools, in either era. + * + * @param cursor The cursor of the page, or null for the first. + * @return The page of tools. + */ public McpSchema.ListToolsResult listTools(String cursor) { - return getSdkClient().listTools(cursor).block(requestTimeout); + McpModernClient current = modernOrNull(); + if (current != null) { + return current.listTools(cursor).block(requestTimeout); + } + return legacy().listTools(cursor).block(requestTimeout); } + /** + * Calls a tool, in either era. An {@code input_required} result the input handler doesn't + * answer fails the call. + * + * @param name The tool. + * @param arguments Its arguments. + * @return The tool's result. + */ public McpSchema.CallToolResult callTool(String name, Map arguments) { - return getSdkClient() - .callTool(new McpSchema.CallToolRequest(name, arguments, null)) - .block(requestTimeout); + McpModernClient current = modernOrNull(); + if (current == null) { + return legacy().callTool( + new McpSchema.CallToolRequest(name, arguments, null, null, null)) + .block(requestTimeout); + } + var outcome = + current.callTool(new McpSchema.CallToolRequest(name, arguments, null, null, null)) + .block(requestTimeout); + if (!(outcome instanceof McpSchema.CallToolResult result)) { + throw new IllegalStateException( + "MCP tool " + name + " requires input that was not provided"); + } + return result; + } + + /** + * Reads a resource, in either era. An {@code input_required} result the input handler doesn't + * answer fails the read. + * + * @param uri The resource URI. + * @return The resource's contents. + */ + public McpSchema.ReadResourceResult readResource(String uri) { + McpModernClient current = modernOrNull(); + if (current == null) { + return legacy().readResource(new McpSchema.ReadResourceRequest(uri, null, null, null)) + .block(requestTimeout); + } + var outcome = + current.readResource(new McpSchema.ReadResourceRequest(uri, null, null, null), null) + .block(requestTimeout); + if (!(outcome instanceof McpSchema.ReadResourceResult result)) { + throw new IllegalStateException( + "MCP resource " + uri + " requires input that was not provided"); + } + return result; + } + + /** + * Gets a prompt, in either era. An {@code input_required} result the input handler doesn't + * answer fails the request. + * + * @param name The prompt. + * @param arguments Its arguments. + * @return The prompt. + */ + public McpSchema.GetPromptResult getPrompt(String name, Map arguments) { + McpModernClient current = modernOrNull(); + var request = new McpSchema.GetPromptRequest(name, arguments, null, null, null); + if (current == null) { + return legacy().getPrompt(request).block(requestTimeout); + } + var outcome = current.getPrompt(request, null).block(requestTimeout); + if (!(outcome instanceof McpSchema.GetPromptResult result)) { + throw new IllegalStateException( + "MCP prompt " + name + " requires input that was not provided"); + } + return result; + } + + private synchronized McpLegacyEdge.Client legacy() { + if (!started || legacy == null) { + throw new IllegalStateException("MCP client is not started"); + } + return legacy; } public synchronized void stop() { @@ -355,10 +669,13 @@ public synchronized void stop() { modern = null; } } - if (sdk != null) { + if (legacy != null) { started = false; - sdk.closeGracefully().block(requestTimeout); - sdk = null; + try { + legacy.close(); + } finally { + legacy = null; + } } } @@ -366,37 +683,4 @@ public synchronized void stop() { public void close() { stop(); } - - private record BorrowedTransport(McpClientTransport delegate) implements McpClientTransport { - @Override - public Mono connect( - Function, Mono> handler) { - return delegate.connect(handler); - } - - @Override - public void setExceptionHandler(Consumer handler) { - delegate.setExceptionHandler(handler); - } - - @Override - public List protocolVersions() { - return delegate.protocolVersions(); - } - - @Override - public Mono sendMessage(McpSchema.JSONRPCMessage message) { - return delegate.sendMessage(message); - } - - @Override - public T unmarshalFrom(Object data, TypeRef type) { - return delegate.unmarshalFrom(data, type); - } - - @Override - public Mono closeGracefully() { - return Mono.empty(); - } - } } diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpInterceptor.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpInterceptor.java index 6b4971b930..b46f40f750 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpInterceptor.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpInterceptor.java @@ -8,18 +8,23 @@ */ package org.restlet.ext.mcp; -import io.modelcontextprotocol.spec.McpSchema; +import io.modelcontextprotocol.modern.JsonRpc; import reactor.core.publisher.Mono; -/** MCP-native envelope interception around the single SDK dispatcher. Notifications stay empty. */ +/** + * MCP-native envelope interception around the server, on the SDK's {@code 2026-07-28} message + * types. Requests of legacy clients are translated to those types before the interceptors run, so + * one interceptor serves both eras; {@link McpRequestContext#protocolVersion()} tells which one a + * request came in. Notifications stay empty. + */ @FunctionalInterface public interface McpInterceptor { - Mono intercept( - McpSchema.JSONRPCMessage message, McpRequestContext context, Chain chain); + Mono intercept( + JsonRpc.JSONRPCMessage message, McpRequestContext context, Chain chain); @FunctionalInterface interface Chain { - Mono proceed( - McpSchema.JSONRPCMessage message, McpRequestContext context); + Mono proceed( + JsonRpc.JSONRPCMessage message, McpRequestContext context); } } 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 7faadfe182..7daa3804bb 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 @@ -8,6 +8,8 @@ */ package org.restlet.ext.mcp; +import java.util.List; + /** * The explicitly supported, sessionless MCP profile. No Ikanos compatibility profile is implied. */ @@ -20,7 +22,25 @@ public final class McpProtocol { */ public static final String MODERN_VERSION = "2026-07-28"; + /** + * Makes {@link McpClientAdapter} speak whichever revision the server does, experimentally: over + * HTTP, {@link #MODERN_VERSION} when the server answers {@code server/discover}, else one of + * {@link #LEGACY_VERSIONS} through an {@code initialize} handshake. + */ + public static final String AUTO = "auto"; + + /** + * The handshake-based revisions an {@link #AUTO} client accepts from a legacy server, newest + * first. + */ + public static final List LEGACY_VERSIONS = + List.of(VERSION, "2025-06-18", "2025-03-26", "2024-11-05"); + + /** The JSON-RPC error code of {@code UnsupportedProtocolVersionError}, from 2026-07-28. */ + public static final int UNSUPPORTED_PROTOCOL_VERSION = -32022; + public static final String VERSION_HEADER = "MCP-Protocol-Version"; + public static final String SESSION_HEADER = "Mcp-Session-Id"; public static final String CONTEXT_KEY = "org.restlet.ext.mcp.requestContext"; public static final String STREAM_TRANSPORT_ATTRIBUTE = "org.restlet.ext.mcp.streamTransport"; public static final int DEFAULT_MAX_MESSAGE_BYTES = 1024 * 1024; diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResource.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpResource.java similarity index 64% rename from org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResource.java rename to org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpResource.java index b384b95afb..e0bb9c3457 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResource.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpResource.java @@ -13,22 +13,22 @@ import java.util.Objects; /** - * A resource served to MCP {@link McpProtocol#MODERN_VERSION} clients only, for {@link - * McpServerAdapter#getModernResources()}: its definition, listed, and its read handler, which may - * answer {@code input_required}. Experimental: the SDK's modern API is a preview. + * A resource of an {@link McpServerAdapter}, for {@link McpServerAdapter#getResources()}: its + * definition, listed, and its read handler, which may answer {@code input_required}. Legacy clients + * read it too, without input. Experimental: the SDK's modern API is a preview. * * @param definition The resource, with its URI. * @param resource The read handler. */ -public record McpModernResource(McpSchema.Resource definition, McpAsyncResource resource) { +public record McpResource(McpSchema.Resource definition, McpAsyncResource resource) { /** - * Creates a modern resource. + * Creates a resource. * * @param definition The resource, with its URI. * @param resource The read handler. */ - public McpModernResource { + public McpResource { Objects.requireNonNull(definition, "definition"); Objects.requireNonNull(resource, "resource"); } diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResourceTemplate.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpResourceTemplate.java similarity index 70% rename from org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResourceTemplate.java rename to org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpResourceTemplate.java index bc46823fb0..b1fe1d6adb 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpModernResourceTemplate.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpResourceTemplate.java @@ -13,23 +13,23 @@ import java.util.Objects; /** - * A resource template served to MCP {@link McpProtocol#MODERN_VERSION} clients only, for {@link - * McpServerAdapter#getModernResourceTemplates()}: its definition, listed, and the read handler of - * the URIs it matches. Experimental: the SDK's modern API is a preview. + * A resource template of an {@link McpServerAdapter}, for {@link + * McpServerAdapter#getResourceTemplates()}: its definition, listed, and the read handler of the + * URIs it matches. Experimental: the SDK's modern API is a preview. * * @param definition The template, with its URI template. * @param resource The read handler of the matching URIs. */ -public record McpModernResourceTemplate( +public record McpResourceTemplate( McpSchema.ResourceTemplate definition, McpAsyncResource resource) { /** - * Creates a modern resource template. + * Creates a resource template. * * @param definition The template, with its URI template. * @param resource The read handler of the matching URIs. */ - public McpModernResourceTemplate { + public McpResourceTemplate { Objects.requireNonNull(definition, "definition"); Objects.requireNonNull(resource, "resource"); } 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 b3f979b33f..4bff5ca171 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,18 +13,15 @@ import io.modelcontextprotocol.json.McpJsonDefaults; import io.modelcontextprotocol.json.McpJsonMapper; import io.modelcontextprotocol.json.jackson2.JacksonMcpJsonMapper; +import io.modelcontextprotocol.modern.JsonRpc; +import io.modelcontextprotocol.modern.McpSchema; import io.modelcontextprotocol.modern.server.HmacRequestStateCodec; +import io.modelcontextprotocol.modern.server.McpParamHeaders; +import io.modelcontextprotocol.modern.server.McpTransportResponse; import io.modelcontextprotocol.modern.server.feature.McpAsyncPrompt; import io.modelcontextprotocol.modern.server.feature.McpAsyncTool; import io.modelcontextprotocol.modern.server.feature.McpChangeBroadcaster; import io.modelcontextprotocol.modern.server.feature.McpChangeFeed; -import io.modelcontextprotocol.server.McpAsyncListFilter; -import io.modelcontextprotocol.server.McpServer; -import io.modelcontextprotocol.server.McpStatelessAsyncServer; -import io.modelcontextprotocol.server.McpStatelessServerFeatures; -import io.modelcontextprotocol.server.McpStatelessServerHandler; -import io.modelcontextprotocol.spec.McpSchema; -import io.modelcontextprotocol.spec.McpStatelessServerTransport; import java.io.IOException; import java.time.Duration; import java.util.List; @@ -46,6 +43,7 @@ import org.restlet.data.Method; import org.restlet.data.Status; import org.restlet.engine.header.PreferenceReader; +import org.restlet.ext.mcp.internal.McpLegacyEdge; import org.restlet.ext.mcp.internal.McpMessages; import org.restlet.ext.mcp.internal.McpModernExchange; import org.restlet.ext.mcp.internal.McpModernFeatures; @@ -55,11 +53,16 @@ import reactor.core.publisher.Sinks; /** - * Sessionless SDK MCP server on an existing Restlet routing chain. HTTP and STDIO share the same - * SDK dispatcher. Configure before start; feature collections are snapshotted at start. + * MCP server on an existing Restlet routing chain, over HTTP and STDIO, built on the MCP SDK's + * {@code 2026-07-28} stateless server. Configure before start; feature collections are snapshotted + * at start. * - *

Only the 2025-11-25 SDK profile is supported. Ikanos revision compatibility, sessions, - * listening GET streams, replay and server-initiated requests are deliberately not advertised. + *

Registrations, tool filters and interceptors all use the SDK's modern types. Requests of + * {@code 2025-11-25} clients are translated at a legacy edge into that model, so the same + * registrations serve both eras and policy runs once. The edge exposes nothing 2026-07-28 + * deprecates or removes: it advertises only tools, resources and prompts, issues no session, opens + * no {@code GET} stream and sends no request to the client; {@code ping} gets an empty result and + * {@code logging/setLevel} is not found. A result requiring input fails for a legacy client. */ public final class McpServerAdapter extends Restlet { private static final McpJsonMapper MAPPER = initMapper(); @@ -76,22 +79,12 @@ private static McpJsonMapper initMapper() { return mapper; } - private final List tools = - new CopyOnWriteArrayList<>(); - private final List resources = - new CopyOnWriteArrayList<>(); - private final List - resourceTemplates = new CopyOnWriteArrayList<>(); - private final List prompts = - new CopyOnWriteArrayList<>(); - private final List> toolFilters = - new CopyOnWriteArrayList<>(); + private final List tools = new CopyOnWriteArrayList<>(); + private final List resources = new CopyOnWriteArrayList<>(); + private final List resourceTemplates = new CopyOnWriteArrayList<>(); + private final List prompts = new CopyOnWriteArrayList<>(); + private final List toolFilters = new CopyOnWriteArrayList<>(); private final List interceptors = new CopyOnWriteArrayList<>(); - private final List modernTools = new CopyOnWriteArrayList<>(); - private final List modernResources = new CopyOnWriteArrayList<>(); - private final List modernResourceTemplates = - new CopyOnWriteArrayList<>(); - private final List modernPrompts = new CopyOnWriteArrayList<>(); private final McpChangeFeed changes = new McpChangeFeed(); private byte[] requestStateKey; private int maxListenStreams = McpProtocol.DEFAULT_MAX_LISTEN_STREAMS; @@ -102,10 +95,9 @@ private static McpJsonMapper initMapper() { private int maxMessageBytes = McpProtocol.DEFAULT_MAX_MESSAGE_BYTES; private int maxConcurrentCalls = McpProtocol.DEFAULT_MAX_CONCURRENT_CALLS; 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 McpLegacyEdge.Server legacyEdge; private volatile McpWorkers listeners; private volatile McpWorkers workers; private volatile boolean accepting; @@ -122,75 +114,61 @@ public McpServerAdapter(Context context) { super(context); } - public List getTools() { + /** + * The tools, served to both eras. A tool's {@code input_required} results and streamed progress + * reach {@code 2026-07-28} clients only: a legacy client gets an error instead of the former, + * and only the final result. Experimental: the SDK's modern API is a preview. + * + * @return The tools, modifiable until the adapter starts. + */ + public List getTools() { return tools; } - public List getResources() { + /** + * The resources, served to both eras; a URI may only be used once. Experimental. + * + * @return The resources, modifiable until the adapter starts. + */ + public List getResources() { return resources; } - public List - getResourceTemplates() { - return resourceTemplates; - } - - public List getPrompts() { - return prompts; - } - - public List> getToolFilters() { - return toolFilters; - } - - 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. + * The resource templates, served to both eras; a URI template may only be used once. + * Experimental. * - * @return The modern tools, modifiable until the adapter starts. + * @return The resource templates, modifiable until the adapter starts. */ - public List getModernTools() { - return modernTools; + public List getResourceTemplates() { + return resourceTemplates; } /** - * Resources served to MCP {@link McpProtocol#MODERN_VERSION} clients only, whose read may - * answer {@code input_required}. The resources of {@link #getResources()} are served to both - * revisions; a URI may only be used once across both lists. Experimental. + * The prompts, served to both eras; a name may only be used once. Experimental. * - * @return The modern resources, modifiable until the adapter starts. + * @return The prompts, modifiable until the adapter starts. */ - public List getModernResources() { - return modernResources; + public List getPrompts() { + return prompts; } /** - * Resource templates served to MCP {@link McpProtocol#MODERN_VERSION} clients only. The - * templates of {@link #getResourceTemplates()} are served to both revisions; a URI template may - * only be used once across both lists. Experimental. + * The filters deciding which tools a request lists, in both eras. * - * @return The modern resource templates, modifiable until the adapter starts. + * @return The filters, modifiable until the adapter starts. */ - public List getModernResourceTemplates() { - return modernResourceTemplates; + public List getToolFilters() { + return toolFilters; } /** - * Prompts served to MCP {@link McpProtocol#MODERN_VERSION} clients only, on the SDK's modern - * types. The prompts of {@link #getPrompts()} are served to both revisions; a name may only be - * used once across both lists. Experimental. + * The interceptors around every request, in both eras, on the SDK's modern types. * - * @return The modern prompts, modifiable until the adapter starts. + * @return The interceptors, modifiable until the adapter starts. */ - public List getModernPrompts() { - return modernPrompts; + public List getInterceptors() { + return interceptors; } /** @@ -314,46 +292,6 @@ public synchronized void start() throws Exception { 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); - } - if (!resources.isEmpty() || !resourceTemplates.isEmpty()) { - caps.resources(false, false); - } - if (!prompts.isEmpty()) { - caps.prompts(false); - } - var transport = - new McpStatelessServerTransport() { - @Override - public void setMcpHandler(McpStatelessServerHandler value) { - handler = value; - } - - @Override - public Mono closeGracefully() { - return Mono.empty(); - } - - @Override - public List protocolVersions() { - return List.of(McpProtocol.VERSION); - } - }; - sdkServer = - McpServer.async(transport) - .jsonMapper(MAPPER) - .serverInfo(serverInfo) - .instructions(instructions) - .capabilities(caps.build()) - .requestTimeout(requestTimeout) - .tools(List.copyOf(tools)) - .resources(List.copyOf(resources)) - .resourceTemplates(List.copyOf(resourceTemplates)) - .prompts(List.copyOf(prompts)) - .toolFilters(filters -> filters.addAll(toolFilters)) - .build(); workers = new McpWorkers(maxConcurrentCalls, requestTimeout); listeners = new McpWorkers(maxListenStreams, null, requestTimeout); stopping = Sinks.one(); @@ -368,6 +306,7 @@ public List protocolVersions() { maxMessageBytes, (message, context, terminal) -> invoke(0, message, context, terminal), () -> stopping.asMono()); + legacyEdge = McpLegacyEdge.server(MAPPER, this::call, maxMessageBytes, requestTimeout); runningInterceptors = List.copyOf(interceptors); runningOrigins = Set.copyOf(allowedOrigins); configured = true; @@ -375,30 +314,21 @@ public List protocolVersions() { accepting = true; } - /** Builds the SDK's modern server from the registrations, bridged and modern. */ + /** Builds the SDK's modern server from the registrations. */ 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)) + .serverInfo(serverInfo) .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), - List.copyOf(modernResources), - List.copyOf(modernResourceTemplates), - List.copyOf(modernPrompts)))) + List.copyOf(toolFilters)))) .subscriptions(changes); var codec = HmacRequestStateCodec.builder() @@ -455,31 +385,34 @@ public synchronized void stop() throws Exception { listeners.close(); } } finally { - try { - sdkServer.closeGracefully().block(requestTimeout); - } finally { - super.stop(); - } + super.stop(); } } /** - * Native SDK dispatcher shared with the transport paths; notifications return Optional.empty. + * Dispatches a {@code 2026-07-28} message through the interceptors to the SDK's modern server, + * within the call limit and the request timeout, keeping only the final response. A request + * needs the {@code _meta} the revision requires. Notifications return Optional.empty. + * + * @param message The message. + * @param context The request context. + * @return The final response, empty for a notification. */ - public Optional dispatch( - McpSchema.JSONRPCMessage message, McpRequestContext context) { + public Optional dispatch( + JsonRpc.JSONRPCMessage message, McpRequestContext context) { Objects.requireNonNull(message, "message"); Objects.requireNonNull(context, "context"); if (!accepting) { throw new IllegalStateException("MCP server is not accepting calls"); } - validateVersion(message); var completed = admit( context, - () -> invoke(0, message, context).blockOptional(requestTimeout), + () -> + invoke(0, message, context, this::terminal) + .blockOptional(requestTimeout), () -> {}); - return message instanceof McpSchema.JSONRPCNotification ? Optional.empty() : completed; + return message instanceof JsonRpc.JSONRPCNotification ? Optional.empty() : completed; } /** @@ -511,15 +444,18 @@ private Optional admit( .blockOptional(requestTimeout); } - private Mono invoke( - int index, McpSchema.JSONRPCMessage message, McpRequestContext context) { - return invoke(index, message, context, this::dispatchStateless); + /** + * Runs a modern request through the interceptors and the modern server, for the legacy edge. + */ + private Mono call( + JsonRpc.JSONRPCRequest request, McpRequestContext context) { + return invoke(0, request, context, this::terminal); } /** Runs the interceptors from {@code index}, then {@code terminal}. */ - private Mono invoke( + private Mono invoke( int index, - McpSchema.JSONRPCMessage message, + JsonRpc.JSONRPCMessage message, McpRequestContext context, McpInterceptor.Chain terminal) { if (index < runningInterceptors.size()) { @@ -533,30 +469,47 @@ private Mono invoke( 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) { + /** + * Dispatches a message without headers to the modern server, keeping only the final response of + * a stream. + */ + private Mono terminal( + JsonRpc.JSONRPCMessage message, McpRequestContext context) { McpTransportContext sdkContext = - McpTransportContext.create(Map.of(McpProtocol.CONTEXT_KEY, context)); - if (message instanceof McpSchema.JSONRPCRequest request) { - return handler.handleRequest(sdkContext, request); - } - if (message instanceof McpSchema.JSONRPCNotification notification) { - return handler.handleNotification(sdkContext, notification).then(Mono.empty()); + McpTransportContext.create( + Map.of( + McpProtocol.CONTEXT_KEY, + context, + McpParamHeaders.TRANSPORT_CONTEXT_KEY, + McpParamHeaders.NOT_MIRRORED)); + if (message instanceof JsonRpc.JSONRPCRequest request) { + return modernServer + .handleBlocking(sdkContext, request) + .flatMap( + reply -> + switch (reply) { + case McpTransportResponse.Result result -> + Mono.just(result.response()); + case McpTransportResponse.Error failure -> + Mono.just(failure.response()); + case McpTransportResponse.Streaming streaming -> + streaming + .messages() + .ofType(JsonRpc.JSONRPCResponse.class) + .last(); + default -> + Mono.error( + new IOException( + "MCP request got no response")); + }); + } + if (message instanceof JsonRpc.JSONRPCNotification notification) { + return modernServer.handleNotification(sdkContext, notification).then(Mono.empty()); } return Mono.error( new IllegalArgumentException("Only MCP requests and notifications are accepted")); } - private static void validateVersion(McpSchema.JSONRPCMessage message) { - if (message instanceof McpSchema.JSONRPCRequest request - && "initialize".equals(request.method()) - && request.params() instanceof Map params - && !McpProtocol.VERSION.equals(params.get("protocolVersion"))) { - throw new IllegalArgumentException("Unsupported MCP initialization version"); - } - } - @Override public void handle(Request request, Response response) { if (!accepting) { @@ -589,22 +542,30 @@ public void handle(Request request, Response response) { modernExchange.handle(request, response); return; } - var context = new McpRequestContext(request); - var parsed = new AtomicReference(); - Reply reply; + var context = + new McpRequestContext(request, version == null ? McpProtocol.VERSION : version); + // The id of the request once decoded, for a failure past the edge to answer. + var decoded = new AtomicReference<>(); + McpLegacyEdge.Reply reply; try { reply = admit( context, - () -> Optional.of(exchange(request, version, context, parsed)), + () -> + Optional.of( + legacyEdge.exchange( + request, + version, + context, + decoded::set)), () -> request.getEntity().release()) .orElseThrow(() -> new IllegalStateException("MCP exchange ended")); } catch (RuntimeException e) { getLogger().log(Level.WARNING, "MCP exchange failed", e); reply = - new Reply( + new McpLegacyEdge.Reply( Status.SERVER_ERROR_SERVICE_UNAVAILABLE, - error(parsed.get(), -32603, "MCP exchange unavailable")); + McpMessages.error(decoded.get(), -32603, "MCP exchange unavailable")); } response.setStatus(reply.status()); if (reply.body() != null) { @@ -670,69 +631,6 @@ private Status entityRejection(Request request) { return null; } - /** Reads, decodes and dispatches one HTTP message in an admitted worker. */ - private Reply exchange( - Request request, - String version, - McpRequestContext context, - AtomicReference parsed) { - McpSchema.JSONRPCMessage message; - try (var input = request.getEntity().getStream()) { - byte[] bytes = McpMessages.read(input, maxMessageBytes); - message = McpMessages.decode(MAPPER, McpMessages.utf8(bytes)); - } catch (McpMessages.MessageLimitException e) { - getLogger().log(Level.FINE, "Oversized MCP request", e); - return new Reply(Status.CLIENT_ERROR_REQUEST_ENTITY_TOO_LARGE, null); - } catch (McpMessages.InvalidEnvelopeException e) { - getLogger().log(Level.FINE, "Invalid MCP envelope", e); - return new Reply( - Status.CLIENT_ERROR_BAD_REQUEST, - McpMessages.error(null, -32600, "Invalid MCP request envelope")); - } catch (IOException | IllegalArgumentException e) { - getLogger().log(Level.FINE, "Invalid MCP request", e); - return new Reply( - Status.CLIENT_ERROR_BAD_REQUEST, - McpMessages.error(null, -32700, "Invalid MCP JSON request")); - } - if (request.getEntity().getSize() == 0) { - return new Reply( - Status.CLIENT_ERROR_BAD_REQUEST, - McpMessages.error(null, -32600, "Empty MCP request body")); - } - parsed.set(message); - if (version == null - && !(message instanceof McpSchema.JSONRPCRequest rpc - && "initialize".equals(rpc.method()))) { - return new Reply( - Status.CLIENT_ERROR_BAD_REQUEST, - error(message, -32600, "MCP-Protocol-Version is required")); - } - try { - validateVersion(message); - } catch (IllegalArgumentException e) { - getLogger().log(Level.FINE, "Unsupported MCP request", e); - return new Reply( - Status.CLIENT_ERROR_BAD_REQUEST, - error(message, -32600, "Unsupported MCP version")); - } - Optional result = - invoke(0, message, context).blockOptional(requestTimeout); - if (message instanceof McpSchema.JSONRPCNotification || result.isEmpty()) { - return new Reply(Status.SUCCESS_ACCEPTED, null); - } - return new Reply(Status.SUCCESS_OK, result.orElseThrow()); - } - - /** A JSON-RPC error answering a request, or null when the message expects no response. */ - private static Map error( - McpSchema.JSONRPCMessage message, int code, String text) { - return message instanceof McpSchema.JSONRPCRequest rpc - ? McpMessages.error(rpc.id(), code, text) - : null; - } - - private record Reply(Status status, Object body) {} - private void write(Response response, Object message) { try { byte[] bytes = McpMessages.encode(MAPPER, message, maxMessageBytes); diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpToolFilter.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpToolFilter.java new file mode 100644 index 0000000000..aeae06e4d1 --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/McpToolFilter.java @@ -0,0 +1,29 @@ +/** + * 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 reactor.core.publisher.Mono; + +/** + * Decides which tools a request lists, on the SDK's {@code 2026-07-28} types. Every filter of an + * {@link McpServerAdapter} applies, to the lists of both eras. + */ +@FunctionalInterface +public interface McpToolFilter { + + /** + * Whether a tool is listed for a request. + * + * @param context The request context. + * @param tool The tool's definition. + * @return True to list it. + */ + Mono isVisible(McpRequestContext context, McpSchema.Tool tool); +} diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/RestletMcpClientTransport.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/RestletMcpClientTransport.java deleted file mode 100644 index 0c641c9917..0000000000 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/RestletMcpClientTransport.java +++ /dev/null @@ -1,322 +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 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.InputStream; -import java.io.InputStreamReader; -import java.io.Reader; -import java.nio.charset.CodingErrorAction; -import java.nio.charset.StandardCharsets; -import java.time.Duration; -import java.util.List; -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 java.util.function.Function; -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.Reference; -import org.restlet.data.ServerSentEvent; -import org.restlet.ext.mcp.internal.McpMessages; -import org.restlet.ext.mcp.internal.McpWorkers; -import org.restlet.representation.ByteArrayRepresentation; -import org.restlet.representation.EventStreamReader; -import reactor.core.publisher.Mono; - -/** - * Native SDK transport using a borrowed, started Restlet HTTP connector. It never owns an HTTP - * engine or retries/reconnects a POST. The borrowed connector must disable redirects. - */ -public final class RestletMcpClientTransport implements McpClientTransport { - private final Client client; - private final String target; - private final McpJsonMapper mapper = McpJsonDefaults.getMapper(); - private final McpWorkers workers; - private final Set requests = ConcurrentHashMap.newKeySet(); - private final int maxMessageBytes; - private final Duration timeout; - private final Consumer customizer; - private volatile Function, Mono> - incoming; - private volatile boolean closed; - - public RestletMcpClientTransport(Client client, String target) { - this( - client, - target, - McpProtocol.DEFAULT_MAX_MESSAGE_BYTES, - McpProtocol.DEFAULT_MAX_CONCURRENT_CALLS, - Duration.ofSeconds(30), - request -> {}); - } - - public RestletMcpClientTransport( - Client client, - String target, - int maxMessageBytes, - int maxConcurrentCalls, - Duration timeout, - Consumer customizer) { - this.client = Objects.requireNonNull(client, "client"); - Reference reference = new Reference(target); - if (!"http".equals(reference.getScheme()) && !"https".equals(reference.getScheme())) { - throw new IllegalArgumentException("MCP HTTP target must use http or https"); - } - if (maxMessageBytes <= 0 || maxMessageBytes == Integer.MAX_VALUE) { - throw new IllegalArgumentException("Invalid MCP message byte limit"); - } - this.target = target; - this.maxMessageBytes = maxMessageBytes; - this.timeout = Objects.requireNonNull(timeout, "timeout"); - this.customizer = Objects.requireNonNull(customizer, "customizer"); - workers = new McpWorkers(maxConcurrentCalls, timeout); - } - - @Override - public synchronized Mono connect( - Function, Mono> handler) { - if (closed || incoming != null) { - return Mono.error(new IllegalStateException("MCP transport is single-use")); - } - if (!client.isStarted() - || client.getContext() == null - || !"false" - .equalsIgnoreCase( - client.getContext() - .getParameters() - .getFirstValue("followRedirects", "true"))) { - return Mono.error( - new IllegalStateException( - "Borrowed MCP client must be started with followRedirects=false")); - } - incoming = Objects.requireNonNull(handler, "handler"); - return Mono.empty(); - } - - @Override - public List protocolVersions() { - return List.of(McpProtocol.VERSION); - } - - @Override - public Mono sendMessage(McpSchema.JSONRPCMessage message) { - var active = new AtomicReference(); - return workers.submit( - () -> exchange(message, active), - () -> { - Request request = active.get(); - if (request != null) { - request.abort(); - } - }); - } - - private void exchange(McpSchema.JSONRPCMessage message, AtomicReference active) - throws IOException { - if (closed || incoming == null) { - throw new IOException("MCP transport is not connected"); - } - byte[] bytes = McpMessages.encode(mapper, message, maxMessageBytes); - Request request = - new Request( - Method.POST, - target, - new ByteArrayRepresentation(bytes, MediaType.APPLICATION_JSON)); - customizer.accept(request); - request.getHeaders().set(McpProtocol.VERSION_HEADER, McpProtocol.VERSION); - request.getClientInfo() - .setAcceptedMediaTypes( - List.of( - new Preference<>(MediaType.APPLICATION_JSON), - new Preference<>(MediaType.TEXT_EVENT_STREAM))); - active.set(request); - requests.add(request); - if (closed || Thread.currentThread().isInterrupted()) { - requests.remove(request); - request.abort(); - throw new IOException("MCP exchange cancelled"); - } - Response response = null; - try { - response = client.handle(request); - if (!response.getStatus().isSuccess()) { - throw new IOException( - "MCP HTTP failure: " + response.getStatus(), - response.getStatus().getThrowable()); - } - if (response.getHeaders().getFirstValue("Mcp-Session-Id", true) != null) { - throw new IOException( - "MCP session responses are not supported by the stateless profile"); - } - if (message instanceof McpSchema.JSONRPCNotification) { - if (response.getStatus().getCode() != 202 - && response.getStatus().getCode() != 204) { - throw new IOException( - "MCP notification must return an empty acceptance response"); - } - if (response.getEntity() != null - && response.getEntity().isAvailable() - && response.getEntity().getSize() != 0) { - throw new IOException("MCP notification unexpectedly returned an entity"); - } - return; - } - if (!(message instanceof McpSchema.JSONRPCRequest rpc)) { - throw new IOException( - "Server-directed responses are outside the stateless MCP profile"); - } - readResponse(response, rpc.id()); - } finally { - if (response != null && response.getEntity() != null) { - response.getEntity().release(); - } - requests.remove(request); - } - } - - private void readResponse(Response response, Object id) throws IOException { - if (response.getEntity() == null || response.getEntity().getMediaType() == null) { - throw new IOException("Missing MCP response entity or media type"); - } - MediaType mediaType = response.getEntity().getMediaType(); - try (InputStream input = - Objects.requireNonNull(response.getEntity().getStream(), "response stream")) { - if (MediaType.APPLICATION_JSON.isCompatible(mediaType)) { - if (!deliver( - McpSchema.deserializeJsonRpcMessage( - mapper, McpMessages.utf8(McpMessages.read(input, maxMessageBytes))), - id)) { - throw new IOException("MCP JSON response must contain the correlated response"); - } - } else if (MediaType.TEXT_EVENT_STREAM.isCompatible(mediaType)) { - readEvents(input, id); - } else { - throw new IOException("Unsupported MCP response media type: " + mediaType); - } - } - } - - private void readEvents(InputStream input, Object id) throws IOException { - Reader decoded = - new InputStreamReader( - input, - StandardCharsets.UTF_8 - .newDecoder() - .onMalformedInput(CodingErrorAction.REPORT)); - try (var events = new EventStreamReader(new LineBudgetReader(decoded, maxMessageBytes))) { - events.setMaxEventSize(maxMessageBytes); - ServerSentEvent event; - while ((event = events.readEvent()) != null) { - if (event.getData() != null - && !event.getData().isEmpty() - && deliver( - McpSchema.deserializeJsonRpcMessage(mapper, event.getData()), id)) { - return; - } - } - throw new IOException("MCP SSE ended before the correlated response"); - } - } - - private boolean deliver(McpSchema.JSONRPCMessage message, Object id) throws IOException { - if (message instanceof McpSchema.JSONRPCResponse rpc && !Objects.equals(id, rpc.id())) { - throw new IOException("MCP response id does not match its HTTP request"); - } - if (message instanceof McpSchema.JSONRPCRequest) { - throw new IOException("Server-initiated MCP requests are not supported"); - } - incoming.apply(Mono.just(message)).block(timeout); - return message instanceof McpSchema.JSONRPCResponse; - } - - @Override - public T unmarshalFrom(Object data, TypeRef type) { - return mapper.convertValue(data, type); - } - - @Override - public Mono closeGracefully() { - return Mono.fromCallable( - () -> { - closed = true; - requests.forEach(Request::abort); - workers.close(); - return Boolean.TRUE; - }) - .then(); - } - - /** - * Hands the event parser at most one line per read and charges the UTF-8 bytes of each line it - * takes. The parser only asks for a line until the correlated response is complete, so earlier - * comments count against the budget, while bytes decoded ahead of a complete response don't. - */ - private static final class LineBudgetReader extends Reader { - private final Reader input; - private final char[] buffer = new char[8192]; - private int position; - private int limit; - private long remaining; - - private LineBudgetReader(Reader input, int bytes) { - this.input = input; - remaining = bytes; - } - - @Override - public int read(char[] target, int offset, int length) throws IOException { - if (length == 0) { - return 0; - } - if (position == limit) { - position = 0; - limit = Math.max(0, input.read(buffer, 0, buffer.length)); - if (limit == 0) { - return -1; - } - } - int count = 0; - char c = 0; - while (count < length && position < limit && c != '\n' && c != '\r') { - c = buffer[position++]; - remaining -= utf8Length(c); - if (remaining < 0) { - throw new IOException("MCP response byte budget exceeded"); - } - target[offset + count++] = c; - } - return count; - } - - /** Surrogates count 2 bytes each, the 4 bytes of their supplementary character. */ - private static int utf8Length(char c) { - if (c < 0x80) { - return 1; - } - return c < 0x800 || Character.isSurrogate(c) ? 2 : 3; - } - - @Override - public void close() throws IOException { - input.close(); - } - } -} diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioClientHelper.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioClientHelper.java index 9a07dc6efb..dd53152b66 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioClientHelper.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/StdioClientHelper.java @@ -8,23 +8,24 @@ */ package org.restlet.ext.mcp; -import io.modelcontextprotocol.spec.McpClientTransport; import java.time.Duration; import org.restlet.Client; import org.restlet.Request; import org.restlet.Response; import org.restlet.data.Protocol; import org.restlet.engine.connector.ClientHelper; -import org.restlet.ext.mcp.internal.StdioMcpClientTransport; +import org.restlet.ext.mcp.internal.McpLegacyEdge; /** * Explicit MCP STDIO connector helper. Its owner supplies an unstarted StreamTransport in the - * connector context under STREAM_TRANSPORT_ATTRIBUTE and stops the connector after its SDK client. - * No executable is inferred from the stdio URI and no system stream is installed by default. + * connector context under STREAM_TRANSPORT_ATTRIBUTE and stops the connector after the {@link + * McpClientAdapter} using it. No executable is inferred from the stdio URI and no system stream is + * installed by default. The connector speaks MCP {@link McpProtocol#VERSION}, the modern client + * being HTTP only for now. */ public final class StdioClientHelper extends ClientHelper { static final String TRANSPORT_ATTRIBUTE = "org.restlet.ext.mcp.clientTransport"; - private StdioMcpClientTransport transport; + private McpLegacyEdge.StdioTransport transport; public StdioClientHelper(Client client) { super(client); @@ -38,35 +39,26 @@ public synchronized void start() { throw new IllegalStateException( "MCP STDIO requires an explicitly configured StreamTransport"); } - transport = - new StdioMcpClientTransport( - stream, - McpProtocol.DEFAULT_MAX_MESSAGE_BYTES, - McpProtocol.DEFAULT_MAX_CONCURRENT_CALLS, - Duration.ofSeconds(30)); + transport = McpLegacyEdge.stdioTransport(stream); getContext().getAttributes().put(TRANSPORT_ATTRIBUTE, this); } - /** Applies an MCP client's limits to this connector's transport before it connects. */ - synchronized McpClientTransport bind(int maxMessageBytes, int maxCalls, Duration timeout) { - StdioMcpClientTransport bound = (StdioMcpClientTransport) getTransport(); - bound.setLimits(maxMessageBytes, maxCalls, timeout); - return bound; - } - - /** Returns the native SDK transport owned by this connector. One SDK client may bind it. */ - public synchronized McpClientTransport getTransport() { + /** + * Starts a client on this connector's transport, with an MCP client's limits. One client may + * bind it. + */ + synchronized McpLegacyEdge.Client bind(int maxMessageBytes, int maxCalls, Duration timeout) { if (transport == null) { throw new IllegalStateException("Start the MCP STDIO client connector first"); } - return transport; + return McpLegacyEdge.stdio(transport, maxMessageBytes, maxCalls, timeout); } @Override public synchronized void stop() { try { if (transport != null) { - transport.closeGracefully().block(Duration.ofSeconds(30)); + transport.close(Duration.ofSeconds(30)); } } finally { getContext().getAttributes().remove(TRANSPORT_ATTRIBUTE, this); @@ -75,7 +67,6 @@ public synchronized void stop() { @Override public void handle(Request request, Response response) { - throw new UnsupportedOperationException( - "Use McpClientAdapter or the native MCP SDK on this helper's transport"); + throw new UnsupportedOperationException("Use McpClientAdapter on this helper's transport"); } } diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpEras.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpEras.java new file mode 100644 index 0000000000..b6e4d5ef75 --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpEras.java @@ -0,0 +1,70 @@ +/** + * 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 java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * The era each MCP HTTP endpoint was found to speak, modern or legacy, as the 2026-07-28 revision + * recommends a dual-era client remember it. It is kept by endpoint URI rather than by origin, since + * one origin may host servers of both eras on different paths, and only for this JVM. A client + * starting on a cached era that the server contradicts forgets it and probes again. + */ +public final class McpEras { + /** An endpoint answering {@code server/discover}. */ + public static final String MODERN = "modern"; + + /** An endpoint expecting an {@code initialize} handshake. */ + public static final String LEGACY = "legacy"; + + /** The most endpoints remembered: past it, the cache starts over. */ + static final int MAX_ENTRIES = 1024; + + private static final Map ERAS = new ConcurrentHashMap<>(); + + private McpEras() {} + + /** + * The era last found for an endpoint. + * + * @param target The endpoint URI. + * @return {@link #MODERN}, {@link #LEGACY}, or null if unknown. + */ + public static String get(String target) { + return ERAS.get(target); + } + + /** + * Remembers the era of an endpoint. + * + * @param target The endpoint URI. + * @param era {@link #MODERN} or {@link #LEGACY}. + */ + public static void put(String target, String era) { + if (ERAS.size() >= MAX_ENTRIES && !ERAS.containsKey(target)) { + ERAS.clear(); + } + ERAS.put(target, era); + } + + /** + * Forgets the era of an endpoint. + * + * @param target The endpoint URI. + */ + public static void remove(String target) { + ERAS.remove(target); + } + + /** Forgets every endpoint. */ + public static void clear() { + ERAS.clear(); + } +} diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpHttpStatusException.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpHttpStatusException.java new file mode 100644 index 0000000000..860b826953 --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpHttpStatusException.java @@ -0,0 +1,49 @@ +/** + * 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 java.io.IOException; + +/** + * An MCP HTTP exchange answered without a JSON-RPC response to its request, with the HTTP status + * code it got. A dual-era client tells a legacy server by a {@code 4xx} of this kind. + */ +public final class McpHttpStatusException extends IOException { + private static final long serialVersionUID = 1L; + + private final int status; + + /** + * @param message The message. + * @param status The HTTP status code. + * @param cause The cause, or null. + */ + public McpHttpStatusException(String message, int status, Throwable cause) { + super(message, cause); + this.status = status; + } + + /** + * The HTTP status code. + * + * @return The code. + */ + public int getStatus() { + return status; + } + + /** + * Whether the status is a client error, {@code 4xx}. + * + * @return True if it is. + */ + public boolean isClientError() { + return status >= 400 && status < 500; + } +} diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpLegacyEdge.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpLegacyEdge.java new file mode 100644 index 0000000000..2e132b9ed4 --- /dev/null +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpLegacyEdge.java @@ -0,0 +1,241 @@ +/** + * 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 io.modelcontextprotocol.modern.McpSchema; +import java.time.Duration; +import java.util.List; +import java.util.function.Consumer; +import org.restlet.Request; +import org.restlet.data.Status; +import org.restlet.ext.mcp.McpRequestContext; +import org.restlet.ext.mcp.internal.legacy.McpLegacyClient; +import org.restlet.ext.mcp.internal.legacy.McpLegacyServer; +import org.restlet.ext.mcp.internal.legacy.StdioMcpClientTransport; +import org.restlet.util.StreamTransport; +import reactor.core.publisher.Mono; + +/** + * The one seam between the modern core of the MCP extension and the legacy edge in {@code + * internal.legacy}, which translates MCP {@code 2025-11-25} and earlier into the {@code 2026-07-28} + * model and back. Nothing else references the legacy package, so dropping legacy support means + * deleting that package and this interface. + * + *

The edge exposes nothing 2026-07-28 deprecates or removes: no logging, sampling or roots, no + * HTTP+SSE transport, no listening {@code GET} stream or resumption, and no server-issued session. + * A client may only send a legacy server's session id when its caller opts in. + */ +public interface McpLegacyEdge { + + /** The modern core a legacy server edge translates requests into. */ + @FunctionalInterface + interface Core { + + /** + * Runs a {@code 2026-07-28} request through the interceptors and the modern server, keeping + * only its final response. + * + * @param request The request, with the {@code _meta} the revision requires. + * @param context The request context, naming the legacy revision. + * @return The final response. + */ + Mono call( + JsonRpc.JSONRPCRequest request, McpRequestContext context); + } + + /** + * The reply to a legacy HTTP exchange. + * + * @param status The HTTP status. + * @param body The JSON-RPC message to write, or null for none. + */ + record Reply(Status status, Object body) {} + + /** The server edge, answering legacy clients from the modern core. */ + @FunctionalInterface + interface Server { + + /** + * Reads, decodes and answers one legacy exchange, in an admitted worker. + * + * @param request The request, whose headers the adapter already checked. + * @param version The {@code MCP-Protocol-Version} header, or null. + * @param context The request context. + * @param decoded Given the id of the request once decoded, so that a failure the adapter + * answers past the edge, such as a timeout, still names it. + * @return The reply. + */ + Reply exchange( + Request request, + String version, + McpRequestContext context, + Consumer decoded); + } + + /** + * A started client of a legacy server, on the SDK's modern types: legacy results are converted + * at the edge, a missing {@code resultType} meaning complete. + */ + interface Client { + + /** + * The revision the last {@code initialize} handshake negotiated, which may change when a + * session expires and the client initializes again. + * + * @return The revision. + */ + String negotiatedVersion(); + + /** + * The session id sent with each request, if any. + * + * @return The session id, or null. + */ + String sessionId(); + + Mono listTools(String cursor); + + Mono callTool(McpSchema.CallToolRequest request); + + Mono listResources(String cursor); + + Mono listResourceTemplates(String cursor); + + Mono readResource(McpSchema.ReadResourceRequest request); + + Mono listPrompts(String cursor); + + Mono getPrompt(McpSchema.GetPromptRequest request); + + /** Ends the client, and the session it opened, if any. */ + void close(); + } + + /** + * The options of a legacy HTTP client. + * + * @param client The borrowed, started connector, not following redirects. + * @param target The endpoint URI. + * @param maxMessageBytes The largest message. + * @param maxConcurrentCalls The most exchanges at once. + * @param timeout The request timeout. + * @param customizer Customizes each request. + * @param versions The revisions offered, newest first. + * @param sessions Whether the session id a server returns on {@code initialize} is kept, sent + * and finally deleted. + * @param sessionId A session id the caller owns, sent but never deleted, or null. + */ + record HttpOptions( + org.restlet.Client client, + String target, + int maxMessageBytes, + int maxConcurrentCalls, + Duration timeout, + Consumer customizer, + List versions, + boolean sessions, + String sessionId) {} + + /** The transport a stdio client connector owns, for one legacy client to bind. */ + interface StdioTransport { + + /** Applies a client's limits, before it connects. */ + void setLimits(int maxMessageBytes, int maxConcurrentCalls, Duration timeout); + + /** Ends the transport. */ + void close(Duration timeout); + } + + /** + * Creates the server edge of an adapter. + * + * @param mapper The JSON mapper. + * @param core The modern core. + * @param maxMessageBytes The largest request or response message. + * @param timeout The request timeout. + * @return The server edge. + */ + static Server server(McpJsonMapper mapper, Core core, int maxMessageBytes, Duration timeout) { + return new McpLegacyServer(mapper, core, maxMessageBytes, timeout); + } + + /** + * Whether a legacy client's failed start proves the server isn't a legacy one: a {@code 4xx} or + * a JSON-RPC error answering {@code initialize}, unlike a timeout or a {@code 5xx}. + * + * @param failure The failure. + * @return True if it does. + */ + static boolean wrongEra(Throwable failure) { + return McpLegacyClient.wrongEra(failure); + } + + /** + * Starts a legacy client over HTTP: an {@code initialize} handshake offering the options' + * revisions. + * + * @param options The options. + * @return The started client. + */ + static Client http(HttpOptions options) { + return McpLegacyClient.http(options); + } + + /** + * Starts a legacy client over a stream it owns. + * + * @param stream The stream. + * @param maxMessageBytes The largest message. + * @param maxConcurrentCalls The most exchanges at once. + * @param timeout The request timeout. + * @return The started client. + */ + static Client stdio( + StreamTransport stream, int maxMessageBytes, int maxConcurrentCalls, Duration timeout) { + return McpLegacyClient.stdio( + new StdioMcpClientTransport(stream, maxMessageBytes, maxConcurrentCalls, timeout), + true, + timeout); + } + + /** + * Starts a legacy client over the transport a stdio connector owns, which stays open when the + * client closes. + * + * @param transport The connector's transport. + * @param maxMessageBytes The largest message. + * @param maxConcurrentCalls The most exchanges at once. + * @param timeout The request timeout. + * @return The started client. + */ + static Client stdio( + StdioTransport transport, + int maxMessageBytes, + int maxConcurrentCalls, + Duration timeout) { + transport.setLimits(maxMessageBytes, maxConcurrentCalls, timeout); + return McpLegacyClient.stdio((StdioMcpClientTransport) transport, false, timeout); + } + + /** + * Creates the transport of a stdio client connector. + * + * @param stream The stream. + * @return The transport. + */ + static StdioTransport stdioTransport(StreamTransport stream) { + return new StdioMcpClientTransport( + stream, + org.restlet.ext.mcp.McpProtocol.DEFAULT_MAX_MESSAGE_BYTES, + org.restlet.ext.mcp.McpProtocol.DEFAULT_MAX_CONCURRENT_CALLS, + Duration.ofSeconds(30)); + } +} 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 471061f841..b6bae6ec71 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 @@ -9,8 +9,6 @@ package org.restlet.ext.mcp.internal; import io.modelcontextprotocol.json.McpJsonMapper; -import io.modelcontextprotocol.json.TypeRef; -import io.modelcontextprotocol.spec.McpSchema; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; @@ -53,25 +51,6 @@ public static byte[] encode(McpJsonMapper mapper, Object message, int limit) return bytes; } - public static McpSchema.JSONRPCMessage decode(McpJsonMapper mapper, String json) - throws IOException { - Object decoded = mapper.readValue(json, new TypeRef() {}); - if (!(decoded instanceof Map envelope) - || !"2.0".equals(envelope.get("jsonrpc")) - || !(envelope.get("method") instanceof String method) - || method.isBlank() - || envelope.containsKey("params") - && envelope.get("params") != null - && !(envelope.get("params") instanceof Map) - || envelope.containsKey("id") - && !(envelope.get("id") instanceof String - || envelope.get("id") instanceof Number)) { - throw new InvalidEnvelopeException("Invalid MCP request or notification envelope"); - } - - return McpSchema.deserializeJsonRpcMessage(mapper, json); - } - public static final class MessageLimitException extends IOException { private static final long serialVersionUID = 1L; diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernExchange.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernExchange.java index 1aa4941f92..08848295c5 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernExchange.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernExchange.java @@ -18,7 +18,6 @@ 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; @@ -61,11 +60,10 @@ * 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. + * adapter's interceptors wrap the request and the final result or error. 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 { @@ -76,13 +74,13 @@ public interface Interceptors { /** * Runs the interceptor chain. * - * @param message The message, on the {@code 2025-11-25} types. + * @param message The message. * @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, + Mono run( + JsonRpc.JSONRPCMessage message, McpRequestContext context, McpInterceptor.Chain terminal); } @@ -250,7 +248,7 @@ private Exchange(Request request, McpRequestContext context, boolean listen) { } /** Waits for a chain's response, within the request timeout unless listening. */ - private Optional await(Mono chain) { + private Optional await(Mono chain) { return listen ? chain.blockOptional() : chain.blockOptional(timeout); } @@ -291,15 +289,16 @@ private void run() throws IOException, InterruptedException { private void notification(JsonRpc.JSONRPCNotification notification) { interceptors .run( - mapper.convertValue(notification, McpSchema.JSONRPCNotification.class), + notification, context, (next, nextContext) -> - sdk.handleNotification( - transport(nextContext), - mapper.convertValue( - next, - JsonRpc.JSONRPCNotification.class)) - .then(Mono.empty())) + next instanceof JsonRpc.JSONRPCNotification forwarded + ? sdk.handleNotification( + transport(nextContext), forwarded) + .then(Mono.empty()) + : Mono.error( + new IllegalArgumentException( + "An MCP notification must stay one"))) .blockOptional(timeout); head.complete(new Reply(Status.SUCCESS_ACCEPTED, null, false)); } @@ -311,10 +310,10 @@ private void call(JsonRpc.JSONRPCRequest rpc) throws InterruptedException { head.complete(error(rpc.id(), ErrorCodes.HEADER_MISMATCH, mismatch)); return; } - Optional last = + Optional last = await( interceptors.run( - mapper.convertValue(rpc, McpSchema.JSONRPCRequest.class), + rpc, context, (next, nextContext) -> Mono.fromCallable(() -> terminal(next, nextContext)))); @@ -342,12 +341,14 @@ private void call(JsonRpc.JSONRPCRequest rpc) throws InterruptedException { } /** Dispatches to the SDK after the interceptors, streaming what comes before the result. */ - private McpSchema.JSONRPCResponse terminal( - McpSchema.JSONRPCMessage next, McpRequestContext forwarded) + private JsonRpc.JSONRPCResponse terminal( + JsonRpc.JSONRPCMessage next, McpRequestContext forwarded) throws IOException, InterruptedException { - JsonRpc.JSONRPCRequest rpc = mapper.convertValue(next, JsonRpc.JSONRPCRequest.class); + if (!(next instanceof JsonRpc.JSONRPCRequest rpc)) { + throw new IOException("An MCP request must stay one"); + } McpTransportContext sdkContext = transport(forwarded); - if (McpSchema.METHOD_TOOLS_CALL.equals(rpc.method())) { + if (io.modelcontextprotocol.modern.McpSchema.METHOD_TOOLS_CALL.equals(rpc.method())) { // The tools feature validates x-mcp-header arguments against these. sdkContext = sdkContext.putAll( @@ -362,10 +363,10 @@ private McpSchema.JSONRPCResponse terminal( ? sdk.handle(sdkContext, rpc).block(timeout) : sdk.handleBlocking(sdkContext, rpc).block(timeout); if (reply instanceof McpTransportResponse.Result result) { - return classic(result.response()); + return result.response(); } if (reply instanceof McpTransportResponse.Error failure) { - return classic(failure.response()); + return failure.response(); } if (!(reply instanceof McpTransportResponse.Streaming streaming)) { throw new IOException("MCP request got no response"); @@ -386,11 +387,7 @@ private McpSchema.JSONRPCResponse terminal( 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); + return last; } /** The servlet transport's header checks, on Restlet headers. */ @@ -539,7 +536,7 @@ private static Reply json(Status status, byte[] body) { return new Reply(status, body, false); } - private static Status status(McpSchema.JSONRPCResponse response) { + private static Status status(JsonRpc.JSONRPCResponse response) { return response.error() == null ? Status.SUCCESS_OK : ERROR_STATUSES.getOrDefault(response.error().code(), Status.SUCCESS_OK); diff --git a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernFeatures.java b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernFeatures.java index 0ed6612dfb..6a1a596c1d 100644 --- a/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernFeatures.java +++ b/org.restlet.ext.mcp/src/main/java/org/restlet/ext/mcp/internal/McpModernFeatures.java @@ -8,10 +8,7 @@ */ package org.restlet.ext.mcp.internal; -import io.modelcontextprotocol.common.McpTransportContext; -import io.modelcontextprotocol.json.McpJsonMapper; import io.modelcontextprotocol.modern.McpSchema; -import io.modelcontextprotocol.modern.server.McpAsyncResponse; import io.modelcontextprotocol.modern.server.McpFeature; import io.modelcontextprotocol.modern.server.McpRequestContext; import io.modelcontextprotocol.modern.server.feature.McpAsyncPrompt; @@ -27,25 +24,22 @@ 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 java.util.function.BiFunction; -import org.restlet.ext.mcp.McpModernResource; -import org.restlet.ext.mcp.McpModernResourceTemplate; +import org.restlet.ext.mcp.McpProtocol; +import org.restlet.ext.mcp.McpResource; +import org.restlet.ext.mcp.McpResourceTemplate; +import org.restlet.ext.mcp.McpToolFilter; 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 registrations are served as registered. List filters apply to - * every listed tool. A name, URI or URI template may only be registered once across both kinds. + * The features of the MCP server, built from the adapter's registrations, on the SDK's modern + * types. Tool filters apply to every listed tool, with the adapter's request context. A name, URI + * or URI template may only be registered once. */ public final class McpModernFeatures { @@ -54,109 +48,76 @@ 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 tools The tools. + * @param resources The resources. + * @param resourceTemplates The resource templates. + * @param prompts The prompts. * @param toolFilters The tool list filters. - * @param modernTools The {@code 2026-07-28} tools. - * @param modernResources The {@code 2026-07-28} resources. - * @param modernResourceTemplates The {@code 2026-07-28} resource templates. - * @param modernPrompts The {@code 2026-07-28} prompts. */ public record Registrations( - List tools, - List resources, - List resourceTemplates, - List prompts, - List> toolFilters, - List modernTools, - List modernResources, - List modernResourceTemplates, - List modernPrompts) {} + List tools, + List resources, + List resourceTemplates, + List prompts, + List toolFilters) {} /** * 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 a name, URI or URI template is registered twice. */ - public static List of(McpJsonMapper mapper, Registrations registrations) { + public static List of(Registrations registrations) { List features = new ArrayList<>(); - if (!registrations.tools().isEmpty() || !registrations.modernTools().isEmpty()) { - features.add(ToolsFeature.ofAsync(new Tools(mapper, registrations))); + if (!registrations.tools().isEmpty()) { + features.add(ToolsFeature.ofAsync(new Tools(registrations))); } - if (!registrations.resources().isEmpty() - || !registrations.resourceTemplates().isEmpty() - || !registrations.modernResources().isEmpty() - || !registrations.modernResourceTemplates().isEmpty()) { - features.add(ResourcesFeature.ofAsync(new Resources(mapper, registrations))); + if (!registrations.resources().isEmpty() || !registrations.resourceTemplates().isEmpty()) { + features.add(ResourcesFeature.ofAsync(new Resources(registrations))); } - if (!registrations.prompts().isEmpty() || !registrations.modernPrompts().isEmpty()) { - features.add(PromptsFeature.ofAsync(new Prompts(mapper, registrations))); + if (!registrations.prompts().isEmpty()) { + features.add(PromptsFeature.ofAsync(new Prompts(registrations))); } return features; } - /** The tools, bridged and modern, indexed by name. */ + /** The adapter's context of a request, or null when the transport gave none. */ + private static org.restlet.ext.mcp.McpRequestContext context(McpRequestContext ctx) { + return ctx.transportContext().get(McpProtocol.CONTEXT_KEY) + instanceof org.restlet.ext.mcp.McpRequestContext context + ? context + : null; + } + + /** The tools, 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 final List filters; - 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 Tools(Registrations registrations) { + this.filters = List.copyOf(registrations.toolFilters()); + for (McpAsyncTool tool : registrations.tools()) { + 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) { + org.restlet.ext.mcp.McpRequestContext context = context(ctx); return Flux.fromIterable(byName.values()) .map(McpAsyncTool::definition) .filterWhen( tool -> - filter.isVisible( - ctx.transportContext(), - mapper.convertValue( - tool, - io.modelcontextprotocol.spec.McpSchema.Tool - .class))) + Flux.fromIterable(filters) + .concatMap( + filter -> + filter.isVisible(context, tool) + .defaultIfEmpty(Boolean.FALSE)) + .all(Boolean::booleanValue)) .collectList() .map(ToolsPage::of); } @@ -174,34 +135,19 @@ private record Template( McpAsyncResource resource) {} /** - * The resources, bridged and modern: exact URIs first, then templates, in registration order. - * Subscriptions are supported: the adapter publishes resource updates through its change feed. + * The resources: exact URIs first, then templates, in registration order. Subscriptions are + * supported: the adapter publishes resource updates through its change feed. */ private static final class Resources implements McpAsyncResourceRepository { - private final McpJsonMapper mapper; private final List listed = new ArrayList<>(); private final Map byUri = new LinkedHashMap<>(); private final List