refactor: improve MCP client timeout handling and reactive testing

- Add configurable initialization timeout separate from request timeout
- Rename ServletSse* test classes to HttpSse* for better naming consistency
- Replace direct .block() calls with StepVerifier for better reactive testing
- Change ping() method to return Mono instead of Mono
- Improve error handling and reactive programming patterns throughout tests
- Chain reactive operations for cleaner test flow

Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>

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

@@ -4,8 +4,6 @@
 
 package io.modelcontextprotocol.client;
 
-import java.time.Duration;
-
 import io.modelcontextprotocol.client.transport.WebFluxSseClientTransport;
 import io.modelcontextprotocol.spec.ClientMcpTransport;
 import org.junit.jupiter.api.Timeout;
@@ -48,9 +46,4 @@ class WebFluxSseMcpAsyncClientTests extends AbstractMcpAsyncClientTests {
 		container.stop();
 	}
 
-	@Override
-	protected Duration getTimeoutDuration() {
-		return Duration.ofMillis(300);
-	}
-
 }

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

@@ -4,8 +4,6 @@
 
 package io.modelcontextprotocol.client;
 
-import java.time.Duration;
-
 import io.modelcontextprotocol.client.transport.WebFluxSseClientTransport;
 import io.modelcontextprotocol.spec.ClientMcpTransport;
 import org.junit.jupiter.api.Timeout;
@@ -48,9 +46,4 @@ class WebFluxSseMcpSyncClientTests extends AbstractMcpSyncClientTests {
 		container.stop();
 	}
 
-	@Override
-	protected Duration getTimeoutDuration() {
-		return Duration.ofMillis(300);
-	}
-
 }

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

@@ -58,8 +58,12 @@ public abstract class AbstractMcpAsyncClientTests {
 	protected void onClose() {
 	}
 
-	protected Duration getTimeoutDuration() {
+	protected Duration getRequestTimeout() {
-		return Duration.ofSeconds(2);
+		return Duration.ofSeconds(10);
+	}
+
+	protected Duration getInitializationTimeout() {
+		return Duration.ofSeconds(1);
 	}
 
 	@BeforeEach
@@ -69,7 +73,8 @@ public abstract class AbstractMcpAsyncClientTests {
 
 		assertThatCode(() -> {
 			mcpAsyncClient = McpClient.async(mcpTransport)
-				.requestTimeout(getTimeoutDuration())
+				.requestTimeout(getRequestTimeout())
+				.initializationTimeout(getInitializationTimeout())
 				.capabilities(ClientCapabilities.builder().roots(true).build())
 				.build();
 		}).doesNotThrowAnyException();
@@ -78,8 +83,7 @@ public abstract class AbstractMcpAsyncClientTests {
 	@AfterEach
 	void tearDown() {
 		if (mcpAsyncClient != null) {
-			assertThatCode(() -> mcpAsyncClient.closeGracefully().block(Duration.ofSeconds(10)))
+			StepVerifier.create(mcpAsyncClient.closeGracefully()).verifyComplete();
-				.doesNotThrowAnyException();
 		}
 		onClose();
 	}
@@ -96,87 +100,93 @@ public abstract class AbstractMcpAsyncClientTests {
 
 	@Test
 	void testListToolsWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.listTools(null).block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.listTools(null)).expectErrorSatisfies(error -> {
-			.hasMessage("Client must be initialized before listing tools");
+			assertThat(error).isInstanceOf(McpError.class)
+				.hasMessage("Client must be initialized before listing tools");
+		}).verify();
 	}
 
 	@Test
 	void testListTools() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listTools(null)))
+			.consumeNextWith(result -> {
+				assertThat(result.tools()).isNotNull().isNotEmpty();
 
-		StepVerifier.create(mcpAsyncClient.listTools(null)).consumeNextWith(result -> {
+				Tool firstTool = result.tools().get(0);
-			assertThat(result.tools()).isNotNull().isNotEmpty();
+				assertThat(firstTool.name()).isNotNull();
-
+				assertThat(firstTool.description()).isNotNull();
-			Tool firstTool = result.tools().get(0);
+			})
-			assertThat(firstTool.name()).isNotNull();
+			.verifyComplete();
-			assertThat(firstTool.description()).isNotNull();
-		}).verifyComplete();
 	}
 
 	@Test
 	void testPingWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.ping().block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.ping())
-			.hasMessage("Client must be initialized before pinging the server");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before pinging the server"))
+			.verify();
 	}
 
 	@Test
 	void testPing() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.ping())).verifyComplete();
-		assertThatCode(() -> mcpAsyncClient.ping().block()).doesNotThrowAnyException();
 	}
 
 	@Test
 	void testCallToolWithoutInitialization() {
 		CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
 
-		assertThatThrownBy(() -> mcpAsyncClient.callTool(callToolRequest).block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.callTool(callToolRequest))
-			.hasMessage("Client must be initialized before calling tools");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before calling tools"))
+			.verify();
 	}
 
 	@Test
 	void testCallTool() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
-
 		CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
 
-		StepVerifier.create(mcpAsyncClient.callTool(callToolRequest)).consumeNextWith(callToolResult -> {
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(callToolRequest)))
-			assertThat(callToolResult).isNotNull().satisfies(result -> {
+			.consumeNextWith(callToolResult -> {
-				assertThat(result.content()).isNotNull();
+				assertThat(callToolResult).isNotNull();
-				assertThat(result.isError()).isNull();
+				assertThat(callToolResult.content()).isNotNull();
-			});
+				assertThat(callToolResult.isError()).isNull();
-		}).verifyComplete();
+			})
+			.verifyComplete();
 	}
 
 	@Test
 	void testCallToolWithInvalidTool() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
-
 		CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", ECHO_TEST_MESSAGE));
 
-		assertThatThrownBy(() -> mcpAsyncClient.callTool(invalidRequest).block()).isInstanceOf(Exception.class);
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(invalidRequest)))
+			.expectError(Exception.class)
+			.verify();
 	}
 
 	@Test
 	void testListResourcesWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.listResources(null).block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.listResources(null))
