modelcontextprotocol/java-sdk
Watch
1
0
Fork
You've already forked java-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Improve client test reliability and execution time

Browse source 
This change uses VirtualTimeScheduler and pretends enough time has
passed to trigger a timeout on the initialization.
Another problem with reliability of the tests was that the used
testcontainer for the SSE server does not support multiple clients and
the existence of both the global client for the entire suite and some
customized local clients in some tests caused responses to be delivered
to the other client at some racing situations. Now each test creates a
dedicated client and performs cleanup locally.
While these tests were improved, two other issues were found and fixed.
The first one is that the closeGracefully of DefaultMcpSession was not
lazy and would trigger connection disposal before the returned Mono was
subscribed. The second one was dealing with closing the StdIo client
before the process was started. In such a case there should not be an
error but rather a warning and successful completion.
This commit is contained in:
Dariusz Jędrzejczyk 2025-03-19 15:08:43 +01:00 • committed by Christian Tzolov
commit 37120f2ae5
7 changed files with 1016 additions and 718 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -6,7 +6,10 @@ package io.modelcontextprotocol.client;
import java.time.Duration; import java.time.Duration;
import java.util.Map; import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function; import java.util.function.Function;
import io.modelcontextprotocol.spec.ClientMcpTransport; import io.modelcontextprotocol.spec.ClientMcpTransport;
@ -44,10 +47,6 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
*/ */
public abstract class AbstractMcpAsyncClientTests { public abstract class AbstractMcpAsyncClientTests {
private McpAsyncClient mcpAsyncClient;
protected ClientMcpTransport mcpTransport;
private static final String ECHO_TEST_MESSAGE = "Hello MCP Spring AI!"; private static final String ECHO_TEST_MESSAGE = "Hello MCP Spring AI!";
abstract protected ClientMcpTransport createMcpTransport(); abstract protected ClientMcpTransport createMcpTransport();
@ -66,25 +65,47 @@ public abstract class AbstractMcpAsyncClientTests {
return Duration.ofSeconds(2); return Duration.ofSeconds(2);
} }
McpAsyncClient client(ClientMcpTransport transport) {
return client(transport, Function.identity());
}
McpAsyncClient client(ClientMcpTransport transport, Function<McpClient.AsyncSpec, McpClient.AsyncSpec> customizer) {
AtomicReference<McpAsyncClient> client = new AtomicReference<>();
assertThatCode(() -> {
McpClient.AsyncSpec builder = McpClient.async(transport)
.requestTimeout(getRequestTimeout())
.initializationTimeout(getInitializationTimeout())
.capabilities(ClientCapabilities.builder().roots(true).build());
builder = customizer.apply(builder);
client.set(builder.build());
}).doesNotThrowAnyException();
return client.get();
}
void withClient(ClientMcpTransport transport, Consumer<McpAsyncClient> c) {
withClient(transport, Function.identity(), c);
}
void withClient(ClientMcpTransport transport, Function<McpClient.AsyncSpec, McpClient.AsyncSpec> customizer,
Consumer<McpAsyncClient> c) {
var client = client(transport, customizer);
try {
c.accept(client);
}
finally {
StepVerifier.create(client.closeGracefully()).expectComplete().verify(Duration.ofSeconds(10));
}
}
@BeforeEach @BeforeEach
void setUp() { void setUp() {
onStart(); onStart();
this.mcpTransport = createMcpTransport();
assertThatCode(() -> {
mcpAsyncClient = McpClient.async(mcpTransport)
.requestTimeout(getRequestTimeout())
.initializationTimeout(getInitializationTimeout())
.capabilities(ClientCapabilities.builder().roots(true).build())
.build();
}).doesNotThrowAnyException();
} }
@AfterEach @AfterEach
void tearDown() { void tearDown() {
if (mcpAsyncClient != null) {
StepVerifier.create(mcpAsyncClient.closeGracefully()).verifyComplete();
}
onClose(); onClose();
} }
@ -93,258 +114,323 @@ public abstract class AbstractMcpAsyncClientTests {
assertThatThrownBy(() -> McpClient.async(null).build()).isInstanceOf(IllegalArgumentException.class) assertThatThrownBy(() -> McpClient.async(null).build()).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Transport must not be null"); .hasMessage("Transport must not be null");
assertThatThrownBy(() -> McpClient.async(mcpTransport).requestTimeout(null).build()) assertThatThrownBy(() -> McpClient.async(createMcpTransport()).requestTimeout(null).build())
.isInstanceOf(IllegalArgumentException.class) .isInstanceOf(IllegalArgumentException.class)
.hasMessage("Request timeout must not be null"); .hasMessage("Request timeout must not be null");
} }
@Test @Test
void testListToolsWithoutInitialization() { void testListToolsWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.listTools(null)).expectErrorSatisfies(error -> { withClient(createMcpTransport(), mcpAsyncClient -> {
assertThat(error).isInstanceOf(McpError.class) StepVerifier.withVirtualTime(() -> mcpAsyncClient.listTools(null))
.hasMessage("Client must be initialized before listing tools"); .expectSubscription()
}).verify(); .thenAwait(getInitializationTimeout())
.consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before listing tools"))
.verify();
});
} }
@Test @Test
void testListTools() { void testListTools() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listTools(null))) withClient(createMcpTransport(), mcpAsyncClient -> {
.consumeNextWith(result -> { StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listTools(null)))
assertThat(result.tools()).isNotNull().isNotEmpty(); .consumeNextWith(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0); Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull(); assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull(); assertThat(firstTool.description()).isNotNull();
}) })
.verifyComplete(); .verifyComplete();
});
} }
@Test @Test
void testPingWithoutInitialization() { void testPingWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.ping()) withClient(createMcpTransport(), mcpAsyncClient -> {
.expectErrorMatches(error -> error instanceof McpError StepVerifier.withVirtualTime(() -> mcpAsyncClient.ping())
&& error.getMessage().equals("Client must be initialized before pinging the server")) .expectSubscription()
.verify(); .thenAwait(getInitializationTimeout())
.consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before pinging the " + "server"))
.verify();
});
} }
@Test @Test
void testPing() { void testPing() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.ping())).consumeNextWith(callToolResult -> { withClient(createMcpTransport(), mcpAsyncClient -> {
}).verifyComplete(); StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.ping()))
.expectNextCount(1)
.verifyComplete();
});
} }
@Test @Test
void testCallToolWithoutInitialization() { void testCallToolWithoutInitialization() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE)); withClient(createMcpTransport(), mcpAsyncClient -> {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
StepVerifier.create(mcpAsyncClient.callTool(callToolRequest)) StepVerifier.withVirtualTime(() -> mcpAsyncClient.callTool(callToolRequest))
.expectErrorMatches(error -> error instanceof McpError .expectSubscription()
&& error.getMessage().equals("Client must be initialized before calling tools")) .thenAwait(getInitializationTimeout())
.verify(); .consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before calling tools"))
.verify();
});
} }
@Test @Test
void testCallTool() { void testCallTool() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE)); withClient(createMcpTransport(), mcpAsyncClient -> {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(callToolRequest))) StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(callToolRequest)))
.consumeNextWith(callToolResult -> { .consumeNextWith(callToolResult -> {
assertThat(callToolResult).isNotNull(); assertThat(callToolResult).isNotNull().satisfies(result -> {
assertThat(callToolResult.content()).isNotNull(); assertThat(result.content()).isNotNull();
assertThat(callToolResult.isError()).isNull(); assertThat(result.isError()).isNull();
}) });
.verifyComplete(); })
.verifyComplete();
});
} }
@Test @Test
void testCallToolWithInvalidTool() { void testCallToolWithInvalidTool() {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", ECHO_TEST_MESSAGE)); withClient(createMcpTransport(), mcpAsyncClient -> {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool",
Map.of("message", ECHO_TEST_MESSAGE));
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(invalidRequest))) StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(invalidRequest)))
.expectError(Exception.class) .consumeErrorWith(
.verify(); e -> assertThat(e).isInstanceOf(McpError.class).hasMessage("Unknown tool: nonexistent_tool"))
.verify();
});
} }
@Test @Test
void testListResourcesWithoutInitialization() { void testListResourcesWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.listResources(null)) withClient(createMcpTransport(), mcpAsyncClient -> {
.expectErrorMatches(error -> error instanceof McpError StepVerifier.withVirtualTime(() -> mcpAsyncClient.listResources(null))
&& error.getMessage().equals("Client must be initialized before listing resources")) .expectSubscription()
.verify(); .thenAwait(getInitializationTimeout())
.consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before listing resources"))
.verify();
});
} }
@Test @Test
void testListResources() { void testListResources() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResources(null))) withClient(createMcpTransport(), mcpAsyncClient -> {
.consumeNextWith(resources -> { StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResources(null)))
assertThat(resources).isNotNull().satisfies(result -> { .consumeNextWith(resources -> {
assertThat(result.resources()).isNotNull(); assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) { if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0); Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull(); assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull(); assertThat(firstResource.name()).isNotNull();
} }
}); });
}) })
.verifyComplete(); .verifyComplete();
});
} }
@Test @Test
void testMcpAsyncClientState() { void testMcpAsyncClientState() {
assertThat(mcpAsyncClient).isNotNull(); withClient(createMcpTransport(), mcpAsyncClient -> {
assertThat(mcpAsyncClient).isNotNull();
});
} }
@Test @Test
void testListPromptsWithoutInitialization() { void testListPromptsWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.listPrompts(null)) withClient(createMcpTransport(), mcpAsyncClient -> {
.expectErrorMatches(error -> error instanceof McpError StepVerifier.withVirtualTime(() -> mcpAsyncClient.listPrompts(null))
&& error.getMessage().equals("Client must be initialized before listing prompts")) .expectSubscription()
.verify(); .thenAwait(getInitializationTimeout())
.consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before listing prompts"))
.verify();
});
} }
@Test @Test
void testListPrompts() { void testListPrompts() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listPrompts(null))) withClient(createMcpTransport(), mcpAsyncClient -> {
.consumeNextWith(prompts -> { StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listPrompts(null)))
assertThat(prompts).isNotNull().satisfies(result -> { .consumeNextWith(prompts -> {
assertThat(result.prompts()).isNotNull(); assertThat(prompts).isNotNull().satisfies(result -> {
assertThat(result.prompts()).isNotNull();
if (!result.prompts().isEmpty()) { if (!result.prompts().isEmpty()) {
Prompt firstPrompt = result.prompts().get(0); Prompt firstPrompt = result.prompts().get(0);
assertThat(firstPrompt.name()).isNotNull(); assertThat(firstPrompt.name()).isNotNull();
assertThat(firstPrompt.description()).isNotNull(); assertThat(firstPrompt.description()).isNotNull();
} }
}); });
}) })
.verifyComplete(); .verifyComplete();
});
} }
@Test @Test
void testGetPromptWithoutInitialization() { void testGetPromptWithoutInitialization() {
GetPromptRequest request = new GetPromptRequest("simple_prompt", Map.of()); withClient(createMcpTransport(), mcpAsyncClient -> {
GetPromptRequest request = new GetPromptRequest("simple_prompt", Map.of());
StepVerifier.create(mcpAsyncClient.getPrompt(request)) StepVerifier.withVirtualTime(() -> mcpAsyncClient.getPrompt(request))
.expectErrorMatches(error -> error instanceof McpError .expectSubscription()
&& error.getMessage().equals("Client must be initialized before getting prompts")) .thenAwait(getInitializationTimeout())
.verify(); .consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before getting prompts"))
.verify();
});
} }
@Test @Test
void testGetPrompt() { void testGetPrompt() {
GetPromptRequest request = new GetPromptRequest("simple_prompt", Map.of()); withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.getPrompt(request))) .create(mcpAsyncClient.initialize()
.consumeNextWith(prompt -> { .then(mcpAsyncClient.getPrompt(new GetPromptRequest("simple_prompt", Map.of()))))
assertThat(prompt).isNotNull().satisfies(result -> { .consumeNextWith(prompt -> {
assertThat(result.messages()).isNotEmpty(); assertThat(prompt).isNotNull().satisfies(result -> {
assertThat(result.messages()).hasSize(1); assertThat(result.messages()).isNotEmpty();
}); assertThat(result.messages()).hasSize(1);
}) });
.verifyComplete(); })
.verifyComplete();
});
} }
@Test @Test
void testRootsListChangedWithoutInitialization() { void testRootsListChangedWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.rootsListChangedNotification()) withClient(createMcpTransport(), mcpAsyncClient -> {
.expectErrorMatches(error -> error instanceof McpError && error.getMessage() StepVerifier.withVirtualTime(() -> mcpAsyncClient.rootsListChangedNotification())
.equals("Client must be initialized before sending roots list changed notification")) .expectSubscription()
.verify(); .thenAwait(getInitializationTimeout())
.consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before sending roots list changed notification"))
.verify();
});
} }
@Test @Test
void testRootsListChanged() { void testRootsListChanged() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.rootsListChangedNotification())) withClient(createMcpTransport(), mcpAsyncClient -> {
.verifyComplete(); StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.rootsListChangedNotification()))
.verifyComplete();
});
} }
@Test @Test
void testInitializeWithRootsListProviders() { void testInitializeWithRootsListProviders() {
var transport = createMcpTransport(); withClient(createMcpTransport(), builder -> builder.roots(new Root("file:///test/path", "test-root")),
client -> {
var client = McpClient.async(transport) StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
.requestTimeout(getRequestTimeout()) });
.roots(new Root("file:///test/path", "test-root"))
.build();
StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
} }
@Test @Test
void testAddRoot() { void testAddRoot() {
Root newRoot = new Root("file:///new/test/path", "new-test-root"); withClient(createMcpTransport(), mcpAsyncClient -> {
Root newRoot = new Root("file:///new/test/path", "new-test-root");
StepVerifier.create(mcpAsyncClient.addRoot(newRoot)).verifyComplete(); StepVerifier.create(mcpAsyncClient.addRoot(newRoot)).verifyComplete();
});
} }
@Test @Test
void testAddRootWithNullValue() { void testAddRootWithNullValue() {
StepVerifier.create(mcpAsyncClient.addRoot(null)) withClient(createMcpTransport(), mcpAsyncClient -> {
.expectErrorMatches(error -> error.getMessage().contains("Root must not be null")) StepVerifier.create(mcpAsyncClient.addRoot(null))
.verify(); .consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class).hasMessage("Root must not be null"))
.verify();
});
} }
@Test @Test
void testRemoveRoot() { void testRemoveRoot() {
Root root = new Root("file:///test/path/to/remove", "root-to-remove"); withClient(createMcpTransport(), mcpAsyncClient -> {
Root root = new Root("file:///test/path/to/remove", "root-to-remove");
StepVerifier.create(mcpAsyncClient.addRoot(root)).verifyComplete();
StepVerifier.create(mcpAsyncClient.addRoot(root).then(mcpAsyncClient.removeRoot(root.uri()))).verifyComplete(); StepVerifier.create(mcpAsyncClient.removeRoot(root.uri())).verifyComplete();
});
} }
@Test @Test
void testRemoveNonExistentRoot() { void testRemoveNonExistentRoot() {
StepVerifier.create(mcpAsyncClient.removeRoot("nonexistent-uri")) withClient(createMcpTransport(), mcpAsyncClient -> {
.expectErrorMatches(error -> error.getMessage().contains("Root with uri 'nonexistent-uri' not found")) StepVerifier.create(mcpAsyncClient.removeRoot("nonexistent-uri"))
.verify(); .consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Root with uri 'nonexistent-uri' not found"))
.verify();
});
} }
@Test @Test
@Disabled @Disabled
void testReadResource() { void testReadResource() {
StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> { withClient(createMcpTransport(), mcpAsyncClient -> {
if (!resources.resources().isEmpty()) { StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> {
Resource firstResource = resources.resources().get(0); if (!resources.resources().isEmpty()) {
StepVerifier.create(mcpAsyncClient.readResource(firstResource)).consumeNextWith(result -> { Resource firstResource = resources.resources().get(0);
assertThat(result).isNotNull(); StepVerifier.create(mcpAsyncClient.readResource(firstResource)).consumeNextWith(result -> {
assertThat(result.contents()).isNotNull(); assertThat(result).isNotNull();
}).verifyComplete(); assertThat(result.contents()).isNotNull();
} }).verifyComplete();
}).verifyComplete(); }
}).verifyComplete();
});
} }
@Test @Test
void testListResourceTemplatesWithoutInitialization() { void testListResourceTemplatesWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.listResourceTemplates()) withClient(createMcpTransport(), mcpAsyncClient -> {
.expectErrorMatches(error -> error instanceof McpError StepVerifier.withVirtualTime(() -> mcpAsyncClient.listResourceTemplates())
&& error.getMessage().equals("Client must be initialized before listing resource templates")) .expectSubscription()
.verify(); .thenAwait(getInitializationTimeout())
.consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before listing resource templates"))
.verify();
});
} }
@Test @Test
void testListResourceTemplates() { void testListResourceTemplates() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResourceTemplates())) withClient(createMcpTransport(), mcpAsyncClient -> {
.consumeNextWith(result -> { StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResourceTemplates()))
assertThat(result).isNotNull(); .consumeNextWith(result -> {
assertThat(result.resourceTemplates()).isNotNull(); assertThat(result).isNotNull();
}) assertThat(result.resourceTemplates()).isNotNull();
.verifyComplete(); })
.verifyComplete();
});
} }
// @Test // @Test
void testResourceSubscription() { void testResourceSubscription() {
StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> { withClient(createMcpTransport(), mcpAsyncClient -> {
if (!resources.resources().isEmpty()) { StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> {
Resource firstResource = resources.resources().get(0); if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0);
// Test subscribe // Test subscribe
StepVerifier.create(mcpAsyncClient.subscribeResource(new SubscribeRequest(firstResource.uri()))) StepVerifier.create(mcpAsyncClient.subscribeResource(new SubscribeRequest(firstResource.uri())))
.verifyComplete(); .verifyComplete();
// Test unsubscribe // Test unsubscribe
StepVerifier.create(mcpAsyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri()))) StepVerifier.create(mcpAsyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())))
.verifyComplete(); .verifyComplete();
} }
}).verifyComplete(); }).verifyComplete();
});
} }
@Test @Test
@ -353,36 +439,44 @@ public abstract class AbstractMcpAsyncClientTests {
AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false); AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false);
AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false); AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false);
var transport = createMcpTransport(); withClient(createMcpTransport(),
var client = McpClient.async(transport) builder -> builder
.requestTimeout(getRequestTimeout()) .toolsChangeConsumer(tools -> Mono.fromRunnable(() -> toolsNotificationReceived.set(true)))
.toolsChangeConsumer(tools -> Mono.fromRunnable(() -> toolsNotificationReceived.set(true))) .resourcesChangeConsumer(
.resourcesChangeConsumer(resources -> Mono.fromRunnable(() -> resourcesNotificationReceived.set(true))) resources -> Mono.fromRunnable(() -> resourcesNotificationReceived.set(true)))
.promptsChangeConsumer(prompts -> Mono.fromRunnable(() -> promptsNotificationReceived.set(true))) .promptsChangeConsumer(prompts -> Mono.fromRunnable(() -> promptsNotificationReceived.set(true))),
.build(); mcpAsyncClient -> {
StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete(); var transport = createMcpTransport();
var client = McpClient.async(transport)
.requestTimeout(getRequestTimeout())
.toolsChangeConsumer(tools -> Mono.fromRunnable(() -> toolsNotificationReceived.set(true)))
.resourcesChangeConsumer(
resources -> Mono.fromRunnable(() -> resourcesNotificationReceived.set(true)))
.promptsChangeConsumer(
prompts -> Mono.fromRunnable(() -> promptsNotificationReceived.set(true)))
.build();
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
});
} }
@Test @Test
void testInitializeWithSamplingCapability() { void testInitializeWithSamplingCapability() {
var transport = createMcpTransport(); ClientCapabilities capabilities = ClientCapabilities.builder().sampling().build();
CreateMessageResult createMessageResult = CreateMessageResult.builder()
var capabilities = ClientCapabilities.builder().sampling().build(); .message("test")
.model("test-model")
var client = McpClient.async(transport)
.requestTimeout(getRequestTimeout())
.capabilities(capabilities)
.sampling(request -> Mono.just(CreateMessageResult.builder().message("test").model("test-model").build()))
.build(); .build();
withClient(createMcpTransport(),
StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete(); builder -> builder.capabilities(capabilities).sampling(request -> Mono.just(createMessageResult)),
client -> {
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
});
} }
@Test @Test
void testInitializeWithAllCapabilities() { void testInitializeWithAllCapabilities() {
var transport = createMcpTransport();
var capabilities = ClientCapabilities.builder() var capabilities = ClientCapabilities.builder()
.experimental(Map.of("feature", "test")) .experimental(Map.of("feature", "test"))
.roots(true) .roots(true)
@ -391,18 +485,14 @@ public abstract class AbstractMcpAsyncClientTests {
Function<CreateMessageRequest, Mono<CreateMessageResult>> samplingHandler = request -> Mono Function<CreateMessageRequest, Mono<CreateMessageResult>> samplingHandler = request -> Mono
.just(CreateMessageResult.builder().message("test").model("test-model").build()); .just(CreateMessageResult.builder().message("test").model("test-model").build());
var client = McpClient.async(transport)
.requestTimeout(getRequestTimeout())
.capabilities(capabilities)
.sampling(samplingHandler)
.build();
StepVerifier.create(client.initialize()).consumeNextWith(result -> { withClient(createMcpTransport(), builder -> builder.capabilities(capabilities).sampling(samplingHandler),
assertThat(result).isNotNull(); client ->
assertThat(result.capabilities()).isNotNull();
}).verifyComplete();
StepVerifier.create(client.closeGracefully()).verifyComplete(); StepVerifier.create(client.initialize()).assertNext(result -> {
assertThat(result).isNotNull();
assertThat(result.capabilities()).isNotNull();
}).verifyComplete());
} }
// --------------------------------------- // ---------------------------------------
@ -411,43 +501,52 @@ public abstract class AbstractMcpAsyncClientTests {
@Test @Test
void testLoggingLevelsWithoutInitialization() { void testLoggingLevelsWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.setLoggingLevel(McpSchema.LoggingLevel.DEBUG)) withClient(createMcpTransport(),
.expectErrorMatches(error -> error instanceof McpError mcpAsyncClient -> StepVerifier
&& error.getMessage().equals("Client must be initialized before setting logging level")) .withVirtualTime(() -> mcpAsyncClient.setLoggingLevel(McpSchema.LoggingLevel.DEBUG))
.verify(); .expectSubscription()
.thenAwait(getInitializationTimeout())
.consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before setting logging level"))
.verify());
} }
@Test @Test
void testLoggingLevels() { void testLoggingLevels() {
Mono<Void> testAllLevels = mcpAsyncClient.initialize().then(Mono.defer(() -> { withClient(createMcpTransport(), mcpAsyncClient -> {
Mono<Void> chain = Mono.empty(); Mono<Void> testAllLevels = mcpAsyncClient.initialize().then(Mono.defer(() -> {
for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) { Mono<Void> chain = Mono.empty();
chain = chain.then(mcpAsyncClient.setLoggingLevel(level)); for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
} chain = chain.then(mcpAsyncClient.setLoggingLevel(level));
return chain; }
})); return chain;
}));
StepVerifier.create(testAllLevels).verifyComplete(); StepVerifier.create(testAllLevels).verifyComplete();
});
} }
@Test @Test
void testLoggingConsumer() { void testLoggingConsumer() {
AtomicBoolean logReceived = new AtomicBoolean(false); AtomicBoolean logReceived = new AtomicBoolean(false);
var transport = createMcpTransport();
var client = McpClient.async(transport) withClient(createMcpTransport(),
.requestTimeout(getRequestTimeout()) builder -> builder.loggingConsumer(notification -> Mono.fromRunnable(() -> logReceived.set(true))),
.loggingConsumer(notification -> Mono.fromRunnable(() -> logReceived.set(true))) client -> {
.build(); StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
StepVerifier.create(client.closeGracefully()).verifyComplete();
});
StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
} }
@Test @Test
void testLoggingWithNullNotification() { void testLoggingWithNullNotification() {
StepVerifier.create(mcpAsyncClient.setLoggingLevel(null)) withClient(createMcpTransport(), mcpAsyncClient -> {
.expectErrorMatches(error -> error.getMessage().contains("Logging level must not be null")) StepVerifier.create(mcpAsyncClient.setLoggingLevel(null))
.verify(); .expectErrorMatches(error -> error.getMessage().contains("Logging level must not be null"))
.verify();
});
} }
} }

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

