001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.juneau.rest.client.mcp;
018
019import static org.apache.juneau.commons.utils.Shorts.*;
020
021import java.io.*;
022import java.net.URI;
023import java.net.URISyntaxException;
024import java.util.Locale;
025import java.util.Map;
026import java.util.Set;
027
028import org.apache.juneau.bean.jsonrpc.*;
029import org.apache.juneau.marshall.json.*;
030import org.apache.juneau.marshall.sse.*;
031import org.apache.juneau.rest.client.*;
032
033/**
034 * Revision-neutral base class for Model Context Protocol (MCP) clients.
035 *
036 * <p>
037 * Wraps a single configured {@link RestClient} and executes raw JSON-RPC 2.0 request/response round trips against one
038 * endpoint in JSON response mode. A generic {@link RestCallInterceptor} passthrough is provided on the builder so
039 * per-revision facades can layer auth and other cross-cutting concerns without changing this core.
040 *
041 * <p>
042 * {@link #send(JsonRpcRequest)} returns the complete {@link JsonRpcResponse} envelope untouched, including
043 * {@code error} responses - it does not throw {@link McpException} on a JSON-RPC error. Typed facades are
044 * responsible for translating a response with a non-<jk>null</jk> {@link JsonRpcResponse#getError() error} into a
045 * thrown {@link McpException}.
046 *
047 * <p>
048 * Instances are thread-safe for concurrent reuse: {@link RestClient} and endpoint are final, and each
049 * {@link #send(JsonRpcRequest)} creates a fresh request/response scope. Builders are single-use value assemblers:
050 * calling {@code build()} twice creates two clients and both use the same explicitly configured transport instance.
051 *
052 * @since 10.0.0
053 */
054@SuppressWarnings({
055   "resource" // The RestClient is built once and held in a field that close() closes.
056})
057public abstract class AbstractMcpClient implements Closeable {
058
059   // JSON-RPC methods that are side-effect-free (pure queries), so a pre-response stale-connection failure can be
060   // safely replayed once.  Kept as revision-neutral protocol literals rather than a dependency on any dated
061   // McpMethods class.  Any method NOT listed here — notably tools/call, subscriptions/listen, sampling/createMessage,
062   // notifications/*, and any unknown method — is never replayed, so a mutating operation can never double-execute.
063   private static final Set<String> RETRYABLE_METHODS = Set.of(
064      "initialize",
065      "server/discover",
066      "ping",
067      "tools/list",
068      "prompts/list",
069      "prompts/get",
070      "resources/list",
071      "resources/read",
072      "resources/templates/list",
073      "completion/complete");
074
075   private final RestClient restClient;
076   private final String endpoint;
077
078   /**
079    * Constructor.
080    *
081    * @param builder The builder supplying this client's configuration. Must not be <jk>null</jk>.
082    * @throws IllegalArgumentException If {@code builder} is <jk>null</jk>, or its endpoint is <jk>null</jk>/blank
083    *    or is not a syntactically valid absolute {@code http}/{@code https} URL (see
084    *    {@link #validateEndpoint(String)}).
085    */
086   protected AbstractMcpClient(Builder<?> builder) {
087      reqnn("builder", builder);
088      this.endpoint = validateEndpoint(builder.endpoint);
089      this.restClient = builder.restClientBuilder
090         .defaultSerializer(JsonSerializer.DEFAULT)
091         .defaultParser(JsonParser.DEFAULT)
092         .build();
093   }
094
095   /**
096    * Validates the configured endpoint: non-<jk>null</jk>/non-blank (as before), and additionally that it
097    * parses as a syntactically valid, absolute URL using the {@code http} or {@code https} scheme.
098    *
099    * <p>
100    * Rejecting a malformed or non-http(s) endpoint here, at construction, is preferable to letting it surface
101    * later as an opaque transport failure on the first {@link #send(JsonRpcRequest)} call.
102    *
103    * @param value The candidate endpoint. Must not be <jk>null</jk>/blank.
104    * @return {@code value}, unchanged, once validated.
105    * @throws IllegalArgumentException If {@code value} is <jk>null</jk>/blank, is not a syntactically valid URI,
106    *    has no scheme, or has a scheme other than {@code http}/{@code https} (case-insensitive).
107    */
108   private static String validateEndpoint(String value) {
109      var endpointValue = reqnb("endpoint", value);
110      URI uri;
111      try {
112         uri = new URI(endpointValue);
113      } catch (URISyntaxException e) {
114         throw new IllegalArgumentException("Invalid MCP endpoint URL '" + endpointValue + "': " + e.getMessage(), e);
115      }
116      var scheme = uri.getScheme();
117      if (scheme == null || ! (eqic(scheme, "http") || eqic(scheme, "https")))
118         throw new IllegalArgumentException(
119            "Invalid MCP endpoint URL '" + endpointValue + "': must be an absolute http or https URL, but scheme was '" + scheme + "'.");
120      return endpointValue;
121   }
122
123   /**
124    * Sends a raw JSON-RPC request and returns the raw JSON-RPC response envelope.
125    *
126    * <p>
127    * A request with a <jk>null</jk> {@link JsonRpcRequest#getId() id} is a notification: this method still sends
128    * it over the wire, but always returns <jk>null</jk> without attempting to read a response body, per the
129    * JSON-RPC 2.0 notification contract (the server sends no response body for a notification).
130    *
131    * @param request The JSON-RPC request to send. Must not be <jk>null</jk>.
132    * @return The JSON-RPC response envelope, or <jk>null</jk> if {@code request} is a notification.
133    * @throws IOException If a transport-level or (de)serialization error occurs.
134    */
135   public JsonRpcResponse send(JsonRpcRequest request) throws IOException {
136      return send(request, Map.of());
137   }
138
139   /**
140    * Sends a raw JSON-RPC request and returns the raw JSON-RPC response envelope.
141    *
142    * <p>
143    * A request with a <jk>null</jk> {@link JsonRpcRequest#getId() id} is a notification: this method still sends
144    * it over the wire, but always returns <jk>null</jk> without attempting to read a response body, per the
145    * JSON-RPC 2.0 notification contract (the server sends no response body for a notification).
146    *
147    * <p>
148    * <b>Note:</b> for a notification, this method returns <jk>null</jk> immediately after the transport call
149    * completes, without ever inspecting {@link RestResponse#getStatusCode()}. A non-2xx HTTP status on a
150    * notification is therefore silently swallowed - the caller sees no indication that the server rejected it.
151    * This is intentional per the notification contract above (there is no response envelope to report an error
152    * through), but it does mean transport-level failures on notifications are otherwise invisible to callers.
153    *
154    * @param request The JSON-RPC request to send. Must not be <jk>null</jk>.
155    * @param httpHeaders Optional additional HTTP headers to add on this request. Can be <jk>null</jk>.
156    * @return The JSON-RPC response envelope, or <jk>null</jk> if {@code request} is a notification.
157    * @throws IOException If a transport-level or (de)serialization error occurs.
158    */
159   public JsonRpcResponse send(JsonRpcRequest request, Map<String,String> httpHeaders) throws IOException {
160      reqnn("request", request);
161      try (var res = run(request, httpHeaders)) {
162         if (JsonRpcResponse.notification(request.getId()))
163            return null;
164         var sc = res.getStatusCode();
165         JsonRpcResponse parsed;
166         try {
167            parsed = res.body().as(JsonRpcResponse.class);
168         } catch (IOException e) {
169            if (sc < 200 || sc > 299)
170               throw ioex(e, "MCP server returned HTTP %s and the body was not a JSON-RPC envelope.", sc);
171            throw e;
172         }
173         if (parsed == null)
174            throw ioex("No response body received for JSON-RPC request id '%s' (HTTP %s).", request.getId(), sc);
175         return parsed;
176      }
177   }
178
179   /**
180    * Runs the request, transparently retrying exactly once on a pre-response stale-connection failure when the
181    * JSON-RPC method is side-effect-free.
182    *
183    * <p>
184    * Every MCP call is transported as an HTTP {@code POST}, which a transport can never safely replay (a {@code POST}
185    * may have already executed server-side before the connection dropped).  A pooled keep-alive connection can,
186    * however, be torn down by the server between requests, causing the very next reuse to fail with an immediate
187    * {@code EOF} before any response is received.  Keying the replay decision on JSON-RPC method idempotency here —
188    * rather than on the HTTP method — lets such a failure be recovered for a read-only call (e.g.
189    * {@code resources/read}) without risking a duplicate side effect.  Mutating methods (e.g. {@code tools/call}) and
190    * any method not in {@link #RETRYABLE_METHODS} are never replayed.
191    */
192   private RestResponse run(JsonRpcRequest request, Map<String,String> httpHeaders) throws IOException {
193      try {
194         return runOnce(request, httpHeaders);
195      } catch (IOException e) {
196         // req.run() only returns once response headers have been read, so any IOException here is pre-response.
197         if (! (isRetryableMethod(request.getMethod()) && TransportException.isStaleConnectionFailure(e)))
198            throw e;
199         return runOnce(request, httpHeaders);
200      }
201   }
202
203   private RestResponse runOnce(JsonRpcRequest request, Map<String,String> httpHeaders) throws IOException {
204      var req = restClient.post(endpoint).body(request);
205      if (httpHeaders != null) {
206         for (var e : httpHeaders.entrySet())
207            req.header(e.getKey(), e.getValue());
208      }
209      return req.run();
210   }
211
212   private static boolean isRetryableMethod(String method) {
213      return method != null && RETRYABLE_METHODS.contains(method);
214   }
215
216   /**
217    * Opens a Server-Sent-Events stream on the configured endpoint, for revision-specific facades that support a
218    * duplex (server-initiated) channel alongside the request/response {@link #send} calls.
219    *
220    * @return A reader over the opened event stream. Never <jk>null</jk>. Callers are responsible for closing it.
221    * @throws IOException If a transport-level error occurs opening the stream.
222    */
223   protected SseEventReader openEventStream() throws IOException {
224      return restClient.post(endpoint).openEventStream();
225   }
226
227   /**
228    * Opens a Server-Sent-Events stream on the configured endpoint by first POSTing {@code request} as the request
229    * body — for a revision-specific method (e.g. {@code subscriptions/listen}) whose held-open SSE response is
230    * itself the answer to one specific JSON-RPC request, as opposed to {@link #openEventStream()}'s body-less
231    * open used by the duplex server-push channel.
232    *
233    * @param request The JSON-RPC request to POST as the body that opens the stream. Must not be <jk>null</jk>.
234    * @return A reader over the opened event stream. Never <jk>null</jk>. Callers are responsible for closing it.
235    * @throws IOException If a transport-level or (de)serialization error occurs opening the stream.
236    */
237   protected SseEventReader openEventStream(JsonRpcRequest request) throws IOException {
238      return openEventStream(request, Map.of());
239   }
240
241   /**
242    * Opens a Server-Sent-Events stream on the configured endpoint by first POSTing {@code request} as the request
243    * body, with additional HTTP headers on the opening request — for a revision-specific method whose held-open
244    * SSE response must agree with a header-based method/name contract (e.g. SEP-2243's {@code Mcp-Method}/
245    * {@code Mcp-Name}) the same way {@link #send(JsonRpcRequest, Map)} already does for the ordinary
246    * request/response path.
247    *
248    * <p>
249    * Unlike the body-less {@link #openEventStream()}, this opening POST is itself the answer to one specific
250    * JSON-RPC request, so the server may reject it with a JSON-RPC error envelope instead of opening the stream
251    * (e.g. a capability, {@code Accept}-header, or over-limit gate) — on any HTTP status, not just 2xx. Rather
252    * than handing that envelope to an {@link SseEventReader} (which would see zero events and eventually look
253    * like an indistinguishable transport failure), this checks the response {@code Content-Type} first: only a
254    * body reporting {@code text/event-stream} is opened as a stream. Anything else is parsed as a
255    * {@link JsonRpcResponse}; a non-<jk>null</jk> {@link JsonRpcResponse#getError() error} is thrown synchronously
256    * as an {@link McpException} carrying the server's code/message/data, matching the way mid-stream JSON-RPC
257    * errors are already surfaced to a stream's listener. A response that is neither {@code text/event-stream} nor
258    * a JSON-RPC error envelope still fails as a plain {@link IOException} (or the underlying
259    * {@link RestCallException} from a non-2xx status), never silently masquerading as an opened stream.
260    *
261    * @param request The JSON-RPC request to POST as the body that opens the stream. Must not be <jk>null</jk>.
262    * @param httpHeaders Optional additional HTTP headers to add on this request. Can be <jk>null</jk>.
263    * @return A reader over the opened event stream. Never <jk>null</jk>. Callers are responsible for closing it.
264    * @throws IOException If a transport-level or (de)serialization error occurs opening the stream, or the
265    *    response was neither an SSE stream nor a JSON-RPC error envelope.
266    * @throws McpException If the server rejected {@code request} with a JSON-RPC error instead of opening the
267    *    stream.
268    */
269   protected SseEventReader openEventStream(JsonRpcRequest request, Map<String,String> httpHeaders) throws IOException {
270      reqnn("request", request);
271      var req = restClient.post(endpoint).body(request);
272      if (httpHeaders != null) {
273         for (var e : httpHeaders.entrySet())
274            req.header(e.getKey(), e.getValue());
275      }
276      if (! req.hasHeader("Accept"))
277         req.header("Accept", "text/event-stream");
278      var res = req.run();
279      var opened = false;
280      try {
281         if (isEventStream(res)) {
282            res.assertOk();
283            var r = res.body().asEventStream();
284            opened = true;
285            return r;
286         }
287         var parsed = res.body().as(JsonRpcResponse.class);
288         if (parsed != null && parsed.getError() != null)
289            throw McpException.fromJsonRpcError(parsed.getError());
290         res.assertOk(); // not a JSON-RPC error envelope either: fall back to the ordinary status diagnostic
291         throw ioex("Expected a text/event-stream response opening %s, but got Content-Type '%s'.",
292            request.getMethod(), contentTypeOf(res));
293      } finally {
294         if (! opened)
295            quiet(res::close);
296      }
297   }
298
299   private static boolean isEventStream(RestResponse res) {
300      var contentType = contentTypeOf(res);
301      return contentType != null && contentType.toLowerCase(Locale.ROOT).contains("text/event-stream");
302   }
303
304   private static String contentTypeOf(RestResponse res) {
305      var h = res.getFirstHeader("Content-Type");
306      return h == null ? null : h.value();
307   }
308
309   @Override /* Overridden from Closeable */
310   public void close() throws IOException {
311      restClient.close();
312   }
313
314   // -----------------------------------------------------------------------------------------------------------------
315   // Builder
316   // -----------------------------------------------------------------------------------------------------------------
317
318   /**
319    * Fluent, self-typed builder for {@link AbstractMcpClient} subclasses.
320    *
321    * @param <SELF> The concrete builder type (CRTP self-type), allowing dated adapters to add their own builder
322    *    methods while preserving fluent chaining.
323    * @since 10.0.0
324    */
325   @SuppressWarnings({
326      "java:S119" // 'SELF' (CRTP self-type) is intentional and clearer than a single-letter name.
327   })
328   public abstract static class Builder<SELF extends Builder<SELF>> {
329
330      String endpoint;
331      RestClient.Builder restClientBuilder = RestClient.builder();
332
333      /**
334       * Sets the MCP server endpoint URL.
335       *
336       * @param value The absolute endpoint URL (e.g. {@code "http://localhost:8080/mcp"}). Must not be <jk>null</jk>.
337       * @return This object.
338       */
339      public SELF endpoint(String value) {
340         endpoint = reqnb("endpoint", value);
341         return self();
342      }
343
344      /**
345       * Sets the HTTP transport used by the underlying {@link RestClient}.
346       *
347       * @param value The transport. Can be <jk>null</jk> (a transport is auto-discovered via {@link java.util.ServiceLoader}).
348       * @return This object.
349       */
350      public SELF transport(HttpTransport value) {
351         restClientBuilder.transport(value);
352         return self();
353      }
354
355      /**
356       * Adds a lifecycle interceptor (e.g. the auth-token seam) to the underlying {@link RestClient}.
357       *
358       * @param value The interceptor to add. Must not be <jk>null</jk>.
359       * @return This object.
360       */
361      public SELF interceptor(RestCallInterceptor value) {
362         restClientBuilder.interceptors(reqnn("interceptor", value));
363         return self();
364      }
365
366      @SuppressWarnings({
367         "unchecked" // SELF is always the concrete builder subtype by construction (CRTP).
368      })
369      protected SELF self() {
370         return (SELF) this;
371      }
372   }
373}