-			.hasMessage("Client must be initialized before listing resources");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before listing resources"))
+			.verify();
 	}
 
 	@Test
 	void testListResources() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResources(null)))
+			.consumeNextWith(resources -> {
+				assertThat(resources).isNotNull().satisfies(result -> {
+					assertThat(result.resources()).isNotNull();
 
-		StepVerifier.create(mcpAsyncClient.listResources(null)).consumeNextWith(resources -> {
+					if (!result.resources().isEmpty()) {
-			assertThat(resources).isNotNull().satisfies(result -> {
+						Resource firstResource = result.resources().get(0);
-				assertThat(result.resources()).isNotNull();
+						assertThat(firstResource.uri()).isNotNull();
-
+						assertThat(firstResource.name()).isNotNull();
-				if (!result.resources().isEmpty()) {
+					}
-					Resource firstResource = result.resources().get(0);
+				});
-					assertThat(firstResource.uri()).isNotNull();
+			})
-					assertThat(firstResource.name()).isNotNull();
+			.verifyComplete();
-				}
-			});
-		}).verifyComplete();
 	}
 
 	@Test
@@ -186,40 +196,44 @@ public abstract class AbstractMcpAsyncClientTests {
 
 	@Test
 	void testListPromptsWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.listPrompts(null).block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.listPrompts(null))
-			.hasMessage("Client must be initialized before listing prompts");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before listing prompts"))
+			.verify();
 	}
 
 	@Test
 	void testListPrompts() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listPrompts(null)))
+			.consumeNextWith(prompts -> {
+				assertThat(prompts).isNotNull().satisfies(result -> {
+					assertThat(result.prompts()).isNotNull();
 
-		StepVerifier.create(mcpAsyncClient.listPrompts(null)).consumeNextWith(prompts -> {
+					if (!result.prompts().isEmpty()) {
-			assertThat(prompts).isNotNull().satisfies(result -> {
+						Prompt firstPrompt = result.prompts().get(0);
-				assertThat(result.prompts()).isNotNull();
+						assertThat(firstPrompt.name()).isNotNull();
-
+						assertThat(firstPrompt.description()).isNotNull();
-				if (!result.prompts().isEmpty()) {
+					}
-					Prompt firstPrompt = result.prompts().get(0);
+				});
-					assertThat(firstPrompt.name()).isNotNull();
+			})
-					assertThat(firstPrompt.description()).isNotNull();
+			.verifyComplete();
-				}
-			});
-		}).verifyComplete();
 	}
 
 	@Test
 	void testGetPromptWithoutInitialization() {
 		GetPromptRequest request = new GetPromptRequest("simple_prompt", Map.of());
 
-		assertThatThrownBy(() -> mcpAsyncClient.getPrompt(request).block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.getPrompt(request))
-			.hasMessage("Client must be initialized before getting prompts");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before getting prompts"))
+			.verify();
 	}
 
 	@Test
 	void testGetPrompt() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		GetPromptRequest request = new GetPromptRequest("simple_prompt", Map.of());
 
-		StepVerifier.create(mcpAsyncClient.getPrompt(new GetPromptRequest("simple_prompt", Map.of())))
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.getPrompt(request)))
 			.consumeNextWith(prompt -> {
 				assertThat(prompt).isNotNull().satisfies(result -> {
 					assertThat(result.messages()).isNotEmpty();
@@ -231,15 +245,16 @@ public abstract class AbstractMcpAsyncClientTests {
 
 	@Test
 	void testRootsListChangedWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.rootsListChangedNotification().block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.rootsListChangedNotification())
-			.hasMessage("Client must be initialized before sending roots list changed notification");
+			.expectErrorMatches(error -> error instanceof McpError && error.getMessage()
+				.equals("Client must be initialized before sending roots list changed notification"))
+			.verify();
 	}
 
 	@Test
 	void testRootsListChanged() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.rootsListChangedNotification()))
-
+			.verifyComplete();
-		assertThatCode(() -> mcpAsyncClient.rootsListChangedNotification().block()).doesNotThrowAnyException();
 	}
 
 	@Test
@@ -247,39 +262,39 @@ public abstract class AbstractMcpAsyncClientTests {
 		var transport = createMcpTransport();
 
 		var client = McpClient.async(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.roots(new Root("file:///test/path", "test-root"))
 			.build();
 
-		assertThatCode(() -> client.initialize().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
+		StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
-
-		assertThatCode(() -> client.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
 	}
 
 	@Test
 	void testAddRoot() {
 		Root newRoot = new Root("file:///new/test/path", "new-test-root");
-		assertThatCode(() -> mcpAsyncClient.addRoot(newRoot).block()).doesNotThrowAnyException();
+
+		StepVerifier.create(mcpAsyncClient.addRoot(newRoot)).verifyComplete();
 	}
 
 	@Test
 	void testAddRootWithNullValue() {
-		assertThatThrownBy(() -> mcpAsyncClient.addRoot(null).block()).hasMessageContaining("Root must not be null");
+		StepVerifier.create(mcpAsyncClient.addRoot(null))
+			.expectErrorMatches(error -> error.getMessage().contains("Root must not be null"))
+			.verify();
 	}
 
 	@Test
 	void testRemoveRoot() {
 		Root root = new Root("file:///test/path/to/remove", "root-to-remove");
-		assertThatCode(() -> {
+
-			mcpAsyncClient.addRoot(root).block();
+		StepVerifier.create(mcpAsyncClient.addRoot(root).then(mcpAsyncClient.removeRoot(root.uri()))).verifyComplete();
-			mcpAsyncClient.removeRoot(root.uri()).block();
-		}).doesNotThrowAnyException();
 	}
 
 	@Test
 	void testRemoveNonExistentRoot() {
-		assertThatThrownBy(() -> mcpAsyncClient.removeRoot("nonexistent-uri").block())
+		StepVerifier.create(mcpAsyncClient.removeRoot("nonexistent-uri"))
-			.hasMessageContaining("Root with uri 'nonexistent-uri' not found");
+			.expectErrorMatches(error -> error.getMessage().contains("Root with uri 'nonexistent-uri' not found"))
+			.verify();
 	}
 
 	@Test
@@ -298,18 +313,20 @@ public abstract class AbstractMcpAsyncClientTests {
 
 	@Test
 	void testListResourceTemplatesWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.listResourceTemplates().block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.listResourceTemplates())
-			.hasMessage("Client must be initialized before listing resource templates");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before listing resource templates"))
+			.verify();
 	}
 
 	@Test
 	void testListResourceTemplates() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResourceTemplates()))
