extract mcp-core (#557)

* extract mcp-core
  - This change extracts classes previously in `mcp` to `mcp-core`; a new module which does not depend on `mcp-json-jackon2`.
  - `mcp` module is now an umbrella module that brings `mcp-json-jackon2` and `mcp-core`
* add mcp-core to bom

BREAKING CHANGE: All source code in the previously `mcp` module is moved to the `mcp-core`.  the `mcp` now is just an aggregation for `mcp-core` + `mcp-json-jackson2` for backward compatibility. 
This change won't affect the end users, but will cause conflicts for the existing PRs.
This commit is contained in:
Sergio del Amo 2025-09-17 11:32:08 +02:00 • committed by GitHub
commit 1499df22c7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
65 changed files with 246 additions and 151 deletions

View file

@ -0,0 +1,97 @@
/*
* Copyright 2025-2025 the original author or authors.
*/
package io.modelcontextprotocol;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.List;
import java.util.Map;
import io.modelcontextprotocol.util.DeafaultMcpUriTemplateManagerFactory;
import io.modelcontextprotocol.util.McpUriTemplateManager;
import io.modelcontextprotocol.util.McpUriTemplateManagerFactory;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
/**
* Tests for {@link McpUriTemplateManager} and its implementations.
*
* @author Christian Tzolov
*/
public class McpUriTemplateManagerTests {
private McpUriTemplateManagerFactory uriTemplateFactory;
@BeforeEach
void setUp() {
this.uriTemplateFactory = new DeafaultMcpUriTemplateManagerFactory();
}
@Test
void shouldExtractVariableNamesFromTemplate() {
List<String> variables = this.uriTemplateFactory.create("/api/users/{userId}/posts/{postId}")
.getVariableNames();
assertEquals(2, variables.size());
assertEquals("userId", variables.get(0));
assertEquals("postId", variables.get(1));
}
@Test
void shouldReturnEmptyListWhenTemplateHasNoVariables() {
List<String> variables = this.uriTemplateFactory.create("/api/users/all").getVariableNames();
assertEquals(0, variables.size());
}
@Test
void shouldThrowExceptionWhenExtractingVariablesFromNullTemplate() {
assertThrows(IllegalArgumentException.class, () -> this.uriTemplateFactory.create(null).getVariableNames());
}
@Test
void shouldThrowExceptionWhenExtractingVariablesFromEmptyTemplate() {
assertThrows(IllegalArgumentException.class, () -> this.uriTemplateFactory.create("").getVariableNames());
}
@Test
void shouldThrowExceptionWhenTemplateContainsDuplicateVariables() {
assertThrows(IllegalArgumentException.class,
() -> this.uriTemplateFactory.create("/api/users/{userId}/posts/{userId}").getVariableNames());
}
@Test
void shouldExtractVariableValuesFromRequestUri() {
Map<String, String> values = this.uriTemplateFactory.create("/api/users/{userId}/posts/{postId}")
.extractVariableValues("/api/users/123/posts/456");
assertEquals(2, values.size());
assertEquals("123", values.get("userId"));
assertEquals("456", values.get("postId"));
}
@Test
void shouldReturnEmptyMapWhenTemplateHasNoVariables() {
Map<String, String> values = this.uriTemplateFactory.create("/api/users/all")
.extractVariableValues("/api/users/all");
assertEquals(0, values.size());
}
@Test
void shouldReturnEmptyMapWhenRequestUriIsNull() {
Map<String, String> values = this.uriTemplateFactory.create("/api/users/{userId}/posts/{postId}")
.extractVariableValues(null);
assertEquals(0, values.size());
}
@Test
void shouldMatchUriAgainstTemplatePattern() {
var uriTemplateManager = this.uriTemplateFactory.create("/api/users/{userId}/posts/{postId}");
assertTrue(uriTemplateManager.matches("/api/users/123/posts/456"));
assertFalse(uriTemplateManager.matches("/api/users/123/comments/456"));
}
}

View file

@ -0,0 +1,106 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol;
import java.util.ArrayList;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.Function;
import io.modelcontextprotocol.json.McpJsonMapper;
import io.modelcontextprotocol.json.TypeRef;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCNotification;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCRequest;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
/**
* A mock implementation of the {@link McpClientTransport} interfaces.
*/
public class MockMcpClientTransport implements McpClientTransport {
private final Sinks.Many<McpSchema.JSONRPCMessage> inbound = Sinks.many().unicast().onBackpressureBuffer();
private final List<McpSchema.JSONRPCMessage> sent = new ArrayList<>();
private final BiConsumer<MockMcpClientTransport, McpSchema.JSONRPCMessage> interceptor;
private String protocolVersion = McpSchema.LATEST_PROTOCOL_VERSION;
public MockMcpClientTransport() {
this((t, msg) -> {
});
}
public MockMcpClientTransport(BiConsumer<MockMcpClientTransport, McpSchema.JSONRPCMessage> interceptor) {
this.interceptor = interceptor;
}
public MockMcpClientTransport withProtocolVersion(String protocolVersion) {
return this;
}
@Override
public List<String> protocolVersions() {
return List.of(protocolVersion);
}
public void simulateIncomingMessage(McpSchema.JSONRPCMessage message) {
if (inbound.tryEmitNext(message).isFailure()) {
throw new RuntimeException("Failed to process incoming message " + message);
}
}
@Override
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
sent.add(message);
interceptor.accept(this, message);
return Mono.empty();
}
public McpSchema.JSONRPCRequest getLastSentMessageAsRequest() {
return (JSONRPCRequest) getLastSentMessage();
}
public McpSchema.JSONRPCNotification getLastSentMessageAsNotification() {
return (JSONRPCNotification) getLastSentMessage();
}
public McpSchema.JSONRPCMessage getLastSentMessage() {
return !sent.isEmpty() ? sent.get(sent.size() - 1) : null;
}
private volatile boolean connected = false;
@Override
public Mono<Void> connect(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
if (connected) {
return Mono.error(new IllegalStateException("Already connected"));
}
connected = true;
return inbound.asFlux()
.flatMap(message -> Mono.just(message).transform(handler))
.doFinally(signal -> connected = false)
.then();
}
@Override
public Mono<Void> closeGracefully() {
return Mono.defer(() -> {
connected = false;
inbound.tryEmitComplete();
// Wait for all subscribers to complete
return Mono.empty();
});
}
@Override
public <T> T unmarshalFrom(Object data, TypeRef<T> typeRef) {
return McpJsonMapper.getDefault().convertValue(data, typeRef);
}
}

View file

@ -0,0 +1,74 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol;
import java.util.ArrayList;
import java.util.List;
import java.util.function.BiConsumer;
import io.modelcontextprotocol.json.McpJsonMapper;
import io.modelcontextprotocol.json.TypeRef;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCNotification;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCRequest;
import io.modelcontextprotocol.spec.McpServerTransport;
import reactor.core.publisher.Mono;
/**
* A mock implementation of the {@link McpServerTransport} interfaces.
*/
public class MockMcpServerTransport implements McpServerTransport {
private final List<McpSchema.JSONRPCMessage> sent = new ArrayList<>();
private final BiConsumer<MockMcpServerTransport, McpSchema.JSONRPCMessage> interceptor;
public MockMcpServerTransport() {
this((t, msg) -> {
});
}
public MockMcpServerTransport(BiConsumer<MockMcpServerTransport, McpSchema.JSONRPCMessage> interceptor) {
this.interceptor = interceptor;
}
@Override
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
sent.add(message);
interceptor.accept(this, message);
return Mono.empty();
}
public McpSchema.JSONRPCRequest getLastSentMessageAsRequest() {
return (JSONRPCRequest) getLastSentMessage();
}
public McpSchema.JSONRPCNotification getLastSentMessageAsNotification() {
return (JSONRPCNotification) getLastSentMessage();
}
public McpSchema.JSONRPCMessage getLastSentMessage() {
return !sent.isEmpty() ? sent.get(sent.size() - 1) : null;
}
public void clearSentMessages() {
sent.clear();
}
public List<McpSchema.JSONRPCMessage> getAllSentMessages() {
return new ArrayList<>(sent);
}
@Override
public Mono<Void> closeGracefully() {
return Mono.empty();
}
@Override
public <T> T unmarshalFrom(Object data, TypeRef<T> typeRef) {
return McpJsonMapper.getDefault().convertValue(data, typeRef);
}
}

View file

@ -0,0 +1,50 @@
/*
* Copyright 2025-2025 the original author or authors.
*/
package io.modelcontextprotocol;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpServerSession;
import io.modelcontextprotocol.spec.McpServerSession.Factory;
import io.modelcontextprotocol.spec.McpServerTransportProvider;
import reactor.core.publisher.Mono;
/**
* @author Christian Tzolov
*/
public class MockMcpServerTransportProvider implements McpServerTransportProvider {
private McpServerSession session;
private final MockMcpServerTransport transport;
public MockMcpServerTransportProvider(MockMcpServerTransport transport) {
this.transport = transport;
}
public MockMcpServerTransport getTransport() {
return transport;
}
@Override
public void setSessionFactory(Factory sessionFactory) {
session = sessionFactory.create(transport);
}
@Override
public Mono<Void> notifyClients(String method, Object params) {
return session.sendNotification(method, params);
}
@Override
public Mono<Void> closeGracefully() {
return session.closeGracefully();
}
public void simulateIncomingMessage(McpSchema.JSONRPCMessage message) {
session.handle(message).subscribe();
}
}

View file

@ -0,0 +1,231 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import eu.rekawek.toxiproxy.Proxy;
import eu.rekawek.toxiproxy.ToxiproxyClient;
import eu.rekawek.toxiproxy.model.ToxicDirection;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpTransport;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.ToxiproxyContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import reactor.test.StepVerifier;
import java.io.IOException;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import static org.assertj.core.api.Assertions.assertThatCode;
/**
* Resiliency test suite for the {@link McpAsyncClient} that can be used with different
* {@link McpTransport} implementations that support Streamable HTTP.
*
* The purpose of these tests is to allow validating the transport layer resiliency
* instead of the functionality offered by the logical layer of MCP concepts such as
* tools, resources, prompts, etc.
*
* @author Dariusz Jędrzejczyk
*/
// KEEP IN SYNC with the class in mcp-test module
public abstract class AbstractMcpAsyncClientResiliencyTests {
private static final Logger logger = LoggerFactory.getLogger(AbstractMcpAsyncClientResiliencyTests.class);
static Network network = Network.newNetwork();
static String host = "http://localhost:3001";
// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v3")
.withCommand("node dist/index.js streamableHttp")
.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
.withNetwork(network)
.withNetworkAliases("everything-server")
.withExposedPorts(3001)
.waitingFor(Wait.forHttp("/").forStatusCode(404));
static ToxiproxyContainer toxiproxy = new ToxiproxyContainer("ghcr.io/shopify/toxiproxy:2.5.0").withNetwork(network)
.withExposedPorts(8474, 3000);
static Proxy proxy;
static {
container.start();
toxiproxy.start();
final ToxiproxyClient toxiproxyClient = new ToxiproxyClient(toxiproxy.getHost(), toxiproxy.getControlPort());
try {
proxy = toxiproxyClient.createProxy("everything-server", "0.0.0.0:3000", "everything-server:3001");
}
catch (IOException e) {
throw new RuntimeException("Can't create proxy!", e);
}
final String ipAddressViaToxiproxy = toxiproxy.getHost();
final int portViaToxiproxy = toxiproxy.getMappedPort(3000);
host = "http://" + ipAddressViaToxiproxy + ":" + portViaToxiproxy;
}
static void disconnect() {
long start = System.nanoTime();
try {
// disconnect
// proxy.toxics().bandwidth("CUT_CONNECTION_DOWNSTREAM",
// ToxicDirection.DOWNSTREAM, 0);
// proxy.toxics().bandwidth("CUT_CONNECTION_UPSTREAM",
// ToxicDirection.UPSTREAM, 0);
proxy.toxics().resetPeer("RESET_DOWNSTREAM", ToxicDirection.DOWNSTREAM, 0);
proxy.toxics().resetPeer("RESET_UPSTREAM", ToxicDirection.UPSTREAM, 0);
logger.info("Disconnect took {} ms", Duration.ofNanos(System.nanoTime() - start).toMillis());
}
catch (IOException e) {
throw new RuntimeException("Failed to disconnect", e);
}
}
static void reconnect() {
long start = System.nanoTime();
try {
proxy.toxics().get("RESET_UPSTREAM").remove();
proxy.toxics().get("RESET_DOWNSTREAM").remove();
// proxy.toxics().get("CUT_CONNECTION_DOWNSTREAM").remove();
// proxy.toxics().get("CUT_CONNECTION_UPSTREAM").remove();
logger.info("Reconnect took {} ms", Duration.ofNanos(System.nanoTime() - start).toMillis());
}
catch (IOException e) {
throw new RuntimeException("Failed to reconnect", e);
}
}
static void restartMcpServer() {
container.stop();
container.start();
}
abstract McpClientTransport createMcpTransport();
protected Duration getRequestTimeout() {
return Duration.ofSeconds(14);
}
protected Duration getInitializationTimeout() {
return Duration.ofSeconds(2);
}
McpAsyncClient client(McpClientTransport transport) {
return client(transport, Function.identity());
}
McpAsyncClient client(McpClientTransport transport, Function<McpClient.AsyncSpec, McpClient.AsyncSpec> customizer) {
AtomicReference<McpAsyncClient> client = new AtomicReference<>();
assertThatCode(() -> {
// Do not advertise roots. Otherwise, the server will list roots during
// initialization. The client responds asynchronously, and there might be a
// rest condition in tests where we disconnect right after initialization.
McpClient.AsyncSpec builder = McpClient.async(transport)
.requestTimeout(getRequestTimeout())
.initializationTimeout(getInitializationTimeout())
.capabilities(McpSchema.ClientCapabilities.builder().build());
builder = customizer.apply(builder);
client.set(builder.build());
}).doesNotThrowAnyException();
return client.get();
}
void withClient(McpClientTransport transport, Consumer<McpAsyncClient> c) {
withClient(transport, Function.identity(), c);
}
void withClient(McpClientTransport 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));
}
}
@Test
void testPing() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize()).expectNextCount(1).verifyComplete();
disconnect();
StepVerifier.create(mcpAsyncClient.ping()).expectError().verify();
reconnect();
StepVerifier.create(mcpAsyncClient.ping()).expectNextCount(1).verifyComplete();
});
}
@Test
void testSessionInvalidation() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize()).expectNextCount(1).verifyComplete();
restartMcpServer();
// The first try will face the session mismatch exception and the second one
// will go through the re-initialization process.
StepVerifier.create(mcpAsyncClient.ping().retry(1)).expectNextCount(1).verifyComplete();
});
}
@Test
void testCallTool() {
withClient(createMcpTransport(), mcpAsyncClient -> {
AtomicReference<List<McpSchema.Tool>> tools = new AtomicReference<>();
StepVerifier.create(mcpAsyncClient.initialize()).expectNextCount(1).verifyComplete();
StepVerifier.create(mcpAsyncClient.listTools())
.consumeNextWith(list -> tools.set(list.tools()))
.verifyComplete();
disconnect();
String name = tools.get().get(0).name();
// Assuming this is the echo tool
McpSchema.CallToolRequest request = new McpSchema.CallToolRequest(name, Map.of("message", "hello"));
StepVerifier.create(mcpAsyncClient.callTool(request)).expectError().verify();
reconnect();
StepVerifier.create(mcpAsyncClient.callTool(request)).expectNextCount(1).verifyComplete();
});
}
@Test
void testSessionClose() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize()).expectNextCount(1).verifyComplete();
// In case of Streamable HTTP this call should issue a HTTP DELETE request
// invalidating the session
StepVerifier.create(mcpAsyncClient.closeGracefully()).expectComplete().verify();
// The next use should immediately re-initialize with no issue and send the
// request without any broken connections.
StepVerifier.create(mcpAsyncClient.ping()).expectNextCount(1).verifyComplete();
});
}
}

