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

Move mcp-core integration tests to mcp-test package

Browse source 
- The integration tests pass with both Jackson 2 and Jackson 3.
- Modified McpSchemaTests.testContentDeserializationWrongType to work
  with both Jackson 2 and Jackson 3.

Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>
This commit is contained in:
Daniel Garnier-Moiroux 2026-02-11 14:35:52 +01:00 • committed by Daniel Garnier-Moiroux
commit 6f3906480d
41 changed files with 135 additions and 94 deletions
Download patch file Download diff file Expand all files Collapse all files

106
mcp-test/src/test/java/io/modelcontextprotocol/MockMcpClientTransport.java Normal file
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.McpJsonDefaults;
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 McpJsonDefaults.getDefaultMcpJsonMapper().convertValue(data, typeRef);
}
}

74
mcp-test/src/test/java/io/modelcontextprotocol/MockMcpServerTransport.java Normal file
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.McpJsonDefaults;
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 McpJsonDefaults.getDefaultMcpJsonMapper().convertValue(data, typeRef);
}
}

50
mcp-test/src/test/java/io/modelcontextprotocol/MockMcpServerTransportProvider.java Normal file
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();
}
}

39
mcp-test/src/test/java/io/modelcontextprotocol/client/HttpClientStreamableHttpAsyncClientResiliencyTests.java Normal file
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();
});
}
}

44
mcp-test/src/test/java/io/modelcontextprotocol/client/HttpClientStreamableHttpAsyncClientTests.java Normal file
View file

@ -0,0 +1,44 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client;
import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
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;
@Timeout(15)
public class HttpClientStreamableHttpAsyncClientTests extends AbstractMcpAsyncClientTests {
private static String host = "http://localhost:3001";
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/node:lts-alpine3.23")
.withCommand("npx -y @modelcontextprotocol/server-everything@2025.12.18 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();
}
}

72
mcp-test/src/test/java/io/modelcontextprotocol/client/HttpClientStreamableHttpSyncClientTests.java Normal file
View file

