feat: implement HttpClient Streamable HTTP transport

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

@@ -79,7 +79,7 @@ public abstract class AbstractMcpAsyncClientResiliencyTests {
 		host = "http://" + ipAddressViaToxiproxy + ":" + portViaToxiproxy;
 	}
 
-	private static void disconnect() {
+	static void disconnect() {
 		long start = System.nanoTime();
 		try {
 			// disconnect
@@ -96,7 +96,7 @@ private static void disconnect() {
 		}
 	}
 
-	private static void reconnect() {
+	static void reconnect() {
 		long start = System.nanoTime();
 		try {
 			proxy.toxics().get("RESET_UPSTREAM").remove();
@@ -110,7 +110,7 @@ private static void reconnect() {
 		}
 	}
 
-	private static void restartMcpServer() {
+	static void restartMcpServer() {
 		container.stop();
 		container.start();
 	}

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

@@ -13,6 +13,7 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -135,7 +136,7 @@ public static Builder builder(String baseUri) {
 	public Mono<Void> connect(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
 		return Mono.deferContextual(ctx -> {
 			this.handler.set(handler);
-			if (openConnectionOnStartup) {
+			if (this.openConnectionOnStartup) {
 				logger.debug("Eagerly opening connection on startup");
 				return this.reconnect(null).onErrorComplete(t -> {
 					logger.warn("Eager connect failed ", t);
@@ -286,6 +287,7 @@ else if (statusCode == BAD_REQUEST) {
 
 				}).<McpSchema
 						.JSONRPCMessage>flatMap(jsonrpcMessage -> this.handler.get().apply(Mono.just(jsonrpcMessage)))
+				.onErrorMap(CompletionException.class, t -> t.getCause())
 				.onErrorComplete(t -> {
 					this.handleException(t);
 					return true;
@@ -373,7 +375,7 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sendMessage) {
 						else {
 							logger.debug("SSE connection established successfully");
 						}
-					})).onErrorComplete().subscribe();
+					})).onErrorMap(CompletionException.class, t -> t.getCause()).onErrorComplete().subscribe();
 
 			}).flatMap(responseEvent -> {
 				if (transportSession.markInitialized(
@@ -464,19 +466,25 @@ else if (statusCode == BAD_REQUEST) {
 
 				return Flux.<McpSchema.JSONRPCMessage>error(
 						new RuntimeException("Failed to send message: " + responseEvent));
-			}).flatMap(jsonRpcMessage -> this.handler.get().apply(Mono.just(jsonRpcMessage))).onErrorComplete(t -> {
-				// handle the error first
-				this.handleException(t);
-				// inform the caller of sendMessage
-				messageSink.error(t);
-				return true;
-			}).doFinally(s -> {
-				logger.debug("SendMessage finally: {}", s);
-				Disposable ref = disposableRef.getAndSet(null);
-				if (ref != null) {
-					transportSession.removeConnection(ref);
-				}
-			}).contextWrite(messageSink.contextView()).subscribe();
+			})
+				.flatMap(jsonRpcMessage -> this.handler.get().apply(Mono.just(jsonRpcMessage)))
+				.onErrorMap(CompletionException.class, t -> t.getCause())
+				.onErrorComplete(t -> {
+					// handle the error first
+					this.handleException(t);
+					// inform the caller of sendMessage
+					messageSink.error(t);
+					return true;
+				})
+				.doFinally(s -> {
+					logger.debug("SendMessage finally: {}", s);
+					Disposable ref = disposableRef.getAndSet(null);
+					if (ref != null) {
+						transportSession.removeConnection(ref);
+					}
+				})
+				.contextWrite(messageSink.contextView())
+				.subscribe();
 
 			disposableRef.set(connection);
 			transportSession.addConnection(connection);

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

@@ -15,15 +15,20 @@
 import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.FluxSink;
 
+/**
+ * Utility class providing various {@link BodySubscriber} implementations for handling
+ * different types of HTTP response bodies in the context of Model Context Protocol (MCP)
+ * clients.
+ *
+ * <p>
+ * Defines subscribers for processing Server-Sent Events (SSE), aggregate responses, and
+ * bodiless responses.
+ *
+ * @author Christian Tzolov
+ * @author Dariusz Jędrzejczyk
+ */
 class ResponseSubscribers {
 
-	/**
-	 * Represents a Server-Sent Event with its standard fields.
-	 *
-	 * @param id the event ID, may be {@code null}
-	 * @param event the event type, may be {@code null} (defaults to "message")
-	 * @param data the event payload data, never {@code null}
-	 */
 	record SseEvent(String id, String event, String data) {
 	}
 
@@ -115,10 +120,6 @@ public SseLineSubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink
 			this.responseInfo = responseInfo;
 		}
 
-		/**
-		 * Initializes the subscription and sets up disposal callback.
-		 * @param subscription the {@link Subscription} to the upstream line source
-		 */
 		@Override
 		protected void hookOnSubscribe(Subscription subscription) {
 
@@ -132,12 +133,6 @@ protected void hookOnSubscribe(Subscription subscription) {
 			});
 		}
 
-		/**
-		 * Processes each line from the SSE stream according to the SSE protocol. Empty
-		 * lines trigger event emission, other lines are parsed for data, id, or event
-		 * type.
-		 * @param line the line to process from the SSE stream
-		 */
 		@Override
 		protected void hookOnNext(String line) {
 			if (line.isEmpty()) {
@@ -172,9 +167,6 @@ else if (line.startsWith("event:")) {
 			}
 		}
 
-		/**
-		 * Called when the upstream line source completes normally.
-		 */
 		@Override
 		protected void hookOnComplete() {
 			if (this.eventBuilder.length() > 0) {
@@ -185,10 +177,6 @@ protected void hookOnComplete() {
 			this.sink.complete();
 		}
 
-		/**
-		 * Called when an error occurs in the upstream line source.
-		 * @param throwable the error that occurred
-		 */
 		@Override
 		protected void hookOnError(Throwable throwable) {
 			this.sink.error(throwable);
@@ -225,10 +213,6 @@ public AggregateSubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> si
 			this.responseInfo = responseInfo;
 		}
 
-		/**
-		 * Initializes the subscription and sets up disposal callback.
-		 * @param subscription the {@link Subscription} to the upstream line source
-		 */
 		@Override
 		protected void hookOnSubscribe(Subscription subscription) {
 			sink.onRequest(subscription::request);
@@ -237,18 +221,11 @@ protected void hookOnSubscribe(Subscription subscription) {
 			sink.onDispose(subscription::cancel);
 		}
 
-		/**
-		 * Aggregate each line from the Http response.
-		 * @param line next line to process from the Http response
-		 */
 		@Override
 		protected void hookOnNext(String line) {
 			this.eventBuilder.append(line).append("\n");
 		}
 
-		/**
-		 * Called when the upstream line source completes normally.
-		 */
 		@Override
 		protected void hookOnComplete() {
 			if (this.eventBuilder.length() > 0) {
@@ -258,10 +235,6 @@ protected void hookOnComplete() {
 			this.sink.complete();
 		}
 
-		/**
-		 * Called when an error occurs in the upstream line source.
-		 * @param throwable the error that occurred
-		 */
 		@Override
 		protected void hookOnError(Throwable throwable) {
 			this.sink.error(throwable);
@@ -283,10 +256,6 @@ public BodilessResponseLineSubscriber(ResponseInfo responseInfo, FluxSink<Respon
 			this.responseInfo = responseInfo;
 		}
 
-		/**
-		 * Initializes the subscription and sets up disposal callback.
-		 * @param subscription the {@link Subscription} to the upstream line source
-		 */
 		@Override
 		protected void hookOnSubscribe(Subscription subscription) {
 
@@ -300,9 +269,6 @@ protected void hookOnSubscribe(Subscription subscription) {
 			});
 		}
 
-		/**
-		 * Called when the upstream line source completes normally.
-		 */
 		@Override
 		protected void hookOnComplete() {
 			// emit dummy event to be able to inspect the response info
@@ -313,10 +279,6 @@ protected void hookOnComplete() {
 			this.sink.complete();
 		}
 
-		/**
-		 * Called when an error occurs in the upstream line source.
-		 * @param throwable the error that occurred
-		 */
 		@Override
 		protected void hookOnError(Throwable throwable) {
 			this.sink.error(throwable);

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

@@ -1,3 +1,6 @@
+/*
+ * Copyright 2024-2024 the original author or authors.
+ */
 package io.modelcontextprotocol.client;
 
 import eu.rekawek.toxiproxy.Proxy;
@@ -35,6 +38,7 @@
  *
  * @author Dariusz Jędrzejczyk
  */
+// KEEP IN SYNC with the class in mcp-test module
 public abstract class AbstractMcpAsyncClientResiliencyTests {
 
 	private static final Logger logger = LoggerFactory.getLogger(AbstractMcpAsyncClientResiliencyTests.class);

mcp/src/test/java/io/modelcontextprotocol/client/HttpClientStreamableHttpAsyncClientResiliencyTests.java

@@ -4,7 +4,7 @@
 
 package io.modelcontextprotocol.client;
 
-import java.util.concurrent.CompletionException;
+import java.io.IOException;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
@@ -28,7 +28,7 @@ void testPingWithExactExceptionType() {
 
 			disconnect();
 
-			StepVerifier.create(mcpAsyncClient.ping()).expectError(CompletionException.class).verify();
+			StepVerifier.create(mcpAsyncClient.ping()).expectError(IOException.class).verify();
 
 			reconnect();