WebClient Streamable HTTP support

mcp-spring/mcp-spring-webflux/pom.xml

@@ -99,6 +99,12 @@
 			<version>${testcontainers.version}</version>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>toxiproxy</artifactId>
+			<version>${toxiproxy.version}</version>
+			<scope>test</scope>
+		</dependency>
 
 		<dependency>
 			<groupId>org.awaitility</groupId>

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

@@ -190,6 +190,9 @@ public WebFluxSseClientTransport(WebClient.Builder webClientBuilder, ObjectMappe
 	 */
 	@Override
 	public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> handler) {
+		// TODO: Avoid eager connection opening and enable resilience
+		// -> upon disconnects, re-establish connection
+		// -> allow optimizing for eager connection start using a constructor flag
 		Flux<ServerSentEvent<String>> events = eventStream();
 		this.inboundSubscription = events.concatMap(event -> Mono.just(event).<JSONRPCMessage>handle((e, s) -> {
 			if (ENDPOINT_EVENT_TYPE.equals(event.event())) {

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/client/WebClientStreamableHttpAsyncClientResiliencyTests.java

@@ -0,0 +1,17 @@
+package io.modelcontextprotocol.client;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.modelcontextprotocol.client.transport.WebClientStreamableHttpTransport;
+import io.modelcontextprotocol.spec.McpClientTransport;
+import org.junit.jupiter.api.Timeout;
+import org.springframework.web.reactive.function.client.WebClient;
+
+@Timeout(15)
+public class WebClientStreamableHttpAsyncClientResiliencyTests extends AbstractMcpAsyncClientResiliencyTests {
+
+	@Override
+	protected McpClientTransport createMcpTransport() {
+		return WebClientStreamableHttpTransport.builder(WebClient.builder().baseUrl(host)).build();
+	}
+
+}

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/client/WebClientStreamableHttpAsyncClientTests.java

@@ -0,0 +1,42 @@
+package io.modelcontextprotocol.client;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.modelcontextprotocol.client.transport.WebClientStreamableHttpTransport;
+import io.modelcontextprotocol.spec.McpClientTransport;
+import org.junit.jupiter.api.Timeout;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.images.builder.ImageFromDockerfile;
+
+@Timeout(15)
+public class WebClientStreamableHttpAsyncClientTests extends AbstractMcpAsyncClientTests {
+
+	static String host = "http://localhost:3001";
+
+	// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
+	@SuppressWarnings("resource")
+	GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v2")
+		.withCommand("node dist/index.js streamableHttp")
+		.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
+		.withExposedPorts(3001)
+		.waitingFor(Wait.forHttp("/").forStatusCode(404));
+
+	@Override
+	protected McpClientTransport createMcpTransport() {
+		return WebClientStreamableHttpTransport.builder(WebClient.builder().baseUrl(host)).build();
+	}
+
+	@Override
+	protected void onStart() {
+		container.start();
+		int port = container.getMappedPort(3001);
+		host = "http://" + container.getHost() + ":" + port;
+	}
+
+	@Override
+	public void onClose() {
+		container.stop();
+	}
+
+}

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/client/WebClientStreamableHttpSyncClientTests.java

@@ -0,0 +1,41 @@
+package io.modelcontextprotocol.client;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.modelcontextprotocol.client.transport.WebClientStreamableHttpTransport;
+import io.modelcontextprotocol.spec.McpClientTransport;
+import org.junit.jupiter.api.Timeout;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+
+@Timeout(15)
+public class WebClientStreamableHttpSyncClientTests extends AbstractMcpSyncClientTests {
+
+	static String host = "http://localhost:3001";
+
+	// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
+	@SuppressWarnings("resource")
+	GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v2")
+		.withCommand("node dist/index.js streamableHttp")
+		.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
+		.withExposedPorts(3001)
+		.waitingFor(Wait.forHttp("/").forStatusCode(404));
+
+	@Override
+	protected McpClientTransport createMcpTransport() {
+		return WebClientStreamableHttpTransport.builder(WebClient.builder().baseUrl(host)).build();
+	}
+
+	@Override
+	protected void onStart() {
+		container.start();
+		int port = container.getMappedPort(3001);
+		host = "http://" + container.getHost() + ":" + port;
+	}
+
+	@Override
+	public void onClose() {
+		container.stop();
+	}
+
+}

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/client/WebFluxSseMcpAsyncClientTests.java

@@ -26,7 +26,8 @@ class WebFluxSseMcpAsyncClientTests extends AbstractMcpAsyncClientTests {
 
 	// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
 	@SuppressWarnings("resource")
-	GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v1")
+	GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v2")
+		.withCommand("node dist/index.js sse")
 		.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
 		.withExposedPorts(3001)
 		.waitingFor(Wait.forHttp("/").forStatusCode(404));

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/client/WebFluxSseMcpSyncClientTests.java

@@ -26,7 +26,8 @@ class WebFluxSseMcpSyncClientTests extends AbstractMcpSyncClientTests {
 
 	// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
 	@SuppressWarnings("resource")
-	GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v1")
+	GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v2")
+		.withCommand("node dist/index.js sse")
 		.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
 		.withExposedPorts(3001)
 		.waitingFor(Wait.forHttp("/").forStatusCode(404));

mcp-spring/mcp-spring-webflux/src/test/java/io/modelcontextprotocol/client/transport/WebFluxSseClientTransportTests.java

@@ -41,7 +41,8 @@ class WebFluxSseClientTransportTests {
 	static String host = "http://localhost:3001";
 
 	@SuppressWarnings("resource")
-	GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v1")
+	GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v2")
+		.withCommand("node dist/index.js sse")
 		.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
 		.withExposedPorts(3001)
 		.waitingFor(Wait.forHttp("/").forStatusCode(404));

mcp-spring/mcp-spring-webflux/src/test/resources/logback.xml

@@ -9,13 +9,13 @@
     </appender>
 
     <!-- Main MCP package -->
-    <logger name="org.springframework.ai.mcp" level="INFO"/>
+    <logger name="io.modelcontextprotocol" level="INFO"/>
 
     <!-- Client packages -->
-    <logger name="org.springframework.ai.mcp.client" level="INFO"/>
-    
+    <logger name="io.modelcontextprotocol.client" level="INFO"/>
+
     <!-- Spec package -->
-    <logger name="org.springframework.ai.mcp.spec" level="INFO"/>
+    <logger name="io.modelcontextprotocol.spec" level="INFO"/>
 
 
     <!-- Root logger -->

mcp-test/pom.xml

@@ -68,6 +68,11 @@
 			<artifactId>junit-jupiter</artifactId>
 			<version>${testcontainers.version}</version>
 		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>toxiproxy</artifactId>
+			<version>${toxiproxy.version}</version>
+		</dependency>
 
 		<dependency>
 			<groupId>org.awaitility</groupId>