@ -0,0 +1,72 @@
/*
* 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";
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/node:lts-alpine3.23")
.withCommand("npx -y @modelcontextprotocol/server-everything@2025.12.18 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));
});
}
}

139
mcp-test/src/test/java/io/modelcontextprotocol/client/HttpSseMcpAsyncClientLostConnectionTests.java Normal file
View file

@ -0,0 +1,139 @@
/*
* 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";
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/node:lts-alpine3.23")
.withCommand("npx -y @modelcontextprotocol/server-everything@2025.12.18 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 testPingWithExactExceptionType() {
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();
});
}
}

50
mcp-test/src/test/java/io/modelcontextprotocol/client/HttpSseMcpAsyncClientTests.java Normal file
View file

@ -0,0 +1,50 @@
/*
* 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";
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/node:lts-alpine3.23")
.withCommand("npx -y @modelcontextprotocol/server-everything@2025.12.18 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();
}
}

77
mcp-test/src/test/java/io/modelcontextprotocol/client/HttpSseMcpSyncClientTests.java Normal file
View file

@ -0,0 +1,77 @@
/*
* 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";
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/node:lts-alpine3.23")
.withCommand("npx -y @modelcontextprotocol/server-everything@2025.12.18 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));
});
}
}

559
mcp-test/src/test/java/io/modelcontextprotocol/client/McpAsyncClientResponseHandlerTests.java Normal file
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();
}
}

310
mcp-test/src/test/java/io/modelcontextprotocol/client/McpAsyncClientTests.java Normal file
View file

@ -0,0 +1,310 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
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 reactor.test.StepVerifier;
import static io.modelcontextprotocol.util.McpJsonMapperUtils.JSON_MAPPER;
import static org.assertj.core.api.Assertions.assertThat;
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()
.tools(true)
.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";
private McpClientTransport createMockTransportForToolValidation(boolean hasOutputSchema, boolean invalidOutput) {
// Create tool with or without output schema
Map<String, Object> inputSchemaMap = Map.of("type", "object", "properties",
Map.of("expression", Map.of("type", "string")), "required", List.of("expression"));
McpSchema.JsonSchema inputSchema = new McpSchema.JsonSchema("object", inputSchemaMap, null, null, null, null);
McpSchema.Tool.Builder toolBuilder = McpSchema.Tool.builder()
.name("calculator")
.description("Performs mathematical calculations")
.inputSchema(inputSchema);
if (hasOutputSchema) {
Map<String, Object> outputSchema = Map.of("type", "object", "properties",
Map.of("result", Map.of("type", "number"), "operation", Map.of("type", "string")), "required",
List.of("result", "operation"));
toolBuilder.outputSchema(outputSchema);
}
McpSchema.Tool calculatorTool = toolBuilder.build();
McpSchema.ListToolsResult mockToolsResult = new McpSchema.ListToolsResult(List.of(calculatorTool), null);
// Create call tool result - valid or invalid based on parameter
Map<String, Object> structuredContent = invalidOutput ? Map.of("result", "5", "operation", "add")
: Map.of("result", 5, "operation", "add");
McpSchema.CallToolResult mockCallToolResult = McpSchema.CallToolResult.builder()
.addTextContent("Calculation result")
.structuredContent(structuredContent)
.build();
return new McpClientTransport() {
Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler;
@Override
public Mono<Void> connect(
Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
this.handler = handler;
return Mono.empty();
}
@Override
public Mono<Void> closeGracefully() {
return Mono.empty();
}
@Override
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
if (!(message instanceof McpSchema.JSONRPCRequest request)) {
return Mono.empty();
}
McpSchema.JSONRPCResponse response;
if (McpSchema.METHOD_INITIALIZE.equals(request.method())) {
response = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), MOCK_INIT_RESULT,
null);
}
else if (McpSchema.METHOD_TOOLS_LIST.equals(request.method())) {
response = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), mockToolsResult,
null);
}
else if (McpSchema.METHOD_TOOLS_CALL.equals(request.method())) {
response = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
mockCallToolResult, null);
}
else {
return Mono.empty();
}
return handler.apply(Mono.just(response)).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();
}
});
}
};
}
@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();
}
@Test
void testCallToolWithOutputSchemaValidationSuccess() {
McpClientTransport transport = createMockTransportForToolValidation(true, false);
McpAsyncClient client = McpClient.async(transport).enableCallToolSchemaCaching(true).build();
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
StepVerifier.create(client.callTool(new McpSchema.CallToolRequest("calculator", Map.of("expression", "2 + 3"))))
.expectNextMatches(response -> {
assertThat(response).isNotNull();
assertThat(response.isError()).isFalse();
assertThat(response.structuredContent()).isInstanceOf(Map.class);
assertThat((Map<?, ?>) response.structuredContent()).hasSize(2);
assertThat(response.content()).hasSize(1);
return true;
})
.verifyComplete();
StepVerifier.create(client.closeGracefully()).verifyComplete();
}
@Test
void testCallToolWithNoOutputSchemaSuccess() {
McpClientTransport transport = createMockTransportForToolValidation(false, false);
McpAsyncClient client = McpClient.async(transport).enableCallToolSchemaCaching(true).build();
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
StepVerifier.create(client.callTool(new McpSchema.CallToolRequest("calculator", Map.of("expression", "2 + 3"))))
.expectNextMatches(response -> {
assertThat(response).isNotNull();
assertThat(response.isError()).isFalse();
assertThat(response.structuredContent()).isInstanceOf(Map.class);
assertThat((Map<?, ?>) response.structuredContent()).hasSize(2);
assertThat(response.content()).hasSize(1);
return true;
})
.verifyComplete();
StepVerifier.create(client.closeGracefully()).verifyComplete();
}
@Test
void testCallToolWithOutputSchemaValidationFailure() {
McpClientTransport transport = createMockTransportForToolValidation(true, true);
McpAsyncClient client = McpClient.async(transport).enableCallToolSchemaCaching(true).build();
StepVerifier.create(client.initialize()).expectNextMatches(Objects::nonNull).verifyComplete();
StepVerifier.create(client.callTool(new McpSchema.CallToolRequest("calculator", Map.of("expression", "2 + 3"))))
.expectErrorMatches(ex -> ex instanceof IllegalArgumentException
&& ex.getMessage().contains("Tool call result validation failed"))
.verify();
StepVerifier.create(client.closeGracefully()).verifyComplete();
}
@Test
void testListToolsWithEmptyCursor() {
McpSchema.Tool addTool = McpSchema.Tool.builder().name("add").description("calculate add").build();
McpSchema.Tool subtractTool = McpSchema.Tool.builder()
.name("subtract")
.description("calculate subtract")
.build();
McpSchema.ListToolsResult mockToolsResult = new McpSchema.ListToolsResult(List.of(addTool, subtractTool), "");
McpClientTransport transport = new McpClientTransport() {
Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler;
@Override
public Mono<Void> connect(
Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
return Mono.deferContextual(ctx -> {
this.handler = handler;
return Mono.empty();
});
}
@Override
public Mono<Void> closeGracefully() {
return Mono.empty();
}
@Override
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
if (!(message instanceof McpSchema.JSONRPCRequest request)) {
return Mono.empty();
}
McpSchema.JSONRPCResponse response;
if (McpSchema.METHOD_INITIALIZE.equals(request.method())) {
response = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), MOCK_INIT_RESULT,
null);
}
else if (McpSchema.METHOD_TOOLS_LIST.equals(request.method())) {
response = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), mockToolsResult,
null);
}
else {
return Mono.empty();
}
return handler.apply(Mono.just(response)).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();
}
});
}
};
McpAsyncClient client = McpClient.async(transport).enableCallToolSchemaCaching(true).build();
Mono<McpSchema.ListToolsResult> mono = client.listTools();
McpSchema.ListToolsResult toolsResult = mono.block();
assertThat(toolsResult).isNotNull();
Set<String> names = toolsResult.tools().stream().map(McpSchema.Tool::name).collect(Collectors.toSet());
assertThat(names).containsExactlyInAnyOrder("subtract", "add");
}
}

158
mcp-test/src/test/java/io/modelcontextprotocol/client/McpClientProtocolVersionTests.java Normal file
View file

@ -0,0 +1,158 @@
/*
* 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.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.InitializeResult;
import io.modelcontextprotocol.spec.McpSchema.ServerCapabilities;
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(300);
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, ServerCapabilities.builder().build(),
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, ServerCapabilities.builder().build(),
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, ServerCapabilities.builder().build(),
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, ServerCapabilities.builder().build(),
new McpSchema.Implementation("test-server", "1.0.0"), null),
null));
}).assertNext(result -> {
assertThat(result.protocolVersion()).isEqualTo(latestVersion);
}).verifyComplete();
}
finally {
StepVerifier.create(client.closeGracefully()).verifyComplete();
}
}
}

21
mcp-test/src/test/java/io/modelcontextprotocol/client/ServerParameterUtils.java Normal file
View file

@ -0,0 +1,21 @@
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@2025.12.18", "stdio")
.build();
}
return ServerParameters.builder("npx")
.args("-y", "@modelcontextprotocol/server-everything@2025.12.18", "stdio")
.build();
}
}

47
mcp-test/src/test/java/io/modelcontextprotocol/client/StdioMcpAsyncClientTests.java Normal file
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);
}
}

78
mcp-test/src/test/java/io/modelcontextprotocol/client/StdioMcpSyncClientTests.java Normal file
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);
}
}

480
mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientSseClientTransportTests.java Normal file
View file

@ -0,0 +1,480 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client.transport;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpClientRequestCustomizer;
import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCRequest;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.mockito.ArgumentCaptor;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.test.StepVerifier;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.util.UriComponentsBuilder;
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.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* Tests for the {@link HttpClientSseClientTransport} class.
*
* @author Christian Tzolov
*/
@Timeout(15)
class HttpClientSseClientTransportTests {
static String host = "http://localhost:3001";
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/node:lts-alpine3.23")
.withCommand("npx -y @modelcontextprotocol/server-everything@2025.12.18 sse")
.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
.withExposedPorts(3001)
.waitingFor(Wait.forHttp("/").forStatusCode(404));
private TestHttpClientSseClientTransport transport;
private final McpTransportContext context = McpTransportContext.create(Map.of("some-key", "some-value"));
// Test class to access protected methods
static class TestHttpClientSseClientTransport extends HttpClientSseClientTransport {
private final AtomicInteger inboundMessageCount = new AtomicInteger(0);
private Sinks.Many<ServerSentEvent<String>> events = Sinks.many().unicast().onBackpressureBuffer();
public TestHttpClientSseClientTransport(final String baseUri) {
super(HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build(),
HttpRequest.newBuilder().header("Content-Type", "application/json"), baseUri, "/sse", JSON_MAPPER,
McpAsyncHttpClientRequestCustomizer.NOOP);
}
public int getInboundMessageCount() {
return inboundMessageCount.get();
}
public void simulateEndpointEvent(String jsonMessage) {
events.tryEmitNext(ServerSentEvent.<String>builder().event("endpoint").data(jsonMessage).build());
inboundMessageCount.incrementAndGet();
}
public void simulateMessageEvent(String jsonMessage) {
events.tryEmitNext(ServerSentEvent.<String>builder().event("message").data(jsonMessage).build());
inboundMessageCount.incrementAndGet();
}
}
@BeforeAll
static void startContainer() {
container.start();
int port = container.getMappedPort(3001);
host = "http://" + container.getHost() + ":" + port;
}
@AfterAll
static void stopContainer() {
container.stop();
}
@BeforeEach
void setUp() {
transport = new TestHttpClientSseClientTransport(host);
transport.connect(Function.identity()).block();
}
@AfterEach
void afterEach() {
if (transport != null) {
assertThatCode(() -> transport.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
}
@Test
void testErrorOnBogusMessage() {
// bogus message
JSONRPCRequest bogusMessage = new JSONRPCRequest(null, null, "test-id", Map.of("key", "value"));
StepVerifier.create(transport.sendMessage(bogusMessage))
.verifyErrorMessage(
"Sending message failed with a non-OK HTTP code: 400 - Invalid message: {\"id\":\"test-id\",\"params\":{\"key\":\"value\"}}");
}
@Test
void testMessageProcessing() {
// Create a test message
JSONRPCRequest testMessage = new JSONRPCRequest(McpSchema.JSONRPC_VERSION, "test-method", "test-id",
Map.of("key", "value"));
// Simulate receiving the message
transport.simulateMessageEvent("""
{
"jsonrpc": "2.0",
"method": "test-method",
"id": "test-id",
"params": {"key": "value"}
}
""");
// Subscribe to messages and verify
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
assertThat(transport.getInboundMessageCount()).isEqualTo(1);
}
@Test
void testResponseMessageProcessing() {
// Simulate receiving a response message
transport.simulateMessageEvent("""
{
"jsonrpc": "2.0",
"id": "test-id",
"result": {"status": "success"}
}
""");
// Create and send a request message
JSONRPCRequest testMessage = new JSONRPCRequest(McpSchema.JSONRPC_VERSION, "test-method", "test-id",
Map.of("key", "value"));
// Verify message handling
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
assertThat(transport.getInboundMessageCount()).isEqualTo(1);
}
@Test
void testErrorMessageProcessing() {
// Simulate receiving an error message
transport.simulateMessageEvent("""
{
"jsonrpc": "2.0",
"id": "test-id",
"error": {
"code": -32600,
"message": "Invalid Request"
}
}
""");
// Create and send a request message
JSONRPCRequest testMessage = new JSONRPCRequest(McpSchema.JSONRPC_VERSION, "test-method", "test-id",
Map.of("key", "value"));
// Verify message handling
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
assertThat(transport.getInboundMessageCount()).isEqualTo(1);
}
@Test
void testNotificationMessageProcessing() {
// Simulate receiving a notification message (no id)
transport.simulateMessageEvent("""
{
"jsonrpc": "2.0",
"method": "update",
"params": {"status": "processing"}
}
""");
// Verify the notification was processed
assertThat(transport.getInboundMessageCount()).isEqualTo(1);
}
@Test
void testGracefulShutdown() {
// Test graceful shutdown
StepVerifier.create(transport.closeGracefully()).verifyComplete();
// Create a test message
JSONRPCRequest testMessage = new JSONRPCRequest(McpSchema.JSONRPC_VERSION, "test-method", "test-id",
Map.of("key", "value"));
// Verify message is not processed after shutdown
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
// Message count should remain 0 after shutdown
assertThat(transport.getInboundMessageCount()).isZero();
}
@Test
void testRetryBehavior() {
// Create a client that simulates connection failures
HttpClientSseClientTransport failingTransport = HttpClientSseClientTransport.builder("http://non-existent-host")
.build();
// Verify that the transport attempts to reconnect
StepVerifier.create(Mono.delay(Duration.ofSeconds(2))).expectNextCount(1).verifyComplete();
// Clean up
failingTransport.closeGracefully().block();
}
@Test
void testMultipleMessageProcessing() {
// Simulate receiving multiple messages in sequence
transport.simulateMessageEvent("""
{
"jsonrpc": "2.0",
"method": "method1",
"id": "id1",
"params": {"key": "value1"}
}
""");
transport.simulateMessageEvent("""
{
"jsonrpc": "2.0",
"method": "method2",
"id": "id2",
"params": {"key": "value2"}
}
""");
// Create and send corresponding messages
JSONRPCRequest message1 = new JSONRPCRequest(McpSchema.JSONRPC_VERSION, "method1", "id1",
Map.of("key", "value1"));
JSONRPCRequest message2 = new JSONRPCRequest(McpSchema.JSONRPC_VERSION, "method2", "id2",
Map.of("key", "value2"));
// Verify both messages are processed
StepVerifier.create(transport.sendMessage(message1).then(transport.sendMessage(message2))).verifyComplete();
// Verify message count
assertThat(transport.getInboundMessageCount()).isEqualTo(2);
}
@Test
void testMessageOrderPreservation() {
// Simulate receiving messages in a specific order
transport.simulateMessageEvent("""
{
"jsonrpc": "2.0",
"method": "first",
"id": "1",
"params": {"sequence": 1}
}
""");
transport.simulateMessageEvent("""
{
"jsonrpc": "2.0",
"method": "second",
"id": "2",
"params": {"sequence": 2}
}
""");
transport.simulateMessageEvent("""
{
"jsonrpc": "2.0",
"method": "third",
"id": "3",
"params": {"sequence": 3}
}
""");
// Verify message count and order
assertThat(transport.getInboundMessageCount()).isEqualTo(3);
}
@Test
void testCustomizeClient() {
// Create an atomic boolean to verify the customizer was called
AtomicBoolean customizerCalled = new AtomicBoolean(false);
// Create a transport with the customizer
HttpClientSseClientTransport customizedTransport = HttpClientSseClientTransport.builder(host)
.customizeClient(builder -> {
builder.version(HttpClient.Version.HTTP_2);
customizerCalled.set(true);
})
.build();
// Verify the customizer was called
assertThat(customizerCalled.get()).isTrue();
// Clean up
customizedTransport.closeGracefully().block();
}
@Test
void testCustomizeRequest() {
// Create an atomic boolean to verify the customizer was called
AtomicBoolean customizerCalled = new AtomicBoolean(false);
// Create a reference to store the custom header value
AtomicReference<String> headerName = new AtomicReference<>();
AtomicReference<String> headerValue = new AtomicReference<>();
// Create a transport with the customizer
HttpClientSseClientTransport customizedTransport = HttpClientSseClientTransport.builder(host)
// Create a request customizer that adds a custom header
.customizeRequest(builder -> {
builder.header("X-Custom-Header", "test-value");
customizerCalled.set(true);
// Create a new request to verify the header was set
HttpRequest request = builder.uri(URI.create("http://example.com")).build();
headerName.set("X-Custom-Header");
headerValue.set(request.headers().firstValue("X-Custom-Header").orElse(null));
})
.build();
// Verify the customizer was called
assertThat(customizerCalled.get()).isTrue();
// Verify the header was set correctly
assertThat(headerName.get()).isEqualTo("X-Custom-Header");
assertThat(headerValue.get()).isEqualTo("test-value");
// Clean up
customizedTransport.closeGracefully().block();
}
@Test
void testChainedCustomizations() {
// Create atomic booleans to verify both customizers were called
AtomicBoolean clientCustomizerCalled = new AtomicBoolean(false);
AtomicBoolean requestCustomizerCalled = new AtomicBoolean(false);
// Create a transport with both customizers chained
HttpClientSseClientTransport customizedTransport = HttpClientSseClientTransport.builder(host)
.customizeClient(builder -> {
builder.connectTimeout(Duration.ofSeconds(30));
clientCustomizerCalled.set(true);
})
.customizeRequest(builder -> {
builder.header("X-Api-Key", "test-api-key");
requestCustomizerCalled.set(true);
})
.build();
// Verify both customizers were called
assertThat(clientCustomizerCalled.get()).isTrue();
assertThat(requestCustomizerCalled.get()).isTrue();
// Clean up
customizedTransport.closeGracefully().block();
}
@Test
void testRequestCustomizer() {
var mockCustomizer = mock(McpSyncHttpClientRequestCustomizer.class);
// Create a transport with the customizer
var customizedTransport = HttpClientSseClientTransport.builder(host)
.httpRequestCustomizer(mockCustomizer)
.build();
// Connect
StepVerifier
.create(customizedTransport.connect(Function.identity())
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY, context)))
.verifyComplete();
// Verify the customizer was called
verify(mockCustomizer).customize(any(), eq("GET"),
eq(UriComponentsBuilder.fromUriString(host).path("/sse").build().toUri()), isNull(), eq(context));
clearInvocations(mockCustomizer);
// Send test message
var testMessage = new JSONRPCRequest(McpSchema.JSONRPC_VERSION, "test-method", "test-id",
Map.of("key", "value"));
// Subscribe to messages and verify
StepVerifier
.create(customizedTransport.sendMessage(testMessage)
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY, context)))
.verifyComplete();
// Verify the customizer was called
var uriArgumentCaptor = ArgumentCaptor.forClass(URI.class);
verify(mockCustomizer).customize(any(), eq("POST"), uriArgumentCaptor.capture(), eq(
"{\"jsonrpc\":\"2.0\",\"method\":\"test-method\",\"id\":\"test-id\",\"params\":{\"key\":\"value\"}}"),
eq(context));
assertThat(uriArgumentCaptor.getValue().toString()).startsWith(host + "/message?sessionId=");
// Clean up
customizedTransport.closeGracefully().block();
}
@Test
void testAsyncRequestCustomizer() {
var mockCustomizer = mock(McpAsyncHttpClientRequestCustomizer.class);
when(mockCustomizer.customize(any(), any(), any(), any(), any()))
.thenAnswer(invocation -> Mono.just(invocation.getArguments()[0]));
// Create a transport with the customizer
var customizedTransport = HttpClientSseClientTransport.builder(host)
.asyncHttpRequestCustomizer(mockCustomizer)
.build();
// Connect
StepVerifier
.create(customizedTransport.connect(Function.identity())
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY, context)))
.verifyComplete();
// Verify the customizer was called
verify(mockCustomizer).customize(any(), eq("GET"),
eq(UriComponentsBuilder.fromUriString(host).path("/sse").build().toUri()), isNull(), eq(context));
clearInvocations(mockCustomizer);
// Send test message
var testMessage = new JSONRPCRequest(McpSchema.JSONRPC_VERSION, "test-method", "test-id",
Map.of("key", "value"));
// Subscribe to messages and verify
StepVerifier
.create(customizedTransport.sendMessage(testMessage)
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY, context)))
.verifyComplete();
// Verify the customizer was called
var uriArgumentCaptor = ArgumentCaptor.forClass(URI.class);
verify(mockCustomizer).customize(any(), eq("POST"), uriArgumentCaptor.capture(), eq(
"{\"jsonrpc\":\"2.0\",\"method\":\"test-method\",\"id\":\"test-id\",\"params\":{\"key\":\"value\"}}"),
eq(context));
assertThat(uriArgumentCaptor.getValue().toString()).startsWith(host + "/message?sessionId=");
// Clean up
customizedTransport.closeGracefully().block();
}
}