-
+			.consumeNextWith(result -> {
-		StepVerifier.create(mcpAsyncClient.listResourceTemplates()).consumeNextWith(result -> {
+				assertThat(result).isNotNull();
-			assertThat(result).isNotNull();
+				assertThat(result.resourceTemplates()).isNotNull();
-			assertThat(result.resourceTemplates()).isNotNull();
+			})
-		}).verifyComplete();
+			.verifyComplete();
 	}
 
 	// @Test
@@ -337,16 +354,13 @@ public abstract class AbstractMcpAsyncClientTests {
 
 		var transport = createMcpTransport();
 		var client = McpClient.async(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.toolsChangeConsumer(tools -> Mono.fromRunnable(() -> toolsNotificationReceived.set(true)))
 			.resourcesChangeConsumer(resources -> Mono.fromRunnable(() -> resourcesNotificationReceived.set(true)))
 			.promptsChangeConsumer(prompts -> Mono.fromRunnable(() -> promptsNotificationReceived.set(true)))
 			.build();
 
-		assertThatCode(() -> {
+		StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
-			client.initialize().block();
-			client.closeGracefully().block();
-		}).doesNotThrowAnyException();
 	}
 
 	@Test
@@ -356,15 +370,12 @@ public abstract class AbstractMcpAsyncClientTests {
 		var capabilities = ClientCapabilities.builder().sampling().build();
 
 		var client = McpClient.async(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.capabilities(capabilities)
 			.sampling(request -> Mono.just(CreateMessageResult.builder().message("test").model("test-model").build()))
 			.build();
 
-		assertThatCode(() -> {
+		StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
-			client.initialize().block(Duration.ofSeconds(10));
-			client.closeGracefully().block(Duration.ofSeconds(10));
-		}).doesNotThrowAnyException();
 	}
 
 	@Test
@@ -380,17 +391,17 @@ public abstract class AbstractMcpAsyncClientTests {
 		Function<CreateMessageRequest, Mono<CreateMessageResult>> samplingHandler = request -> Mono
 			.just(CreateMessageResult.builder().message("test").model("test-model").build());
 		var client = McpClient.async(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.capabilities(capabilities)
 			.sampling(samplingHandler)
 			.build();
 
-		assertThatCode(() -> {
+		StepVerifier.create(client.initialize()).consumeNextWith(result -> {
-			var result = client.initialize().block(Duration.ofSeconds(10));
 			assertThat(result).isNotNull();
 			assertThat(result.capabilities()).isNotNull();
-			client.closeGracefully().block(Duration.ofSeconds(10));
+		}).verifyComplete();
-		}).doesNotThrowAnyException();
+
+		StepVerifier.create(client.closeGracefully()).verifyComplete();
 	}
 
 	// ---------------------------------------
@@ -399,19 +410,23 @@ public abstract class AbstractMcpAsyncClientTests {
 
 	@Test
 	void testLoggingLevelsWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.setLoggingLevel(McpSchema.LoggingLevel.DEBUG).block())
+		StepVerifier.create(mcpAsyncClient.setLoggingLevel(McpSchema.LoggingLevel.DEBUG))
-			.isInstanceOf(McpError.class)
+			.expectErrorMatches(error -> error instanceof McpError
-			.hasMessage("Client must be initialized before setting logging level");
+					&& error.getMessage().equals("Client must be initialized before setting logging level"))
+			.verify();
 	}
 
 	@Test
 	void testLoggingLevels() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		Mono<Void> testAllLevels = mcpAsyncClient.initialize().then(Mono.defer(() -> {
+			Mono<Void> chain = Mono.empty();
+			for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
+				chain = chain.then(mcpAsyncClient.setLoggingLevel(level));
+			}
+			return chain;
+		}));
 
-		// Test all logging levels
+		StepVerifier.create(testAllLevels).verifyComplete();
-		for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
-			StepVerifier.create(mcpAsyncClient.setLoggingLevel(level)).verifyComplete();
-		}
 	}
 
 	@Test
@@ -420,20 +435,18 @@ public abstract class AbstractMcpAsyncClientTests {
 		var transport = createMcpTransport();
 
 		var client = McpClient.async(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.loggingConsumer(notification -> Mono.fromRunnable(() -> logReceived.set(true)))
 			.build();
 
-		assertThatCode(() -> {
+		StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
-			client.initialize().block(Duration.ofSeconds(10));
-			client.closeGracefully().block(Duration.ofSeconds(10));
-		}).doesNotThrowAnyException();
 	}
 
 	@Test
 	void testLoggingWithNullNotification() {
-		assertThatThrownBy(() -> mcpAsyncClient.setLoggingLevel(null).block())
+		StepVerifier.create(mcpAsyncClient.setLoggingLevel(null))
-			.hasMessageContaining("Logging level must not be null");
+			.expectErrorMatches(error -> error.getMessage().contains("Logging level must not be null"))
+			.verify();
 	}
 
 }

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