View file

@ -0,0 +1,823 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import static io.modelcontextprotocol.util.McpJsonMapperUtils.JSON_MAPPER;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.fail;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.BlobResourceContents;
import io.modelcontextprotocol.spec.McpSchema.CallToolRequest;
import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
import io.modelcontextprotocol.spec.McpSchema.CreateMessageRequest;
import io.modelcontextprotocol.spec.McpSchema.CreateMessageResult;
import io.modelcontextprotocol.spec.McpSchema.ElicitRequest;
import io.modelcontextprotocol.spec.McpSchema.ElicitResult;
import io.modelcontextprotocol.spec.McpSchema.GetPromptRequest;
import io.modelcontextprotocol.spec.McpSchema.Prompt;
import io.modelcontextprotocol.spec.McpSchema.ReadResourceResult;
import io.modelcontextprotocol.spec.McpSchema.Resource;
import io.modelcontextprotocol.spec.McpSchema.ResourceContents;
import io.modelcontextprotocol.spec.McpSchema.Root;
import io.modelcontextprotocol.spec.McpSchema.SubscribeRequest;
import io.modelcontextprotocol.spec.McpSchema.TextResourceContents;
import io.modelcontextprotocol.spec.McpSchema.Tool;
import io.modelcontextprotocol.spec.McpSchema.UnsubscribeRequest;
import io.modelcontextprotocol.spec.McpTransport;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.test.StepVerifier;
/**
* Test suite for the {@link McpAsyncClient} that can be used with different
* {@link McpTransport} implementations.
*
* @author Christian Tzolov
* @author Dariusz Jędrzejczyk
*/
// KEEP IN SYNC with the class in mcp-test module
public abstract class AbstractMcpAsyncClientTests {
private static final String ECHO_TEST_MESSAGE = "Hello MCP Spring AI!";
abstract protected McpClientTransport createMcpTransport();
protected Duration getRequestTimeout() {
return Duration.ofSeconds(14);
}
protected Duration getInitializationTimeout() {
return Duration.ofSeconds(2);
}
McpAsyncClient client(McpClientTransport transport) {
return client(transport, Function.identity());
}
McpAsyncClient client(McpClientTransport transport, Function<McpClient.AsyncSpec, McpClient.AsyncSpec> customizer) {
AtomicReference<McpAsyncClient> client = new AtomicReference<>();
assertThatCode(() -> {
McpClient.AsyncSpec builder = McpClient.async(transport)
.requestTimeout(getRequestTimeout())
.initializationTimeout(getInitializationTimeout())
.sampling(req -> Mono.just(new CreateMessageResult(McpSchema.Role.USER,
new McpSchema.TextContent("Oh, hi!"), "modelId", CreateMessageResult.StopReason.END_TURN)))
.capabilities(ClientCapabilities.builder().roots(true).sampling().build());
builder = customizer.apply(builder);
client.set(builder.build());
}).doesNotThrowAnyException();
return client.get();
}
void withClient(McpClientTransport transport, Consumer<McpAsyncClient> c) {
withClient(transport, Function.identity(), c);
}
void withClient(McpClientTransport 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));
}
}
<T> void verifyNotificationSucceedsWithImplicitInitialization(Function<McpAsyncClient, Mono<T>> operation,
String action) {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(operation.apply(mcpAsyncClient)).verifyComplete();
});
}
<T> void verifyCallSucceedsWithImplicitInitialization(Function<McpAsyncClient, Mono<T>> operation, String action) {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(operation.apply(mcpAsyncClient)).expectNextCount(1).verifyComplete();
});
}
@Test
void testConstructorWithInvalidArguments() {
assertThatThrownBy(() -> McpClient.async(null).build()).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Transport must not be null");
assertThatThrownBy(() -> McpClient.async(createMcpTransport()).requestTimeout(null).build())
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Request timeout must not be null");
}
@Test
void testListToolsWithoutInitialization() {
verifyCallSucceedsWithImplicitInitialization(client -> client.listTools(McpSchema.FIRST_PAGE), "listing tools");
}
@Test
void testListTools() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listTools(McpSchema.FIRST_PAGE)))
.consumeNextWith(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull();
})
.verifyComplete();
});
}
@Test
void testListAllTools() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listTools()))
.consumeNextWith(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull();
})
.verifyComplete();
});
}
@Test
void testListAllToolsReturnsImmutableList() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listTools()))
.consumeNextWith(result -> {
assertThat(result.tools()).isNotNull();
// Verify that the returned list is immutable
assertThatThrownBy(() -> result.tools()
.add(Tool.builder()
.name("test")
.title("test")
.inputSchema(JSON_MAPPER, "{\"type\":\"object\"}")
.build()))
.isInstanceOf(UnsupportedOperationException.class);
})
.verifyComplete();
});
}
@Test
void testPingWithoutInitialization() {
verifyCallSucceedsWithImplicitInitialization(client -> client.ping(), "pinging the server");
}
@Test
void testPing() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.ping()))
.expectNextCount(1)
.verifyComplete();
});
}
@Test
void testCallToolWithoutInitialization() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
verifyCallSucceedsWithImplicitInitialization(client -> client.callTool(callToolRequest), "calling tools");
}
@Test
void testCallTool() {
withClient(createMcpTransport(), mcpAsyncClient -> {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(callToolRequest)))
.consumeNextWith(callToolResult -> {
assertThat(callToolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).isNotNull();
assertThat(result.isError()).isNull();
});
})
.verifyComplete();
});
}
@Test
void testCallToolWithInvalidTool() {
withClient(createMcpTransport(), mcpAsyncClient -> {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool",
Map.of("message", ECHO_TEST_MESSAGE));
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.callTool(invalidRequest)))
.consumeErrorWith(
e -> assertThat(e).isInstanceOf(McpError.class).hasMessage("Unknown tool: nonexistent_tool"))
.verify();
});
}
@ParameterizedTest
@ValueSource(strings = { "success", "error", "debug" })
void testCallToolWithMessageAnnotations(String messageType) {
McpClientTransport transport = createMcpTransport();
withClient(transport, mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize()
.then(mcpAsyncClient.callTool(new McpSchema.CallToolRequest("annotatedMessage",
Map.of("messageType", messageType, "includeImage", true)))))
.consumeNextWith(result -> {
assertThat(result).isNotNull();
assertThat(result.isError()).isNotEqualTo(true);
assertThat(result.content()).isNotEmpty();
assertThat(result.content()).allSatisfy(content -> {
switch (content.type()) {
case "text":
McpSchema.TextContent textContent = assertInstanceOf(McpSchema.TextContent.class,
content);
assertThat(textContent.text()).isNotEmpty();
assertThat(textContent.annotations()).isNotNull();
switch (messageType) {
case "error":
assertThat(textContent.annotations().priority()).isEqualTo(1.0);
assertThat(textContent.annotations().audience())
.containsOnly(McpSchema.Role.USER, McpSchema.Role.ASSISTANT);
break;
case "success":
assertThat(textContent.annotations().priority()).isEqualTo(0.7);
assertThat(textContent.annotations().audience())
.containsExactly(McpSchema.Role.USER);
break;
case "debug":
assertThat(textContent.annotations().priority()).isEqualTo(0.3);
assertThat(textContent.annotations().audience())
.containsExactly(McpSchema.Role.ASSISTANT);
break;
default:
throw new IllegalStateException("Unexpected value: " + content.type());
}
break;
case "image":
McpSchema.ImageContent imageContent = assertInstanceOf(McpSchema.ImageContent.class,
content);
assertThat(imageContent.data()).isNotEmpty();
assertThat(imageContent.annotations()).isNotNull();
assertThat(imageContent.annotations().priority()).isEqualTo(0.5);
assertThat(imageContent.annotations().audience()).containsExactly(McpSchema.Role.USER);
break;
default:
fail("Unexpected content type: " + content.type());
}
});
})
.verifyComplete();
});
}
@Test
void testListResourcesWithoutInitialization() {
verifyCallSucceedsWithImplicitInitialization(client -> client.listResources(McpSchema.FIRST_PAGE),
"listing resources");
}
@Test
void testListResources() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResources(McpSchema.FIRST_PAGE)))
.consumeNextWith(resources -> {
assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull();
}
});
})
.verifyComplete();
});
}
@Test
void testListAllResources() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResources()))
.consumeNextWith(resources -> {
assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull();
}
});
})
.verifyComplete();
});
}
@Test
void testListAllResourcesReturnsImmutableList() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResources()))
.consumeNextWith(result -> {
assertThat(result.resources()).isNotNull();
// Verify that the returned list is immutable
assertThatThrownBy(
() -> result.resources().add(Resource.builder().uri("test://uri").name("test").build()))
.isInstanceOf(UnsupportedOperationException.class);
})
.verifyComplete();
});
}
@Test
void testMcpAsyncClientState() {
withClient(createMcpTransport(), mcpAsyncClient -> {
assertThat(mcpAsyncClient).isNotNull();
});
}
@Test
void testListPromptsWithoutInitialization() {
verifyCallSucceedsWithImplicitInitialization(client -> client.listPrompts(McpSchema.FIRST_PAGE),
"listing " + "prompts");
}
@Test
void testListPrompts() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listPrompts(McpSchema.FIRST_PAGE)))
.consumeNextWith(prompts -> {
assertThat(prompts).isNotNull().satisfies(result -> {
assertThat(result.prompts()).isNotNull();
if (!result.prompts().isEmpty()) {
Prompt firstPrompt = result.prompts().get(0);
assertThat(firstPrompt.name()).isNotNull();
assertThat(firstPrompt.description()).isNotNull();
}
});
})
.verifyComplete();
});
}
@Test
void testListAllPrompts() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listPrompts()))
.consumeNextWith(prompts -> {
assertThat(prompts).isNotNull().satisfies(result -> {
assertThat(result.prompts()).isNotNull();
if (!result.prompts().isEmpty()) {
Prompt firstPrompt = result.prompts().get(0);
assertThat(firstPrompt.name()).isNotNull();
assertThat(firstPrompt.description()).isNotNull();
}
});
})
.verifyComplete();
});
}
@Test
void testListAllPromptsReturnsImmutableList() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listPrompts()))
.consumeNextWith(result -> {
assertThat(result.prompts()).isNotNull();
// Verify that the returned list is immutable
assertThatThrownBy(() -> result.prompts().add(new Prompt("test", "test", "test", null)))
.isInstanceOf(UnsupportedOperationException.class);
})
.verifyComplete();
});
}
@Test
void testGetPromptWithoutInitialization() {
GetPromptRequest request = new GetPromptRequest("simple_prompt", Map.of());
verifyCallSucceedsWithImplicitInitialization(client -> client.getPrompt(request), "getting " + "prompts");
}
@Test
void testGetPrompt() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier
.create(mcpAsyncClient.initialize()
.then(mcpAsyncClient.getPrompt(new GetPromptRequest("simple_prompt", Map.of()))))
.consumeNextWith(prompt -> {
assertThat(prompt).isNotNull().satisfies(result -> {
assertThat(result.messages()).isNotEmpty();
assertThat(result.messages()).hasSize(1);
});
})
.verifyComplete();
});
}
@Test
void testRootsListChangedWithoutInitialization() {
verifyNotificationSucceedsWithImplicitInitialization(client -> client.rootsListChangedNotification(),
"sending roots list changed notification");
}
@Test
void testRootsListChanged() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.rootsListChangedNotification()))
.verifyComplete();
});
}
@Test
void testInitializeWithRootsListProviders() {
withClient(createMcpTransport(), builder -> builder.roots(new Root("file:///test/path", "test-root")),
client -> {
StepVerifier.create(client.initialize().then(client.closeGracefully())).verifyComplete();
});
}
@Test
void testAddRoot() {
withClient(createMcpTransport(), mcpAsyncClient -> {
Root newRoot = new Root("file:///new/test/path", "new-test-root");
StepVerifier.create(mcpAsyncClient.addRoot(newRoot)).verifyComplete();
});
}
@Test
void testAddRootWithNullValue() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.addRoot(null))
.consumeErrorWith(e -> assertThat(e).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Root must not be null"))
.verify();
});
}
@Test
void testRemoveRoot() {
withClient(createMcpTransport(), mcpAsyncClient -> {
Root root = new Root("file:///test/path/to/remove", "root-to-remove");
StepVerifier.create(mcpAsyncClient.addRoot(root)).verifyComplete();
StepVerifier.create(mcpAsyncClient.removeRoot(root.uri())).verifyComplete();
});
}
@Test
void testRemoveNonExistentRoot() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.removeRoot("nonexistent-uri"))
.consumeErrorWith(e -> assertThat(e).isInstanceOf(IllegalStateException.class)
.hasMessage("Root with uri 'nonexistent-uri' not found"))
.verify();
});
}
@Test
void testReadResource() {
withClient(createMcpTransport(), client -> {
Flux<McpSchema.ReadResourceResult> resources = client.initialize()
.then(client.listResources(null))
.flatMapMany(r -> Flux.fromIterable(r.resources()))
.flatMap(r -> client.readResource(r));
StepVerifier.create(resources).recordWith(ArrayList::new).consumeRecordedWith(readResourceResults -> {
for (ReadResourceResult result : readResourceResults) {
assertThat(result).isNotNull();
assertThat(result.contents()).isNotNull().isNotEmpty();
// Validate each content item
for (ResourceContents content : result.contents()) {
assertThat(content).isNotNull();
assertThat(content.uri()).isNotNull().isNotEmpty();
assertThat(content.mimeType()).isNotNull().isNotEmpty();
// Validate content based on its type with more comprehensive
// checks
switch (content.mimeType()) {
case "text/plain" -> {
TextResourceContents textContent = assertInstanceOf(TextResourceContents.class,
content);
assertThat(textContent.text()).isNotNull().isNotEmpty();
assertThat(textContent.uri()).isNotEmpty();
}
case "application/octet-stream" -> {
BlobResourceContents blobContent = assertInstanceOf(BlobResourceContents.class,
content);
assertThat(blobContent.blob()).isNotNull().isNotEmpty();
assertThat(blobContent.uri()).isNotNull().isNotEmpty();
// Validate base64 encoding format
assertThat(blobContent.blob()).matches("^[A-Za-z0-9+/]*={0,2}$");
}
default -> {
// Still validate basic properties
if (content instanceof TextResourceContents textContent) {
assertThat(textContent.text()).isNotNull();
}
else if (content instanceof BlobResourceContents blobContent) {
assertThat(blobContent.blob()).isNotNull();
}
}
}
}
}
})
.expectNextCount(10) // Expect 10 elements
.verifyComplete();
});
}
@Test
void testListResourceTemplatesWithoutInitialization() {
verifyCallSucceedsWithImplicitInitialization(client -> client.listResourceTemplates(McpSchema.FIRST_PAGE),
"listing resource templates");
}
@Test
void testListResourceTemplates() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier
.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResourceTemplates(McpSchema.FIRST_PAGE)))
.consumeNextWith(result -> {
assertThat(result).isNotNull();
assertThat(result.resourceTemplates()).isNotNull();
})
.verifyComplete();
});
}
@Test
void testListAllResourceTemplates() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResourceTemplates()))
.consumeNextWith(result -> {
assertThat(result).isNotNull();
assertThat(result.resourceTemplates()).isNotNull();
})
.verifyComplete();
});
}
@Test
void testListAllResourceTemplatesReturnsImmutableList() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize().then(mcpAsyncClient.listResourceTemplates()))
.consumeNextWith(result -> {
assertThat(result.resourceTemplates()).isNotNull();
// Verify that the returned list is immutable
assertThatThrownBy(() -> result.resourceTemplates()
.add(new McpSchema.ResourceTemplate("test://template", "test", "test", null, null, null)))
.isInstanceOf(UnsupportedOperationException.class);
})
.verifyComplete();
});
}
// @Test
void testResourceSubscription() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> {
if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0);
// Test subscribe
StepVerifier.create(mcpAsyncClient.subscribeResource(new SubscribeRequest(firstResource.uri())))
.verifyComplete();
// Test unsubscribe
StepVerifier.create(mcpAsyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())))
.verifyComplete();
}
}).verifyComplete();
});
}
@Test
void testNotificationHandlers() {
AtomicBoolean toolsNotificationReceived = new AtomicBoolean(false);
AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false);
AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false);
withClient(createMcpTransport(),
builder -> builder
.toolsChangeConsumer(tools -> Mono.fromRunnable(() -> toolsNotificationReceived.set(true)))
.resourcesChangeConsumer(
resources -> Mono.fromRunnable(() -> resourcesNotificationReceived.set(true)))
.promptsChangeConsumer(prompts -> Mono.fromRunnable(() -> promptsNotificationReceived.set(true))),
mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize())
.expectNextMatches(Objects::nonNull)
.verifyComplete();
});
}
@Test
void testInitializeWithSamplingCapability() {
ClientCapabilities capabilities = ClientCapabilities.builder().sampling().build();
CreateMessageResult createMessageResult = CreateMessageResult.builder()
.message("test")
.model("test-model")
.build();
withClient(createMcpTransport(),
builder -> builder.capabilities(capabilities).sampling(request -> Mono.just(createMessageResult)),
client -> {
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
});
}
@Test
void testInitializeWithElicitationCapability() {
ClientCapabilities capabilities = ClientCapabilities.builder().elicitation().build();
ElicitResult elicitResult = ElicitResult.builder()
.message(ElicitResult.Action.ACCEPT)
.content(Map.of("foo", "bar"))
.build();
withClient(createMcpTransport(),
builder -> builder.capabilities(capabilities).elicitation(request -> Mono.just(elicitResult)),
client -> {
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
});
}
@Test
void testInitializeWithAllCapabilities() {
var capabilities = ClientCapabilities.builder()
.experimental(Map.of("feature", "test"))
.roots(true)
.sampling()
.build();
Function<CreateMessageRequest, Mono<CreateMessageResult>> samplingHandler = request -> Mono
.just(CreateMessageResult.builder().message("test").model("test-model").build());
Function<ElicitRequest, Mono<ElicitResult>> elicitationHandler = request -> Mono
.just(ElicitResult.builder().message(ElicitResult.Action.ACCEPT).content(Map.of("foo", "bar")).build());
withClient(createMcpTransport(),
builder -> builder.capabilities(capabilities).sampling(samplingHandler).elicitation(elicitationHandler),
client ->
StepVerifier.create(client.initialize()).assertNext(result -> {
assertThat(result).isNotNull();
assertThat(result.capabilities()).isNotNull();
}).verifyComplete());
}
// ---------------------------------------
// Logging Tests
// ---------------------------------------
@Test
void testLoggingLevelsWithoutInitialization() {
verifyNotificationSucceedsWithImplicitInitialization(
client -> client.setLoggingLevel(McpSchema.LoggingLevel.DEBUG), "setting logging level");
}
@Test
void testLoggingLevels() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier
.create(mcpAsyncClient.initialize()
.thenMany(Flux.fromArray(McpSchema.LoggingLevel.values()).flatMap(mcpAsyncClient::setLoggingLevel)))
.verifyComplete();
});
}
@Test
void testLoggingConsumer() {
AtomicBoolean logReceived = new AtomicBoolean(false);
withClient(createMcpTransport(),
builder -> builder.loggingConsumer(notification -> Mono.fromRunnable(() -> logReceived.set(true))),
client -> {
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
StepVerifier.create(client.closeGracefully()).verifyComplete();
});
}
@Test
void testLoggingWithNullNotification() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.setLoggingLevel(null))
.expectErrorMatches(error -> error.getMessage().contains("Logging level must not be null"))
.verify();
});
}
@Test
void testSampling() {
McpClientTransport transport = createMcpTransport();
final String message = "Hello, world!";
final String response = "Goodbye, world!";
final int maxTokens = 100;
AtomicReference<String> receivedPrompt = new AtomicReference<>();
AtomicReference<String> receivedMessage = new AtomicReference<>();
AtomicInteger receivedMaxTokens = new AtomicInteger();
withClient(transport, spec -> spec.capabilities(McpSchema.ClientCapabilities.builder().sampling().build())
.sampling(request -> {
McpSchema.TextContent messageText = assertInstanceOf(McpSchema.TextContent.class,
request.messages().get(0).content());
receivedPrompt.set(request.systemPrompt());
receivedMessage.set(messageText.text());
receivedMaxTokens.set(request.maxTokens());
return Mono
.just(new McpSchema.CreateMessageResult(McpSchema.Role.USER, new McpSchema.TextContent(response),
"modelId", McpSchema.CreateMessageResult.StopReason.END_TURN));
}), client -> {
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
StepVerifier.create(client.callTool(
new McpSchema.CallToolRequest("sampleLLM", Map.of("prompt", message, "maxTokens", maxTokens))))
.consumeNextWith(result -> {
// Verify tool response to ensure our sampling response was passed
// through
assertThat(result.content()).hasAtLeastOneElementOfType(McpSchema.TextContent.class);
assertThat(result.content()).allSatisfy(content -> {
if (!(content instanceof McpSchema.TextContent text))
return;
assertThat(text.text()).endsWith(response); // Prefixed
});
// Verify sampling request parameters received in our callback
assertThat(receivedPrompt.get()).isNotEmpty();
assertThat(receivedMessage.get()).endsWith(message); // Prefixed
assertThat(receivedMaxTokens.get()).isEqualTo(maxTokens);
})
.verifyComplete();
});
}
// ---------------------------------------
// Progress Notification Tests
// ---------------------------------------
@Test
void testProgressConsumer() {
Sinks.Many<McpSchema.ProgressNotification> sink = Sinks.many().unicast().onBackpressureBuffer();
List<McpSchema.ProgressNotification> receivedNotifications = new CopyOnWriteArrayList<>();
withClient(createMcpTransport(), builder -> builder.progressConsumer(notification -> {
receivedNotifications.add(notification);
sink.tryEmitNext(notification);
return Mono.empty();
}), client -> {
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
// Call a tool that sends progress notifications
CallToolRequest request = CallToolRequest.builder()
.name("longRunningOperation")
.arguments(Map.of("duration", 1, "steps", 2))
.progressToken("test-token")
.build();
StepVerifier.create(client.callTool(request)).consumeNextWith(result -> {
assertThat(result).isNotNull();
}).verifyComplete();
// Use StepVerifier to verify the progress notifications via the sink
StepVerifier.create(sink.asFlux()).expectNextCount(2).thenCancel().verify(Duration.ofSeconds(3));
assertThat(receivedNotifications).hasSize(2);
assertThat(receivedNotifications.get(0).progressToken()).isEqualTo("test-token");
});
}
}