94
mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportEmptyJsonResponseTest.java Normal file
View file

@ -0,0 +1,94 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client.transport;
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;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
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 com.sun.net.httpserver.HttpServer;
import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
import io.modelcontextprotocol.server.transport.TomcatTestUtil;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.ProtocolVersions;
import reactor.test.StepVerifier;
/**
* Handles emplty application/json response with 200 OK status code.
*
* @author codezkk
*/
public class HttpClientStreamableHttpTransportEmptyJsonResponseTest {
static int PORT = TomcatTestUtil.findAvailablePort();
static String host = "http://localhost:" + PORT;
static HttpServer server;
@BeforeAll
static void startContainer() throws IOException {
server = HttpServer.create(new InetSocketAddress(PORT), 0);
// Empty, 200 OK response for the /mcp endpoint
server.createContext("/mcp", exchange -> {
exchange.getResponseHeaders().set("Content-Type", "application/json");
exchange.sendResponseHeaders(200, 0);
exchange.close();
});
server.setExecutor(null);
server.start();
}
@AfterAll
static void stopContainer() {
server.stop(1);
}
/**
* Regardless of the response (even if the response is null and the content-type is
* present), notify should handle it correctly.
*/
@Test
@Timeout(3)
void testNotificationInitialized() throws URISyntaxException {
var uri = new URI(host + "/mcp");
var mockRequestCustomizer = mock(McpSyncHttpClientRequestCustomizer.class);
var transport = HttpClientStreamableHttpTransport.builder(host)
.httpRequestCustomizer(mockRequestCustomizer)
.build();
var initializeRequest = new McpSchema.InitializeRequest(ProtocolVersions.MCP_2025_03_26,
McpSchema.ClientCapabilities.builder().roots(true).build(),
new McpSchema.Implementation("MCP Client", "0.3.1"));
var testMessage = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, McpSchema.METHOD_INITIALIZE,
"test-id", initializeRequest);
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
// Verify the customizer was called
verify(mockRequestCustomizer, atLeastOnce()).customize(any(), eq("POST"), eq(uri), eq(
"{\"jsonrpc\":\"2.0\",\"method\":\"initialize\",\"id\":\"test-id\",\"params\":{\"protocolVersion\":\"2025-03-26\",\"capabilities\":{\"roots\":{\"listChanged\":true}},\"clientInfo\":{\"name\":\"MCP Client\",\"version\":\"0.3.1\"}}}"),
any());
}
}