@@ -52,8 +52,12 @@ public abstract class AbstractMcpSyncClientTests {
 
 	abstract protected void onClose();
 
-	protected Duration getTimeoutDuration() {
+	protected Duration getRequestTimeout() {
-		return Duration.ofSeconds(2);
+		return Duration.ofSeconds(10);
+	}
+
+	protected Duration getInitializationTimeout() {
+		return Duration.ofSeconds(1);
 	}
 
 	@BeforeEach
@@ -63,7 +67,8 @@ public abstract class AbstractMcpSyncClientTests {
 
 		assertThatCode(() -> {
 			mcpSyncClient = McpClient.sync(mcpTransport)
-				.requestTimeout(getTimeoutDuration())
+				.requestTimeout(getRequestTimeout())
+				.initializationTimeout(getInitializationTimeout())
 				.capabilities(ClientCapabilities.builder().roots(true).build())
 				.build();
 		}).doesNotThrowAnyException();
@@ -215,7 +220,7 @@ public abstract class AbstractMcpSyncClientTests {
 		var transport = createMcpTransport();
 
 		var client = McpClient.sync(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.roots(new Root("file:///test/path", "test-root"))
 			.build();
 
@@ -313,7 +318,7 @@ public abstract class AbstractMcpSyncClientTests {
 
 		var transport = createMcpTransport();
 		var client = McpClient.sync(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.toolsChangeConsumer(tools -> toolsNotificationReceived.set(true))
 			.resourcesChangeConsumer(resources -> resourcesNotificationReceived.set(true))
 			.promptsChangeConsumer(prompts -> promptsNotificationReceived.set(true))
@@ -351,7 +356,7 @@ public abstract class AbstractMcpSyncClientTests {
 		var transport = createMcpTransport();
 
 		var client = McpClient.sync(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.loggingConsumer(notification -> logReceived.set(true))
 			.build();
 

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

@@ -88,7 +88,6 @@ public class McpAsyncClient {
 
 	/**
 	 * The max timeout to await for the client-server connection to be initialized.
-	 * Usually x2 the request timeout. // TODO should we make it configurable?
 	 */
 	private final Duration initializationTimeout;
 
@@ -151,18 +150,21 @@ public class McpAsyncClient {
 	 * timeout.
 	 * @param transport the transport to use.
 	 * @param requestTimeout the session request-response timeout.
+	 * @param initializationTimeout the max timeout to await for the client-server
 	 * @param features the MCP Client supported features.
 	 */
-	McpAsyncClient(ClientMcpTransport transport, Duration requestTimeout, McpClientFeatures.Async features) {
+	McpAsyncClient(ClientMcpTransport transport, Duration requestTimeout, Duration initializationTimeout,
+			McpClientFeatures.Async features) {
 
 		Assert.notNull(transport, "Transport must not be null");
 		Assert.notNull(requestTimeout, "Request timeout must not be null");
+		Assert.notNull(initializationTimeout, "Initialization timeout must not be null");
 
 		this.clientInfo = features.clientInfo();
 		this.clientCapabilities = features.clientCapabilities();
 		this.transport = transport;
 		this.roots = new ConcurrentHashMap<>(features.roots());
-		this.initializationTimeout = requestTimeout.multipliedBy(2);
+		this.initializationTimeout = initializationTimeout;
 
 		// Request Handlers
 		Map<String, RequestHandler<?>> requestHandlers = new HashMap<>();
@@ -367,12 +369,13 @@ public class McpAsyncClient {
 
 	/**
 	 * Sends a ping request to the server.
-	 * @return A Mono that completes with the server's ping response
+	 * @return A Mono that completes when the server responds to the ping
 	 */
-	public Mono<Object> ping() {
+	public Mono<Void> ping() {
 		return this.withInitializationCheck("pinging the server", initializedResult -> this.mcpSession
 			.sendRequest(McpSchema.METHOD_PING, null, new TypeReference<Object>() {
-			}));
+			})
+			.then());
 	}
 
 	// --------------------------
@@ -771,7 +774,9 @@ public class McpAsyncClient {
 	 * @see McpSchema.LoggingLevel
 	 */
 	public Mono<Void> setLoggingLevel(LoggingLevel loggingLevel) {
-		Assert.notNull(loggingLevel, "Logging level must not be null");
+		if (loggingLevel == null) {
+			return Mono.error(new McpError("Logging level must not be null"));
+		}
 
 		return this.withInitializationCheck("setting logging level", initializedResult -> {
 			String levelName = this.transport.unmarshalFrom(loggingLevel, new TypeReference<String>() {

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

@@ -157,6 +157,8 @@ public interface McpClient {
 
 		private Duration requestTimeout = Duration.ofSeconds(20); // Default timeout
 
+		private Duration initializationTimeout = Duration.ofSeconds(20);
+
 		private ClientCapabilities capabilities;
 
 		private Implementation clientInfo = new Implementation("Java SDK MCP Client", "1.0.0");
@@ -193,6 +195,18 @@ public interface McpClient {
 			return this;
 		}
 
+		/**
+		 * @param initializationTimeout The duration to wait for the initializaiton
+		 * lifecycle step to complete.
+		 * @return This builder instance for method chaining
+		 * @throws IllegalArgumentException if initializationTimeout is null
+		 */
+		public SyncSpec initializationTimeout(Duration initializationTimeout) {
+			Assert.notNull(initializationTimeout, "Initialization timeout must not be null");
+			this.initializationTimeout = initializationTimeout;
+			return this;
+		}
+
 		/**
 		 * Sets the client capabilities that will be advertised to the server during
 		 * connection initialization. Capabilities define what features the client
@@ -354,7 +368,8 @@ public interface McpClient {
 
 			McpClientFeatures.Async asyncFeatures = McpClientFeatures.Async.fromSync(syncFeatures);
 
-			return new McpSyncClient(new McpAsyncClient(transport, this.requestTimeout, asyncFeatures));
+			return new McpSyncClient(
+					new McpAsyncClient(transport, this.requestTimeout, this.initializationTimeout, asyncFeatures));
 		}
 
 	}
@@ -381,6 +396,8 @@ public interface McpClient {
 
 		private Duration requestTimeout = Duration.ofSeconds(20); // Default timeout
 
+		private Duration initializationTimeout = Duration.ofSeconds(20);
+
 		private ClientCapabilities capabilities;
 
 		private Implementation clientInfo = new Implementation("Spring AI MCP Client", "0.3.1");
@@ -417,6 +434,18 @@ public interface McpClient {
 			return this;
 		}
 
+		/**
+		 * @param initializationTimeout The duration to wait for the initializaiton
+		 * lifecycle step to complete.
+		 * @return This builder instance for method chaining
+		 * @throws IllegalArgumentException if initializationTimeout is null
+		 */
+		public AsyncSpec initializationTimeout(Duration initializationTimeout) {
+			Assert.notNull(initializationTimeout, "Initialization timeout must not be null");
+			this.initializationTimeout = initializationTimeout;
+			return this;
+		}
+
 		/**
 		 * Sets the client capabilities that will be advertised to the server during
 		 * connection initialization. Capabilities define what features the client
@@ -574,7 +603,7 @@ public interface McpClient {
 		 * @return a new instance of {@link McpAsyncClient}.
 		 */
 		public McpAsyncClient build() {
-			return new McpAsyncClient(this.transport, this.requestTimeout,
+			return new McpAsyncClient(this.transport, this.requestTimeout, this.initializationTimeout,
 					new McpClientFeatures.Async(this.clientInfo, this.capabilities, this.roots,
 							this.toolsChangeConsumers, this.resourcesChangeConsumers, this.promptsChangeConsumers,
 							this.loggingConsumers, this.samplingHandler));

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

@@ -179,11 +179,10 @@ public class McpSyncClient implements AutoCloseable {
 	}
 
 	/**
-	 * Send a synchronous ping request.
+	 * Send a synchronous ping request to the server.
-	 * @return
 	 */
-	public Object ping() {
+	public void ping() {
-		return this.delegate.ping().block();
+		this.delegate.ping().block();
 	}
 
 	// --------------------------

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

@@ -59,7 +59,11 @@ public abstract class AbstractMcpAsyncClientTests {
 	protected void onClose() {
 	}
 
-	protected Duration getTimeoutDuration() {
+	protected Duration getRequestTimeout() {
+		return Duration.ofSeconds(10);
+	}
+
+	protected Duration getInitializationTimeout() {
 		return Duration.ofSeconds(2);
 	}
 
@@ -70,7 +74,8 @@ public abstract class AbstractMcpAsyncClientTests {
 
 		assertThatCode(() -> {
 			mcpAsyncClient = McpClient.async(mcpTransport)
-				.requestTimeout(getTimeoutDuration())
+				.requestTimeout(getRequestTimeout())
+				.initializationTimeout(getInitializationTimeout())
 				.capabilities(ClientCapabilities.builder().roots(true).build())
 				.build();
 		}).doesNotThrowAnyException();
@@ -79,105 +84,110 @@ public abstract class AbstractMcpAsyncClientTests {
 	@AfterEach
 	void tearDown() {
 		if (mcpAsyncClient != null) {
-			assertThatCode(() -> mcpAsyncClient.closeGracefully().block(Duration.ofSeconds(10)))
+			StepVerifier.create(mcpAsyncClient.closeGracefully()).verifyComplete();
-				.doesNotThrowAnyException();
 		}
 		onClose();
 	}
 
 	@Test
 	void testConstructorWithInvalidArguments() {
-		assertThatThrownBy(() -> McpClient.sync(null).build()).isInstanceOf(IllegalArgumentException.class)
+		assertThatThrownBy(() -> McpClient.async(null).build()).isInstanceOf(IllegalArgumentException.class)
 			.hasMessage("Transport must not be null");
 
-		assertThatThrownBy(() -> McpClient.sync(mcpTransport).requestTimeout(null).build())
+		assertThatThrownBy(() -> McpClient.async(mcpTransport).requestTimeout(null).build())
 			.isInstanceOf(IllegalArgumentException.class)
 			.hasMessage("Request timeout must not be null");
 	}
 
 	@Test
 	void testListToolsWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.listTools(null).block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.listTools(null)).expectErrorSatisfies(error -> {
-			.hasMessage("Client must be initialized before listing tools");
+			assertThat(error).isInstanceOf(McpError.class)
+				.hasMessage("Client must be initialized before listing tools");
+		}).verify();
 	}
 
 	@Test
 	void testListTools() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listTools(null)))
+			.consumeNextWith(result -> {
+				assertThat(result.tools()).isNotNull().isNotEmpty();
 
-		StepVerifier.create(mcpAsyncClient.listTools(null)).consumeNextWith(result -> {
+				Tool firstTool = result.tools().get(0);
-			assertThat(result.tools()).isNotNull().isNotEmpty();
+				assertThat(firstTool.name()).isNotNull();
-
+				assertThat(firstTool.description()).isNotNull();
-			Tool firstTool = result.tools().get(0);
+			})
-			assertThat(firstTool.name()).isNotNull();
+			.verifyComplete();
-			assertThat(firstTool.description()).isNotNull();
-		}).verifyComplete();
 	}
 
 	@Test
 	void testPingWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.ping().block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.ping())
-			.hasMessage("Client must be initialized before pinging the server");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before pinging the server"))
+			.verify();
 	}
 
 	@Test
 	void testPing() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.ping())).verifyComplete();
-		assertThatCode(() -> mcpAsyncClient.ping().block()).doesNotThrowAnyException();
 	}
 
 	@Test
 	void testCallToolWithoutInitialization() {
 		CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
 
-		assertThatThrownBy(() -> mcpAsyncClient.callTool(callToolRequest).block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.callTool(callToolRequest))
-			.hasMessage("Client must be initialized before calling tools");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before calling tools"))
+			.verify();
 	}
 
 	@Test
 	void testCallTool() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
-
 		CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
 
-		StepVerifier.create(mcpAsyncClient.callTool(callToolRequest)).consumeNextWith(callToolResult -> {
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(callToolRequest)))
-			assertThat(callToolResult).isNotNull().satisfies(result -> {
+			.consumeNextWith(callToolResult -> {
-				assertThat(result.content()).isNotNull();
+				assertThat(callToolResult).isNotNull();
-				assertThat(result.isError()).isNull();
+				assertThat(callToolResult.content()).isNotNull();
-			});
+				assertThat(callToolResult.isError()).isNull();
-		}).verifyComplete();
+			})
+			.verifyComplete();
 	}
 
 	@Test
 	void testCallToolWithInvalidTool() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
-
 		CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", ECHO_TEST_MESSAGE));
 
-		assertThatThrownBy(() -> mcpAsyncClient.callTool(invalidRequest).block()).isInstanceOf(Exception.class);
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(invalidRequest)))
+			.expectError(Exception.class)
+			.verify();
 	}
 
 	@Test
 	void testListResourcesWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.listResources(null).block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.listResources(null))
-			.hasMessage("Client must be initialized before listing resources");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before listing resources"))
+			.verify();
 	}
 
 	@Test
 	void testListResources() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResources(null)))
+			.consumeNextWith(resources -> {
+				assertThat(resources).isNotNull().satisfies(result -> {
+					assertThat(result.resources()).isNotNull();
 
-		StepVerifier.create(mcpAsyncClient.listResources(null)).consumeNextWith(resources -> {
+					if (!result.resources().isEmpty()) {
-			assertThat(resources).isNotNull().satisfies(result -> {
+						Resource firstResource = result.resources().get(0);
-				assertThat(result.resources()).isNotNull();
+						assertThat(firstResource.uri()).isNotNull();
-
+						assertThat(firstResource.name()).isNotNull();
-				if (!result.resources().isEmpty()) {
+					}
-					Resource firstResource = result.resources().get(0);
+				});
-					assertThat(firstResource.uri()).isNotNull();
+			})
-					assertThat(firstResource.name()).isNotNull();
+			.verifyComplete();
-				}
-			});
-		}).verifyComplete();
 	}
 
 	@Test
@@ -187,40 +197,44 @@ public abstract class AbstractMcpAsyncClientTests {
 
 	@Test
 	void testListPromptsWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.listPrompts(null).block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.listPrompts(null))
-			.hasMessage("Client must be initialized before listing prompts");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before listing prompts"))
+			.verify();
 	}
 
 	@Test
 	void testListPrompts() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listPrompts(null)))
+			.consumeNextWith(prompts -> {
+				assertThat(prompts).isNotNull().satisfies(result -> {
+					assertThat(result.prompts()).isNotNull();
 
-		StepVerifier.create(mcpAsyncClient.listPrompts(null)).consumeNextWith(prompts -> {
+					if (!result.prompts().isEmpty()) {
-			assertThat(prompts).isNotNull().satisfies(result -> {
+						Prompt firstPrompt = result.prompts().get(0);
-				assertThat(result.prompts()).isNotNull();
+						assertThat(firstPrompt.name()).isNotNull();
-
+						assertThat(firstPrompt.description()).isNotNull();
-				if (!result.prompts().isEmpty()) {
+					}
-					Prompt firstPrompt = result.prompts().get(0);
+				});
-					assertThat(firstPrompt.name()).isNotNull();
+			})
-					assertThat(firstPrompt.description()).isNotNull();
+			.verifyComplete();
-				}
-			});
-		}).verifyComplete();
 	}
 
 	@Test
 	void testGetPromptWithoutInitialization() {
 		GetPromptRequest request = new GetPromptRequest("simple_prompt", Map.of());
 
-		assertThatThrownBy(() -> mcpAsyncClient.getPrompt(request).block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.getPrompt(request))
-			.hasMessage("Client must be initialized before getting prompts");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before getting prompts"))
+			.verify();
 	}
 
 	@Test
 	void testGetPrompt() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		GetPromptRequest request = new GetPromptRequest("simple_prompt", Map.of());
 
-		StepVerifier.create(mcpAsyncClient.getPrompt(new GetPromptRequest("simple_prompt", Map.of())))
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.getPrompt(request)))
 			.consumeNextWith(prompt -> {
 				assertThat(prompt).isNotNull().satisfies(result -> {
 					assertThat(result.messages()).isNotEmpty();
@@ -232,15 +246,16 @@ public abstract class AbstractMcpAsyncClientTests {
 
 	@Test
 	void testRootsListChangedWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.rootsListChangedNotification().block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.rootsListChangedNotification())
-			.hasMessage("Client must be initialized before sending roots list changed notification");
+			.expectErrorMatches(error -> error instanceof McpError && error.getMessage()
+				.equals("Client must be initialized before sending roots list changed notification"))
+			.verify();
 	}
 
 	@Test
 	void testRootsListChanged() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.rootsListChangedNotification()))
-
+			.verifyComplete();
-		assertThatCode(() -> mcpAsyncClient.rootsListChangedNotification().block()).doesNotThrowAnyException();
 	}
 
 	@Test