View file

@ -0,0 +1,680 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.fail;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.BlobResourceContents;
import io.modelcontextprotocol.spec.McpSchema.CallToolRequest;
import io.modelcontextprotocol.spec.McpSchema.CallToolResult;
import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
import io.modelcontextprotocol.spec.McpSchema.ListResourceTemplatesResult;
import io.modelcontextprotocol.spec.McpSchema.ListResourcesResult;
import io.modelcontextprotocol.spec.McpSchema.ListToolsResult;
import io.modelcontextprotocol.spec.McpSchema.ReadResourceResult;
import io.modelcontextprotocol.spec.McpSchema.Resource;
import io.modelcontextprotocol.spec.McpSchema.ResourceContents;
import io.modelcontextprotocol.spec.McpSchema.Root;
import io.modelcontextprotocol.spec.McpSchema.SubscribeRequest;
import io.modelcontextprotocol.spec.McpSchema.TextContent;
import io.modelcontextprotocol.spec.McpSchema.TextResourceContents;
import io.modelcontextprotocol.spec.McpSchema.Tool;
import io.modelcontextprotocol.spec.McpSchema.UnsubscribeRequest;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;
/**
* Unit tests for MCP Client Session functionality.
*
* @author Christian Tzolov
* @author Dariusz Jędrzejczyk
*/
// KEEP IN SYNC with the class in mcp-test module
public abstract class AbstractMcpSyncClientTests {
private static final Logger logger = LoggerFactory.getLogger(AbstractMcpSyncClientTests.class);
private static final String TEST_MESSAGE = "Hello MCP Spring AI!";
abstract protected McpClientTransport createMcpTransport();
protected Duration getRequestTimeout() {
return Duration.ofSeconds(14);
}
protected Duration getInitializationTimeout() {
return Duration.ofSeconds(2);
}
McpSyncClient client(McpClientTransport transport) {
return client(transport, Function.identity());
}
McpSyncClient client(McpClientTransport 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(McpClientTransport transport, Consumer<McpSyncClient> c) {
withClient(transport, Function.identity(), c);
}
void withClient(McpClientTransport transport, Function<McpClient.SyncSpec, McpClient.SyncSpec> customizer,
Consumer<McpSyncClient> c) {
var client = client(transport, customizer);
try {
c.accept(client);
}
finally {
assertThat(client.closeGracefully()).isTrue();
}
}
static final Object DUMMY_RETURN_VALUE = new Object();
<T> void verifyNotificationSucceedsWithImplicitInitialization(Consumer<McpSyncClient> operation, String action) {
verifyCallSucceedsWithImplicitInitialization(client -> {
operation.accept(client);
return DUMMY_RETURN_VALUE;
}, action);
}
<T> void verifyCallSucceedsWithImplicitInitialization(Function<McpSyncClient, T> blockingOperation, String action) {
withClient(createMcpTransport(), mcpSyncClient -> {
StepVerifier.create(Mono.fromSupplier(() -> blockingOperation.apply(mcpSyncClient))
// Offload the blocking call to the real scheduler
.subscribeOn(Schedulers.boundedElastic())).expectNextCount(1).verifyComplete();
});
}
@Test
void testConstructorWithInvalidArguments() {
assertThatThrownBy(() -> McpClient.sync(null).build()).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Transport must not be null");
assertThatThrownBy(() -> McpClient.sync(createMcpTransport()).requestTimeout(null).build())
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Request timeout must not be null");
}
@Test
void testListToolsWithoutInitialization() {
verifyCallSucceedsWithImplicitInitialization(client -> client.listTools(McpSchema.FIRST_PAGE), "listing tools");
}
@Test
void testListTools() {
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
ListToolsResult tools = mcpSyncClient.listTools(McpSchema.FIRST_PAGE);
assertThat(tools).isNotNull().satisfies(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull();
});
});
}
@Test
void testListAllTools() {
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
ListToolsResult tools = mcpSyncClient.listTools();
assertThat(tools).isNotNull().satisfies(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull();
});
});
}
@Test
void testCallToolsWithoutInitialization() {
verifyCallSucceedsWithImplicitInitialization(
client -> client.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4))), "calling tools");
}
@Test
void testCallTools() {
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
CallToolResult toolResult = mcpSyncClient.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4)));
assertThat(toolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).hasSize(1);
TextContent content = (TextContent) result.content().get(0);
assertThat(content).isNotNull();
assertThat(content.text()).isNotNull();
assertThat(content.text()).contains("7");
});
});
}
@Test
void testPingWithoutInitialization() {
verifyCallSucceedsWithImplicitInitialization(client -> client.ping(), "pinging the server");
}
@Test
void testPing() {
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
assertThatCode(() -> mcpSyncClient.ping()).doesNotThrowAnyException();
});
}
@Test
void testCallToolWithoutInitialization() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE));
verifyCallSucceedsWithImplicitInitialization(client -> client.callTool(callToolRequest), "calling tools");
}
@Test
void testCallTool() {
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE));
CallToolResult callToolResult = mcpSyncClient.callTool(callToolRequest);
assertThat(callToolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).isNotNull();
assertThat(result.isError()).isNull();
});
});
}
@Test
void testCallToolWithInvalidTool() {
withClient(createMcpTransport(), mcpSyncClient -> {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", TEST_MESSAGE));
assertThatThrownBy(() -> mcpSyncClient.callTool(invalidRequest)).isInstanceOf(Exception.class);
});
}
@ParameterizedTest
@ValueSource(strings = { "success", "error", "debug" })
void testCallToolWithMessageAnnotations(String messageType) {
McpClientTransport transport = createMcpTransport();
withClient(transport, client -> {
client.initialize();
McpSchema.CallToolResult result = client.callTool(new McpSchema.CallToolRequest("annotatedMessage",
Map.of("messageType", messageType, "includeImage", true)));
assertThat(result).isNotNull();
assertThat(result.isError()).isNotEqualTo(true);
assertThat(result.content()).isNotEmpty();
assertThat(result.content()).allSatisfy(content -> {
switch (content.type()) {
case "text":
McpSchema.TextContent textContent = assertInstanceOf(McpSchema.TextContent.class, content);
assertThat(textContent.text()).isNotEmpty();
assertThat(textContent.annotations()).isNotNull();
switch (messageType) {
case "error":
assertThat(textContent.annotations().priority()).isEqualTo(1.0);
assertThat(textContent.annotations().audience()).containsOnly(McpSchema.Role.USER,
McpSchema.Role.ASSISTANT);
break;
case "success":
assertThat(textContent.annotations().priority()).isEqualTo(0.7);
assertThat(textContent.annotations().audience()).containsExactly(McpSchema.Role.USER);
break;
case "debug":
assertThat(textContent.annotations().priority()).isEqualTo(0.3);
assertThat(textContent.annotations().audience())
.containsExactly(McpSchema.Role.ASSISTANT);
break;
default:
throw new IllegalStateException("Unexpected value: " + content.type());
}
break;
case "image":
McpSchema.ImageContent imageContent = assertInstanceOf(McpSchema.ImageContent.class, content);
assertThat(imageContent.data()).isNotEmpty();
assertThat(imageContent.annotations()).isNotNull();
assertThat(imageContent.annotations().priority()).isEqualTo(0.5);
assertThat(imageContent.annotations().audience()).containsExactly(McpSchema.Role.USER);
break;
default:
fail("Unexpected content type: " + content.type());
}
});
});
}
@Test
void testRootsListChangedWithoutInitialization() {
verifyNotificationSucceedsWithImplicitInitialization(client -> client.rootsListChangedNotification(),
"sending roots list changed notification");
}
@Test
void testRootsListChanged() {
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
assertThatCode(() -> mcpSyncClient.rootsListChangedNotification()).doesNotThrowAnyException();
});
}
@Test
void testListResourcesWithoutInitialization() {
verifyCallSucceedsWithImplicitInitialization(client -> client.listResources(McpSchema.FIRST_PAGE),
"listing resources");
}
@Test
void testListResources() {
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
ListResourcesResult resources = mcpSyncClient.listResources(McpSchema.FIRST_PAGE);
assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull();
}
});
});
}
@Test
void testListAllResources() {
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
ListResourcesResult resources = mcpSyncClient.listResources();
assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull();
}
});
});
}
@Test
void testClientSessionState() {
withClient(createMcpTransport(), mcpSyncClient -> {
assertThat(mcpSyncClient).isNotNull();
});
}
@Test
void testInitializeWithRootsListProviders() {
withClient(createMcpTransport(), builder -> builder.roots(new Root("file:///test/path", "test-root")),
mcpSyncClient -> {
assertThatCode(() -> {
mcpSyncClient.initialize();
mcpSyncClient.close();
}).doesNotThrowAnyException();
});
}
@Test
void testAddRoot() {
withClient(createMcpTransport(), mcpSyncClient -> {
Root newRoot = new Root("file:///new/test/path", "new-test-root");
assertThatCode(() -> mcpSyncClient.addRoot(newRoot)).doesNotThrowAnyException();
});
}
@Test
void testAddRootWithNullValue() {
withClient(createMcpTransport(), mcpSyncClient -> {
assertThatThrownBy(() -> mcpSyncClient.addRoot(null)).hasMessageContaining("Root must not be null");
});
}
@Test
void testRemoveRoot() {
withClient(createMcpTransport(), mcpSyncClient -> {
Root root = new Root("file:///test/path/to/remove", "root-to-remove");
assertThatCode(() -> {
mcpSyncClient.addRoot(root);
mcpSyncClient.removeRoot(root.uri());
}).doesNotThrowAnyException();
});
}
@Test
void testRemoveNonExistentRoot() {
withClient(createMcpTransport(), mcpSyncClient -> {
assertThatThrownBy(() -> mcpSyncClient.removeRoot("nonexistent-uri"))
.hasMessageContaining("Root with uri 'nonexistent-uri' not found");
});
}
@Test
void testReadResourceWithoutInitialization() {
AtomicReference<List<Resource>> resources = new AtomicReference<>();
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
resources.set(mcpSyncClient.listResources().resources());
});
verifyCallSucceedsWithImplicitInitialization(client -> client.readResource(resources.get().get(0)),
"reading resources");
}
@Test
void testReadResource() {
withClient(createMcpTransport(), mcpSyncClient -> {
int readResourceCount = 0;
mcpSyncClient.initialize();
ListResourcesResult resources = mcpSyncClient.listResources(null);
assertThat(resources).isNotNull();
assertThat(resources.resources()).isNotNull();
assertThat(resources.resources()).isNotNull().isNotEmpty();
// Test reading each resource individually for better error isolation
for (Resource resource : resources.resources()) {
ReadResourceResult result = mcpSyncClient.readResource(resource);
assertThat(result).isNotNull();
assertThat(result.contents()).isNotNull().isNotEmpty();
readResourceCount++;
// Validate each content item
for (ResourceContents content : result.contents()) {
assertThat(content).isNotNull();
assertThat(content.uri()).isNotNull().isNotEmpty();
assertThat(content.mimeType()).isNotNull().isNotEmpty();
// Validate content based on its type with more comprehensive
// checks
switch (content.mimeType()) {
case "text/plain" -> {
TextResourceContents textContent = assertInstanceOf(TextResourceContents.class, content);
assertThat(textContent.text()).isNotNull().isNotEmpty();
// Verify URI consistency
assertThat(textContent.uri()).isEqualTo(resource.uri());
}
case "application/octet-stream" -> {
BlobResourceContents blobContent = assertInstanceOf(BlobResourceContents.class, content);
assertThat(blobContent.blob()).isNotNull().isNotEmpty();
// Verify URI consistency
assertThat(blobContent.uri()).isEqualTo(resource.uri());
// Validate base64 encoding format
assertThat(blobContent.blob()).matches("^[A-Za-z0-9+/]*={0,2}$");
}
default -> {
// More flexible handling of additional MIME types
// Log the unexpected type for debugging but don't fail
// the test
logger.warn("Warning: Encountered unexpected MIME type: {} for resource: {}",
content.mimeType(), resource.uri());
// Still validate basic properties
if (content instanceof TextResourceContents textContent) {
assertThat(textContent.text()).isNotNull();
}
else if (content instanceof BlobResourceContents blobContent) {
assertThat(blobContent.blob()).isNotNull();
}
}
}
}
}
// Assert that we read exactly 10 resources
assertThat(readResourceCount).isEqualTo(10);
});
}
@Test
void testListResourceTemplatesWithoutInitialization() {
verifyCallSucceedsWithImplicitInitialization(client -> client.listResourceTemplates(McpSchema.FIRST_PAGE),
"listing resource templates");
}
@Test
void testListResourceTemplates() {
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
ListResourceTemplatesResult result = mcpSyncClient.listResourceTemplates(McpSchema.FIRST_PAGE);
assertThat(result).isNotNull();
assertThat(result.resourceTemplates()).isNotNull();
});
}
@Test
void testListAllResourceTemplates() {
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
ListResourceTemplatesResult result = mcpSyncClient.listResourceTemplates();
assertThat(result).isNotNull();
assertThat(result.resourceTemplates()).isNotNull();
});
}
// @Test
void testResourceSubscription() {
withClient(createMcpTransport(), mcpSyncClient -> {
ListResourcesResult resources = mcpSyncClient.listResources(null);
if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0);
// Test subscribe
assertThatCode(() -> mcpSyncClient.subscribeResource(new SubscribeRequest(firstResource.uri())))
.doesNotThrowAnyException();
// Test unsubscribe
assertThatCode(() -> mcpSyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())))
.doesNotThrowAnyException();
}
});
}
@Test
void testNotificationHandlers() {
AtomicBoolean toolsNotificationReceived = new AtomicBoolean(false);
AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false);
AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false);
withClient(createMcpTransport(),
builder -> builder.toolsChangeConsumer(tools -> toolsNotificationReceived.set(true))
.resourcesChangeConsumer(resources -> resourcesNotificationReceived.set(true))
.promptsChangeConsumer(prompts -> promptsNotificationReceived.set(true)),
client -> {
assertThatCode(() -> {
client.initialize();
client.close();
}).doesNotThrowAnyException();
});
}
// ---------------------------------------
// Logging Tests
// ---------------------------------------
@Test
void testLoggingLevelsWithoutInitialization() {
verifyNotificationSucceedsWithImplicitInitialization(
client -> client.setLoggingLevel(McpSchema.LoggingLevel.DEBUG), "setting logging level");
}
@Test
void testLoggingLevels() {
withClient(createMcpTransport(), mcpSyncClient -> {
mcpSyncClient.initialize();
// Test all logging levels
for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
assertThatCode(() -> mcpSyncClient.setLoggingLevel(level)).doesNotThrowAnyException();
}
});
}
@Test
void testLoggingConsumer() {
AtomicBoolean logReceived = new AtomicBoolean(false);
withClient(createMcpTransport(), builder -> builder.requestTimeout(getRequestTimeout())
.loggingConsumer(notification -> logReceived.set(true)), client -> {
assertThatCode(() -> {
client.initialize();
client.close();
}).doesNotThrowAnyException();
});
}
@Test
void testLoggingWithNullNotification() {
withClient(createMcpTransport(), mcpSyncClient -> assertThatThrownBy(() -> mcpSyncClient.setLoggingLevel(null))
.hasMessageContaining("Logging level must not be null"));
}
@Test
void testSampling() {
McpClientTransport transport = createMcpTransport();
final String message = "Hello, world!";
final String response = "Goodbye, world!";
final int maxTokens = 100;
AtomicReference<String> receivedPrompt = new AtomicReference<>();
AtomicReference<String> receivedMessage = new AtomicReference<>();
AtomicInteger receivedMaxTokens = new AtomicInteger();
withClient(transport, spec -> spec.capabilities(McpSchema.ClientCapabilities.builder().sampling().build())
.sampling(request -> {
McpSchema.TextContent messageText = assertInstanceOf(McpSchema.TextContent.class,
request.messages().get(0).content());
receivedPrompt.set(request.systemPrompt());
receivedMessage.set(messageText.text());
receivedMaxTokens.set(request.maxTokens());
return new McpSchema.CreateMessageResult(McpSchema.Role.USER, new McpSchema.TextContent(response),
"modelId", McpSchema.CreateMessageResult.StopReason.END_TURN);
}), client -> {
client.initialize();
McpSchema.CallToolResult result = client.callTool(
new McpSchema.CallToolRequest("sampleLLM", Map.of("prompt", message, "maxTokens", maxTokens)));
// Verify tool response to ensure our sampling response was passed through
assertThat(result.content()).hasAtLeastOneElementOfType(McpSchema.TextContent.class);
assertThat(result.content()).allSatisfy(content -> {
if (!(content instanceof McpSchema.TextContent text))
return;
assertThat(text.text()).endsWith(response); // Prefixed
});
// Verify sampling request parameters received in our callback
assertThat(receivedPrompt.get()).isNotEmpty();
assertThat(receivedMessage.get()).endsWith(message); // Prefixed
assertThat(receivedMaxTokens.get()).isEqualTo(maxTokens);
});
}
// ---------------------------------------
// Progress Notification Tests
// ---------------------------------------
@Test
void testProgressConsumer() {
AtomicInteger progressNotificationCount = new AtomicInteger(0);
List<McpSchema.ProgressNotification> receivedNotifications = new CopyOnWriteArrayList<>();
CountDownLatch latch = new CountDownLatch(2);
withClient(createMcpTransport(), builder -> builder.progressConsumer(notification -> {
System.out.println("Received progress notification: " + notification);
receivedNotifications.add(notification);
progressNotificationCount.incrementAndGet();
latch.countDown();
}), client -> {
client.initialize();
// Call a tool that sends progress notifications
CallToolRequest request = CallToolRequest.builder()
.name("longRunningOperation")
.arguments(Map.of("duration", 1, "steps", 2))
.progressToken("test-token")
.build();
CallToolResult result = client.callTool(request);
assertThat(result).isNotNull();
try {
// Wait for progress notifications to be processed
latch.await(3, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
e.printStackTrace();
}
assertThat(progressNotificationCount.get()).isEqualTo(2);
assertThat(receivedNotifications).isNotEmpty();
assertThat(receivedNotifications.get(0).progressToken()).isEqualTo("test-token");
});
}
}