345
mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportErrorHandlingTest.java Normal file
View file

@ -0,0 +1,345 @@
/*
* Copyright 2025-2025 the original author or authors.
*/
package io.modelcontextprotocol.client.transport;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import com.sun.net.httpserver.HttpServer;
import io.modelcontextprotocol.server.transport.TomcatTestUtil;
import io.modelcontextprotocol.spec.HttpHeaders;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpTransportException;
import io.modelcontextprotocol.spec.McpTransportSessionNotFoundException;
import io.modelcontextprotocol.spec.ProtocolVersions;
import reactor.test.StepVerifier;
/**
* Tests for error handling changes in HttpClientStreamableHttpTransport. Specifically
* tests the distinction between session-related errors and general transport errors for
* 404 and 400 status codes.
*
* @author Christian Tzolov
*/
@Timeout(15)
public class HttpClientStreamableHttpTransportErrorHandlingTest {
private static final int PORT = TomcatTestUtil.findAvailablePort();
private static final String HOST = "http://localhost:" + PORT;
private HttpServer server;
private AtomicReference<Integer> serverResponseStatus = new AtomicReference<>(200);
private AtomicReference<String> currentServerSessionId = new AtomicReference<>(null);
private AtomicReference<String> lastReceivedSessionId = new AtomicReference<>(null);
private McpClientTransport transport;
@BeforeEach
void startServer() throws IOException {
server = HttpServer.create(new InetSocketAddress(PORT), 0);
// Configure the /mcp endpoint with dynamic response
server.createContext("/mcp", httpExchange -> {
if ("DELETE".equals(httpExchange.getRequestMethod())) {
httpExchange.sendResponseHeaders(200, 0);
}
else if ("POST".equals(httpExchange.getRequestMethod())) {
// Capture session ID from request if present
String requestSessionId = httpExchange.getRequestHeaders().getFirst(HttpHeaders.MCP_SESSION_ID);
lastReceivedSessionId.set(requestSessionId);
int status = serverResponseStatus.get();
// Set response headers
httpExchange.getResponseHeaders().set("Content-Type", "application/json");
// Add session ID to response if configured
String responseSessionId = currentServerSessionId.get();
if (responseSessionId != null) {
httpExchange.getResponseHeaders().set(HttpHeaders.MCP_SESSION_ID, responseSessionId);
}
// Send response based on configured status
if (status == 200) {
String response = "{\"jsonrpc\":\"2.0\",\"result\":{},\"id\":\"test-id\"}";
httpExchange.sendResponseHeaders(200, response.length());
httpExchange.getResponseBody().write(response.getBytes());
}
else {
httpExchange.sendResponseHeaders(status, 0);
}
}
httpExchange.close();
});
server.setExecutor(null);
server.start();
transport = HttpClientStreamableHttpTransport.builder(HOST).build();
}
@AfterEach
void stopServer() {
if (server != null) {
server.stop(0);
}
}
/**
* Test that 404 response WITHOUT session ID throws McpTransportException (not
* SessionNotFoundException)
*/
@Test
void test404WithoutSessionId() {
serverResponseStatus.set(404);
currentServerSessionId.set(null); // No session ID in response
var testMessage = createTestRequestMessage();
StepVerifier.create(transport.sendMessage(testMessage))
.expectErrorMatches(throwable -> throwable instanceof McpTransportException
&& throwable.getMessage().contains("Not Found") && throwable.getMessage().contains("404")
&& !(throwable instanceof McpTransportSessionNotFoundException))
.verify();
StepVerifier.create(transport.closeGracefully()).verifyComplete();
}
/**
* Test that 404 response WITH session ID throws McpTransportSessionNotFoundException
*/
@Test
void test404WithSessionId() {
// First establish a session
serverResponseStatus.set(200);
currentServerSessionId.set("test-session-123");
// Set up exception handler to verify session invalidation
@SuppressWarnings("unchecked")
Consumer<Throwable> exceptionHandler = mock(Consumer.class);
transport.setExceptionHandler(exceptionHandler);
// Connect with handler
StepVerifier.create(transport.connect(msg -> msg)).verifyComplete();
// Send initial message to establish session
var testMessage = createTestRequestMessage();
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
// The session should now be established, next request will include session ID
// Now return 404 for next request
serverResponseStatus.set(404);
// Send another message - should get SessionNotFoundException
StepVerifier.create(transport.sendMessage(testMessage))
.expectError(McpTransportSessionNotFoundException.class)
.verify();
// Verify exception handler was called with SessionNotFoundException
verify(exceptionHandler).accept(any(McpTransportSessionNotFoundException.class));
StepVerifier.create(transport.closeGracefully()).verifyComplete();
}
/**
* Test that 400 response WITHOUT session ID throws McpTransportException (not
* SessionNotFoundException)
*/
@Test
void test400WithoutSessionId() {
serverResponseStatus.set(400);
currentServerSessionId.set(null); // No session ID
var testMessage = createTestRequestMessage();
StepVerifier.create(transport.sendMessage(testMessage))
.expectErrorMatches(throwable -> throwable instanceof McpTransportException
&& throwable.getMessage().contains("Bad Request") && throwable.getMessage().contains("400")
&& !(throwable instanceof McpTransportSessionNotFoundException))
.verify();
StepVerifier.create(transport.closeGracefully()).verifyComplete();
}
/**
* Test that 400 response WITH session ID throws McpTransportSessionNotFoundException
* This handles the case mentioned in the code comment about some implementations
* returning 400 for unknown session IDs.
*/
@Test
void test400WithSessionId() {
// First establish a session
serverResponseStatus.set(200);
currentServerSessionId.set("test-session-456");
// Set up exception handler
@SuppressWarnings("unchecked")
Consumer<Throwable> exceptionHandler = mock(Consumer.class);
transport.setExceptionHandler(exceptionHandler);
// Connect with handler
StepVerifier.create(transport.connect(msg -> msg)).verifyComplete();
// Send initial message to establish session
var testMessage = createTestRequestMessage();
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
// The session should now be established, next request will include session ID
// Now return 400 for next request (simulating unknown session ID)
serverResponseStatus.set(400);
// Send another message - should get SessionNotFoundException
StepVerifier.create(transport.sendMessage(testMessage))
.expectError(McpTransportSessionNotFoundException.class)
.verify();
// Verify exception handler was called
verify(exceptionHandler).accept(any(McpTransportSessionNotFoundException.class));
StepVerifier.create(transport.closeGracefully()).verifyComplete();
}
/**
* Test session recovery after SessionNotFoundException Verifies that a new session
* can be established after the old one is invalidated
*/
@Test
void testSessionRecoveryAfter404() {
// First establish a session
serverResponseStatus.set(200);
currentServerSessionId.set("session-1");
// Send initial message to establish session
var testMessage = createTestRequestMessage();
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
assertThat(lastReceivedSessionId.get()).isNull();
// The session should now be established
// Simulate session loss - return 404
serverResponseStatus.set(404);
// This should fail with SessionNotFoundException
StepVerifier.create(transport.sendMessage(testMessage))
.expectError(McpTransportSessionNotFoundException.class)
.verify();
// Now server is back with new session
serverResponseStatus.set(200);
currentServerSessionId.set("session-2");
lastReceivedSessionId.set(null); // Reset to verify new session
// Should be able to establish new session
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
// Verify no session ID was sent (since old session was invalidated)
assertThat(lastReceivedSessionId.get()).isNull();
// Next request should use the new session ID
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
// Session ID should now be sent with requests
assertThat(lastReceivedSessionId.get()).isEqualTo("session-2");
StepVerifier.create(transport.closeGracefully()).verifyComplete();
}
/**
* Test that reconnect (GET request) also properly handles 404/400 errors
*/
@Test
void testReconnectErrorHandling() {
// Set up SSE endpoint for GET requests
server.createContext("/mcp-sse", exchange -> {
String method = exchange.getRequestMethod();
String requestSessionId = exchange.getRequestHeaders().getFirst(HttpHeaders.MCP_SESSION_ID);
if ("GET".equals(method)) {
int status = serverResponseStatus.get();
if (status == 404 && requestSessionId != null) {
// 404 with session ID - should trigger SessionNotFoundException
exchange.sendResponseHeaders(404, 0);
}
else if (status == 404) {
// 404 without session ID - should trigger McpTransportException
exchange.sendResponseHeaders(404, 0);
}
else {
// Normal SSE response
exchange.getResponseHeaders().set("Content-Type", "text/event-stream");
exchange.sendResponseHeaders(200, 0);
// Send a test SSE event
String sseData = "event: message\ndata: {\"jsonrpc\":\"2.0\",\"method\":\"test\",\"params\":{}}\n\n";
exchange.getResponseBody().write(sseData.getBytes());
}
}
else {
// POST request handling
exchange.getResponseHeaders().set("Content-Type", "application/json");
String responseSessionId = currentServerSessionId.get();
if (responseSessionId != null) {
exchange.getResponseHeaders().set(HttpHeaders.MCP_SESSION_ID, responseSessionId);
}
String response = "{\"jsonrpc\":\"2.0\",\"result\":{},\"id\":\"test-id\"}";
exchange.sendResponseHeaders(200, response.length());
exchange.getResponseBody().write(response.getBytes());
}
exchange.close();
});
// Test with session ID - should get SessionNotFoundException
serverResponseStatus.set(200);
currentServerSessionId.set("sse-session-1");
var transport = HttpClientStreamableHttpTransport.builder(HOST)
.endpoint("/mcp-sse")
.openConnectionOnStartup(true) // This will trigger GET request on connect
.build();
// First connect successfully
StepVerifier.create(transport.connect(msg -> msg)).verifyComplete();
// Send message to establish session
var testMessage = createTestRequestMessage();
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
// Now simulate server returning 404 on reconnect
serverResponseStatus.set(404);
// This should trigger reconnect which will fail
// The error should be handled internally and passed to exception handler
StepVerifier.create(transport.closeGracefully()).verifyComplete();
}
private McpSchema.JSONRPCRequest createTestRequestMessage() {
var initializeRequest = new McpSchema.InitializeRequest(ProtocolVersions.MCP_2025_03_26,
McpSchema.ClientCapabilities.builder().roots(true).build(),
new McpSchema.Implementation("Test Client", "1.0.0"));
return new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, McpSchema.METHOD_INITIALIZE, "test-id",
initializeRequest);
}
}