@ -7,6 +7,9 @@ package io.modelcontextprotocol.client;
import java.time.Duration; import java.time.Duration;
import java.util.Map; import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import io.modelcontextprotocol.spec.ClientMcpTransport; import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.McpError; import io.modelcontextprotocol.spec.McpError;
@ -27,6 +30,10 @@ import io.modelcontextprotocol.spec.McpSchema.UnsubscribeRequest;
import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatCode;
@ -40,12 +47,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
*/ */
public abstract class AbstractMcpSyncClientTests { public abstract class AbstractMcpSyncClientTests {
private McpSyncClient mcpSyncClient;
private static final String TEST_MESSAGE = "Hello MCP Spring AI!"; private static final String TEST_MESSAGE = "Hello MCP Spring AI!";
protected ClientMcpTransport mcpTransport;
abstract protected ClientMcpTransport createMcpTransport(); abstract protected ClientMcpTransport createMcpTransport();
protected void onStart() { protected void onStart() {
@ -62,254 +65,322 @@ public abstract class AbstractMcpSyncClientTests {
return Duration.ofSeconds(2); return Duration.ofSeconds(2);
} }
McpSyncClient client(ClientMcpTransport transport) {
return client(transport, Function.identity());
}
McpSyncClient client(ClientMcpTransport transport, Function<McpClient.SyncSpec, McpClient.SyncSpec> customizer) {
AtomicReference<McpSyncClient> client = new AtomicReference<>();
assertThatCode(() -> {
McpClient.SyncSpec builder = McpClient.sync(transport)
.requestTimeout(getRequestTimeout())
.initializationTimeout(getInitializationTimeout())
.capabilities(ClientCapabilities.builder().roots(true).build());
builder = customizer.apply(builder);
client.set(builder.build());
}).doesNotThrowAnyException();
return client.get();
}
void withClient(ClientMcpTransport transport, Consumer<McpSyncClient> c) {
withClient(transport, Function.identity(), c);
}
void withClient(ClientMcpTransport transport, Function<McpClient.SyncSpec, McpClient.SyncSpec> customizer,
Consumer<McpSyncClient> c) {
var client = client(transport, customizer);
try {
c.accept(client);
}
finally {
assertThat(client.closeGracefully()).isTrue();
}
}
@BeforeEach @BeforeEach
void setUp() { void setUp() {
onStart(); onStart();
this.mcpTransport = createMcpTransport();
assertThatCode(() -> {
mcpSyncClient = McpClient.sync(mcpTransport)
.requestTimeout(getRequestTimeout())
.initializationTimeout(getInitializationTimeout())
.capabilities(ClientCapabilities.builder().roots(true).build())
.build();
}).doesNotThrowAnyException();
} }
@AfterEach @AfterEach
void tearDown() { void tearDown() {
if (mcpSyncClient != null) {
assertThatCode(() -> mcpSyncClient.close()).doesNotThrowAnyException();
}
onClose(); onClose();
} }
static final Object DUMMY_RETURN_VALUE = new Object();
<T> void verifyNotificationTimesOut(Consumer<McpSyncClient> operation, String action) {
verifyCallTimesOut(client -> {
operation.accept(client);
return DUMMY_RETURN_VALUE;
}, action);
}
<T> void verifyCallTimesOut(Function<McpSyncClient, T> operation, String action) {
withClient(createMcpTransport(), mcpSyncClient -> {
// This scheduler is not replaced by virtual time scheduler
Scheduler customScheduler = Schedulers.newBoundedElastic(1, 1, "actualBoundedElastic");
StepVerifier.withVirtualTime(() -> Mono.fromSupplier(() -> operation.apply(mcpSyncClient))
// offload the blocking call to the real scheduler
.subscribeOn(customScheduler))
.expectSubscription()
.thenAwait(getInitializationTimeout())
.consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before " + action))
.verify();
customScheduler.dispose();
});
}
@Test @Test
void testConstructorWithInvalidArguments() { void testConstructorWithInvalidArguments() {
assertThatThrownBy(() -> McpClient.sync(null).build()).isInstanceOf(IllegalArgumentException.class) assertThatThrownBy(() -> McpClient.sync(null).build()).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Transport must not be null"); .hasMessage("Transport must not be null");
assertThatThrownBy(() -> McpClient.sync(mcpTransport).requestTimeout(null).build()) assertThatThrownBy(() -> McpClient.sync(createMcpTransport()).requestTimeout(null).build())
.isInstanceOf(IllegalArgumentException.class) .isInstanceOf(IllegalArgumentException.class)
.hasMessage("Request timeout must not be null"); .hasMessage("Request timeout must not be null");
} }
@Test @Test
void testListToolsWithoutInitialization() { void testListToolsWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.listTools(null)).isInstanceOf(McpError.class) verifyCallTimesOut(client -> client.listTools(null), "listing tools");
.hasMessage("Client must be initialized before listing tools");
} }
@Test @Test
void testListTools() { void testListTools() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
ListToolsResult tools = mcpSyncClient.listTools(null); mcpSyncClient.initialize();
ListToolsResult tools = mcpSyncClient.listTools(null);
assertThat(tools).isNotNull().satisfies(result -> { assertThat(tools).isNotNull().satisfies(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty(); assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0); Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull(); assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull(); assertThat(firstTool.description()).isNotNull();
});
}); });
} }
@Test @Test
void testCallToolsWithoutInitialization() { void testCallToolsWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4)))) verifyCallTimesOut(client -> client.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4))),
.isInstanceOf(McpError.class) "calling tools");
.hasMessage("Client must be initialized before calling tools");
} }
@Test @Test
void testCallTools() { void testCallTools() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
CallToolResult toolResult = mcpSyncClient.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4))); mcpSyncClient.initialize();
CallToolResult toolResult = mcpSyncClient.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4)));
assertThat(toolResult).isNotNull().satisfies(result -> { assertThat(toolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).hasSize(1); assertThat(result.content()).hasSize(1);
TextContent content = (TextContent) result.content().get(0); TextContent content = (TextContent) result.content().get(0);
assertThat(content).isNotNull(); assertThat(content).isNotNull();
assertThat(content.text()).isNotNull(); assertThat(content.text()).isNotNull();
assertThat(content.text()).contains("7"); assertThat(content.text()).contains("7");
});
}); });
} }
@Test @Test
void testPingWithoutInitialization() { void testPingWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.ping()).isInstanceOf(McpError.class) verifyCallTimesOut(client -> client.ping(), "pinging the server");
.hasMessage("Client must be initialized before pinging the server");
} }
@Test @Test
void testPing() { void testPing() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
assertThatCode(() -> mcpSyncClient.ping()).doesNotThrowAnyException(); mcpSyncClient.initialize();
assertThatCode(() -> mcpSyncClient.ping()).doesNotThrowAnyException();
});
} }
@Test @Test
void testCallToolWithoutInitialization() { void testCallToolWithoutInitialization() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE)); CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE));
verifyCallTimesOut(client -> client.callTool(callToolRequest), "calling tools");
assertThatThrownBy(() -> mcpSyncClient.callTool(callToolRequest)).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before calling tools");
} }
@Test @Test
void testCallTool() { void testCallTool() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE)); mcpSyncClient.initialize();
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE));
CallToolResult callToolResult = mcpSyncClient.callTool(callToolRequest); CallToolResult callToolResult = mcpSyncClient.callTool(callToolRequest);
assertThat(callToolResult).isNotNull().satisfies(result -> { assertThat(callToolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).isNotNull(); assertThat(result.content()).isNotNull();
assertThat(result.isError()).isNull(); assertThat(result.isError()).isNull();
});
}); });
} }
@Test @Test
void testCallToolWithInvalidTool() { void testCallToolWithInvalidTool() {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", TEST_MESSAGE)); withClient(createMcpTransport(), mcpSyncClient -> {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", TEST_MESSAGE));
assertThatThrownBy(() -> mcpSyncClient.callTool(invalidRequest)).isInstanceOf(Exception.class); assertThatThrownBy(() -> mcpSyncClient.callTool(invalidRequest)).isInstanceOf(Exception.class);
});
} }
@Test @Test
void testRootsListChangedWithoutInitialization() { void testRootsListChangedWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.rootsListChangedNotification()).isInstanceOf(McpError.class) verifyNotificationTimesOut(client -> client.rootsListChangedNotification(),
.hasMessage("Client must be initialized before sending roots list changed notification"); "sending roots list changed notification");
} }
@Test @Test
void testRootsListChanged() { void testRootsListChanged() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
assertThatCode(() -> mcpSyncClient.rootsListChangedNotification()).doesNotThrowAnyException(); mcpSyncClient.initialize();
assertThatCode(() -> mcpSyncClient.rootsListChangedNotification()).doesNotThrowAnyException();
});
} }
@Test @Test
void testListResourcesWithoutInitialization() { void testListResourcesWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.listResources(null)).isInstanceOf(McpError.class) verifyCallTimesOut(client -> client.listResources(null), "listing resources");
.hasMessage("Client must be initialized before listing resources");
} }
@Test @Test
void testListResources() { void testListResources() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
ListResourcesResult resources = mcpSyncClient.listResources(null); mcpSyncClient.initialize();
ListResourcesResult resources = mcpSyncClient.listResources(null);
assertThat(resources).isNotNull().satisfies(result -> { assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull(); assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) { if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0); Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull(); assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull(); assertThat(firstResource.name()).isNotNull();
} }
});
}); });
} }
@Test @Test
void testClientSessionState() { void testClientSessionState() {
assertThat(mcpSyncClient).isNotNull(); withClient(createMcpTransport(), mcpSyncClient -> {
assertThat(mcpSyncClient).isNotNull();
});
} }
@Test @Test
void testInitializeWithRootsListProviders() { void testInitializeWithRootsListProviders() {
var transport = createMcpTransport(); withClient(createMcpTransport(), builder -> builder.roots(new Root("file:///test/path", "test-root")),
mcpSyncClient -> {
var client = McpClient.sync(transport) assertThatCode(() -> {
.requestTimeout(getRequestTimeout()) mcpSyncClient.initialize();
.roots(new Root("file:///test/path", "test-root")) mcpSyncClient.close();
.build(); }).doesNotThrowAnyException();
});
assertThatCode(() -> {
client.initialize();
client.close();
}).doesNotThrowAnyException();
} }
@Test @Test
void testAddRoot() { void testAddRoot() {
Root newRoot = new Root("file:///new/test/path", "new-test-root"); withClient(createMcpTransport(), mcpSyncClient -> {
assertThatCode(() -> mcpSyncClient.addRoot(newRoot)).doesNotThrowAnyException(); Root newRoot = new Root("file:///new/test/path", "new-test-root");
assertThatCode(() -> mcpSyncClient.addRoot(newRoot)).doesNotThrowAnyException();
});
} }
@Test @Test
void testAddRootWithNullValue() { void testAddRootWithNullValue() {
assertThatThrownBy(() -> mcpSyncClient.addRoot(null)).hasMessageContaining("Root must not be null"); withClient(createMcpTransport(), mcpSyncClient -> {
assertThatThrownBy(() -> mcpSyncClient.addRoot(null)).hasMessageContaining("Root must not be null");
});
} }
@Test @Test
void testRemoveRoot() { void testRemoveRoot() {
Root root = new Root("file:///test/path/to/remove", "root-to-remove"); withClient(createMcpTransport(), mcpSyncClient -> {
assertThatCode(() -> { Root root = new Root("file:///test/path/to/remove", "root-to-remove");
mcpSyncClient.addRoot(root); assertThatCode(() -> {
mcpSyncClient.removeRoot(root.uri()); mcpSyncClient.addRoot(root);
}).doesNotThrowAnyException(); mcpSyncClient.removeRoot(root.uri());
}).doesNotThrowAnyException();
});
} }
@Test @Test
void testRemoveNonExistentRoot() { void testRemoveNonExistentRoot() {
assertThatThrownBy(() -> mcpSyncClient.removeRoot("nonexistent-uri")) withClient(createMcpTransport(), mcpSyncClient -> {
.hasMessageContaining("Root with uri 'nonexistent-uri' not found"); assertThatThrownBy(() -> mcpSyncClient.removeRoot("nonexistent-uri"))
.hasMessageContaining("Root with uri 'nonexistent-uri' not found");
});
} }
@Test @Test
void testReadResourceWithoutInitialization() { void testReadResourceWithoutInitialization() {
assertThatThrownBy(() -> { Resource resource = new Resource("test://uri", "Test Resource", null, null, null);
Resource resource = new Resource("test://uri", "Test Resource", null, null, null); verifyCallTimesOut(client -> client.readResource(resource), "reading resources");
mcpSyncClient.readResource(resource);
}).isInstanceOf(McpError.class).hasMessage("Client must be initialized before reading resources");
} }
@Test @Test
void testReadResource() { void testReadResource() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
ListResourcesResult resources = mcpSyncClient.listResources(null); mcpSyncClient.initialize();
ListResourcesResult resources = mcpSyncClient.listResources(null);
if (!resources.resources().isEmpty()) { if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0); Resource firstResource = resources.resources().get(0);
ReadResourceResult result = mcpSyncClient.readResource(firstResource); ReadResourceResult result = mcpSyncClient.readResource(firstResource);
assertThat(result).isNotNull(); assertThat(result).isNotNull();
assertThat(result.contents()).isNotNull(); assertThat(result.contents()).isNotNull();
} }
});
} }
@Test @Test
void testListResourceTemplatesWithoutInitialization() { void testListResourceTemplatesWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.listResourceTemplates(null)).isInstanceOf(McpError.class) verifyCallTimesOut(client -> client.listResourceTemplates(null), "listing resource templates");
.hasMessage("Client must be initialized before listing resource templates");
} }
@Test @Test
void testListResourceTemplates() { void testListResourceTemplates() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
ListResourceTemplatesResult result = mcpSyncClient.listResourceTemplates(null); mcpSyncClient.initialize();
ListResourceTemplatesResult result = mcpSyncClient.listResourceTemplates(null);
assertThat(result).isNotNull(); assertThat(result).isNotNull();
assertThat(result.resourceTemplates()).isNotNull(); assertThat(result.resourceTemplates()).isNotNull();
});
} }
// @Test // @Test
void testResourceSubscription() { void testResourceSubscription() {
ListResourcesResult resources = mcpSyncClient.listResources(null); withClient(createMcpTransport(), mcpSyncClient -> {
ListResourcesResult resources = mcpSyncClient.listResources(null);
if (!resources.resources().isEmpty()) { if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0); Resource firstResource = resources.resources().get(0);
// Test subscribe // Test subscribe
assertThatCode(() -> mcpSyncClient.subscribeResource(new SubscribeRequest(firstResource.uri()))) assertThatCode(() -> mcpSyncClient.subscribeResource(new SubscribeRequest(firstResource.uri())))
.doesNotThrowAnyException(); .doesNotThrowAnyException();
// Test unsubscribe // Test unsubscribe
assertThatCode(() -> mcpSyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri()))) assertThatCode(() -> mcpSyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())))
.doesNotThrowAnyException(); .doesNotThrowAnyException();
} }
});
} }
@Test @Test
@ -318,18 +389,17 @@ public abstract class AbstractMcpSyncClientTests {
AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false); AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false);
AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false); AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false);
var transport = createMcpTransport(); withClient(createMcpTransport(),
var client = McpClient.sync(transport) builder -> builder.toolsChangeConsumer(tools -> toolsNotificationReceived.set(true))
.requestTimeout(getRequestTimeout()) .resourcesChangeConsumer(resources -> resourcesNotificationReceived.set(true))
.toolsChangeConsumer(tools -> toolsNotificationReceived.set(true)) .promptsChangeConsumer(prompts -> promptsNotificationReceived.set(true)),
.resourcesChangeConsumer(resources -> resourcesNotificationReceived.set(true)) client -> {
.promptsChangeConsumer(prompts -> promptsNotificationReceived.set(true))
.build();
assertThatCode(() -> { assertThatCode(() -> {
client.initialize(); client.initialize();
client.close(); client.close();
}).doesNotThrowAnyException(); }).doesNotThrowAnyException();
});
} }
// --------------------------------------- // ---------------------------------------
@ -338,40 +408,37 @@ public abstract class AbstractMcpSyncClientTests {
@Test @Test
void testLoggingLevelsWithoutInitialization() { void testLoggingLevelsWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.setLoggingLevel(McpSchema.LoggingLevel.DEBUG)) verifyNotificationTimesOut(client -> client.setLoggingLevel(McpSchema.LoggingLevel.DEBUG),
.isInstanceOf(McpError.class) "setting logging level");
.hasMessage("Client must be initialized before setting logging level");
} }
@Test @Test
void testLoggingLevels() { void testLoggingLevels() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
// Test all logging levels mcpSyncClient.initialize();
for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) { // Test all logging levels
assertThatCode(() -> mcpSyncClient.setLoggingLevel(level)).doesNotThrowAnyException(); for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
} assertThatCode(() -> mcpSyncClient.setLoggingLevel(level)).doesNotThrowAnyException();
}
});
} }
@Test @Test
void testLoggingConsumer() { void testLoggingConsumer() {
AtomicBoolean logReceived = new AtomicBoolean(false); AtomicBoolean logReceived = new AtomicBoolean(false);
var transport = createMcpTransport(); withClient(createMcpTransport(), builder -> builder.requestTimeout(getRequestTimeout())
.loggingConsumer(notification -> logReceived.set(true)), client -> {
var client = McpClient.sync(transport) assertThatCode(() -> {
.requestTimeout(getRequestTimeout()) client.initialize();
.loggingConsumer(notification -> logReceived.set(true)) client.close();
.build(); }).doesNotThrowAnyException();
});
assertThatCode(() -> {
client.initialize();
client.close();
}).doesNotThrowAnyException();
} }
@Test @Test
void testLoggingWithNullNotification() { void testLoggingWithNullNotification() {
assertThatThrownBy(() -> mcpSyncClient.setLoggingLevel(null)) withClient(createMcpTransport(), mcpSyncClient -> assertThatThrownBy(() -> mcpSyncClient.setLoggingLevel(null))
.hasMessageContaining("Logging level must not be null"); .hasMessageContaining("Logging level must not be null"));
} }
} }