View file

@ -0,0 +1,39 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import java.io.IOException;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
import reactor.test.StepVerifier;
@Timeout(15)
public class HttpClientStreamableHttpAsyncClientResiliencyTests extends AbstractMcpAsyncClientResiliencyTests {
@Override
protected McpClientTransport createMcpTransport() {
return HttpClientStreamableHttpTransport.builder(host).build();
}
@Test
void testPingWithExactExceptionType() {
withClient(createMcpTransport(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize()).expectNextCount(1).verifyComplete();
disconnect();
StepVerifier.create(mcpAsyncClient.ping()).expectError(IOException.class).verify();
reconnect();
StepVerifier.create(mcpAsyncClient.ping()).expectNextCount(1).verifyComplete();
});
}
}

View file

@ -0,0 +1,46 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Timeout;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
@Timeout(15)
public class HttpClientStreamableHttpAsyncClientTests extends AbstractMcpAsyncClientTests {
private static String host = "http://localhost:3001";
// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v3")
.withCommand("node dist/index.js streamableHttp")
.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
.withExposedPorts(3001)
.waitingFor(Wait.forHttp("/").forStatusCode(404));
@Override
protected McpClientTransport createMcpTransport() {
return HttpClientStreamableHttpTransport.builder(host).build();
}
@BeforeAll
static void startContainer() {
container.start();
int port = container.getMappedPort(3001);
host = "http://" + container.getHost() + ":" + port;
}
@AfterAll
static void stopContainer() {
container.stop();
}
}

View file