163
mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransportTest.java Normal file
View file

@ -0,0 +1,163 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.client.transport;
import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpClientRequestCustomizer;
import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
import io.modelcontextprotocol.common.McpTransportContext;
import io.modelcontextprotocol.spec.McpSchema;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Map;
import java.util.function.Consumer;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
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;
import static org.mockito.Mockito.when;
/**
* Tests for the {@link HttpClientStreamableHttpTransport} class.
*
* @author Daniel Garnier-Moiroux
*/
class HttpClientStreamableHttpTransportTest {
static String host = "http://localhost:3001";
private McpTransportContext context = McpTransportContext
.create(Map.of("test-transport-context-key", "some-value"));
@SuppressWarnings("resource")
static GenericContainer<?> container = new GenericContainer<>("docker.io/node:lts-alpine3.23")
.withCommand("npx -y @modelcontextprotocol/server-everything@2025.12.18 streamableHttp")
.withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
.withExposedPorts(3001)
.waitingFor(Wait.forHttp("/").forStatusCode(404));
@BeforeAll
static void startContainer() {
container.start();
int port = container.getMappedPort(3001);
host = "http://" + container.getHost() + ":" + port;
}
@AfterAll
static void stopContainer() {
container.stop();
}
void withTransport(HttpClientStreamableHttpTransport transport, Consumer<HttpClientStreamableHttpTransport> c) {
try {
c.accept(transport);
}
finally {
StepVerifier.create(transport.closeGracefully()).verifyComplete();
}
}
@Test
void testRequestCustomizer() throws URISyntaxException {
var uri = new URI(host + "/mcp");
var mockRequestCustomizer = mock(McpSyncHttpClientRequestCustomizer.class);
var transport = HttpClientStreamableHttpTransport.builder(host)
.httpRequestCustomizer(mockRequestCustomizer)
.build();
withTransport(transport, (t) -> {
// Send test message
var initializeRequest = new McpSchema.InitializeRequest(McpSchema.LATEST_PROTOCOL_VERSION,
McpSchema.ClientCapabilities.builder().roots(true).build(),
new McpSchema.Implementation("MCP Client", "0.3.1"));
var testMessage = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, McpSchema.METHOD_INITIALIZE,
"test-id", initializeRequest);
StepVerifier
.create(t.sendMessage(testMessage).contextWrite(ctx -> ctx.put(McpTransportContext.KEY, context)))
.verifyComplete();
// Verify the customizer was called
verify(mockRequestCustomizer, atLeastOnce()).customize(any(), eq("POST"), eq(uri), eq(
"{\"jsonrpc\":\"2.0\",\"method\":\"initialize\",\"id\":\"test-id\",\"params\":{\"protocolVersion\":\"2025-11-25\",\"capabilities\":{\"roots\":{\"listChanged\":true}},\"clientInfo\":{\"name\":\"MCP Client\",\"version\":\"0.3.1\"}}}"),
eq(context));
});
}
@Test
void testAsyncRequestCustomizer() throws URISyntaxException {
var uri = new URI(host + "/mcp");
var mockRequestCustomizer = mock(McpAsyncHttpClientRequestCustomizer.class);
when(mockRequestCustomizer.customize(any(), any(), any(), any(), any()))
.thenAnswer(invocation -> Mono.just(invocation.getArguments()[0]));
var transport = HttpClientStreamableHttpTransport.builder(host)
.asyncHttpRequestCustomizer(mockRequestCustomizer)
.build();
withTransport(transport, (t) -> {
// Send test message
var initializeRequest = new McpSchema.InitializeRequest(McpSchema.LATEST_PROTOCOL_VERSION,
McpSchema.ClientCapabilities.builder().roots(true).build(),
new McpSchema.Implementation("MCP Client", "0.3.1"));
var testMessage = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, McpSchema.METHOD_INITIALIZE,
"test-id", initializeRequest);
StepVerifier
.create(t.sendMessage(testMessage).contextWrite(ctx -> ctx.put(McpTransportContext.KEY, context)))
.verifyComplete();
// Verify the customizer was called
verify(mockRequestCustomizer, atLeastOnce()).customize(any(), eq("POST"), eq(uri), eq(
"{\"jsonrpc\":\"2.0\",\"method\":\"initialize\",\"id\":\"test-id\",\"params\":{\"protocolVersion\":\"2025-11-25\",\"capabilities\":{\"roots\":{\"listChanged\":true}},\"clientInfo\":{\"name\":\"MCP Client\",\"version\":\"0.3.1\"}}}"),
eq(context));
});
}
@Test
void testCloseUninitialized() {
var transport = HttpClientStreamableHttpTransport.builder(host).build();
StepVerifier.create(transport.closeGracefully()).verifyComplete();
var initializeRequest = new McpSchema.InitializeRequest(McpSchema.LATEST_PROTOCOL_VERSION,
McpSchema.ClientCapabilities.builder().roots(true).build(),
new McpSchema.Implementation("MCP Client", "0.3.1"));
var testMessage = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, McpSchema.METHOD_INITIALIZE,
"test-id", initializeRequest);
StepVerifier.create(transport.sendMessage(testMessage))
.expectErrorMessage("MCP session has been closed")
.verify();
}
@Test
void testCloseInitialized() {
var transport = HttpClientStreamableHttpTransport.builder(host).build();
var initializeRequest = new McpSchema.InitializeRequest(McpSchema.LATEST_PROTOCOL_VERSION,
McpSchema.ClientCapabilities.builder().roots(true).build(),
new McpSchema.Implementation("MCP Client", "0.3.1"));
var testMessage = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, McpSchema.METHOD_INITIALIZE,
"test-id", initializeRequest);
StepVerifier.create(transport.sendMessage(testMessage)).verifyComplete();
StepVerifier.create(transport.closeGracefully()).verifyComplete();
StepVerifier.create(transport.sendMessage(testMessage))
.expectErrorMatches(err -> err.getMessage().matches("MCP session with ID [a-zA-Z0-9-]* has been closed"))
.verify();
}
}