7
mcp/src/main/java/io/modelcontextprotocol/client/transport/StdioClientTransport.java
View file

@ -353,14 +353,15 @@ public class StdioClientTransport implements ClientMcpTransport {
// Give a short time for any pending messages to be processed // Give a short time for any pending messages to be processed
return Mono.delay(Duration.ofMillis(100)); return Mono.delay(Duration.ofMillis(100));
})).then(Mono.fromFuture(() -> { })).then(Mono.defer(() -> {
logger.debug("Sending TERM to process"); logger.debug("Sending TERM to process");
if (this.process != null) { if (this.process != null) {
this.process.destroy(); this.process.destroy();
return process.onExit(); return Mono.fromFuture(process.onExit());
} }
else { else {
return CompletableFuture.failedFuture(new RuntimeException("Process not started")); logger.warn("Process not started");
return Mono.empty();
} }
})).doOnNext(process -> { })).doOnNext(process -> {
if (process.exitValue() != 0) { if (process.exitValue() != 0) {

6
mcp/src/main/java/io/modelcontextprotocol/spec/DefaultMcpSession.java
View file

@ -270,8 +270,10 @@ public class DefaultMcpSession implements McpSession {
*/ */
@Override @Override
public Mono<Void> closeGracefully() { public Mono<Void> closeGracefully() {
this.connection.dispose(); return Mono.defer(() -> {
return transport.closeGracefully(); this.connection.dispose();
return transport.closeGracefully();
});
} }
/** /**

460
mcp/src/test/java/io/modelcontextprotocol/client/AbstractMcpAsyncClientTests.java
View file

@ -6,8 +6,12 @@ package io.modelcontextprotocol.client;
import java.time.Duration; import java.time.Duration;
import java.util.Map; import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function; import java.util.function.Function;
import java.util.function.Supplier;
import io.modelcontextprotocol.spec.ClientMcpTransport; import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.McpError; import io.modelcontextprotocol.spec.McpError;
@ -45,10 +49,6 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
// KEEP IN SYNC with the class in mcp-test module // KEEP IN SYNC with the class in mcp-test module
public abstract class AbstractMcpAsyncClientTests { public abstract class AbstractMcpAsyncClientTests {
private McpAsyncClient mcpAsyncClient;
protected ClientMcpTransport mcpTransport;
private static final String ECHO_TEST_MESSAGE = "Hello MCP Spring AI!"; private static final String ECHO_TEST_MESSAGE = "Hello MCP Spring AI!";
abstract protected ClientMcpTransport createMcpTransport(); abstract protected ClientMcpTransport createMcpTransport();
@ -67,285 +67,326 @@ public abstract class AbstractMcpAsyncClientTests {
return Duration.ofSeconds(2); return Duration.ofSeconds(2);
} }
McpAsyncClient client(ClientMcpTransport transport) {
return client(transport, Function.identity());
}
McpAsyncClient client(ClientMcpTransport transport, Function<McpClient.AsyncSpec, McpClient.AsyncSpec> customizer) {
AtomicReference<McpAsyncClient> client = new AtomicReference<>();
assertThatCode(() -> {
McpClient.AsyncSpec builder = McpClient.async(transport)
.requestTimeout(getRequestTimeout())
.initializationTimeout(getInitializationTimeout())
.capabilities(ClientCapabilities.builder().roots(true).build());
builder = customizer.apply(builder);
client.set(builder.build());
}).doesNotThrowAnyException();
return client.get();
}
void withClient(ClientMcpTransport transport, Consumer<McpAsyncClient> c) {
withClient(transport, Function.identity(), c);
}
void withClient(ClientMcpTransport transport, Function<McpClient.AsyncSpec, McpClient.AsyncSpec> customizer,
Consumer<McpAsyncClient> c) {
var client = client(transport, customizer);
try {
c.accept(client);
}
finally {
StepVerifier.create(client.closeGracefully()).expectComplete().verify(Duration.ofSeconds(10));
}
}
@BeforeEach @BeforeEach
void setUp() { void setUp() {
onStart(); onStart();
this.mcpTransport = createMcpTransport();
assertThatCode(() -> {
mcpAsyncClient = McpClient.async(mcpTransport)
.requestTimeout(getRequestTimeout())
.initializationTimeout(getInitializationTimeout())
.capabilities(ClientCapabilities.builder().roots(true).build())
.build();
}).doesNotThrowAnyException();
} }
@AfterEach @AfterEach
void tearDown() { void tearDown() {
if (mcpAsyncClient != null) {
StepVerifier.create(mcpAsyncClient.closeGracefully()).verifyComplete();
}
onClose(); onClose();
} }
<T> void verifyInitializationTimeout(Function<McpAsyncClient, Mono<T>> operation, String action) {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.withVirtualTime(() -> operation.apply(mcpAsyncClient))
.expectSubscription()
.thenAwait(getInitializationTimeout())
.consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before " + action))
.verify();
});
}
@Test @Test
void testConstructorWithInvalidArguments() { void testConstructorWithInvalidArguments() {
assertThatThrownBy(() -> McpClient.async(null).build()).isInstanceOf(IllegalArgumentException.class) assertThatThrownBy(() -> McpClient.async(null).build()).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Transport must not be null"); .hasMessage("Transport must not be null");
assertThatThrownBy(() -> McpClient.async(mcpTransport).requestTimeout(null).build()) assertThatThrownBy(() -> McpClient.async(createMcpTransport()).requestTimeout(null).build())
.isInstanceOf(IllegalArgumentException.class) .isInstanceOf(IllegalArgumentException.class)
.hasMessage("Request timeout must not be null"); .hasMessage("Request timeout must not be null");
} }
@Test @Test
void testListToolsWithoutInitialization() { void testListToolsWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.listTools(null)).expectErrorSatisfies(error -> { verifyInitializationTimeout(client -> client.listTools(null), "listing tools");
assertThat(error).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before listing tools");
}).verify();
} }
@Test @Test
void testListTools() { void testListTools() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listTools(null))) withClient(createMcpTransport(), mcpAsyncClient -> {
.consumeNextWith(result -> { StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listTools(null)))
assertThat(result.tools()).isNotNull().isNotEmpty(); .consumeNextWith(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0); Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull(); assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull(); assertThat(firstTool.description()).isNotNull();
}) })
.verifyComplete(); .verifyComplete();
});
} }
@Test @Test
void testPingWithoutInitialization() { void testPingWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.ping()) verifyInitializationTimeout(client -> client.ping(), "pinging the server");
.expectErrorMatches(error -> error instanceof McpError
&& error.getMessage().equals("Client must be initialized before pinging the server"))
.verify();
} }
@Test @Test
void testPing() { void testPing() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.ping())).consumeNextWith(callToolResult -> { withClient(createMcpTransport(), mcpAsyncClient -> {
}).verifyComplete(); StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.ping()))
.expectNextCount(1)
.verifyComplete();
});
} }
@Test @Test
void testCallToolWithoutInitialization() { void testCallToolWithoutInitialization() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE)); CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
verifyInitializationTimeout(client -> client.callTool(callToolRequest), "calling tools");
StepVerifier.create(mcpAsyncClient.callTool(callToolRequest))
.expectErrorMatches(error -> error instanceof McpError
&& error.getMessage().equals("Client must be initialized before calling tools"))
.verify();
} }
@Test @Test
void testCallTool() { void testCallTool() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE)); withClient(createMcpTransport(), mcpAsyncClient -> {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(callToolRequest))) StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(callToolRequest)))
.consumeNextWith(callToolResult -> { .consumeNextWith(callToolResult -> {
assertThat(callToolResult).isNotNull(); assertThat(callToolResult).isNotNull().satisfies(result -> {
assertThat(callToolResult.content()).isNotNull(); assertThat(result.content()).isNotNull();
assertThat(callToolResult.isError()).isNull(); assertThat(result.isError()).isNull();
}) });
.verifyComplete(); })
.verifyComplete();
});
} }
@Test @Test
void testCallToolWithInvalidTool() { void testCallToolWithInvalidTool() {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", ECHO_TEST_MESSAGE)); withClient(createMcpTransport(), mcpAsyncClient -> {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool",
Map.of("message", ECHO_TEST_MESSAGE));
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(invalidRequest))) StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(invalidRequest)))
.expectError(Exception.class) .consumeErrorWith(
.verify(); e -> assertThat(e).isInstanceOf(McpError.class).hasMessage("Unknown tool: nonexistent_tool"))
.verify();
});
} }
@Test @Test
void testListResourcesWithoutInitialization() { void testListResourcesWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.listResources(null)) verifyInitializationTimeout(client -> client.listResources(null), "listing resources");
.expectErrorMatches(error -> error instanceof McpError
&& error.getMessage().equals("Client must be initialized before listing resources"))
.verify();
} }
@Test @Test
void testListResources() { void testListResources() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResources(null))) withClient(createMcpTransport(), mcpAsyncClient -> {
.consumeNextWith(resources -> { StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResources(null)))
assertThat(resources).isNotNull().satisfies(result -> { .consumeNextWith(resources -> {
assertThat(result.resources()).isNotNull(); assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) { if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0); Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull(); assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull(); assertThat(firstResource.name()).isNotNull();
} }
}); });
}) })
.verifyComplete(); .verifyComplete();
});
} }
@Test @Test
void testMcpAsyncClientState() { void testMcpAsyncClientState() {
assertThat(mcpAsyncClient).isNotNull(); withClient(createMcpTransport(), mcpAsyncClient -> {
assertThat(mcpAsyncClient).isNotNull();
});
} }
@Test @Test
void testListPromptsWithoutInitialization() { void testListPromptsWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.listPrompts(null)) verifyInitializationTimeout(client -> client.listPrompts(null), "listing " + "prompts");
.expectErrorMatches(error -> error instanceof McpError
&& error.getMessage().equals("Client must be initialized before listing prompts"))
.verify();
} }
@Test @Test
void testListPrompts() { void testListPrompts() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listPrompts(null))) withClient(createMcpTransport(), mcpAsyncClient -> {
.consumeNextWith(prompts -> { StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listPrompts(null)))
assertThat(prompts).isNotNull().satisfies(result -> { .consumeNextWith(prompts -> {
assertThat(result.prompts()).isNotNull(); assertThat(prompts).isNotNull().satisfies(result -> {
assertThat(result.prompts()).isNotNull();
if (!result.prompts().isEmpty()) { if (!result.prompts().isEmpty()) {
Prompt firstPrompt = result.prompts().get(0); Prompt firstPrompt = result.prompts().get(0);
assertThat(firstPrompt.name()).isNotNull(); assertThat(firstPrompt.name()).isNotNull();
assertThat(firstPrompt.description()).isNotNull(); assertThat(firstPrompt.description()).isNotNull();
} }
}); });
}) })
.verifyComplete(); .verifyComplete();
});
} }
@Test @Test
void testGetPromptWithoutInitialization() { void testGetPromptWithoutInitialization() {
GetPromptRequest request = new GetPromptRequest("simple_prompt", Map.of()); GetPromptRequest request = new GetPromptRequest("simple_prompt", Map.of());
verifyInitializationTimeout(client -> client.getPrompt(request), "getting " + "prompts");
StepVerifier.create(mcpAsyncClient.getPrompt(request))
.expectErrorMatches(error -> error instanceof McpError
&& error.getMessage().equals("Client must be initialized before getting prompts"))
.verify();
} }
@Test @Test
void testGetPrompt() { void testGetPrompt() {
GetPromptRequest request = new GetPromptRequest("simple_prompt", Map.of()); withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.getPrompt(request))) .create(mcpAsyncClient.initialize()
.consumeNextWith(prompt -> { .then(mcpAsyncClient.getPrompt(new GetPromptRequest("simple_prompt", Map.of()))))
assertThat(prompt).isNotNull().satisfies(result -> { .consumeNextWith(prompt -> {
assertThat(result.messages()).isNotEmpty(); assertThat(prompt).isNotNull().satisfies(result -> {
assertThat(result.messages()).hasSize(1); assertThat(result.messages()).isNotEmpty();
}); assertThat(result.messages()).hasSize(1);
}) });
.verifyComplete(); })
.verifyComplete();
});
} }
@Test @Test
void testRootsListChangedWithoutInitialization() { void testRootsListChangedWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.rootsListChangedNotification()) verifyInitializationTimeout(client -> client.rootsListChangedNotification(),
.expectErrorMatches(error -> error instanceof McpError && error.getMessage() "sending roots list changed notification");
.equals("Client must be initialized before sending roots list changed notification"))
.verify();
} }
@Test @Test
void testRootsListChanged() { void testRootsListChanged() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.rootsListChangedNotification())) withClient(createMcpTransport(), mcpAsyncClient -> {
.verifyComplete(); StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.rootsListChangedNotification()))
.verifyComplete();
});
} }
@Test @Test
void testInitializeWithRootsListProviders() { void testInitializeWithRootsListProviders() {
var transport = createMcpTransport(); withClient(createMcpTransport(), builder -> builder.roots(new Root("file:///test/path", "test-root")),
client -> {
var client = McpClient.async(transport) StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
.requestTimeout(getRequestTimeout()) });
.roots(new Root("file:///test/path", "test-root"))
.build();
StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
} }
@Test @Test
void testAddRoot() { void testAddRoot() {
Root newRoot = new Root("file:///new/test/path", "new-test-root"); withClient(createMcpTransport(), mcpAsyncClient -> {
Root newRoot = new Root("file:///new/test/path", "new-test-root");
StepVerifier.create(mcpAsyncClient.addRoot(newRoot)).verifyComplete(); StepVerifier.create(mcpAsyncClient.addRoot(newRoot)).verifyComplete();
});
} }
@Test @Test
void testAddRootWithNullValue() { void testAddRootWithNullValue() {
StepVerifier.create(mcpAsyncClient.addRoot(null)) withClient(createMcpTransport(), mcpAsyncClient -> {
.expectErrorMatches(error -> error.getMessage().contains("Root must not be null")) StepVerifier.create(mcpAsyncClient.addRoot(null))
.verify(); .consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class).hasMessage("Root must not be null"))
.verify();
});
} }
@Test @Test
void testRemoveRoot() { void testRemoveRoot() {
Root root = new Root("file:///test/path/to/remove", "root-to-remove"); withClient(createMcpTransport(), mcpAsyncClient -> {
Root root = new Root("file:///test/path/to/remove", "root-to-remove");
StepVerifier.create(mcpAsyncClient.addRoot(root)).verifyComplete();
StepVerifier.create(mcpAsyncClient.addRoot(root).then(mcpAsyncClient.removeRoot(root.uri()))).verifyComplete(); StepVerifier.create(mcpAsyncClient.removeRoot(root.uri())).verifyComplete();
});
} }
@Test @Test
void testRemoveNonExistentRoot() { void testRemoveNonExistentRoot() {
StepVerifier.create(mcpAsyncClient.removeRoot("nonexistent-uri")) withClient(createMcpTransport(), mcpAsyncClient -> {
.expectErrorMatches(error -> error.getMessage().contains("Root with uri 'nonexistent-uri' not found")) StepVerifier.create(mcpAsyncClient.removeRoot("nonexistent-uri"))
.verify(); .consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Root with uri 'nonexistent-uri' not found"))
.verify();
});
} }
@Test @Test
@Disabled @Disabled
void testReadResource() { void testReadResource() {
StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> { withClient(createMcpTransport(), mcpAsyncClient -> {
if (!resources.resources().isEmpty()) { StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> {
Resource firstResource = resources.resources().get(0); if (!resources.resources().isEmpty()) {
StepVerifier.create(mcpAsyncClient.readResource(firstResource)).consumeNextWith(result -> { Resource firstResource = resources.resources().get(0);
assertThat(result).isNotNull(); StepVerifier.create(mcpAsyncClient.readResource(firstResource)).consumeNextWith(result -> {
assertThat(result.contents()).isNotNull(); assertThat(result).isNotNull();
}).verifyComplete(); assertThat(result.contents()).isNotNull();
} }).verifyComplete();
}).verifyComplete(); }
}).verifyComplete();
});
} }
@Test @Test
void testListResourceTemplatesWithoutInitialization() { void testListResourceTemplatesWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.listResourceTemplates()) verifyInitializationTimeout(client -> client.listResourceTemplates(), "listing resource templates");
.expectErrorMatches(error -> error instanceof McpError
&& error.getMessage().equals("Client must be initialized before listing resource templates"))
.verify();
} }
@Test @Test
void testListResourceTemplates() { void testListResourceTemplates() {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResourceTemplates())) withClient(createMcpTransport(), mcpAsyncClient -> {
.consumeNextWith(result -> { StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResourceTemplates()))
assertThat(result).isNotNull(); .consumeNextWith(result -> {
assertThat(result.resourceTemplates()).isNotNull(); assertThat(result).isNotNull();
}) assertThat(result.resourceTemplates()).isNotNull();
.verifyComplete(); })
.verifyComplete();
});
} }
// @Test // @Test
void testResourceSubscription() { void testResourceSubscription() {
StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> { withClient(createMcpTransport(), mcpAsyncClient -> {
if (!resources.resources().isEmpty()) { StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> {
Resource firstResource = resources.resources().get(0); if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0);
// Test subscribe // Test subscribe
StepVerifier.create(mcpAsyncClient.subscribeResource(new SubscribeRequest(firstResource.uri()))) StepVerifier.create(mcpAsyncClient.subscribeResource(new SubscribeRequest(firstResource.uri())))
.verifyComplete(); .verifyComplete();
// Test unsubscribe // Test unsubscribe
StepVerifier.create(mcpAsyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri()))) StepVerifier.create(mcpAsyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())))
.verifyComplete(); .verifyComplete();
} }
}).verifyComplete(); }).verifyComplete();
});
} }
@Test @Test
@ -354,36 +395,44 @@ public abstract class AbstractMcpAsyncClientTests {
AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false); AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false);
AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false); AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false);
var transport = createMcpTransport(); withClient(createMcpTransport(),
var client = McpClient.async(transport) builder -> builder
.requestTimeout(getRequestTimeout()) .toolsChangeConsumer(tools -> Mono.fromRunnable(() -> toolsNotificationReceived.set(true)))
.toolsChangeConsumer(tools -> Mono.fromRunnable(() -> toolsNotificationReceived.set(true))) .resourcesChangeConsumer(
.resourcesChangeConsumer(resources -> Mono.fromRunnable(() -> resourcesNotificationReceived.set(true))) resources -> Mono.fromRunnable(() -> resourcesNotificationReceived.set(true)))
.promptsChangeConsumer(prompts -> Mono.fromRunnable(() -> promptsNotificationReceived.set(true))) .promptsChangeConsumer(prompts -> Mono.fromRunnable(() -> promptsNotificationReceived.set(true))),
.build(); mcpAsyncClient -> {
StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete(); var transport = createMcpTransport();
var client = McpClient.async(transport)
.requestTimeout(getRequestTimeout())
.toolsChangeConsumer(tools -> Mono.fromRunnable(() -> toolsNotificationReceived.set(true)))
.resourcesChangeConsumer(
resources -> Mono.fromRunnable(() -> resourcesNotificationReceived.set(true)))
.promptsChangeConsumer(
prompts -> Mono.fromRunnable(() -> promptsNotificationReceived.set(true)))
.build();
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
});
} }
@Test @Test
void testInitializeWithSamplingCapability() { void testInitializeWithSamplingCapability() {
var transport = createMcpTransport(); ClientCapabilities capabilities = ClientCapabilities.builder().sampling().build();
CreateMessageResult createMessageResult = CreateMessageResult.builder()
var capabilities = ClientCapabilities.builder().sampling().build(); .message("test")
.model("test-model")
var client = McpClient.async(transport)
.requestTimeout(getRequestTimeout())
.capabilities(capabilities)
.sampling(request -> Mono.just(CreateMessageResult.builder().message("test").model("test-model").build()))
.build(); .build();
withClient(createMcpTransport(),
StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete(); builder -> builder.capabilities(capabilities).sampling(request -> Mono.just(createMessageResult)),
client -> {
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
});
} }
@Test @Test
void testInitializeWithAllCapabilities() { void testInitializeWithAllCapabilities() {
var transport = createMcpTransport();
var capabilities = ClientCapabilities.builder() var capabilities = ClientCapabilities.builder()
.experimental(Map.of("feature", "test")) .experimental(Map.of("feature", "test"))
.roots(true) .roots(true)
@ -392,18 +441,14 @@ public abstract class AbstractMcpAsyncClientTests {
Function<CreateMessageRequest, Mono<CreateMessageResult>> samplingHandler = request -> Mono Function<CreateMessageRequest, Mono<CreateMessageResult>> samplingHandler = request -> Mono
.just(CreateMessageResult.builder().message("test").model("test-model").build()); .just(CreateMessageResult.builder().message("test").model("test-model").build());
var client = McpClient.async(transport)
.requestTimeout(getRequestTimeout())
.capabilities(capabilities)
.sampling(samplingHandler)
.build();
StepVerifier.create(client.initialize()).consumeNextWith(result -> { withClient(createMcpTransport(), builder -> builder.capabilities(capabilities).sampling(samplingHandler),
assertThat(result).isNotNull(); client ->
assertThat(result.capabilities()).isNotNull();
}).verifyComplete();
StepVerifier.create(client.closeGracefully()).verifyComplete(); StepVerifier.create(client.initialize()).assertNext(result -> {
assertThat(result).isNotNull();
assertThat(result.capabilities()).isNotNull();
}).verifyComplete());
} }
// --------------------------------------- // ---------------------------------------
@ -412,43 +457,46 @@ public abstract class AbstractMcpAsyncClientTests {
@Test @Test
void testLoggingLevelsWithoutInitialization() { void testLoggingLevelsWithoutInitialization() {
StepVerifier.create(mcpAsyncClient.setLoggingLevel(McpSchema.LoggingLevel.DEBUG)) verifyInitializationTimeout(client -> client.setLoggingLevel(McpSchema.LoggingLevel.DEBUG),
.expectErrorMatches(error -> error instanceof McpError "setting logging level");
&& error.getMessage().equals("Client must be initialized before setting logging level"))
.verify();
} }
@Test @Test
void testLoggingLevels() { void testLoggingLevels() {
Mono<Void> testAllLevels = mcpAsyncClient.initialize().then(Mono.defer(() -> { withClient(createMcpTransport(), mcpAsyncClient -> {
Mono<Void> chain = Mono.empty(); Mono<Void> testAllLevels = mcpAsyncClient.initialize().then(Mono.defer(() -> {
for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) { Mono<Void> chain = Mono.empty();
chain = chain.then(mcpAsyncClient.setLoggingLevel(level)); for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
} chain = chain.then(mcpAsyncClient.setLoggingLevel(level));
return chain; }
})); return chain;
}));
StepVerifier.create(testAllLevels).verifyComplete(); StepVerifier.create(testAllLevels).verifyComplete();
});
} }
@Test @Test
void testLoggingConsumer() { void testLoggingConsumer() {
AtomicBoolean logReceived = new AtomicBoolean(false); AtomicBoolean logReceived = new AtomicBoolean(false);
var transport = createMcpTransport();
var client = McpClient.async(transport) withClient(createMcpTransport(),
.requestTimeout(getRequestTimeout()) builder -> builder.loggingConsumer(notification -> Mono.fromRunnable(() -> logReceived.set(true))),
.loggingConsumer(notification -> Mono.fromRunnable(() -> logReceived.set(true))) client -> {
.build(); StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
StepVerifier.create(client.closeGracefully()).verifyComplete();
});
StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
} }
@Test @Test
void testLoggingWithNullNotification() { void testLoggingWithNullNotification() {
StepVerifier.create(mcpAsyncClient.setLoggingLevel(null)) withClient(createMcpTransport(), mcpAsyncClient -> {
.expectErrorMatches(error -> error.getMessage().contains("Logging level must not be null")) StepVerifier.create(mcpAsyncClient.setLoggingLevel(null))
.verify(); .expectErrorMatches(error -> error.getMessage().contains("Logging level must not be null"))
.verify();
});
} }
} }

361
mcp/src/test/java/io/modelcontextprotocol/client/AbstractMcpSyncClientTests.java
View file

@ -7,6 +7,9 @@ package io.modelcontextprotocol.client;
import java.time.Duration; import java.time.Duration;
import java.util.Map; import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import io.modelcontextprotocol.spec.ClientMcpTransport; import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.McpError; import io.modelcontextprotocol.spec.McpError;
@ -27,6 +30,10 @@ import io.modelcontextprotocol.spec.McpSchema.UnsubscribeRequest;
import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatCode;
@ -41,12 +48,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
// KEEP IN SYNC with the class in mcp-test module // KEEP IN SYNC with the class in mcp-test module
public abstract class AbstractMcpSyncClientTests { public abstract class AbstractMcpSyncClientTests {
private McpSyncClient mcpSyncClient;
private static final String TEST_MESSAGE = "Hello MCP Spring AI!"; private static final String TEST_MESSAGE = "Hello MCP Spring AI!";
protected ClientMcpTransport mcpTransport;
abstract protected ClientMcpTransport createMcpTransport(); abstract protected ClientMcpTransport createMcpTransport();
protected void onStart() { protected void onStart() {
@ -63,254 +66,322 @@ public abstract class AbstractMcpSyncClientTests {
return Duration.ofSeconds(2); return Duration.ofSeconds(2);
} }
McpSyncClient client(ClientMcpTransport transport) {
return client(transport, Function.identity());
}
McpSyncClient client(ClientMcpTransport transport, Function<McpClient.SyncSpec, McpClient.SyncSpec> customizer) {
AtomicReference<McpSyncClient> client = new AtomicReference<>();
assertThatCode(() -> {
McpClient.SyncSpec builder = McpClient.sync(transport)
.requestTimeout(getRequestTimeout())
.initializationTimeout(getInitializationTimeout())
.capabilities(ClientCapabilities.builder().roots(true).build());
builder = customizer.apply(builder);
client.set(builder.build());
}).doesNotThrowAnyException();
return client.get();
}
void withClient(ClientMcpTransport transport, Consumer<McpSyncClient> c) {
withClient(transport, Function.identity(), c);
}
void withClient(ClientMcpTransport transport, Function<McpClient.SyncSpec, McpClient.SyncSpec> customizer,
Consumer<McpSyncClient> c) {
var client = client(transport, customizer);
try {
c.accept(client);
}
finally {
assertThat(client.closeGracefully()).isTrue();
}
}
@BeforeEach @BeforeEach
void setUp() { void setUp() {
onStart(); onStart();
this.mcpTransport = createMcpTransport();
assertThatCode(() -> {
mcpSyncClient = McpClient.sync(mcpTransport)
.requestTimeout(getRequestTimeout())
.initializationTimeout(getInitializationTimeout())
.capabilities(ClientCapabilities.builder().roots(true).build())
.build();
}).doesNotThrowAnyException();
} }
@AfterEach @AfterEach
void tearDown() { void tearDown() {
if (mcpSyncClient != null) {
assertThatCode(() -> mcpSyncClient.close()).doesNotThrowAnyException();
}
onClose(); onClose();
} }
static final Object DUMMY_RETURN_VALUE = new Object();
<T> void verifyNotificationTimesOut(Consumer<McpSyncClient> operation, String action) {
verifyCallTimesOut(client -> {
operation.accept(client);
return DUMMY_RETURN_VALUE;
}, action);
}
<T> void verifyCallTimesOut(Function<McpSyncClient, T> operation, String action) {
withClient(createMcpTransport(), mcpSyncClient -> {
// This scheduler is not replaced by virtual time scheduler
Scheduler customScheduler = Schedulers.newBoundedElastic(1, 1, "actualBoundedElastic");
StepVerifier.withVirtualTime(() -> Mono.fromSupplier(() -> operation.apply(mcpSyncClient))
// offload the blocking call to the real scheduler
.subscribeOn(customScheduler))
.expectSubscription()
.thenAwait(getInitializationTimeout())
.consumeErrorWith(e -> assertThat(e).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before " + action))
.verify();
customScheduler.dispose();
});
}
@Test @Test
void testConstructorWithInvalidArguments() { void testConstructorWithInvalidArguments() {
assertThatThrownBy(() -> McpClient.sync(null).build()).isInstanceOf(IllegalArgumentException.class) assertThatThrownBy(() -> McpClient.sync(null).build()).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Transport must not be null"); .hasMessage("Transport must not be null");
assertThatThrownBy(() -> McpClient.sync(mcpTransport).requestTimeout(null).build()) assertThatThrownBy(() -> McpClient.sync(createMcpTransport()).requestTimeout(null).build())
.isInstanceOf(IllegalArgumentException.class) .isInstanceOf(IllegalArgumentException.class)
.hasMessage("Request timeout must not be null"); .hasMessage("Request timeout must not be null");
} }
@Test @Test
void testListToolsWithoutInitialization() { void testListToolsWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.listTools(null)).isInstanceOf(McpError.class) verifyCallTimesOut(client -> client.listTools(null), "listing tools");
.hasMessage("Client must be initialized before listing tools");
} }
@Test @Test
void testListTools() { void testListTools() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
ListToolsResult tools = mcpSyncClient.listTools(null); mcpSyncClient.initialize();
ListToolsResult tools = mcpSyncClient.listTools(null);
assertThat(tools).isNotNull().satisfies(result -> { assertThat(tools).isNotNull().satisfies(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty(); assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0); Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull(); assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull(); assertThat(firstTool.description()).isNotNull();
});
}); });
} }
@Test @Test
void testCallToolsWithoutInitialization() { void testCallToolsWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4)))) verifyCallTimesOut(client -> client.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4))),
.isInstanceOf(McpError.class) "calling tools");
.hasMessage("Client must be initialized before calling tools");
} }
@Test @Test
void testCallTools() { void testCallTools() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
CallToolResult toolResult = mcpSyncClient.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4))); mcpSyncClient.initialize();
CallToolResult toolResult = mcpSyncClient.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4)));
assertThat(toolResult).isNotNull().satisfies(result -> { assertThat(toolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).hasSize(1); assertThat(result.content()).hasSize(1);
TextContent content = (TextContent) result.content().get(0); TextContent content = (TextContent) result.content().get(0);
assertThat(content).isNotNull(); assertThat(content).isNotNull();
assertThat(content.text()).isNotNull(); assertThat(content.text()).isNotNull();
assertThat(content.text()).contains("7"); assertThat(content.text()).contains("7");
});
}); });
} }
@Test @Test
void testPingWithoutInitialization() { void testPingWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.ping()).isInstanceOf(McpError.class) verifyCallTimesOut(client -> client.ping(), "pinging the server");
.hasMessage("Client must be initialized before pinging the server");
} }
@Test @Test
void testPing() { void testPing() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
assertThatCode(() -> mcpSyncClient.ping()).doesNotThrowAnyException(); mcpSyncClient.initialize();
assertThatCode(() -> mcpSyncClient.ping()).doesNotThrowAnyException();
});
} }
@Test @Test
void testCallToolWithoutInitialization() { void testCallToolWithoutInitialization() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE)); CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE));
verifyCallTimesOut(client -> client.callTool(callToolRequest), "calling tools");
assertThatThrownBy(() -> mcpSyncClient.callTool(callToolRequest)).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized before calling tools");
} }
@Test @Test
void testCallTool() { void testCallTool() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE)); mcpSyncClient.initialize();
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE));
CallToolResult callToolResult = mcpSyncClient.callTool(callToolRequest); CallToolResult callToolResult = mcpSyncClient.callTool(callToolRequest);
assertThat(callToolResult).isNotNull().satisfies(result -> { assertThat(callToolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).isNotNull(); assertThat(result.content()).isNotNull();
assertThat(result.isError()).isNull(); assertThat(result.isError()).isNull();
});
}); });
} }
@Test @Test
void testCallToolWithInvalidTool() { void testCallToolWithInvalidTool() {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", TEST_MESSAGE)); withClient(createMcpTransport(), mcpSyncClient -> {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", TEST_MESSAGE));
assertThatThrownBy(() -> mcpSyncClient.callTool(invalidRequest)).isInstanceOf(Exception.class); assertThatThrownBy(() -> mcpSyncClient.callTool(invalidRequest)).isInstanceOf(Exception.class);
});
} }
@Test @Test
void testRootsListChangedWithoutInitialization() { void testRootsListChangedWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.rootsListChangedNotification()).isInstanceOf(McpError.class) verifyNotificationTimesOut(client -> client.rootsListChangedNotification(),
.hasMessage("Client must be initialized before sending roots list changed notification"); "sending roots list changed notification");
} }
@Test @Test
void testRootsListChanged() { void testRootsListChanged() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
assertThatCode(() -> mcpSyncClient.rootsListChangedNotification()).doesNotThrowAnyException(); mcpSyncClient.initialize();
assertThatCode(() -> mcpSyncClient.rootsListChangedNotification()).doesNotThrowAnyException();
});
} }
@Test @Test
void testListResourcesWithoutInitialization() { void testListResourcesWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.listResources(null)).isInstanceOf(McpError.class) verifyCallTimesOut(client -> client.listResources(null), "listing resources");
.hasMessage("Client must be initialized before listing resources");
} }
@Test @Test
void testListResources() { void testListResources() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
ListResourcesResult resources = mcpSyncClient.listResources(null); mcpSyncClient.initialize();
ListResourcesResult resources = mcpSyncClient.listResources(null);
assertThat(resources).isNotNull().satisfies(result -> { assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull(); assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) { if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0); Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull(); assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull(); assertThat(firstResource.name()).isNotNull();
} }
});
}); });
} }
@Test @Test
void testClientSessionState() { void testClientSessionState() {
assertThat(mcpSyncClient).isNotNull(); withClient(createMcpTransport(), mcpSyncClient -> {
assertThat(mcpSyncClient).isNotNull();
});
} }
@Test @Test
void testInitializeWithRootsListProviders() { void testInitializeWithRootsListProviders() {
var transport = createMcpTransport(); withClient(createMcpTransport(), builder -> builder.roots(new Root("file:///test/path", "test-root")),
mcpSyncClient -> {
var client = McpClient.sync(transport) assertThatCode(() -> {
.requestTimeout(getRequestTimeout()) mcpSyncClient.initialize();
.roots(new Root("file:///test/path", "test-root")) mcpSyncClient.close();
.build(); }).doesNotThrowAnyException();
});
assertThatCode(() -> {
client.initialize();
client.close();
}).doesNotThrowAnyException();
} }
@Test @Test
void testAddRoot() { void testAddRoot() {
Root newRoot = new Root("file:///new/test/path", "new-test-root"); withClient(createMcpTransport(), mcpSyncClient -> {
assertThatCode(() -> mcpSyncClient.addRoot(newRoot)).doesNotThrowAnyException(); Root newRoot = new Root("file:///new/test/path", "new-test-root");
assertThatCode(() -> mcpSyncClient.addRoot(newRoot)).doesNotThrowAnyException();
});
} }
@Test @Test
void testAddRootWithNullValue() { void testAddRootWithNullValue() {
assertThatThrownBy(() -> mcpSyncClient.addRoot(null)).hasMessageContaining("Root must not be null"); withClient(createMcpTransport(), mcpSyncClient -> {
assertThatThrownBy(() -> mcpSyncClient.addRoot(null)).hasMessageContaining("Root must not be null");
});
} }
@Test @Test
void testRemoveRoot() { void testRemoveRoot() {
Root root = new Root("file:///test/path/to/remove", "root-to-remove"); withClient(createMcpTransport(), mcpSyncClient -> {
assertThatCode(() -> { Root root = new Root("file:///test/path/to/remove", "root-to-remove");
mcpSyncClient.addRoot(root); assertThatCode(() -> {
mcpSyncClient.removeRoot(root.uri()); mcpSyncClient.addRoot(root);
}).doesNotThrowAnyException(); mcpSyncClient.removeRoot(root.uri());
}).doesNotThrowAnyException();
});
} }
@Test @Test
void testRemoveNonExistentRoot() { void testRemoveNonExistentRoot() {
assertThatThrownBy(() -> mcpSyncClient.removeRoot("nonexistent-uri")) withClient(createMcpTransport(), mcpSyncClient -> {
.hasMessageContaining("Root with uri 'nonexistent-uri' not found"); assertThatThrownBy(() -> mcpSyncClient.removeRoot("nonexistent-uri"))
.hasMessageContaining("Root with uri 'nonexistent-uri' not found");
});
} }
@Test @Test
void testReadResourceWithoutInitialization() { void testReadResourceWithoutInitialization() {
assertThatThrownBy(() -> { Resource resource = new Resource("test://uri", "Test Resource", null, null, null);
Resource resource = new Resource("test://uri", "Test Resource", null, null, null); verifyCallTimesOut(client -> client.readResource(resource), "reading resources");
mcpSyncClient.readResource(resource);
}).isInstanceOf(McpError.class).hasMessage("Client must be initialized before reading resources");
} }
@Test @Test
void testReadResource() { void testReadResource() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
ListResourcesResult resources = mcpSyncClient.listResources(null); mcpSyncClient.initialize();
ListResourcesResult resources = mcpSyncClient.listResources(null);
if (!resources.resources().isEmpty()) { if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0); Resource firstResource = resources.resources().get(0);
ReadResourceResult result = mcpSyncClient.readResource(firstResource); ReadResourceResult result = mcpSyncClient.readResource(firstResource);
assertThat(result).isNotNull(); assertThat(result).isNotNull();
assertThat(result.contents()).isNotNull(); assertThat(result.contents()).isNotNull();
} }
});
} }
@Test @Test
void testListResourceTemplatesWithoutInitialization() { void testListResourceTemplatesWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.listResourceTemplates(null)).isInstanceOf(McpError.class) verifyCallTimesOut(client -> client.listResourceTemplates(null), "listing resource templates");
.hasMessage("Client must be initialized before listing resource templates");
} }
@Test @Test
void testListResourceTemplates() { void testListResourceTemplates() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
ListResourceTemplatesResult result = mcpSyncClient.listResourceTemplates(null); mcpSyncClient.initialize();
ListResourceTemplatesResult result = mcpSyncClient.listResourceTemplates(null);
assertThat(result).isNotNull(); assertThat(result).isNotNull();
assertThat(result.resourceTemplates()).isNotNull(); assertThat(result.resourceTemplates()).isNotNull();
});
} }
// @Test // @Test
void testResourceSubscription() { void testResourceSubscription() {
ListResourcesResult resources = mcpSyncClient.listResources(null); withClient(createMcpTransport(), mcpSyncClient -> {
ListResourcesResult resources = mcpSyncClient.listResources(null);
if (!resources.resources().isEmpty()) { if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0); Resource firstResource = resources.resources().get(0);
// Test subscribe // Test subscribe
assertThatCode(() -> mcpSyncClient.subscribeResource(new SubscribeRequest(firstResource.uri()))) assertThatCode(() -> mcpSyncClient.subscribeResource(new SubscribeRequest(firstResource.uri())))
.doesNotThrowAnyException(); .doesNotThrowAnyException();
// Test unsubscribe // Test unsubscribe
assertThatCode(() -> mcpSyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri()))) assertThatCode(() -> mcpSyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())))
.doesNotThrowAnyException(); .doesNotThrowAnyException();
} }
});
} }
@Test @Test
@ -319,18 +390,17 @@ public abstract class AbstractMcpSyncClientTests {
AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false); AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false);
AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false); AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false);
var transport = createMcpTransport(); withClient(createMcpTransport(),
var client = McpClient.sync(transport) builder -> builder.toolsChangeConsumer(tools -> toolsNotificationReceived.set(true))
.requestTimeout(getRequestTimeout()) .resourcesChangeConsumer(resources -> resourcesNotificationReceived.set(true))
.toolsChangeConsumer(tools -> toolsNotificationReceived.set(true)) .promptsChangeConsumer(prompts -> promptsNotificationReceived.set(true)),
.resourcesChangeConsumer(resources -> resourcesNotificationReceived.set(true)) client -> {
.promptsChangeConsumer(prompts -> promptsNotificationReceived.set(true))
.build();
assertThatCode(() -> { assertThatCode(() -> {
client.initialize(); client.initialize();
client.close(); client.close();
}).doesNotThrowAnyException(); }).doesNotThrowAnyException();
});
} }
// --------------------------------------- // ---------------------------------------
@ -339,40 +409,37 @@ public abstract class AbstractMcpSyncClientTests {
@Test @Test
void testLoggingLevelsWithoutInitialization() { void testLoggingLevelsWithoutInitialization() {
assertThatThrownBy(() -> mcpSyncClient.setLoggingLevel(McpSchema.LoggingLevel.DEBUG)) verifyNotificationTimesOut(client -> client.setLoggingLevel(McpSchema.LoggingLevel.DEBUG),
.isInstanceOf(McpError.class) "setting logging level");
.hasMessage("Client must be initialized before setting logging level");
} }
@Test @Test
void testLoggingLevels() { void testLoggingLevels() {
mcpSyncClient.initialize(); withClient(createMcpTransport(), mcpSyncClient -> {
// Test all logging levels mcpSyncClient.initialize();
for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) { // Test all logging levels
assertThatCode(() -> mcpSyncClient.setLoggingLevel(level)).doesNotThrowAnyException(); for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
} assertThatCode(() -> mcpSyncClient.setLoggingLevel(level)).doesNotThrowAnyException();
}
});
} }
@Test @Test
void testLoggingConsumer() { void testLoggingConsumer() {
AtomicBoolean logReceived = new AtomicBoolean(false); AtomicBoolean logReceived = new AtomicBoolean(false);
var transport = createMcpTransport(); withClient(createMcpTransport(), builder -> builder.requestTimeout(getRequestTimeout())
.loggingConsumer(notification -> logReceived.set(true)), client -> {
var client = McpClient.sync(transport) assertThatCode(() -> {
.requestTimeout(getRequestTimeout()) client.initialize();
.loggingConsumer(notification -> logReceived.set(true)) client.close();
.build(); }).doesNotThrowAnyException();
});
assertThatCode(() -> {
client.initialize();
client.close();
}).doesNotThrowAnyException();
} }
@Test @Test
void testLoggingWithNullNotification() { void testLoggingWithNullNotification() {
assertThatThrownBy(() -> mcpSyncClient.setLoggingLevel(null)) withClient(createMcpTransport(), mcpSyncClient -> assertThatThrownBy(() -> mcpSyncClient.setLoggingLevel(null))
.hasMessageContaining("Logging level must not be null"); .hasMessageContaining("Logging level must not be null"));
} }
} }

20
mcp/src/test/java/io/modelcontextprotocol/client/StdioMcpSyncClientTests.java
View file

@ -5,6 +5,8 @@
package io.modelcontextprotocol.client; package io.modelcontextprotocol.client;
import java.time.Duration; import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import io.modelcontextprotocol.client.transport.ServerParameters; import io.modelcontextprotocol.client.transport.ServerParameters;
@ -13,6 +15,7 @@ import io.modelcontextprotocol.spec.ClientMcpTransport;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout; import org.junit.jupiter.api.Timeout;
import reactor.core.publisher.Sinks; import reactor.core.publisher.Sinks;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
@ -35,15 +38,26 @@ class StdioMcpSyncClientTests extends AbstractMcpSyncClientTests {
} }
@Test @Test
void customErrorHandlerShouldReceiveErrors() { void customErrorHandlerShouldReceiveErrors() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<String> receivedError = new AtomicReference<>(); AtomicReference<String> receivedError = new AtomicReference<>();
((StdioClientTransport) mcpTransport).setStdErrorHandler(error -> receivedError.set(error)); ClientMcpTransport transport = createMcpTransport();
StepVerifier.create(transport.connect(msg -> msg)).verifyComplete();
((StdioClientTransport) transport).setStdErrorHandler(error -> {
receivedError.set(error);
latch.countDown();
});
String errorMessage = "Test error"; String errorMessage = "Test error";
((StdioClientTransport) mcpTransport).getErrorSink().emitNext(errorMessage, Sinks.EmitFailureHandler.FAIL_FAST); ((StdioClientTransport) transport).getErrorSink().emitNext(errorMessage, Sinks.EmitFailureHandler.FAIL_FAST);
assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(receivedError.get()).isNotNull().isEqualTo(errorMessage); assertThat(receivedError.get()).isNotNull().isEqualTo(errorMessage);
StepVerifier.create(transport.closeGracefully()).expectComplete().verify(Duration.ofSeconds(5));
} }
protected Duration getInitializationTimeout() { protected Duration getInitializationTimeout() {