@ -0,0 +1,73 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client;
import java.net.URI;
import java.util.Map;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport;
import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.spec.McpClientTransport;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@Timeout(15)
public class HttpClientStreamableHttpSyncClientTests extends AbstractMcpSyncClientTests {
static String host = "http://localhost:3001";
// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v3")
.withCommand("node dist/index.js streamableHttp")
.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
.withExposedPorts(3001)
.waitingFor(Wait.forHttp("/").forStatusCode(404));
private final McpSyncHttpClientRequestCustomizer requestCustomizer = mock(McpSyncHttpClientRequestCustomizer.class);
@Override
protected McpClientTransport createMcpTransport() {
return HttpClientStreamableHttpTransport.builder(host).httpRequestCustomizer(requestCustomizer).build();
}
@BeforeAll
static void startContainer() {
container.start();
int port = container.getMappedPort(3001);
host = "http://" + container.getHost() + ":" + port;
}
@AfterAll
static void stopContainer() {
container.stop();
}
@Test
void customizesRequests() {
var mcpTransportContext = McpTransportContext.create(Map.of("some-key", "some-value"));
withClient(createMcpTransport(), syncSpec -> syncSpec.transportContextProvider(() -> mcpTransportContext),
mcpSyncClient -> {
mcpSyncClient.initialize();
verify(requestCustomizer, atLeastOnce()).customize(any(), eq("POST"), eq(URI.create(host + "/mcp")),
eq("{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}"),
eq(mcpTransportContext));
});
}
}

View file

@ -0,0 +1,140 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import static org.assertj.core.api.Assertions.assertThatCode;
import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.ToxiproxyContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import eu.rekawek.toxiproxy.Proxy;
import eu.rekawek.toxiproxy.ToxiproxyClient;
import eu.rekawek.toxiproxy.model.ToxicDirection;
import io.modelcontextprotocol.client.transport.HttpClientSseClientTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import reactor.test.StepVerifier;
@Timeout(20)
public class HttpSseMcpAsyncClientLostConnectionTests {
private static final Logger logger = LoggerFactory.getLogger(HttpSseMcpAsyncClientLostConnectionTests.class);
static Network network = Network.newNetwork();
static String host = "http://localhost:3001";
// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v3")
.withCommand("node dist/index.js sse")
.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
.withNetwork(network)
.withNetworkAliases("everything-server")
.withExposedPorts(3001)
.waitingFor(Wait.forHttp("/").forStatusCode(404));
static ToxiproxyContainer toxiproxy = new ToxiproxyContainer("ghcr.io/shopify/toxiproxy:2.5.0").withNetwork(network)
.withExposedPorts(8474, 3000);
static Proxy proxy;
static {
container.start();
toxiproxy.start();
final ToxiproxyClient toxiproxyClient = new ToxiproxyClient(toxiproxy.getHost(), toxiproxy.getControlPort());
try {
proxy = toxiproxyClient.createProxy("everything-server", "0.0.0.0:3000", "everything-server:3001");
}
catch (IOException e) {
throw new RuntimeException("Can't create proxy!", e);
}
final String ipAddressViaToxiproxy = toxiproxy.getHost();
final int portViaToxiproxy = toxiproxy.getMappedPort(3000);
host = "http://" + ipAddressViaToxiproxy + ":" + portViaToxiproxy;
}
static void disconnect() {
long start = System.nanoTime();
try {
proxy.toxics().resetPeer("RESET_DOWNSTREAM", ToxicDirection.DOWNSTREAM, 0);
proxy.toxics().resetPeer("RESET_UPSTREAM", ToxicDirection.UPSTREAM, 0);
logger.info("Disconnect took {} ms", Duration.ofNanos(System.nanoTime() - start).toMillis());
}
catch (IOException e) {
throw new RuntimeException("Failed to disconnect", e);
}
}
static void reconnect() {
long start = System.nanoTime();
try {
proxy.toxics().get("RESET_UPSTREAM").remove();
proxy.toxics().get("RESET_DOWNSTREAM").remove();
logger.info("Reconnect took {} ms", Duration.ofNanos(System.nanoTime() - start).toMillis());
}
catch (IOException e) {
throw new RuntimeException("Failed to reconnect", e);
}
}
McpAsyncClient client(McpClientTransport transport) {
AtomicReference<McpAsyncClient> client = new AtomicReference<>();
assertThatCode(() -> {
// Do not advertise roots. Otherwise, the server will list roots during
// initialization. The client responds asynchronously, and there might be a
// rest condition in tests where we disconnect right after initialization.
McpClient.AsyncSpec builder = McpClient.async(transport)
.requestTimeout(Duration.ofSeconds(14))
.initializationTimeout(Duration.ofSeconds(2))
.capabilities(McpSchema.ClientCapabilities.builder().build());
client.set(builder.build());
}).doesNotThrowAnyException();
return client.get();
}
void withClient(McpClientTransport transport, Consumer<McpAsyncClient> c) {
var client = client(transport);
try {
c.accept(client);
}
finally {
StepVerifier.create(client.closeGracefully()).expectComplete().verify(Duration.ofSeconds(10));
}
}
@Test
void testPingWithEaxctExceptionType() {
withClient(HttpClientSseClientTransport.builder(host).build(), mcpAsyncClient -> {
StepVerifier.create(mcpAsyncClient.initialize()).expectNextCount(1).verifyComplete();
disconnect();
// Veryfiy that the exception type is IOException and not TimeoutException
StepVerifier.create(mcpAsyncClient.ping()).expectError(IOException.class).verify();
reconnect();
StepVerifier.create(mcpAsyncClient.ping()).expectNextCount(1).verifyComplete();
});
}
}

View file

@ -0,0 +1,51 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Timeout;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import io.modelcontextprotocol.client.transport.HttpClientSseClientTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
/**
* Tests for the {@link McpSyncClient} with {@link HttpClientSseClientTransport}.
*
* @author Christian Tzolov
*/
@Timeout(15)
class HttpSseMcpAsyncClientTests extends AbstractMcpAsyncClientTests {
private static String host = "http://localhost:3004";
// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v3")
.withCommand("node dist/index.js sse")
.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
.withExposedPorts(3001)
.waitingFor(Wait.forHttp("/").forStatusCode(404));
@Override
protected McpClientTransport createMcpTransport() {
return HttpClientSseClientTransport.builder(host).build();
}
@BeforeAll
static void startContainer() {
container.start();
int port = container.getMappedPort(3001);
host = "http://" + container.getHost() + ":" + port;
}
@AfterAll
static void stopContainer() {
container.stop();
}
}

View file

@ -0,0 +1,78 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import java.net.URI;
import java.util.Map;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import io.modelcontextprotocol.client.transport.HttpClientSseClientTransport;
import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.spec.McpClientTransport;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
/**
* Tests for the {@link McpSyncClient} with {@link HttpClientSseClientTransport}.
*
* @author Christian Tzolov
*/
@Timeout(15) // Giving extra time beyond the client timeout
class HttpSseMcpSyncClientTests extends AbstractMcpSyncClientTests {
static String host = "http://localhost:3003";
// Uses the https://github.com/tzolov/mcp-everything-server-docker-image
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/tzolov/mcp-everything-server:v3")
.withCommand("node dist/index.js sse")
.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
.withExposedPorts(3001)
.waitingFor(Wait.forHttp("/").forStatusCode(404));
private final McpSyncHttpClientRequestCustomizer requestCustomizer = mock(McpSyncHttpClientRequestCustomizer.class);
@Override
protected McpClientTransport createMcpTransport() {
return HttpClientSseClientTransport.builder(host).httpRequestCustomizer(requestCustomizer).build();
}
@BeforeAll
static void startContainer() {
container.start();
int port = container.getMappedPort(3001);
host = "http://" + container.getHost() + ":" + port;
}
@AfterAll
static void stopContainer() {
container.stop();
}
@Test
void customizesRequests() {
var mcpTransportContext = McpTransportContext.create(Map.of("some-key", "some-value"));
withClient(createMcpTransport(), syncSpec -> syncSpec.transportContextProvider(() -> mcpTransportContext),
mcpSyncClient -> {
mcpSyncClient.initialize();
verify(requestCustomizer, atLeastOnce()).customize(any(), eq("GET"), eq(URI.create(host + "/sse")),
isNull(), eq(mcpTransportContext));
});
}
}

View file