280
mcp-test/src/test/java/io/modelcontextprotocol/common/AsyncServerMcpTransportContextIntegrationTests.java Normal file
View file

@ -0,0 +1,280 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.common;
import java.util.Map;
import java.util.function.BiFunction;
import io.modelcontextprotocol.client.McpAsyncClient;
import io.modelcontextprotocol.client.McpClient;
import io.modelcontextprotocol.client.transport.HttpClientSseClientTransport;
import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport;
import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpClientRequestCustomizer;
import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
import io.modelcontextprotocol.server.McpAsyncServerExchange;
import io.modelcontextprotocol.server.McpServer;
import io.modelcontextprotocol.server.McpServerFeatures;
import io.modelcontextprotocol.server.McpStatelessServerFeatures;
import io.modelcontextprotocol.server.McpTransportContextExtractor;
import io.modelcontextprotocol.server.transport.HttpServletSseServerTransportProvider;
import io.modelcontextprotocol.server.transport.HttpServletStatelessServerTransport;
import io.modelcontextprotocol.server.transport.HttpServletStreamableServerTransportProvider;
import io.modelcontextprotocol.server.transport.TomcatTestUtil;
import io.modelcontextprotocol.spec.McpSchema;
import jakarta.servlet.Servlet;
import jakarta.servlet.http.HttpServletRequest;
import org.apache.catalina.LifecycleException;
import org.apache.catalina.LifecycleState;
import org.apache.catalina.startup.Tomcat;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Integration tests for {@link McpTransportContext} propagation between MCP clients and
* async servers.
*
* <p>
* This test class validates the end-to-end flow of transport context propagation in MCP
* communication, demonstrating how contextual information can be passed from client to
* server through HTTP headers and accessed within server-side handlers.
*
* <h2>Test Scenarios</h2>
* <p>
* The tests cover multiple transport configurations with async servers:
* <ul>
* <li>Stateless server with async streamable HTTP clients</li>
* <li>Streamable server with async streamable HTTP clients</li>
* <li>SSE (Server-Sent Events) server with async SSE clients</li>
* </ul>
*
* <h2>Context Propagation Flow</h2>
* <ol>
* <li>Client-side: Context data is stored in the Reactor Context and injected into HTTP
* headers via {@link McpSyncHttpClientRequestCustomizer}</li>
* <li>Transport: The context travels as HTTP headers (specifically "x-test" header in
* these tests)</li>
* <li>Server-side: A {@link McpTransportContextExtractor} extracts the header value and
* makes it available to request handlers through {@link McpTransportContext}</li>
* <li>Verification: The server echoes back the received context value as the tool call
* result</li>
* </ol>
*
* <p>
* All tests use an embedded Tomcat server running on a dynamically allocated port to
* ensure isolation and prevent port conflicts during parallel test execution.
*
* @author Daniel Garnier-Moiroux
* @author Christian Tzolov
*/
@Timeout(15)
public class AsyncServerMcpTransportContextIntegrationTests {
private static final int PORT = TomcatTestUtil.findAvailablePort();
private Tomcat tomcat;
private static final String HEADER_NAME = "x-test";
private final McpAsyncHttpClientRequestCustomizer asyncClientRequestCustomizer = (builder, method, endpoint, body,
context) -> {
var headerValue = context.get("client-side-header-value");
if (headerValue != null) {
builder.header(HEADER_NAME, headerValue.toString());
}
return Mono.just(builder);
};
private final McpTransportContextExtractor<HttpServletRequest> serverContextExtractor = (HttpServletRequest r) -> {
var headerValue = r.getHeader(HEADER_NAME);
return headerValue != null ? McpTransportContext.create(Map.of("server-side-header-value", headerValue))
: McpTransportContext.EMPTY;
};
private final HttpServletStatelessServerTransport statelessServerTransport = HttpServletStatelessServerTransport
.builder()
.contextExtractor(serverContextExtractor)
.build();
private final HttpServletStreamableServerTransportProvider streamableServerTransport = HttpServletStreamableServerTransportProvider
.builder()
.contextExtractor(serverContextExtractor)
.build();
private final HttpServletSseServerTransportProvider sseServerTransport = HttpServletSseServerTransportProvider
.builder()
.contextExtractor(serverContextExtractor)
.messageEndpoint("/message")
.build();
private final McpAsyncClient asyncStreamableClient = McpClient
.async(HttpClientStreamableHttpTransport.builder("http://localhost:" + PORT)
.asyncHttpRequestCustomizer(asyncClientRequestCustomizer)
.build())
.build();
private final McpAsyncClient asyncSseClient = McpClient
.async(HttpClientSseClientTransport.builder("http://localhost:" + PORT)
.asyncHttpRequestCustomizer(asyncClientRequestCustomizer)
.build())
.build();
private final McpSchema.Tool tool = McpSchema.Tool.builder()
.name("test-tool")
.description("return the value of the x-test header from call tool request")
.build();
private final BiFunction<McpTransportContext, McpSchema.CallToolRequest, Mono<McpSchema.CallToolResult>> asyncStatelessHandler = (
transportContext, request) -> {
return Mono
.just(new McpSchema.CallToolResult(transportContext.get("server-side-header-value").toString(), null));
};
private final BiFunction<McpAsyncServerExchange, McpSchema.CallToolRequest, Mono<McpSchema.CallToolResult>> asyncStatefulHandler = (
exchange, request) -> {
return asyncStatelessHandler.apply(exchange.transportContext(), request);
};
@AfterEach
public void after() {
if (statelessServerTransport != null) {
statelessServerTransport.closeGracefully().block();
}
if (streamableServerTransport != null) {
streamableServerTransport.closeGracefully().block();
}
if (sseServerTransport != null) {
sseServerTransport.closeGracefully().block();
}
if (asyncStreamableClient != null) {
asyncStreamableClient.closeGracefully().block();
}
if (asyncSseClient != null) {
asyncSseClient.closeGracefully().block();
}
stopTomcat();
}
@Test
void asyncClinetStatelessServer() {
startTomcat(statelessServerTransport);
var mcpServer = McpServer.async(statelessServerTransport)
.capabilities(McpSchema.ServerCapabilities.builder().tools(true).build())
.tools(new McpStatelessServerFeatures.AsyncToolSpecification(tool, asyncStatelessHandler))
.build();
StepVerifier.create(asyncStreamableClient.initialize()).assertNext(initResult -> {
assertThat(initResult).isNotNull();
}).verifyComplete();
// Test tool call with context
StepVerifier
.create(asyncStreamableClient.callTool(new McpSchema.CallToolRequest("test-tool", Map.of()))
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY,
McpTransportContext.create(Map.of("client-side-header-value", "some important value")))))
.assertNext(response -> {
assertThat(response).isNotNull();
assertThat(response.content()).hasSize(1)
.first()
.extracting(McpSchema.TextContent.class::cast)
.extracting(McpSchema.TextContent::text)
.isEqualTo("some important value");
})
.verifyComplete();
mcpServer.close();
}
@Test
void asyncClientStreamableServer() {
startTomcat(streamableServerTransport);
var mcpServer = McpServer.async(streamableServerTransport)
.capabilities(McpSchema.ServerCapabilities.builder().tools(true).build())
.tools(new McpServerFeatures.AsyncToolSpecification(tool, null, asyncStatefulHandler))
.build();
StepVerifier.create(asyncStreamableClient.initialize()).assertNext(initResult -> {
assertThat(initResult).isNotNull();
}).verifyComplete();
// Test tool call with context
StepVerifier
.create(asyncStreamableClient.callTool(new McpSchema.CallToolRequest("test-tool", Map.of()))
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY,
McpTransportContext.create(Map.of("client-side-header-value", "some important value")))))
.assertNext(response -> {
assertThat(response).isNotNull();
assertThat(response.content()).hasSize(1)
.first()
.extracting(McpSchema.TextContent.class::cast)
.extracting(McpSchema.TextContent::text)
.isEqualTo("some important value");
})
.verifyComplete();
mcpServer.close();
}
@Test
void asyncClientSseServer() {
startTomcat(sseServerTransport);
var mcpServer = McpServer.async(sseServerTransport)
.capabilities(McpSchema.ServerCapabilities.builder().tools(true).build())
.tools(new McpServerFeatures.AsyncToolSpecification(tool, null, asyncStatefulHandler))
.build();
StepVerifier.create(asyncSseClient.initialize()).assertNext(initResult -> {
assertThat(initResult).isNotNull();
}).verifyComplete();
// Test tool call with context
StepVerifier
.create(asyncSseClient.callTool(new McpSchema.CallToolRequest("test-tool", Map.of()))
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY,
McpTransportContext.create(Map.of("client-side-header-value", "some important value")))))
.assertNext(response -> {
assertThat(response).isNotNull();
assertThat(response.content()).hasSize(1)
.first()
.extracting(McpSchema.TextContent.class::cast)
.extracting(McpSchema.TextContent::text)
.isEqualTo("some important value");
})
.verifyComplete();
mcpServer.close();
}
private void startTomcat(Servlet transport) {
tomcat = TomcatTestUtil.createTomcatServer("", PORT, transport);
try {
tomcat.start();
assertThat(tomcat.getServer().getState()).isEqualTo(LifecycleState.STARTED);
}
catch (Exception e) {
throw new RuntimeException("Failed to start Tomcat", e);
}
}
private void stopTomcat() {
if (tomcat != null) {
try {
tomcat.stop();
tomcat.destroy();
}
catch (LifecycleException e) {
throw new RuntimeException("Failed to stop Tomcat", e);
}
}
}
}

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