Add javadoc, clean up

Author: chemicL

mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/client/transport/WebClientStreamableHttpTransport.java

@@ -35,6 +35,31 @@
 import java.util.function.Function;
 import java.util.function.Supplier;
 
+/**
+ * An implementation of the Streamable HTTP protocol as defined by the
+ * <code>2025-03-26</code> version of the MCP specification.
+ *
+ * <p>
+ * The transport is capable of resumability and reconnects. It reacts to transport-level
+ * session invalidation and will propagate {@link McpTransportSessionNotFoundException
+ * appropriate exceptions} to the higher level abstraction layer when needed in order to
+ * allow proper state management. The implementation handles servers that are stateful and
+ * provide session meta information, but can also communicate with stateless servers that
+ * do not provide a session identifier and do not support SSE streams.
+ * </p>
+ * <p>
+ * This implementation does not handle backwards compatibility with the <a href=
+ * "https://modelcontextprotocol.io/specification/2024-11-05/basic/transports#http-with-sse">"HTTP
+ * with SSE" transport</a>. In order to communicate over the phased-out
+ * <code>2024-11-05</code> protocol, use {@link HttpClientSseClientTransport} or
+ * {@link WebFluxSseClientTransport}.
+ * </p>
+ *
+ * @author Dariusz Jędrzejczyk
+ * @see <a href=
+ * "https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http">Streamable
+ * HTTP transport specification</a>
+ */
 public class WebClientStreamableHttpTransport implements McpClientTransport {
 
 	private static final Logger logger = LoggerFactory.getLogger(WebClientStreamableHttpTransport.class);
@@ -47,7 +72,7 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
 	 */
 	private static final String MESSAGE_EVENT_TYPE = "message";
 
-	public static final ParameterizedTypeReference<ServerSentEvent<String>> PARAMETERIZED_TYPE_REF = new ParameterizedTypeReference<>() {
+	private static final ParameterizedTypeReference<ServerSentEvent<String>> PARAMETERIZED_TYPE_REF = new ParameterizedTypeReference<>() {
 	};
 
 	private final ObjectMapper objectMapper;
@@ -66,7 +91,6 @@ public class WebClientStreamableHttpTransport implements McpClientTransport {
 
 	private final AtomicReference<Consumer<Throwable>> exceptionHandler = new AtomicReference<>();
 
-	// TODO: builder
 	private WebClientStreamableHttpTransport(ObjectMapper objectMapper, WebClient.Builder webClientBuilder,
 			String endpoint, boolean resumableStreams, boolean openConnectionOnStartup) {
 		this.objectMapper = objectMapper;
@@ -77,6 +101,13 @@ private WebClientStreamableHttpTransport(ObjectMapper objectMapper, WebClient.Bu
 		this.activeSession.set(createTransportSession());
 	}
 
+	/**
+	 * Create a stateful builder for creating {@link WebClientStreamableHttpTransport}
+	 * instances.
+	 * @param webClientBuilder the {@link WebClient.Builder} to use
+	 * @return a builder which will create an instance of
+	 * {@link WebClientStreamableHttpTransport} once {@link Builder#build()} is called
+	 */
 	public static Builder builder(WebClient.Builder webClientBuilder) {
 		return new Builder(webClientBuilder);
 	}
@@ -392,6 +423,9 @@ private Tuple2<Optional<String>, Iterable<McpSchema.JSONRPCMessage>> parse(Serve
 		}
 	}
 
+	/**
+	 * Builder for {@link WebClientStreamableHttpTransport}.
+	 */
 	public static class Builder {
 
 		private ObjectMapper objectMapper;
@@ -409,34 +443,71 @@ private Builder(WebClient.Builder webClientBuilder) {
 			this.webClientBuilder = webClientBuilder;
 		}
 
+		/**
+		 * Configure the {@link ObjectMapper} to use.
+		 * @param objectMapper instance to use
+		 * @return the builder instance
+		 */
 		public Builder objectMapper(ObjectMapper objectMapper) {
 			Assert.notNull(objectMapper, "ObjectMapper must not be null");
 			this.objectMapper = objectMapper;
 			return this;
 		}
 
+		/**
+		 * Configure the {@link WebClient.Builder} to construct the {@link WebClient}.
+		 * @param webClientBuilder instance to use
+		 * @return the builder instance
+		 */
 		public Builder webClientBuilder(WebClient.Builder webClientBuilder) {
 			Assert.notNull(webClientBuilder, "WebClient.Builder must not be null");
 			this.webClientBuilder = webClientBuilder;
 			return this;
 		}
 
+		/**
+		 * Configure the endpoint to make HTTP requests against.
+		 * @param endpoint endpoint to use
+		 * @return the builder instance
+		 */
 		public Builder endpoint(String endpoint) {
 			Assert.hasText(endpoint, "endpoint must be a non-empty String");
 			this.endpoint = endpoint;
 			return this;
 		}
 
+		/**
+		 * Configure whether to use the stream resumability feature by keeping track of
+		 * SSE event ids.
+		 * @param resumableStreams if {@code true} event ids will be tracked and upon
+		 * disconnection, the last seen id will be used upon reconnection as a header to
+		 * resume consuming messages.
+		 * @return the builder instance
+		 */
 		public Builder resumableStreams(boolean resumableStreams) {
 			this.resumableStreams = resumableStreams;
 			return this;
 		}
 
+		/**
+		 * Configure whether the client should open an SSE connection upon startup. Not
+		 * all servers support this (although it is in theory possible with the current
+		 * specification), so use with caution. By default, this value is {@code false}.
+		 * @param openConnectionOnStartup if {@code true} the {@link #connect(Function)}
+		 * method call will try to open an SSE connection before sending any JSON-RPC
+		 * request
+		 * @return the builder instance
+		 */
 		public Builder openConnectionOnStartup(boolean openConnectionOnStartup) {
 			this.openConnectionOnStartup = openConnectionOnStartup;
 			return this;
 		}
 
+		/**
+		 * Construct a fresh instance of {@link WebClientStreamableHttpTransport} using
+		 * the current builder configuration.
+		 * @return a new instance of {@link WebClientStreamableHttpTransport}
+		 */
 		public WebClientStreamableHttpTransport build() {
 			ObjectMapper objectMapper = this.objectMapper != null ? this.objectMapper : new ObjectMapper();
 

mcp-test/src/main/java/io/modelcontextprotocol/client/AbstractMcpAsyncClientResiliencyTests.java

@@ -5,6 +5,7 @@
 import eu.rekawek.toxiproxy.model.ToxicDirection;
 import io.modelcontextprotocol.spec.McpClientTransport;
 import io.modelcontextprotocol.spec.McpSchema;
+import io.modelcontextprotocol.spec.McpTransport;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -24,6 +25,16 @@
 
 import static org.assertj.core.api.Assertions.assertThatCode;
 
+/**
+ * Resiliency test suite for the {@link McpAsyncClient} that can be used with different
+ * {@link McpTransport} implementations that support Streamable HTTP.
+ *
+ * The purpose of these tests is to allow validating the transport layer resiliency
+ * instead of the functionality offered by the logical layer of MCP concepts such as
+ * tools, resources, prompts, etc.
+ *
+ * @author Dariusz Jędrzejczyk
+ */
 public abstract class AbstractMcpAsyncClientResiliencyTests {
 
 	private static final Logger logger = LoggerFactory.getLogger(AbstractMcpAsyncClientResiliencyTests.class);

mcp/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java

@@ -75,6 +75,7 @@
  * @see McpClient
  * @see McpSchema
  * @see McpClientSession
+ * @see McpClientTransport
  */
 public class McpAsyncClient {
 

mcp/src/main/java/io/modelcontextprotocol/client/transport/StdioClientTransport.java

@@ -396,15 +396,4 @@ public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
 		return this.objectMapper.convertValue(data, typeRef);
 	}
 
-	private static void measureTime(Runnable op, String opName) {
-		long start = System.nanoTime();
-		try {
-			op.run();
-		}
-		finally {
-			long delta = System.nanoTime() - start;
-			logger.info("{} took {}ms", opName, Duration.ofNanos(delta).toMillis());
-		}
-	}
-
 }

mcp/src/main/java/io/modelcontextprotocol/spec/DefaultMcpTransportSession.java

@@ -12,6 +12,13 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 
+/**
+ * Default implementation of {@link McpTransportSession} which manages the open
+ * connections using tye {@link Disposable} type and allows to perform clean up using the
+ * {@link Disposable#dispose()} method.
+ *
+ * @author Dariusz Jędrzejczyk
+ */
 public class DefaultMcpTransportSession implements McpTransportSession<Disposable> {
 
 	private static final Logger logger = LoggerFactory.getLogger(DefaultMcpTransportSession.class);

mcp/src/main/java/io/modelcontextprotocol/spec/DefaultMcpTransportStream.java

@@ -12,6 +12,12 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
+/**
+ * An implementation of {@link McpTransportStream} using Project Reactor types.
+ *
+ * @param <CONNECTION> the resource serving the stream
+ * @author Dariusz Jędrzejczyk
+ */
 public class DefaultMcpTransportStream<CONNECTION> implements McpTransportStream<CONNECTION> {
 
 	private static final Logger logger = LoggerFactory.getLogger(DefaultMcpTransportStream.class);
@@ -27,6 +33,14 @@ public class DefaultMcpTransportStream<CONNECTION> implements McpTransportStream
 
 	private final Function<McpTransportStream<CONNECTION>, Publisher<CONNECTION>> reconnect;
 
+	/**
+	 * Constructs a new instance representing a particular stream that can resume using
+	 * the provided reconnect mechanism.
+	 * @param resumable whether the stream is resumable and should try to reconnect
+	 * @param reconnect the mechanism to use in case an error is observed on the current
+	 * event stream to asynchronously kick off a resumed stream consumption, potentially
+	 * using the stored {@link #lastId()}.
+	 */
 	public DefaultMcpTransportStream(boolean resumable,
 			Function<McpTransportStream<CONNECTION>, Publisher<CONNECTION>> reconnect) {
 		this.reconnect = reconnect;

mcp/src/main/java/io/modelcontextprotocol/spec/McpTransportSessionNotFoundException.java

@@ -8,11 +8,20 @@
  */
 public class McpTransportSessionNotFoundException extends RuntimeException {
 
+	/**
+	 * Construct an instance with a known {@link Exception cause}.
+	 * @param sessionId transport session identifier
+	 * @param cause the cause that was identified as a session not found error
+	 */
 	public McpTransportSessionNotFoundException(String sessionId, Exception cause) {
 		super("Session " + sessionId + " not found on the server", cause);
-
 	}
 
+	/**
+	 * Construct an instance with the session identifier but without a {@link Exception
+	 * cause}.
+	 * @param sessionId transport session identifier
+	 */
 	public McpTransportSessionNotFoundException(String sessionId) {
 		super("Session " + sessionId + " not found on the server");
 	}

mcp/src/main/java/io/modelcontextprotocol/spec/McpTransportStream.java

@@ -5,12 +5,40 @@
 
 import java.util.Optional;
 
+/**
+ * A representation of a stream at the transport layer of the MCP protocol. In particular,
+ * it is currently used in the Streamable HTTP implementation to potentially be able to
+ * resume a broken connection from where it left off by optionally keeping track of
+ * attached SSE event ids.
+ *
+ * @param <CONNECTION> the resource on which the stream is being served and consumed via
+ * this mechanism
+ * @author Dariusz Jędrzejczyk
+ */
 public interface McpTransportStream<CONNECTION> {
 
+	/**
+	 * The last observed event identifier.
+	 * @return if not empty, contains the most recent event that was consumed
+	 */
 	Optional<String> lastId();
 
+	/**
+	 * An internal stream identifier used to distinguish streams while debugging.
+	 * @return a {@code long} stream identifier value
+	 */
 	long streamId();
 
+	/**
+	 * Allows keeping track of the transport stream of events (currently an SSE stream
+	 * from Streamable HTTP specification) and enable resumability and reconnects in case
+	 * of stream errors.
+	 * @param eventStream a {@link Publisher} of tuples (pairs) of an optional identifier
+	 * associated with a collection of messages
+	 * @return a flattened {@link Publisher} of
+	 * {@link io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage JSON-RPC messages}
+	 * with the identifier stripped away
+	 */
 	Publisher<McpSchema.JSONRPCMessage> consumeSseStream(
 			Publisher<Tuple2<Optional<String>, Iterable<McpSchema.JSONRPCMessage>>> eventStream);