@ -0,0 +1,423 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import io.modelcontextprotocol.spec.McpClientSession;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpTransportSessionNotFoundException;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;
import reactor.test.scheduler.VirtualTimeScheduler;
import reactor.util.context.Context;
import reactor.util.context.ContextView;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* Tests for {@link LifecycleInitializer}.
*/
class LifecycleInitializerTests {
private static final Duration INITIALIZATION_TIMEOUT = Duration.ofSeconds(5);
private static final McpSchema.ClientCapabilities CLIENT_CAPABILITIES = McpSchema.ClientCapabilities.builder()
.build();
private static final McpSchema.Implementation CLIENT_INFO = new McpSchema.Implementation("test-client", "1.0.0");
private static final List<String> PROTOCOL_VERSIONS = List.of("1.0.0", "2.0.0");
private static final McpSchema.InitializeResult MOCK_INIT_RESULT = new McpSchema.InitializeResult("2.0.0",
McpSchema.ServerCapabilities.builder().build(), new McpSchema.Implementation("test-server", "1.0.0"),
"Test instructions");
@Mock
private McpClientSession mockClientSession;
@Mock
private Function<ContextView, McpClientSession> mockSessionSupplier;
private LifecycleInitializer initializer;
@BeforeEach
void setUp() {
MockitoAnnotations.openMocks(this);
when(mockSessionSupplier.apply(any(ContextView.class))).thenReturn(mockClientSession);
when(mockClientSession.sendRequest(eq(McpSchema.METHOD_INITIALIZE), any(), any()))
.thenReturn(Mono.just(MOCK_INIT_RESULT));
when(mockClientSession.sendNotification(eq(McpSchema.METHOD_NOTIFICATION_INITIALIZED), any()))
.thenReturn(Mono.empty());
when(mockClientSession.closeGracefully()).thenReturn(Mono.empty());
initializer = new LifecycleInitializer(CLIENT_CAPABILITIES, CLIENT_INFO, PROTOCOL_VERSIONS,
INITIALIZATION_TIMEOUT, mockSessionSupplier);
}
@Test
void constructorShouldValidateParameters() {
assertThatThrownBy(() -> new LifecycleInitializer(null, CLIENT_INFO, PROTOCOL_VERSIONS, INITIALIZATION_TIMEOUT,
mockSessionSupplier))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Client capabilities must not be null");
assertThatThrownBy(() -> new LifecycleInitializer(CLIENT_CAPABILITIES, null, PROTOCOL_VERSIONS,
INITIALIZATION_TIMEOUT, mockSessionSupplier))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Client info must not be null");
assertThatThrownBy(() -> new LifecycleInitializer(CLIENT_CAPABILITIES, CLIENT_INFO, null,
INITIALIZATION_TIMEOUT, mockSessionSupplier))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Protocol versions must not be empty");
assertThatThrownBy(() -> new LifecycleInitializer(CLIENT_CAPABILITIES, CLIENT_INFO, List.of(),
INITIALIZATION_TIMEOUT, mockSessionSupplier))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Protocol versions must not be empty");
assertThatThrownBy(() -> new LifecycleInitializer(CLIENT_CAPABILITIES, CLIENT_INFO, PROTOCOL_VERSIONS, null,
mockSessionSupplier))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Initialization timeout must not be null");
assertThatThrownBy(() -> new LifecycleInitializer(CLIENT_CAPABILITIES, CLIENT_INFO, PROTOCOL_VERSIONS,
INITIALIZATION_TIMEOUT, null))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Session supplier must not be null");
}
@Test
void shouldInitializeSuccessfully() {
StepVerifier.create(initializer.withIntitialization("test", init -> Mono.just(init.initializeResult())))
.assertNext(result -> {
assertThat(result).isEqualTo(MOCK_INIT_RESULT);
assertThat(initializer.isInitialized()).isTrue();
assertThat(initializer.currentInitializationResult()).isEqualTo(MOCK_INIT_RESULT);
})
.verifyComplete();
verify(mockClientSession).sendRequest(eq(McpSchema.METHOD_INITIALIZE), any(McpSchema.InitializeRequest.class),
any());
verify(mockClientSession).sendNotification(eq(McpSchema.METHOD_NOTIFICATION_INITIALIZED), eq(null));
}
@Test
void shouldUseLatestProtocolVersionInInitializeRequest() {
AtomicReference<McpSchema.InitializeRequest> capturedRequest = new AtomicReference<>();
when(mockClientSession.sendRequest(eq(McpSchema.METHOD_INITIALIZE), any(), any())).thenAnswer(invocation -> {
capturedRequest.set((McpSchema.InitializeRequest) invocation.getArgument(1));
return Mono.just(MOCK_INIT_RESULT);
});
StepVerifier.create(initializer.withIntitialization("test", init -> Mono.just(init.initializeResult())))
.assertNext(result -> {
assertThat(capturedRequest.get().protocolVersion()).isEqualTo("2.0.0"); // Latest
// version
assertThat(capturedRequest.get().capabilities()).isEqualTo(CLIENT_CAPABILITIES);
assertThat(capturedRequest.get().clientInfo()).isEqualTo(CLIENT_INFO);
})
.verifyComplete();
}
@Test
void shouldFailForUnsupportedProtocolVersion() {
McpSchema.InitializeResult unsupportedResult = new McpSchema.InitializeResult("999.0.0", // Unsupported
// version
McpSchema.ServerCapabilities.builder().build(), new McpSchema.Implementation("test-server", "1.0.0"),
"Test instructions");
when(mockClientSession.sendRequest(eq(McpSchema.METHOD_INITIALIZE), any(), any()))
.thenReturn(Mono.just(unsupportedResult));
StepVerifier.create(initializer.withIntitialization("test", init -> Mono.just(init.initializeResult())))
.expectError(RuntimeException.class)
.verify();
verify(mockClientSession, never()).sendNotification(eq(McpSchema.METHOD_NOTIFICATION_INITIALIZED), any());
}
@Test
void shouldTimeoutOnSlowInitialization() {
VirtualTimeScheduler virtualTimeScheduler = VirtualTimeScheduler.getOrSet();
Duration INITIALIZE_TIMEOUT = Duration.ofSeconds(1);
Duration SLOW_RESPONSE_DELAY = Duration.ofSeconds(5);
LifecycleInitializer shortTimeoutInitializer = new LifecycleInitializer(CLIENT_CAPABILITIES, CLIENT_INFO,
PROTOCOL_VERSIONS, INITIALIZE_TIMEOUT, mockSessionSupplier);
when(mockClientSession.<McpSchema.InitializeResult>sendRequest(eq(McpSchema.METHOD_INITIALIZE), any(), any()))
.thenReturn(Mono.just(MOCK_INIT_RESULT).delayElement(SLOW_RESPONSE_DELAY, virtualTimeScheduler));
StepVerifier
.withVirtualTime(() -> shortTimeoutInitializer.withIntitialization("test",
init -> Mono.just(init.initializeResult())), () -> virtualTimeScheduler, Long.MAX_VALUE)
.expectSubscription()
.expectNoEvent(INITIALIZE_TIMEOUT)
.expectError(RuntimeException.class)
.verify();
}
@Test
void shouldReuseExistingInitialization() {
// First initialization
StepVerifier.create(initializer.withIntitialization("test1", init -> Mono.just("result1")))
.expectNext("result1")
.verifyComplete();
// Second call should reuse the same initialization
StepVerifier.create(initializer.withIntitialization("test2", init -> Mono.just("result2")))
.expectNext("result2")
.verifyComplete();
// Verify session was created only once
verify(mockSessionSupplier, times(1)).apply(any(ContextView.class));
verify(mockClientSession, times(1)).sendRequest(eq(McpSchema.METHOD_INITIALIZE), any(), any());
}
@Test
void shouldHandleConcurrentInitializationRequests() {
AtomicInteger sessionCreationCount = new AtomicInteger(0);
when(mockSessionSupplier.apply(any(ContextView.class))).thenAnswer(invocation -> {
sessionCreationCount.incrementAndGet();
return mockClientSession;
});
// Start multiple concurrent initializations using subscribeOn with parallel
// scheduler
Mono<String> init1 = initializer.withIntitialization("test1", init -> Mono.just("result1"))
.subscribeOn(Schedulers.parallel());
Mono<String> init2 = initializer.withIntitialization("test2", init -> Mono.just("result2"))
.subscribeOn(Schedulers.parallel());
Mono<String> init3 = initializer.withIntitialization("test3", init -> Mono.just("result3"))
.subscribeOn(Schedulers.parallel());
StepVerifier.create(Mono.zip(init1, init2, init3)).assertNext(tuple -> {
assertThat(tuple.getT1()).isEqualTo("result1");
assertThat(tuple.getT2()).isEqualTo("result2");
assertThat(tuple.getT3()).isEqualTo("result3");
}).verifyComplete();
// Should only create one session despite concurrent requests
assertThat(sessionCreationCount.get()).isEqualTo(1);
verify(mockClientSession, times(1)).sendRequest(eq(McpSchema.METHOD_INITIALIZE), any(), any());
}
@Test
void shouldHandleInitializationFailure() {
when(mockClientSession.sendRequest(eq(McpSchema.METHOD_INITIALIZE), any(), any()))
// fail once
.thenReturn(Mono.error(new RuntimeException("Connection failed")))
// succeeds on the second call
.thenReturn(Mono.just(MOCK_INIT_RESULT));
StepVerifier.create(initializer.withIntitialization("test", init -> Mono.just(init.initializeResult())))
.expectError(RuntimeException.class)
.verify();
assertThat(initializer.isInitialized()).isFalse();
assertThat(initializer.currentInitializationResult()).isNull();
// The initializer can recover from previous errors
StepVerifier
.create(initializer.withIntitialization("successful init", init -> Mono.just(init.initializeResult())))
.expectNext(MOCK_INIT_RESULT)
.verifyComplete();
assertThat(initializer.isInitialized()).isTrue();
assertThat(initializer.currentInitializationResult()).isEqualTo(MOCK_INIT_RESULT);
}
@Test
void shouldHandleTransportSessionNotFoundException() {
// successful initialization first
StepVerifier.create(initializer.withIntitialization("test", init -> Mono.just(init.initializeResult())))
.expectNext(MOCK_INIT_RESULT)
.verifyComplete();
assertThat(initializer.isInitialized()).isTrue();
// Simulate transport session not found
initializer.handleException(new McpTransportSessionNotFoundException("Session not found"));
assertThat(initializer.isInitialized()).isTrue();
// Verify that the session was closed and re-initialized
verify(mockClientSession).close();
// Verify session was created 2 times (once for initial and once for
// re-initialization)
verify(mockSessionSupplier, times(2)).apply(any(ContextView.class));
}
@Test
void shouldHandleOtherExceptions() {
// Simulate a successful initialization first
StepVerifier.create(initializer.withIntitialization("test", init -> Mono.just(init.initializeResult())))
.expectNext(MOCK_INIT_RESULT)
.verifyComplete();
assertThat(initializer.isInitialized()).isTrue();
// Simulate other exception (should not trigger re-initialization)
initializer.handleException(new RuntimeException("Some other error"));
// Should still be initialized
assertThat(initializer.isInitialized()).isTrue();
verify(mockClientSession, never()).close();
// Verify that the session was not re-created
verify(mockSessionSupplier, times(1)).apply(any(ContextView.class));
}
@Test
void shouldCloseGracefully() {
StepVerifier.create(initializer.withIntitialization("test", init -> Mono.just(init.initializeResult())))
.expectNext(MOCK_INIT_RESULT)
.verifyComplete();
StepVerifier.create(initializer.closeGracefully()).verifyComplete();
verify(mockClientSession).closeGracefully();
assertThat(initializer.isInitialized()).isFalse();
}
@Test
void shouldCloseImmediately() {
StepVerifier.create(initializer.withIntitialization("test", init -> Mono.just(init.initializeResult())))
.expectNext(MOCK_INIT_RESULT)
.verifyComplete();
// Close immediately
initializer.close();
verify(mockClientSession).close();
assertThat(initializer.isInitialized()).isFalse();
}
@Test
void shouldHandleCloseWithoutInitialization() {
// Close without initialization should not throw
initializer.close();
StepVerifier.create(initializer.closeGracefully()).verifyComplete();
verify(mockClientSession, never()).close();
verify(mockClientSession, never()).closeGracefully();
}
@Test
void shouldSetProtocolVersionsForTesting() {
List<String> newVersions = List.of("3.0.0", "4.0.0");
initializer.setProtocolVersions(newVersions);
AtomicReference<McpSchema.InitializeRequest> capturedRequest = new AtomicReference<>();
when(mockClientSession.sendRequest(eq(McpSchema.METHOD_INITIALIZE), any(), any())).thenAnswer(invocation -> {
capturedRequest.set((McpSchema.InitializeRequest) invocation.getArgument(1));
return Mono.just(new McpSchema.InitializeResult("4.0.0", McpSchema.ServerCapabilities.builder().build(),
new McpSchema.Implementation("test-server", "1.0.0"), "Test instructions"));
});
StepVerifier.create(initializer.withIntitialization("test", init -> Mono.just(init.initializeResult())))
.assertNext(result -> {
// Latest from new versions
assertThat(capturedRequest.get().protocolVersion()).isEqualTo("4.0.0");
})
.verifyComplete();
}
@Test
void shouldPassContextToSessionSupplier() {
String contextKey = "test.key";
String contextValue = "test.value";
AtomicReference<ContextView> capturedContext = new AtomicReference<>();
when(mockSessionSupplier.apply(any(ContextView.class))).thenAnswer(invocation -> {
capturedContext.set(invocation.getArgument(0));
return mockClientSession;
});
StepVerifier
.create(initializer.withIntitialization("test", init -> Mono.just(init.initializeResult()))
.contextWrite(Context.of(contextKey, contextValue)))
.expectNext(MOCK_INIT_RESULT)
.verifyComplete();
assertThat(capturedContext.get().hasKey(contextKey)).isTrue();
assertThat((String) capturedContext.get().get(contextKey)).isEqualTo(contextValue);
}
@Test
void shouldProvideAccessToMcpSessionAndInitializeResult() {
StepVerifier.create(initializer.withIntitialization("test", init -> {
assertThat(init.mcpSession()).isEqualTo(mockClientSession);
assertThat(init.initializeResult()).isEqualTo(MOCK_INIT_RESULT);
return Mono.just("success");
})).expectNext("success").verifyComplete();
}
@Test
void shouldHandleNotificationFailure() {
when(mockClientSession.sendNotification(eq(McpSchema.METHOD_NOTIFICATION_INITIALIZED), any()))
.thenReturn(Mono.error(new RuntimeException("Notification failed")));
StepVerifier.create(initializer.withIntitialization("test", init -> Mono.just(init.initializeResult())))
.expectError(RuntimeException.class)
.verify();
verify(mockClientSession).sendRequest(eq(McpSchema.METHOD_INITIALIZE), any(), any());
verify(mockClientSession).sendNotification(eq(McpSchema.METHOD_NOTIFICATION_INITIALIZED), eq(null));
}
@Test
void shouldReturnNullWhenNotInitialized() {
assertThat(initializer.isInitialized()).isFalse();
assertThat(initializer.currentInitializationResult()).isNull();
}
@Test
void shouldReinitializeAfterTransportSessionException() {
// First initialization
StepVerifier.create(initializer.withIntitialization("test1", init -> Mono.just("result1")))
.expectNext("result1")
.verifyComplete();
// Simulate transport session exception
initializer.handleException(new McpTransportSessionNotFoundException("Session lost"));
// Should be able to initialize again
StepVerifier.create(initializer.withIntitialization("test2", init -> Mono.just("result2")))
.expectNext("result2")
.verifyComplete();
// Verify two separate initializations occurred
verify(mockSessionSupplier, times(2)).apply(any(ContextView.class));
verify(mockClientSession, times(2)).sendRequest(eq(McpSchema.METHOD_INITIALIZE), any(), any());
}
}

View file