@@ -248,39 +263,39 @@ public abstract class AbstractMcpAsyncClientTests {
 		var transport = createMcpTransport();
 
 		var client = McpClient.async(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.roots(new Root("file:///test/path", "test-root"))
 			.build();
 
-		assertThatCode(() -> client.initialize().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
+		StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
-
-		assertThatCode(() -> client.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
 	}
 
 	@Test
 	void testAddRoot() {
 		Root newRoot = new Root("file:///new/test/path", "new-test-root");
-		assertThatCode(() -> mcpAsyncClient.addRoot(newRoot).block()).doesNotThrowAnyException();
+
+		StepVerifier.create(mcpAsyncClient.addRoot(newRoot)).verifyComplete();
 	}
 
 	@Test
 	void testAddRootWithNullValue() {
-		assertThatThrownBy(() -> mcpAsyncClient.addRoot(null).block()).hasMessageContaining("Root must not be null");
+		StepVerifier.create(mcpAsyncClient.addRoot(null))
+			.expectErrorMatches(error -> error.getMessage().contains("Root must not be null"))
+			.verify();
 	}
 
 	@Test
 	void testRemoveRoot() {
 		Root root = new Root("file:///test/path/to/remove", "root-to-remove");
-		assertThatCode(() -> {
+
-			mcpAsyncClient.addRoot(root).block();
+		StepVerifier.create(mcpAsyncClient.addRoot(root).then(mcpAsyncClient.removeRoot(root.uri()))).verifyComplete();
-			mcpAsyncClient.removeRoot(root.uri()).block();
-		}).doesNotThrowAnyException();
 	}
 
 	@Test
 	void testRemoveNonExistentRoot() {
-		assertThatThrownBy(() -> mcpAsyncClient.removeRoot("nonexistent-uri").block())
+		StepVerifier.create(mcpAsyncClient.removeRoot("nonexistent-uri"))
-			.hasMessageContaining("Root with uri 'nonexistent-uri' not found");
+			.expectErrorMatches(error -> error.getMessage().contains("Root with uri 'nonexistent-uri' not found"))
+			.verify();
 	}
 
 	@Test
@@ -299,18 +314,20 @@ public abstract class AbstractMcpAsyncClientTests {
 
 	@Test
 	void testListResourceTemplatesWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.listResourceTemplates().block()).isInstanceOf(McpError.class)
+		StepVerifier.create(mcpAsyncClient.listResourceTemplates())
-			.hasMessage("Client must be initialized before listing resource templates");
+			.expectErrorMatches(error -> error instanceof McpError
+					&& error.getMessage().equals("Client must be initialized before listing resource templates"))
+			.verify();
 	}
 
 	@Test
 	void testListResourceTemplates() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResourceTemplates()))