@ -0,0 +1,559 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import io.modelcontextprotocol.json.TypeRef;
import io.modelcontextprotocol.MockMcpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
import io.modelcontextprotocol.spec.McpSchema.InitializeResult;
import io.modelcontextprotocol.spec.McpSchema.PaginatedRequest;
import io.modelcontextprotocol.spec.McpSchema.Root;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import reactor.core.publisher.Mono;
import static io.modelcontextprotocol.spec.McpSchema.METHOD_INITIALIZE;
import static io.modelcontextprotocol.util.McpJsonMapperUtils.JSON_MAPPER;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
class McpAsyncClientResponseHandlerTests {
private static final McpSchema.Implementation SERVER_INFO = new McpSchema.Implementation("test-server", "1.0.0");
private static final McpSchema.ServerCapabilities SERVER_CAPABILITIES = McpSchema.ServerCapabilities.builder()
.tools(true)
.resources(true, true) // Enable both resources and resource templates
.build();
private static MockMcpClientTransport initializationEnabledTransport() {
return initializationEnabledTransport(SERVER_CAPABILITIES, SERVER_INFO);
}
private static MockMcpClientTransport initializationEnabledTransport(
McpSchema.ServerCapabilities mockServerCapabilities, McpSchema.Implementation mockServerInfo) {
McpSchema.InitializeResult mockInitResult = new McpSchema.InitializeResult(McpSchema.LATEST_PROTOCOL_VERSION,
mockServerCapabilities, mockServerInfo, "Test instructions");
return new MockMcpClientTransport((t, message) -> {
if (message instanceof McpSchema.JSONRPCRequest r && METHOD_INITIALIZE.equals(r.method())) {
McpSchema.JSONRPCResponse initResponse = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION,
r.id(), mockInitResult, null);
t.simulateIncomingMessage(initResponse);
}
}).withProtocolVersion(McpSchema.LATEST_PROTOCOL_VERSION);
}
@Test
void testSuccessfulInitialization() {
McpSchema.Implementation serverInfo = new McpSchema.Implementation("mcp-test-server", "0.0.1");
McpSchema.ServerCapabilities serverCapabilities = McpSchema.ServerCapabilities.builder()
.tools(false)
.resources(true, true) // Enable both resources and resource templates
.build();
MockMcpClientTransport transport = initializationEnabledTransport(serverCapabilities, serverInfo);
McpAsyncClient asyncMcpClient = McpClient.async(transport).build();
// Verify client is not initialized initially
assertThat(asyncMcpClient.isInitialized()).isFalse();
// Start initialization with reactive handling
InitializeResult result = asyncMcpClient.initialize().block();
// Verify initialized notification was sent
McpSchema.JSONRPCMessage notificationMessage = transport.getLastSentMessage();
assertThat(notificationMessage).isInstanceOf(McpSchema.JSONRPCNotification.class);
McpSchema.JSONRPCNotification notification = (McpSchema.JSONRPCNotification) notificationMessage;
assertThat(notification.method()).isEqualTo(McpSchema.METHOD_NOTIFICATION_INITIALIZED);
// Verify initialization result
assertThat(result).isNotNull();
assertThat(result.protocolVersion()).isEqualTo(transport.protocolVersions().get(0));
assertThat(result.capabilities()).isEqualTo(serverCapabilities);
assertThat(result.capabilities().logging()).isNull();
assertThat(result.serverInfo()).isEqualTo(serverInfo);
assertThat(result.instructions()).isEqualTo("Test instructions");
// Verify client state after initialization
assertThat(asyncMcpClient.isInitialized()).isTrue();
assertThat(asyncMcpClient.getServerCapabilities()).isEqualTo(serverCapabilities);
assertThat(asyncMcpClient.getServerInfo()).isEqualTo(serverInfo);
asyncMcpClient.closeGracefully();
}
@Test
void testToolsChangeNotificationHandling() throws IOException {
MockMcpClientTransport transport = initializationEnabledTransport();
// Create a list to store received tools for verification
List<McpSchema.Tool> receivedTools = new ArrayList<>();
// Create a consumer that will be called when tools change
Function<List<McpSchema.Tool>, Mono<Void>> toolsChangeConsumer = tools -> Mono
.fromRunnable(() -> receivedTools.addAll(tools));
// Create client with tools change consumer
McpAsyncClient asyncMcpClient = McpClient.async(transport).toolsChangeConsumer(toolsChangeConsumer).build();
assertThat(asyncMcpClient.initialize().block()).isNotNull();
// Create a mock tools list that the server will return
Map<String, Object> inputSchema = Map.of("type", "object", "properties", Map.of(), "required", List.of());
McpSchema.Tool mockTool = McpSchema.Tool.builder()
.name("test-tool-1")
.description("Test Tool 1 Description")
.inputSchema(JSON_MAPPER, JSON_MAPPER.writeValueAsString(inputSchema))
.build();
// Create page 1 response with nextPageToken
String nextPageToken = "page2Token";
McpSchema.ListToolsResult mockToolsResult1 = new McpSchema.ListToolsResult(List.of(mockTool), nextPageToken);
// Simulate server sending tools/list_changed notification
McpSchema.JSONRPCNotification notification = new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION,
McpSchema.METHOD_NOTIFICATION_TOOLS_LIST_CHANGED, null);
transport.simulateIncomingMessage(notification);
// Simulate server response to first tools/list request
McpSchema.JSONRPCRequest toolsListRequest1 = transport.getLastSentMessageAsRequest();
assertThat(toolsListRequest1.method()).isEqualTo(McpSchema.METHOD_TOOLS_LIST);
McpSchema.JSONRPCResponse toolsListResponse1 = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION,
toolsListRequest1.id(), mockToolsResult1, null);
transport.simulateIncomingMessage(toolsListResponse1);
// Create mock tools for page 2
McpSchema.Tool mockTool2 = McpSchema.Tool.builder()
.name("test-tool-2")
.description("Test Tool 2 Description")
.inputSchema(JSON_MAPPER, JSON_MAPPER.writeValueAsString(inputSchema))
.build();
// Create page 2 response with no nextPageToken (last page)
McpSchema.ListToolsResult mockToolsResult2 = new McpSchema.ListToolsResult(List.of(mockTool2), null);
// Simulate server response to second tools/list request with page token
McpSchema.JSONRPCRequest toolsListRequest2 = transport.getLastSentMessageAsRequest();
assertThat(toolsListRequest2.method()).isEqualTo(McpSchema.METHOD_TOOLS_LIST);
// Verify the page token was included in the request
PaginatedRequest params = (PaginatedRequest) toolsListRequest2.params();
assertThat(params).isNotNull();
assertThat(params.cursor()).isEqualTo(nextPageToken);
McpSchema.JSONRPCResponse toolsListResponse2 = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION,
toolsListRequest2.id(), mockToolsResult2, null);
transport.simulateIncomingMessage(toolsListResponse2);
// Verify the consumer received all expected tools from both pages
assertThat(receivedTools).hasSize(2);
assertThat(receivedTools.get(0).name()).isEqualTo("test-tool-1");
assertThat(receivedTools.get(0).description()).isEqualTo("Test Tool 1 Description");
assertThat(receivedTools.get(1).name()).isEqualTo("test-tool-2");
assertThat(receivedTools.get(1).description()).isEqualTo("Test Tool 2 Description");
asyncMcpClient.closeGracefully();
}
@Test
void testRootsListRequestHandling() {
MockMcpClientTransport transport = initializationEnabledTransport();
McpAsyncClient asyncMcpClient = McpClient.async(transport)
.roots(new Root("file:///test/path", "test-root"))
.build();
assertThat(asyncMcpClient.initialize().block()).isNotNull();
// Simulate incoming request
McpSchema.JSONRPCRequest request = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION,
McpSchema.METHOD_ROOTS_LIST, "test-id", null);
transport.simulateIncomingMessage(request);
// Verify response
McpSchema.JSONRPCMessage sentMessage = transport.getLastSentMessage();
assertThat(sentMessage).isInstanceOf(McpSchema.JSONRPCResponse.class);
McpSchema.JSONRPCResponse response = (McpSchema.JSONRPCResponse) sentMessage;
assertThat(response.id()).isEqualTo("test-id");
assertThat(response.result())
.isEqualTo(new McpSchema.ListRootsResult(List.of(new Root("file:///test/path", "test-root"))));
assertThat(response.error()).isNull();
asyncMcpClient.closeGracefully();
}
@Test
void testResourcesChangeNotificationHandling() {
MockMcpClientTransport transport = initializationEnabledTransport();
// Create a list to store received resources for verification
List<McpSchema.Resource> receivedResources = new ArrayList<>();
// Create a consumer that will be called when resources change
Function<List<McpSchema.Resource>, Mono<Void>> resourcesChangeConsumer = resources -> Mono
.fromRunnable(() -> receivedResources.addAll(resources));
// Create client with resources change consumer
McpAsyncClient asyncMcpClient = McpClient.async(transport)
.resourcesChangeConsumer(resourcesChangeConsumer)
.build();
assertThat(asyncMcpClient.initialize().block()).isNotNull();
// Create a mock resources list that the server will return
McpSchema.Resource mockResource = new McpSchema.Resource("test://resource", "Test Resource", "A test resource",
"text/plain", null);
McpSchema.ListResourcesResult mockResourcesResult = new McpSchema.ListResourcesResult(List.of(mockResource),
null);
// Simulate server sending resources/list_changed notification
McpSchema.JSONRPCNotification notification = new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION,
McpSchema.METHOD_NOTIFICATION_RESOURCES_LIST_CHANGED, null);
transport.simulateIncomingMessage(notification);
// Simulate server response to resources/list request
McpSchema.JSONRPCRequest resourcesListRequest = transport.getLastSentMessageAsRequest();
assertThat(resourcesListRequest.method()).isEqualTo(McpSchema.METHOD_RESOURCES_LIST);
McpSchema.JSONRPCResponse resourcesListResponse = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION,
resourcesListRequest.id(), mockResourcesResult, null);
transport.simulateIncomingMessage(resourcesListResponse);
// Verify the consumer received the expected resources
assertThat(receivedResources).hasSize(1);
assertThat(receivedResources.get(0).uri()).isEqualTo("test://resource");
assertThat(receivedResources.get(0).name()).isEqualTo("Test Resource");
assertThat(receivedResources.get(0).description()).isEqualTo("A test resource");
asyncMcpClient.closeGracefully();
}
@Test
void testPromptsChangeNotificationHandling() {
MockMcpClientTransport transport = initializationEnabledTransport();
// Create a list to store received prompts for verification
List<McpSchema.Prompt> receivedPrompts = new ArrayList<>();
// Create a consumer that will be called when prompts change
Function<List<McpSchema.Prompt>, Mono<Void>> promptsChangeConsumer = prompts -> Mono
.fromRunnable(() -> receivedPrompts.addAll(prompts));
// Create client with prompts change consumer
McpAsyncClient asyncMcpClient = McpClient.async(transport).promptsChangeConsumer(promptsChangeConsumer).build();
assertThat(asyncMcpClient.initialize().block()).isNotNull();
// Create a mock prompts list that the server will return
McpSchema.Prompt mockPrompt = new McpSchema.Prompt("test-prompt", "Test Prompt", "Test Prompt Description",
List.of(new McpSchema.PromptArgument("arg1", "Test argument", "Test argument", true)));
McpSchema.ListPromptsResult mockPromptsResult = new McpSchema.ListPromptsResult(List.of(mockPrompt), null);
// Simulate server sending prompts/list_changed notification
McpSchema.JSONRPCNotification notification = new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION,
McpSchema.METHOD_NOTIFICATION_PROMPTS_LIST_CHANGED, null);
transport.simulateIncomingMessage(notification);
// Simulate server response to prompts/list request
McpSchema.JSONRPCRequest promptsListRequest = transport.getLastSentMessageAsRequest();
assertThat(promptsListRequest.method()).isEqualTo(McpSchema.METHOD_PROMPT_LIST);
McpSchema.JSONRPCResponse promptsListResponse = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION,
promptsListRequest.id(), mockPromptsResult, null);
transport.simulateIncomingMessage(promptsListResponse);
// Verify the consumer received the expected prompts
assertThat(receivedPrompts).hasSize(1);
assertThat(receivedPrompts.get(0).name()).isEqualTo("test-prompt");
assertThat(receivedPrompts.get(0).description()).isEqualTo("Test Prompt Description");
assertThat(receivedPrompts.get(0).arguments()).hasSize(1);
assertThat(receivedPrompts.get(0).arguments().get(0).name()).isEqualTo("arg1");
asyncMcpClient.closeGracefully();
}
@Test
void testSamplingCreateMessageRequestHandling() {
MockMcpClientTransport transport = initializationEnabledTransport();
// Create a test sampling handler that echoes back the input
Function<McpSchema.CreateMessageRequest, Mono<McpSchema.CreateMessageResult>> samplingHandler = request -> {
var content = request.messages().get(0).content();
return Mono.just(new McpSchema.CreateMessageResult(McpSchema.Role.ASSISTANT, content, "test-model",
McpSchema.CreateMessageResult.StopReason.END_TURN));
};
// Create client with sampling capability and handler
McpAsyncClient asyncMcpClient = McpClient.async(transport)
.capabilities(ClientCapabilities.builder().sampling().build())
.sampling(samplingHandler)
.build();
assertThat(asyncMcpClient.initialize().block()).isNotNull();
// Create a mock create message request
var messageRequest = new McpSchema.CreateMessageRequest(
List.of(new McpSchema.SamplingMessage(McpSchema.Role.USER, new McpSchema.TextContent("Test message"))),
null, // modelPreferences
"Test system prompt", McpSchema.CreateMessageRequest.ContextInclusionStrategy.NONE, 0.7, // temperature
100, // maxTokens
null, // stopSequences
null // metadata
);
// Simulate incoming request
McpSchema.JSONRPCRequest request = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION,
McpSchema.METHOD_SAMPLING_CREATE_MESSAGE, "test-id", messageRequest);
transport.simulateIncomingMessage(request);
// Verify response
McpSchema.JSONRPCMessage sentMessage = transport.getLastSentMessage();
assertThat(sentMessage).isInstanceOf(McpSchema.JSONRPCResponse.class);
McpSchema.JSONRPCResponse response = (McpSchema.JSONRPCResponse) sentMessage;
assertThat(response.id()).isEqualTo("test-id");
assertThat(response.error()).isNull();
McpSchema.CreateMessageResult result = transport.unmarshalFrom(response.result(),
new TypeRef<McpSchema.CreateMessageResult>() {
});
assertThat(result).isNotNull();
assertThat(result.role()).isEqualTo(McpSchema.Role.ASSISTANT);
assertThat(result.content()).isNotNull();
assertThat(result.model()).isEqualTo("test-model");
assertThat(result.stopReason()).isEqualTo(McpSchema.CreateMessageResult.StopReason.END_TURN);
asyncMcpClient.closeGracefully();
}
@Test
void testSamplingCreateMessageRequestHandlingWithoutCapability() {
MockMcpClientTransport transport = initializationEnabledTransport();
// Create client without sampling capability
McpAsyncClient asyncMcpClient = McpClient.async(transport)
.capabilities(ClientCapabilities.builder().build()) // No sampling capability
.build();
assertThat(asyncMcpClient.initialize().block()).isNotNull();
// Create a mock create message request
var messageRequest = new McpSchema.CreateMessageRequest(
List.of(new McpSchema.SamplingMessage(McpSchema.Role.USER, new McpSchema.TextContent("Test message"))),
null, null, null, null, 0, null, null);
// Simulate incoming request
McpSchema.JSONRPCRequest request = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION,
McpSchema.METHOD_SAMPLING_CREATE_MESSAGE, "test-id", messageRequest);
transport.simulateIncomingMessage(request);
// Verify error response
McpSchema.JSONRPCMessage sentMessage = transport.getLastSentMessage();
assertThat(sentMessage).isInstanceOf(McpSchema.JSONRPCResponse.class);
McpSchema.JSONRPCResponse response = (McpSchema.JSONRPCResponse) sentMessage;
assertThat(response.id()).isEqualTo("test-id");
assertThat(response.result()).isNull();
assertThat(response.error()).isNotNull();
assertThat(response.error().message()).contains("Method not found: sampling/createMessage");
asyncMcpClient.closeGracefully();
}
@Test
void testSamplingCreateMessageRequestHandlingWithNullHandler() {
MockMcpClientTransport transport = new MockMcpClientTransport();
// Create client with sampling capability but null handler
assertThatThrownBy(
() -> McpClient.async(transport).capabilities(ClientCapabilities.builder().sampling().build()).build())
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Sampling handler must not be null when client capabilities include sampling");
}
@Test
@SuppressWarnings("unchecked")
void testElicitationCreateRequestHandling() {
MockMcpClientTransport transport = initializationEnabledTransport();
// Create a test elicitation handler that echoes back the input
Function<McpSchema.ElicitRequest, Mono<McpSchema.ElicitResult>> elicitationHandler = request -> {
assertThat(request.message()).isNotEmpty();
assertThat(request.requestedSchema()).isInstanceOf(Map.class);
assertThat(request.requestedSchema().get("type")).isEqualTo("object");
var properties = request.requestedSchema().get("properties");
assertThat(properties).isNotNull();
assertThat(((Map<String, Object>) properties).get("message")).isInstanceOf(Map.class);
return Mono.just(McpSchema.ElicitResult.builder()
.message(McpSchema.ElicitResult.Action.ACCEPT)
.content(Map.of("message", request.message()))
.build());
};
// Create client with elicitation capability and handler
McpAsyncClient asyncMcpClient = McpClient.async(transport)
.capabilities(ClientCapabilities.builder().elicitation().build())
.elicitation(elicitationHandler)
.build();
assertThat(asyncMcpClient.initialize().block()).isNotNull();
// Create a mock elicitation
var elicitRequest = McpSchema.ElicitRequest.builder()
.message("Test message")
.requestedSchema(Map.of("type", "object", "properties", Map.of("message", Map.of("type", "string"))))
.build();
// Simulate incoming request
McpSchema.JSONRPCRequest request = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION,
McpSchema.METHOD_ELICITATION_CREATE, "test-id", elicitRequest);
transport.simulateIncomingMessage(request);
// Verify response
McpSchema.JSONRPCMessage sentMessage = transport.getLastSentMessage();
assertThat(sentMessage).isInstanceOf(McpSchema.JSONRPCResponse.class);
McpSchema.JSONRPCResponse response = (McpSchema.JSONRPCResponse) sentMessage;
assertThat(response.id()).isEqualTo("test-id");
assertThat(response.error()).isNull();
McpSchema.ElicitResult result = transport.unmarshalFrom(response.result(), new TypeRef<>() {
});
assertThat(result).isNotNull();
assertThat(result.action()).isEqualTo(McpSchema.ElicitResult.Action.ACCEPT);
assertThat(result.content()).isEqualTo(Map.of("message", "Test message"));
asyncMcpClient.closeGracefully();
}
@ParameterizedTest
@EnumSource(value = McpSchema.ElicitResult.Action.class, names = { "DECLINE", "CANCEL" })
void testElicitationFailRequestHandling(McpSchema.ElicitResult.Action action) {
MockMcpClientTransport transport = initializationEnabledTransport();
// Create a test elicitation handler to decline the request
Function<McpSchema.ElicitRequest, Mono<McpSchema.ElicitResult>> elicitationHandler = request -> Mono
.just(McpSchema.ElicitResult.builder().message(action).build());
// Create client with elicitation capability and handler
McpAsyncClient asyncMcpClient = McpClient.async(transport)
.capabilities(ClientCapabilities.builder().elicitation().build())
.elicitation(elicitationHandler)
.build();
assertThat(asyncMcpClient.initialize().block()).isNotNull();
// Create a mock elicitation
var elicitRequest = McpSchema.ElicitRequest.builder()
.message("Test message")
.requestedSchema(Map.of("type", "object", "properties", Map.of("message", Map.of("type", "string"))))
.build();
// Simulate incoming request
McpSchema.JSONRPCRequest request = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION,
McpSchema.METHOD_ELICITATION_CREATE, "test-id", elicitRequest);
transport.simulateIncomingMessage(request);
// Verify response
McpSchema.JSONRPCMessage sentMessage = transport.getLastSentMessage();
assertThat(sentMessage).isInstanceOf(McpSchema.JSONRPCResponse.class);
McpSchema.JSONRPCResponse response = (McpSchema.JSONRPCResponse) sentMessage;
assertThat(response.id()).isEqualTo("test-id");
assertThat(response.error()).isNull();
McpSchema.ElicitResult result = transport.unmarshalFrom(response.result(), new TypeRef<>() {
});
assertThat(result).isNotNull();
assertThat(result.action()).isEqualTo(action);
assertThat(result.content()).isNull();
asyncMcpClient.closeGracefully();
}
@Test
void testElicitationCreateRequestHandlingWithoutCapability() {
MockMcpClientTransport transport = initializationEnabledTransport();
// Create client without elicitation capability
McpAsyncClient asyncMcpClient = McpClient.async(transport)
.capabilities(ClientCapabilities.builder().build()) // No elicitation
// capability
.build();
assertThat(asyncMcpClient.initialize().block()).isNotNull();
// Create a mock elicitation
var elicitRequest = new McpSchema.ElicitRequest("test",
Map.of("type", "object", "properties", Map.of("test", Map.of("type", "boolean", "defaultValue", true,
"description", "test-description", "title", "test-title"))));
// Simulate incoming request
McpSchema.JSONRPCRequest request = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION,
McpSchema.METHOD_ELICITATION_CREATE, "test-id", elicitRequest);
transport.simulateIncomingMessage(request);
// Verify error response
McpSchema.JSONRPCMessage sentMessage = transport.getLastSentMessage();
assertThat(sentMessage).isInstanceOf(McpSchema.JSONRPCResponse.class);
McpSchema.JSONRPCResponse response = (McpSchema.JSONRPCResponse) sentMessage;
assertThat(response.id()).isEqualTo("test-id");
assertThat(response.result()).isNull();
assertThat(response.error()).isNotNull();
assertThat(response.error().message()).contains("Method not found: elicitation/create");
asyncMcpClient.closeGracefully();
}
@Test
void testElicitationCreateRequestHandlingWithNullHandler() {
MockMcpClientTransport transport = new MockMcpClientTransport();
// Create client with elicitation capability but null handler
assertThatThrownBy(() -> McpClient.async(transport)
.capabilities(ClientCapabilities.builder().elicitation().build())
.build()).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Elicitation handler must not be null when client capabilities include elicitation");
}
@Test
void testPingMessageRequestHandling() {
MockMcpClientTransport transport = initializationEnabledTransport();
McpAsyncClient asyncMcpClient = McpClient.async(transport).build();
assertThat(asyncMcpClient.initialize().block()).isNotNull();
// Simulate incoming ping request from server
McpSchema.JSONRPCRequest pingRequest = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION,
McpSchema.METHOD_PING, "ping-id", null);
transport.simulateIncomingMessage(pingRequest);
// Verify response
McpSchema.JSONRPCMessage sentMessage = transport.getLastSentMessage();
assertThat(sentMessage).isInstanceOf(McpSchema.JSONRPCResponse.class);
McpSchema.JSONRPCResponse response = (McpSchema.JSONRPCResponse) sentMessage;
assertThat(response.id()).isEqualTo("ping-id");
assertThat(response.error()).isNull();
assertThat(response.result()).isInstanceOf(Map.class);
assertThat(((Map<?, ?>) response.result())).isEmpty();
asyncMcpClient.closeGracefully();
}
}