-
+			.consumeNextWith(result -> {
-		StepVerifier.create(mcpAsyncClient.listResourceTemplates()).consumeNextWith(result -> {
+				assertThat(result).isNotNull();
-			assertThat(result).isNotNull();
+				assertThat(result.resourceTemplates()).isNotNull();
-			assertThat(result.resourceTemplates()).isNotNull();
+			})
-		}).verifyComplete();
+			.verifyComplete();
 	}
 
 	// @Test
@@ -338,16 +355,13 @@ public abstract class AbstractMcpAsyncClientTests {
 
 		var transport = createMcpTransport();
 		var client = McpClient.async(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.toolsChangeConsumer(tools -> Mono.fromRunnable(() -> toolsNotificationReceived.set(true)))
 			.resourcesChangeConsumer(resources -> Mono.fromRunnable(() -> resourcesNotificationReceived.set(true)))
 			.promptsChangeConsumer(prompts -> Mono.fromRunnable(() -> promptsNotificationReceived.set(true)))
 			.build();
 
-		assertThatCode(() -> {
+		StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
-			client.initialize().block();
-			client.closeGracefully().block();
-		}).doesNotThrowAnyException();
 	}
 
 	@Test
@@ -357,15 +371,12 @@ public abstract class AbstractMcpAsyncClientTests {
 		var capabilities = ClientCapabilities.builder().sampling().build();
 
 		var client = McpClient.async(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.capabilities(capabilities)
 			.sampling(request -> Mono.just(CreateMessageResult.builder().message("test").model("test-model").build()))
 			.build();
 
-		assertThatCode(() -> {
+		StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
-			client.initialize().block(Duration.ofSeconds(10));
-			client.closeGracefully().block(Duration.ofSeconds(10));
-		}).doesNotThrowAnyException();
 	}
 
 	@Test
@@ -381,17 +392,17 @@ public abstract class AbstractMcpAsyncClientTests {
 		Function<CreateMessageRequest, Mono<CreateMessageResult>> samplingHandler = request -> Mono
 			.just(CreateMessageResult.builder().message("test").model("test-model").build());
 		var client = McpClient.async(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.capabilities(capabilities)
 			.sampling(samplingHandler)
 			.build();
 
-		assertThatCode(() -> {
+		StepVerifier.create(client.initialize()).consumeNextWith(result -> {
-			var result = client.initialize().block(Duration.ofSeconds(10));
 			assertThat(result).isNotNull();
 			assertThat(result.capabilities()).isNotNull();
-			client.closeGracefully().block(Duration.ofSeconds(10));
+		}).verifyComplete();
-		}).doesNotThrowAnyException();
+
+		StepVerifier.create(client.closeGracefully()).verifyComplete();
 	}
 
 	// ---------------------------------------
@@ -400,19 +411,23 @@ public abstract class AbstractMcpAsyncClientTests {
 
 	@Test
 	void testLoggingLevelsWithoutInitialization() {
-		assertThatThrownBy(() -> mcpAsyncClient.setLoggingLevel(McpSchema.LoggingLevel.DEBUG).block())
+		StepVerifier.create(mcpAsyncClient.setLoggingLevel(McpSchema.LoggingLevel.DEBUG))
-			.isInstanceOf(McpError.class)
+			.expectErrorMatches(error -> error instanceof McpError
-			.hasMessage("Client must be initialized before setting logging level");
+					&& error.getMessage().equals("Client must be initialized before setting logging level"))
+			.verify();
 	}
 
 	@Test
 	void testLoggingLevels() {
-		mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
+		Mono<Void> testAllLevels = mcpAsyncClient.initialize().then(Mono.defer(() -> {
+			Mono<Void> chain = Mono.empty();
+			for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
+				chain = chain.then(mcpAsyncClient.setLoggingLevel(level));
+			}
+			return chain;
+		}));
 
-		// Test all logging levels
+		StepVerifier.create(testAllLevels).verifyComplete();
-		for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
-			StepVerifier.create(mcpAsyncClient.setLoggingLevel(level)).verifyComplete();
-		}
 	}
 
 	@Test
@@ -421,20 +436,18 @@ public abstract class AbstractMcpAsyncClientTests {
 		var transport = createMcpTransport();
 
 		var client = McpClient.async(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.loggingConsumer(notification -> Mono.fromRunnable(() -> logReceived.set(true)))
 			.build();
 
-		assertThatCode(() -> {
+		StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
-			client.initialize().block(Duration.ofSeconds(10));
-			client.closeGracefully().block(Duration.ofSeconds(10));
-		}).doesNotThrowAnyException();
 	}
 
 	@Test
 	void testLoggingWithNullNotification() {
-		assertThatThrownBy(() -> mcpAsyncClient.setLoggingLevel(null).block())
+		StepVerifier.create(mcpAsyncClient.setLoggingLevel(null))
-			.hasMessageContaining("Logging level must not be null");
+			.expectErrorMatches(error -> error.getMessage().contains("Logging level must not be null"))
+			.verify();
 	}
 
 }

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

@@ -53,7 +53,11 @@ public abstract class AbstractMcpSyncClientTests {
 
 	abstract protected void onClose();
 
-	protected Duration getTimeoutDuration() {
+	protected Duration getRequestTimeout() {
+		return Duration.ofSeconds(10);
+	}
+
+	protected Duration getInitializationTimeout() {
 		return Duration.ofSeconds(2);
 	}
 
@@ -64,7 +68,8 @@ public abstract class AbstractMcpSyncClientTests {
 
 		assertThatCode(() -> {
 			mcpSyncClient = McpClient.sync(mcpTransport)
-				.requestTimeout(getTimeoutDuration())
+				.requestTimeout(getRequestTimeout())
+				.initializationTimeout(getInitializationTimeout())
 				.capabilities(ClientCapabilities.builder().roots(true).build())
 				.build();
 		}).doesNotThrowAnyException();
@@ -216,7 +221,7 @@ public abstract class AbstractMcpSyncClientTests {
 		var transport = createMcpTransport();
 
 		var client = McpClient.sync(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.roots(new Root("file:///test/path", "test-root"))
 			.build();
 
@@ -314,7 +319,7 @@ public abstract class AbstractMcpSyncClientTests {
 
 		var transport = createMcpTransport();
 		var client = McpClient.sync(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.toolsChangeConsumer(tools -> toolsNotificationReceived.set(true))
 			.resourcesChangeConsumer(resources -> resourcesNotificationReceived.set(true))
 			.promptsChangeConsumer(prompts -> promptsNotificationReceived.set(true))
@@ -352,7 +357,7 @@ public abstract class AbstractMcpSyncClientTests {
 		var transport = createMcpTransport();
 
 		var client = McpClient.sync(transport)
-			.requestTimeout(getTimeoutDuration())
+			.requestTimeout(getRequestTimeout())
 			.loggingConsumer(notification -> logReceived.set(true))
 			.build();
 

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

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2024-2024 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client;
+
+import java.time.Duration;
+
+import io.modelcontextprotocol.client.transport.HttpClientSseClientTransport;
+import io.modelcontextprotocol.spec.ClientMcpTransport;
+import org.junit.jupiter.api.Timeout;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+
+/**
+ * Tests for the {@link McpSyncClient} with {@link HttpClientSseClientTransport}.
+ *
+ * @author Christian Tzolov
+ */
+@Timeout(15) // Giving extra time beyond the client timeout
+class HttpSseMcpAsyncClientTests extends AbstractMcpAsyncClientTests {
+
+	String host = "http://localhost:3004";
+
+	// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
+	@SuppressWarnings("resource")
+	GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v1")
+		.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
+		.withExposedPorts(3001)
+		.waitingFor(Wait.forHttp("/").forStatusCode(404));
+
+	@Override
+	protected ClientMcpTransport createMcpTransport() {
+		return new HttpClientSseClientTransport(host);
+	}
+
+	@Override
+	protected void onStart() {
+		container.start();
+		int port = container.getMappedPort(3001);
+		host = "http://" + container.getHost() + ":" + port;
+	}
+
+	@Override
+	protected void onClose() {
+		container.stop();
+	}
+
+}

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

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2024-2024 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client;
+
+import java.time.Duration;
+
+import io.modelcontextprotocol.client.transport.HttpClientSseClientTransport;
+import io.modelcontextprotocol.spec.ClientMcpTransport;
+import org.junit.jupiter.api.Timeout;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+
+/**
+ * Tests for the {@link McpSyncClient} with {@link HttpClientSseClientTransport}.
+ *
+ * @author Christian Tzolov
+ */
+@Timeout(15) // Giving extra time beyond the client timeout
+class HttpSseMcpSyncClientTests extends AbstractMcpSyncClientTests {
+
+	String host = "http://localhost:3003";
+
+	// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
+	@SuppressWarnings("resource")
+	GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v1")
+		.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
+		.withExposedPorts(3001)
+		.waitingFor(Wait.forHttp("/").forStatusCode(404));
+
+	@Override
+	protected ClientMcpTransport createMcpTransport() {
+		return new HttpClientSseClientTransport(host);
+	}
+
+	@Override
+	protected void onStart() {
+		container.start();
+		int port = container.getMappedPort(3001);
+		host = "http://" + container.getHost() + ":" + port;
+	}
+
+	@Override
+	protected void onClose() {
+		container.stop();
+	}
+
+}

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

@@ -39,7 +39,7 @@ class StdioMcpSyncClientTests extends AbstractMcpSyncClientTests {
 		((StdioClientTransport) mcpTransport).setStdErrorHandler(error -> receivedError.set(error));
 
 		String errorMessage = "Test error";
-		((StdioClientTransport) mcpTransport).getErrorSink().tryEmitNext(errorMessage);
+		((StdioClientTransport) mcpTransport).getErrorSink().emitNext(errorMessage, null);
 
 		assertThat(receivedError.get()).isNotNull().isEqualTo(errorMessage);
 	}