View file

@ -0,0 +1,90 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client;
import io.modelcontextprotocol.json.TypeRef;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.ProtocolVersions;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import static io.modelcontextprotocol.util.McpJsonMapperUtils.JSON_MAPPER;
import static org.assertj.core.api.Assertions.assertThatCode;
class McpAsyncClientTests {
public static final McpSchema.Implementation MOCK_SERVER_INFO = new McpSchema.Implementation("test-server",
"1.0.0");
public static final McpSchema.ServerCapabilities MOCK_SERVER_CAPABILITIES = McpSchema.ServerCapabilities.builder()
.build();
public static final McpSchema.InitializeResult MOCK_INIT_RESULT = new McpSchema.InitializeResult(
ProtocolVersions.MCP_2024_11_05, MOCK_SERVER_CAPABILITIES, MOCK_SERVER_INFO, "Test instructions");
private static final String CONTEXT_KEY = "context.key";
@Test
void validateContextPassedToTransportConnect() {
McpClientTransport transport = new McpClientTransport() {
Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler;
final AtomicReference<String> contextValue = new AtomicReference<>();
@Override
public Mono<Void> connect(
Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
return Mono.deferContextual(ctx -> {
this.handler = handler;
if (ctx.hasKey(CONTEXT_KEY)) {
this.contextValue.set(ctx.get(CONTEXT_KEY));
}
return Mono.empty();
});
}
@Override
public Mono<Void> closeGracefully() {
return Mono.empty();
}
@Override
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
if (!"hello".equals(this.contextValue.get())) {
return Mono.error(new RuntimeException("Context value not propagated via #connect method"));
}
// We're only interested in handling the init request to provide an init
// response
if (!(message instanceof McpSchema.JSONRPCRequest)) {
return Mono.empty();
}
McpSchema.JSONRPCResponse initResponse = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION,
((McpSchema.JSONRPCRequest) message).id(), MOCK_INIT_RESULT, null);
return handler.apply(Mono.just(initResponse)).then();
}
@Override
public <T> T unmarshalFrom(Object data, TypeRef<T> typeRef) {
return JSON_MAPPER.convertValue(data, new TypeRef<>() {
@Override
public java.lang.reflect.Type getType() {
return typeRef.getType();
}
});
}
};
assertThatCode(() -> {
McpAsyncClient client = McpClient.async(transport).build();
client.initialize().contextWrite(ctx -> ctx.put(CONTEXT_KEY, "hello")).block();
}).doesNotThrowAnyException();
}
}

View file

@ -0,0 +1,159 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import java.time.Duration;
import java.util.List;
import io.modelcontextprotocol.MockMcpClientTransport;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.InitializeResult;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for MCP protocol version negotiation and compatibility.
*/
class McpClientProtocolVersionTests {
private static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(30);
private static final McpSchema.Implementation CLIENT_INFO = new McpSchema.Implementation("test-client", "1.0.0");
@Test
void shouldUseLatestVersionByDefault() {
MockMcpClientTransport transport = new MockMcpClientTransport();
McpAsyncClient client = McpClient.async(transport)
.clientInfo(CLIENT_INFO)
.requestTimeout(REQUEST_TIMEOUT)
.build();
try {
Mono<InitializeResult> initializeResultMono = client.initialize();
String protocolVersion = transport.protocolVersions().get(transport.protocolVersions().size() - 1);
StepVerifier.create(initializeResultMono).then(() -> {
McpSchema.JSONRPCRequest request = transport.getLastSentMessageAsRequest();
assertThat(request.params()).isInstanceOf(McpSchema.InitializeRequest.class);
McpSchema.InitializeRequest initRequest = (McpSchema.InitializeRequest) request.params();
assertThat(initRequest.protocolVersion()).isEqualTo(transport.protocolVersions().get(0));
transport.simulateIncomingMessage(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
new McpSchema.InitializeResult(protocolVersion, null,
new McpSchema.Implementation("test-server", "1.0.0"), null),
null));
}).assertNext(result -> {
assertThat(result.protocolVersion()).isEqualTo(protocolVersion);
}).verifyComplete();
}
finally {
// Ensure cleanup happens even if test fails
StepVerifier.create(client.closeGracefully()).verifyComplete();
}
}
@Test
void shouldNegotiateSpecificVersion() {
String oldVersion = "0.1.0";
MockMcpClientTransport transport = new MockMcpClientTransport();
McpAsyncClient client = McpClient.async(transport)
.clientInfo(CLIENT_INFO)
.requestTimeout(REQUEST_TIMEOUT)
.build();
client.setProtocolVersions(List.of(oldVersion, McpSchema.LATEST_PROTOCOL_VERSION));
try {
Mono<InitializeResult> initializeResultMono = client.initialize();
StepVerifier.create(initializeResultMono).then(() -> {
McpSchema.JSONRPCRequest request = transport.getLastSentMessageAsRequest();
assertThat(request.params()).isInstanceOf(McpSchema.InitializeRequest.class);
McpSchema.InitializeRequest initRequest = (McpSchema.InitializeRequest) request.params();
assertThat(initRequest.protocolVersion()).isIn(List.of(oldVersion, McpSchema.LATEST_PROTOCOL_VERSION));
transport.simulateIncomingMessage(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
new McpSchema.InitializeResult(oldVersion, null,
new McpSchema.Implementation("test-server", "1.0.0"), null),
null));
}).assertNext(result -> {
assertThat(result.protocolVersion()).isEqualTo(oldVersion);
}).verifyComplete();
}
finally {
StepVerifier.create(client.closeGracefully()).verifyComplete();
}
}
@Test
void shouldFailForUnsupportedVersion() {
String unsupportedVersion = "999.999.999";
MockMcpClientTransport transport = new MockMcpClientTransport();
McpAsyncClient client = McpClient.async(transport)
.clientInfo(CLIENT_INFO)
.requestTimeout(REQUEST_TIMEOUT)
.build();
try {
Mono<InitializeResult> initializeResultMono = client.initialize();
StepVerifier.create(initializeResultMono).then(() -> {
McpSchema.JSONRPCRequest request = transport.getLastSentMessageAsRequest();
assertThat(request.params()).isInstanceOf(McpSchema.InitializeRequest.class);
transport.simulateIncomingMessage(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
new McpSchema.InitializeResult(unsupportedVersion, null,
new McpSchema.Implementation("test-server", "1.0.0"), null),
null));
}).expectError(RuntimeException.class).verify();
}
finally {
StepVerifier.create(client.closeGracefully()).verifyComplete();
}
}
@Test
void shouldUseHighestVersionWhenMultipleSupported() {
String oldVersion = "0.1.0";
String middleVersion = "0.2.0";
String latestVersion = McpSchema.LATEST_PROTOCOL_VERSION;
MockMcpClientTransport transport = new MockMcpClientTransport();
McpAsyncClient client = McpClient.async(transport)
.clientInfo(CLIENT_INFO)
.requestTimeout(REQUEST_TIMEOUT)
.build();
client.setProtocolVersions(List.of(oldVersion, middleVersion, latestVersion));
try {
Mono<InitializeResult> initializeResultMono = client.initialize();
StepVerifier.create(initializeResultMono).then(() -> {
McpSchema.JSONRPCRequest request = transport.getLastSentMessageAsRequest();
McpSchema.InitializeRequest initRequest = (McpSchema.InitializeRequest) request.params();
assertThat(initRequest.protocolVersion()).isEqualTo(latestVersion);
transport.simulateIncomingMessage(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
new McpSchema.InitializeResult(latestVersion, null,
new McpSchema.Implementation("test-server", "1.0.0"), null),
null));
}).assertNext(result -> {
assertThat(result.protocolVersion()).isEqualTo(latestVersion);
}).verifyComplete();
}
finally {
StepVerifier.create(client.closeGracefully()).verifyComplete();
}
}
}

View file

@ -0,0 +1,19 @@
package io.modelcontextprotocol.client;
import io.modelcontextprotocol.client.transport.ServerParameters;
public final class ServerParameterUtils {
private ServerParameterUtils() {
}
public static ServerParameters createServerParameters() {
if (System.getProperty("os.name").toLowerCase().contains("win")) {
return ServerParameters.builder("cmd.exe")
.args("/c", "npx.cmd", "-y", "@modelcontextprotocol/server-everything", "stdio")
.build();
}
return ServerParameters.builder("npx").args("-y", "@modelcontextprotocol/server-everything", "stdio").build();
}
}

View file

@ -0,0 +1,47 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import java.time.Duration;
import io.modelcontextprotocol.client.transport.ServerParameters;
import io.modelcontextprotocol.client.transport.StdioClientTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
import org.junit.jupiter.api.Timeout;
import static io.modelcontextprotocol.client.ServerParameterUtils.createServerParameters;
import static io.modelcontextprotocol.util.McpJsonMapperUtils.JSON_MAPPER;
/**
* Tests for the {@link McpAsyncClient} with {@link StdioClientTransport}.
*
* <p>
* These tests use npx to download and run the MCP "everything" server locally. The first
* test execution will download the everything server scripts and cache them locally,
* which can take more than 15 seconds. Subsequent test runs will use the cached version
* and execute faster.
*
* @author Christian Tzolov
* @author Dariusz Jędrzejczyk
*/
@Timeout(25) // Giving extra time beyond the client timeout to account for initial server
// download
class StdioMcpAsyncClientTests extends AbstractMcpAsyncClientTests {
@Override
protected McpClientTransport createMcpTransport() {
return new StdioClientTransport(createServerParameters(), JSON_MAPPER);
}
protected Duration getInitializationTimeout() {
return Duration.ofSeconds(20);
}
@Override
protected Duration getRequestTimeout() {
return Duration.ofSeconds(25);
}
}

View file

@ -0,0 +1,78 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import io.modelcontextprotocol.client.transport.ServerParameters;
import io.modelcontextprotocol.client.transport.StdioClientTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import reactor.core.publisher.Sinks;
import reactor.test.StepVerifier;
import static io.modelcontextprotocol.client.ServerParameterUtils.createServerParameters;
import static io.modelcontextprotocol.util.McpJsonMapperUtils.JSON_MAPPER;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for the {@link McpSyncClient} with {@link StdioClientTransport}.
*
* <p>
* These tests use npx to download and run the MCP "everything" server locally. The first
* test execution will download the everything server scripts and cache them locally,
* which can take more than 15 seconds. Subsequent test runs will use the cached version
* and execute faster.
*
* @author Christian Tzolov
* @author Dariusz Jędrzejczyk
*/
@Timeout(25) // Giving extra time beyond the client timeout to account for initial server
// download
class StdioMcpSyncClientTests extends AbstractMcpSyncClientTests {
@Override
protected McpClientTransport createMcpTransport() {
ServerParameters stdioParams = createServerParameters();
return new StdioClientTransport(stdioParams, JSON_MAPPER);
}
@Test
void customErrorHandlerShouldReceiveErrors() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<String> receivedError = new AtomicReference<>();
McpClientTransport transport = createMcpTransport();
StepVerifier.create(transport.connect(msg -> msg)).verifyComplete();
((StdioClientTransport) transport).setStdErrorHandler(error -> {
receivedError.set(error);
latch.countDown();
});
String errorMessage = "Test error";
((StdioClientTransport) transport).getErrorSink().emitNext(errorMessage, Sinks.EmitFailureHandler.FAIL_FAST);
assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(receivedError.get()).isNotNull().isEqualTo(errorMessage);
StepVerifier.create(transport.closeGracefully()).expectComplete().verify(Duration.ofSeconds(5));
}
protected Duration getInitializationTimeout() {
return Duration.ofSeconds(10);
}
@Override
protected Duration getRequestTimeout() {
return Duration.ofSeconds(25);
}
}

Some files were not shown because too many files have changed in this diff Show more