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

Change the top domain from org to io

Browse source
This commit is contained in:
Christian Tzolov 2025-01-28 17:23:07 +01:00 • committed by Christian Tzolov
commit 181e1f25ab
77 changed files with 470 additions and 417 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/continuous-integration.yml vendored
View file

@ -35,7 +35,7 @@ jobs:
MAVEN_PASSWORD: ${{ secrets.OSSRH_TOKEN }}
MAVEN_GPG_PASSPHRASE: ${{ secrets.SIGNING_PASSPHRASE }}
run: |
mvn -Pjavadoc -Prelease --batch-mode --update-snapshots deploy
mvn -Pjavadoc -Prelease -DskipTests --batch-mode --update-snapshots deploy
- name: Generate Java docs
run: mvn -Pjavadoc -B javadoc:aggregate

6
README.md
View file

@ -47,19 +47,19 @@ Add the following dependencies to your Maven project:
```xml
<!-- Core MCP -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp</artifactId>
</dependency>
<!-- Optional: WebFlux SSE transport -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-webflux-sse-transport</artifactId>
</dependency>
<!-- Optional: WebMVC SSE transport -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-webmvc-sse-transport</artifactId>
</dependency>
```

10
mcp-bom/pom.xml
View file

@ -5,7 +5,7 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-parent</artifactId>
<version>0.7.0-SNAPSHOT</version>
</parent>
@ -28,28 +28,28 @@
<dependencies>
<!-- Core MCP -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp</artifactId>
<version>${project.version}</version>
</dependency>
<!-- MCP Test -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-test</artifactId>
<version>${project.version}</version>
</dependency>
<!-- MCP Transport - WebFlux SSE -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-webflux-sse-transport</artifactId>
<version>${project.version}</version>
</dependency>
<!-- MCP Transport - WebMVC SSE -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-webmvc-sse-transport</artifactId>
<version>${project.version}</version>
</dependency>

2
mcp-docs/pom.xml
View file

@ -18,7 +18,7 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-parent</artifactId>
<version>0.7.0-SNAPSHOT</version>
</parent>

12
mcp-docs/src/main/antora/modules/ROOT/pages/dependency-management.adoc
View file

@ -20,7 +20,7 @@ Maven::
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-bom</artifactId>
<version>0.6.0-SNAPSHOT</version>
<type>pom</type>
@ -35,7 +35,7 @@ Gradle::
[source,groovy,indent=0,subs="verbatim,quotes"]
----
dependencies {
implementation platform("org.modelcontextprotocol.sdk:mcp-bom:0.6.0-SNAPSHOT")
implementation platform("io.modelcontextprotocol.sdk:mcp-bom:0.6.0-SNAPSHOT")
//...
}
----
@ -53,16 +53,16 @@ The following dependencies are available and managed by the BOM:
=== Core Dependencies
* `org.modelcontextprotocol.sdk:mcp` - Core MCP library providing the base functionality and APIs for Model Context Protocol implementation.
* `io.modelcontextprotocol.sdk:mcp` - Core MCP library providing the base functionality and APIs for Model Context Protocol implementation.
=== Transport Dependencies
* `org.modelcontextprotocol.sdk:mcp-webflux-sse-transport` - WebFlux-based Server-Sent Events (SSE) transport implementation for reactive applications.
* `org.modelcontextprotocol.sdk:mcp-webmvc-sse-transport` - WebMVC-based Server-Sent Events (SSE) transport implementation for servlet-based applications.
* `io.modelcontextprotocol.sdk:mcp-webflux-sse-transport` - WebFlux-based Server-Sent Events (SSE) transport implementation for reactive applications.
* `io.modelcontextprotocol.sdk:mcp-webmvc-sse-transport` - WebMVC-based Server-Sent Events (SSE) transport implementation for servlet-based applications.
=== Testing Dependencies
* `org.modelcontextprotocol.sdk:mcp-test` - Testing utilities and support for MCP-based applications.
* `io.modelcontextprotocol.sdk:mcp-test` - Testing utilities and support for MCP-based applications.
[[repositories]]
=== Milestone and Snapshot Repositories

12
mcp-docs/src/main/antora/modules/ROOT/pages/mcp.adoc
View file

@ -34,7 +34,7 @@ The core MCP functionality:
[source,xml]
----
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp</artifactId>
</dependency>
----
@ -46,13 +46,13 @@ For HTTP SSE transport implementations, add one of the following dependencies
----
<!-- Spring WebFlux-based SSE client and server transport -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-webflux-sse-transport</artifactId>
</dependency>
<!-- Spring WebMVC-based SSE server transport -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-webmvc-sse-transport</artifactId>
</dependency>
----
@ -64,7 +64,7 @@ The core MCP functionality:
[source,groovy]
----
dependencies {
implementation 'org.modelcontextprotocol.sdk:mcp'
implementation 'io.modelcontextprotocol.sdk:mcp'
}
----
+
@ -74,10 +74,10 @@ For HTTP SSE transport implementations, add one of the following dependencies
[source,groovy]
----
// Spring WebFlux-based SSE client and server transport
implementation 'org.modelcontextprotocol.sdk:mcp-webflux-sse-transport'
implementation 'io.modelcontextprotocol.sdk:mcp-webflux-sse-transport'
// Spring WebMVC-based SSE server transport
implementation 'org.modelcontextprotocol.sdk:mcp-webmvc-sse-transport'
implementation 'io.modelcontextprotocol.sdk:mcp-webmvc-sse-transport'
----
======

12
mcp-docs/src/main/antora/modules/ROOT/pages/overview.adoc
View file

@ -39,19 +39,19 @@ Maven::
----
<!-- Core MCP -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp</artifactId>
</dependency>
<!-- Optional: WebFlux SSE transport -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-webflux-sse-transport</artifactId>
</dependency>
<!-- Optional: WebMVC SSE transport -->
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-webmvc-sse-transport</artifactId>
</dependency>
@ -78,9 +78,9 @@ Gradle::
[source,groovy]
----
dependencies {
implementation 'org.modelcontextprotocol.sdk:mcp' // Core
implementation 'org.modelcontextprotocol.sdk:mcp-webflux-sse-transport' // Optional
implementation 'org.modelcontextprotocol.sdk:mcp-webmvc-sse-transport' // Optional
implementation 'io.modelcontextprotocol.sdk:mcp' // Core
implementation 'io.modelcontextprotocol.sdk:mcp-webflux-sse-transport' // Optional
implementation 'io.modelcontextprotocol.sdk:mcp-webmvc-sse-transport' // Optional
}
repositories {

4
mcp-test/pom.xml
View file

@ -4,7 +4,7 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-parent</artifactId>
<version>0.7.0-SNAPSHOT</version>
</parent>
@ -22,7 +22,7 @@
<dependencies>
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp</artifactId>
<version>0.7.0-SNAPSHOT</version>
</dependency>

90
mcp-test/src/main/java/io/modelcontextprotocol/MockMcpTransport.java Normal file
View file

@ -0,0 +1,90 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.ServerMcpTransport;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCNotification;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCRequest;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
@SuppressWarnings("unused")
public class MockMcpTransport implements ClientMcpTransport, ServerMcpTransport {
private final AtomicInteger inboundMessageCount = new AtomicInteger(0);
private final Sinks.Many<McpSchema.JSONRPCMessage> outgoing = Sinks.many().multicast().onBackpressureBuffer();
private final Sinks.Many<McpSchema.JSONRPCMessage> inbound = Sinks.many().unicast().onBackpressureBuffer();
private final Flux<McpSchema.JSONRPCMessage> outboundView = outgoing.asFlux().cache(1);
public void simulateIncomingMessage(McpSchema.JSONRPCMessage message) {
if (inbound.tryEmitNext(message).isFailure()) {
throw new RuntimeException("Failed to emit message " + message);
}
inboundMessageCount.incrementAndGet();
}
@Override
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
if (outgoing.tryEmitNext(message).isFailure()) {
return Mono.error(new RuntimeException("Can't emit outgoing message " + message));
}
return Mono.empty();
}
public McpSchema.JSONRPCRequest getLastSentMessageAsRequest() {
return (JSONRPCRequest) outboundView.blockFirst();
}
public McpSchema.JSONRPCNotification getLastSentMessageAsNotifiation() {
return (JSONRPCNotification) outboundView.blockFirst();
}
public McpSchema.JSONRPCMessage getLastSentMessage() {
return outboundView.blockFirst();
}
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()
.publishOn(Schedulers.boundedElastic())
.flatMap(message -> Mono.just(message).transform(handler))
.doFinally(signal -> connected = false)
.then();
}
@Override
public Mono<Void> closeGracefully() {
return Mono.defer(() -> {
connected = false;
outgoing.tryEmitComplete();
inbound.tryEmitComplete();
return Mono.empty();
});
}
@Override
public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
return new ObjectMapper().convertValue(data, typeRef);
}
}

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

@ -0,0 +1,364 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import io.modelcontextprotocol.client.McpAsyncClient;
import io.modelcontextprotocol.client.McpClient;
import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpTransport;
import io.modelcontextprotocol.spec.McpSchema.CallToolRequest;
import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
import io.modelcontextprotocol.spec.McpSchema.CreateMessageRequest;
import io.modelcontextprotocol.spec.McpSchema.CreateMessageResult;
import io.modelcontextprotocol.spec.McpSchema.GetPromptRequest;
import io.modelcontextprotocol.spec.McpSchema.Prompt;
import io.modelcontextprotocol.spec.McpSchema.Resource;
import io.modelcontextprotocol.spec.McpSchema.Root;
import io.modelcontextprotocol.spec.McpSchema.SubscribeRequest;
import io.modelcontextprotocol.spec.McpSchema.Tool;
import io.modelcontextprotocol.spec.McpSchema.UnsubscribeRequest;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Test suite for the {@link McpAsyncClient} that can be used with different
* {@link McpTransport} implementations.
*
* @author Christian Tzolov
* @author Dariusz Jędrzejczyk
*/
public abstract class AbstractMcpAsyncClientTests {
private McpAsyncClient mcpAsyncClient;
protected ClientMcpTransport mcpTransport;
private static final Duration TIMEOUT = Duration.ofSeconds(20);
private static final String ECHO_TEST_MESSAGE = "Hello MCP Spring AI!";
abstract protected ClientMcpTransport createMcpTransport();
protected void onStart() {
}
protected void onClose() {
}
@BeforeEach
void setUp() {
onStart();
this.mcpTransport = createMcpTransport();
assertThatCode(() -> {
mcpAsyncClient = McpClient.async(mcpTransport)
.requestTimeout(TIMEOUT)
.capabilities(ClientCapabilities.builder().roots(true).build())
.build();
mcpAsyncClient.initialize().block(Duration.ofSeconds(10));
}).doesNotThrowAnyException();
}
@AfterEach
void tearDown() {
if (mcpAsyncClient != null) {
assertThatCode(() -> mcpAsyncClient.closeGracefully().block(Duration.ofSeconds(10)))
.doesNotThrowAnyException();
}
onClose();
}
@Test
void testConstructorWithInvalidArguments() {
assertThatThrownBy(() -> McpClient.sync(null).build()).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Transport must not be null");
assertThatThrownBy(() -> McpClient.sync(mcpTransport).requestTimeout(null).build())
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Request timeout must not be null");
}
@Test
void testListTools() {
StepVerifier.create(mcpAsyncClient.listTools(null)).consumeNextWith(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull();
}).verifyComplete();
}
@Test
void testPing() {
assertThatCode(() -> mcpAsyncClient.ping().block()).doesNotThrowAnyException();
}
@Test
void testCallTool() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", ECHO_TEST_MESSAGE));
StepVerifier.create(mcpAsyncClient.callTool(callToolRequest)).consumeNextWith(callToolResult -> {
assertThat(callToolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).isNotNull();
assertThat(result.isError()).isNull();
});
}).verifyComplete();
}
@Test
void testCallToolWithInvalidTool() {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", ECHO_TEST_MESSAGE));
assertThatThrownBy(() -> mcpAsyncClient.callTool(invalidRequest).block()).isInstanceOf(Exception.class);
}
@Test
void testListResources() {
StepVerifier.create(mcpAsyncClient.listResources(null)).consumeNextWith(resources -> {
assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull();
}
});
}).verifyComplete();
}
@Test
void testMcpAsyncClientState() {
assertThat(mcpAsyncClient).isNotNull();
}
@Test
void testListPrompts() {
StepVerifier.create(mcpAsyncClient.listPrompts(null)).consumeNextWith(prompts -> {
assertThat(prompts).isNotNull().satisfies(result -> {
assertThat(result.prompts()).isNotNull();
if (!result.prompts().isEmpty()) {
Prompt firstPrompt = result.prompts().get(0);
assertThat(firstPrompt.name()).isNotNull();
assertThat(firstPrompt.description()).isNotNull();
}
});
}).verifyComplete();
}
@Test
void testGetPrompt() {
StepVerifier.create(mcpAsyncClient.getPrompt(new GetPromptRequest("simple_prompt", Map.of())))
.consumeNextWith(prompt -> {
assertThat(prompt).isNotNull().satisfies(result -> {
assertThat(result.messages()).isNotEmpty();
assertThat(result.messages()).hasSize(1);
});
})
.verifyComplete();
}
@Test
void testRootsListChanged() {
assertThatCode(() -> mcpAsyncClient.rootsListChangedNotification().block()).doesNotThrowAnyException();
}
@Test
void testInitializeWithRootsListProviders() {
var transport = createMcpTransport();
var client = McpClient.async(transport)
.requestTimeout(TIMEOUT)
.roots(new Root("file:///test/path", "test-root"))
.build();
assertThatCode(() -> client.initialize().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
assertThatCode(() -> client.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
@Test
void testAddRoot() {
Root newRoot = new Root("file:///new/test/path", "new-test-root");
assertThatCode(() -> mcpAsyncClient.addRoot(newRoot).block()).doesNotThrowAnyException();
}
@Test
void testAddRootWithNullValue() {
assertThatThrownBy(() -> mcpAsyncClient.addRoot(null).block()).hasMessageContaining("Root must not be null");
}
@Test
void testRemoveRoot() {
Root root = new Root("file:///test/path/to/remove", "root-to-remove");
assertThatCode(() -> {
mcpAsyncClient.addRoot(root).block();
mcpAsyncClient.removeRoot(root.uri()).block();
}).doesNotThrowAnyException();
}
@Test
void testRemoveNonExistentRoot() {
assertThatThrownBy(() -> mcpAsyncClient.removeRoot("nonexistent-uri").block())
.hasMessageContaining("Root with uri 'nonexistent-uri' not found");
}
@Test
@Disabled
void testReadResource() {
StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> {
if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0);
StepVerifier.create(mcpAsyncClient.readResource(firstResource)).consumeNextWith(result -> {
assertThat(result).isNotNull();
assertThat(result.contents()).isNotNull();
}).verifyComplete();
}
}).verifyComplete();
}
@Test
void testListResourceTemplates() {
StepVerifier.create(mcpAsyncClient.listResourceTemplates()).consumeNextWith(result -> {
assertThat(result).isNotNull();
assertThat(result.resourceTemplates()).isNotNull();
}).verifyComplete();
}
// @Test
void testResourceSubscription() {
StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> {
if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0);
// Test subscribe
StepVerifier.create(mcpAsyncClient.subscribeResource(new SubscribeRequest(firstResource.uri())))
.verifyComplete();
// Test unsubscribe
StepVerifier.create(mcpAsyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())))
.verifyComplete();
}
}).verifyComplete();
}
@Test
void testNotificationHandlers() {
AtomicBoolean toolsNotificationReceived = new AtomicBoolean(false);
AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false);
AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false);
var transport = createMcpTransport();
var client = McpClient.async(transport)
.requestTimeout(TIMEOUT)
.toolsChangeConsumer(tools -> Mono.fromRunnable(() -> toolsNotificationReceived.set(true)))
.resourcesChangeConsumer(resources -> Mono.fromRunnable(() -> resourcesNotificationReceived.set(true)))
.promptsChangeConsumer(prompts -> Mono.fromRunnable(() -> promptsNotificationReceived.set(true)))
.build();
assertThatCode(() -> {
client.initialize().block();
// Trigger notifications
client.sendResourcesListChanged().block();
client.promptListChangedNotification().block();
client.closeGracefully().block();
}).doesNotThrowAnyException();
}
@Test
void testInitializeWithSamplingCapability() {
var transport = createMcpTransport();
var capabilities = ClientCapabilities.builder().sampling().build();
var client = McpClient.async(transport)
.requestTimeout(TIMEOUT)
.capabilities(capabilities)
.sampling(request -> Mono.just(CreateMessageResult.builder().message("test").model("test-model").build()))
.build();
assertThatCode(() -> {
client.initialize().block(Duration.ofSeconds(10));
client.closeGracefully().block(Duration.ofSeconds(10));
}).doesNotThrowAnyException();
}
@Test
void testInitializeWithAllCapabilities() {
var transport = createMcpTransport();
var capabilities = ClientCapabilities.builder()
.experimental(Map.of("feature", "test"))
.roots(true)
.sampling()
.build();
Function<CreateMessageRequest, Mono<CreateMessageResult>> samplingHandler = request -> Mono
.just(CreateMessageResult.builder().message("test").model("test-model").build());
var client = McpClient.async(transport)
.requestTimeout(TIMEOUT)
.capabilities(capabilities)
.sampling(samplingHandler)
.build();
assertThatCode(() -> {
var result = client.initialize().block(Duration.ofSeconds(10));
assertThat(result).isNotNull();
assertThat(result.capabilities()).isNotNull();
client.closeGracefully().block(Duration.ofSeconds(10));
}).doesNotThrowAnyException();
}
// ---------------------------------------
// Logging Tests
// ---------------------------------------
@Test
void testLoggingLevels() {
// Test all logging levels
for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
StepVerifier.create(mcpAsyncClient.setLoggingLevel(level)).verifyComplete();
}
}
@Test
void testLoggingConsumer() {
AtomicBoolean logReceived = new AtomicBoolean(false);
var transport = createMcpTransport();
var client = McpClient.async(transport)
.requestTimeout(TIMEOUT)
.loggingConsumer(notification -> Mono.fromRunnable(() -> logReceived.set(true)))
.build();
assertThatCode(() -> {
client.initialize().block(Duration.ofSeconds(10));
client.closeGracefully().block(Duration.ofSeconds(10));
}).doesNotThrowAnyException();
}
@Test
void testLoggingWithNullNotification() {
assertThatThrownBy(() -> mcpAsyncClient.setLoggingLevel(null).block())
.hasMessageContaining("Logging level must not be null");
}
}

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

@ -0,0 +1,304 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.client;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import io.modelcontextprotocol.client.McpClient;
import io.modelcontextprotocol.client.McpSyncClient;
import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.CallToolRequest;
import io.modelcontextprotocol.spec.McpSchema.CallToolResult;
import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
import io.modelcontextprotocol.spec.McpSchema.ListResourceTemplatesResult;
import io.modelcontextprotocol.spec.McpSchema.ListResourcesResult;
import io.modelcontextprotocol.spec.McpSchema.ListToolsResult;
import io.modelcontextprotocol.spec.McpSchema.ReadResourceResult;
import io.modelcontextprotocol.spec.McpSchema.Resource;
import io.modelcontextprotocol.spec.McpSchema.Root;
import io.modelcontextprotocol.spec.McpSchema.SubscribeRequest;
import io.modelcontextprotocol.spec.McpSchema.TextContent;
import io.modelcontextprotocol.spec.McpSchema.Tool;
import io.modelcontextprotocol.spec.McpSchema.UnsubscribeRequest;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Unit tests for MCP Client Session functionality.
*
* @author Christian Tzolov
* @author Dariusz Jędrzejczyk
*/
public abstract class AbstractMcpSyncClientTests {
private McpSyncClient mcpSyncClient;
private static final Duration TIMEOUT = Duration.ofSeconds(10);
private static final String TEST_MESSAGE = "Hello MCP Spring AI!";
protected ClientMcpTransport mcpTransport;
abstract protected ClientMcpTransport createMcpTransport();
abstract protected void onStart();
abstract protected void onClose();
@BeforeEach
void setUp() {
onStart();
this.mcpTransport = createMcpTransport();
assertThatCode(() -> {
mcpSyncClient = McpClient.sync(mcpTransport)
.requestTimeout(TIMEOUT)
.capabilities(ClientCapabilities.builder().roots(true).build())
.build();
mcpSyncClient.initialize();
}).doesNotThrowAnyException();
}
@AfterEach
void tearDown() {
if (mcpSyncClient != null) {
assertThatCode(() -> mcpSyncClient.close()).doesNotThrowAnyException();
}
onClose();
}
@Test
void testConstructorWithInvalidArguments() {
assertThatThrownBy(() -> McpClient.sync(null).build()).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Transport must not be null");
assertThatThrownBy(() -> McpClient.sync(mcpTransport).requestTimeout(null).build())
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Request timeout must not be null");
}
@Test
void testListTools() {
ListToolsResult tools = mcpSyncClient.listTools(null);
assertThat(tools).isNotNull().satisfies(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull();
});
}
@Test
void testCallTools() {
CallToolResult toolResult = mcpSyncClient.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4)));
assertThat(toolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).hasSize(1);
TextContent content = (TextContent) result.content().get(0);
assertThat(content).isNotNull();
assertThat(content.text()).isNotNull();
assertThat(content.text()).contains("7");
});
}
@Test
void testPing() {
assertThatCode(() -> mcpSyncClient.ping()).doesNotThrowAnyException();
}
@Test
void testCallTool() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE));
CallToolResult callToolResult = mcpSyncClient.callTool(callToolRequest);
assertThat(callToolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).isNotNull();
assertThat(result.isError()).isNull();
});
}
@Test
void testCallToolWithInvalidTool() {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", TEST_MESSAGE));
assertThatThrownBy(() -> mcpSyncClient.callTool(invalidRequest)).isInstanceOf(Exception.class);
}
@Test
void testRootsListChanged() {
assertThatCode(() -> mcpSyncClient.rootsListChangedNotification()).doesNotThrowAnyException();
}
@Test
void testListResources() {
ListResourcesResult resources = mcpSyncClient.listResources(null);
assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull();
}
});
}
@Test
void testClientSessionState() {
assertThat(mcpSyncClient).isNotNull();
}
@Test
void testInitializeWithRootsListProviders() {
var transport = createMcpTransport();
var client = McpClient.sync(transport)
.requestTimeout(TIMEOUT)
.roots(new Root("file:///test/path", "test-root"))
.build();
assertThatCode(() -> {
client.initialize();
client.close();
}).doesNotThrowAnyException();
}
@Test
void testAddRoot() {
Root newRoot = new Root("file:///new/test/path", "new-test-root");
assertThatCode(() -> mcpSyncClient.addRoot(newRoot)).doesNotThrowAnyException();
}
@Test
void testAddRootWithNullValue() {
assertThatThrownBy(() -> mcpSyncClient.addRoot(null)).hasMessageContaining("Root must not be null");
}
@Test
void testRemoveRoot() {
Root root = new Root("file:///test/path/to/remove", "root-to-remove");
assertThatCode(() -> {
mcpSyncClient.addRoot(root);
mcpSyncClient.removeRoot(root.uri());
}).doesNotThrowAnyException();
}
@Test
void testRemoveNonExistentRoot() {
assertThatThrownBy(() -> mcpSyncClient.removeRoot("nonexistent-uri"))
.hasMessageContaining("Root with uri 'nonexistent-uri' not found");
}
@Test
void testReadResource() {
ListResourcesResult resources = mcpSyncClient.listResources(null);
if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0);
ReadResourceResult result = mcpSyncClient.readResource(firstResource);
assertThat(result).isNotNull();
assertThat(result.contents()).isNotNull();
}
}
@Test
void testListResourceTemplates() {
ListResourceTemplatesResult result = mcpSyncClient.listResourceTemplates(null);
assertThat(result).isNotNull();
assertThat(result.resourceTemplates()).isNotNull();
}
// @Test
void testResourceSubscription() {
ListResourcesResult resources = mcpSyncClient.listResources(null);
if (!resources.resources().isEmpty()) {
Resource firstResource = resources.resources().get(0);
// Test subscribe
assertThatCode(() -> mcpSyncClient.subscribeResource(new SubscribeRequest(firstResource.uri())))
.doesNotThrowAnyException();
// Test unsubscribe
assertThatCode(() -> mcpSyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())))
.doesNotThrowAnyException();
}
}
@Test
void testNotificationHandlers() {
AtomicBoolean toolsNotificationReceived = new AtomicBoolean(false);
AtomicBoolean resourcesNotificationReceived = new AtomicBoolean(false);
AtomicBoolean promptsNotificationReceived = new AtomicBoolean(false);
var transport = createMcpTransport();
var client = McpClient.sync(transport)
.requestTimeout(TIMEOUT)
.toolsChangeConsumer(tools -> toolsNotificationReceived.set(true))
.resourcesChangeConsumer(resources -> resourcesNotificationReceived.set(true))
.promptsChangeConsumer(prompts -> promptsNotificationReceived.set(true))
.build();
assertThatCode(() -> {
client.initialize();
// Trigger notifications
client.sendResourcesListChanged();
client.promptListChangedNotification();
client.close();
}).doesNotThrowAnyException();
}
// ---------------------------------------
// Logging Tests
// ---------------------------------------
@Test
void testLoggingLevels() {
// Test all logging levels
for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
assertThatCode(() -> mcpSyncClient.setLoggingLevel(level)).doesNotThrowAnyException();
}
}
@Test
void testLoggingConsumer() {
AtomicBoolean logReceived = new AtomicBoolean(false);
var transport = createMcpTransport();
var client = McpClient.sync(transport)
.requestTimeout(TIMEOUT)
.loggingConsumer(notification -> logReceived.set(true))
.build();
assertThatCode(() -> {
client.initialize();
client.close();
}).doesNotThrowAnyException();
}
@Test
void testLoggingWithNullNotification() {
assertThatThrownBy(() -> mcpSyncClient.setLoggingLevel(null))
.hasMessageContaining("Logging level must not be null");
}
}

466
mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpAsyncServerTests.java Normal file
View file

@ -0,0 +1,466 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.server;
import java.time.Duration;
import java.util.List;
import io.modelcontextprotocol.server.McpAsyncServer;
import io.modelcontextprotocol.server.McpServer;
import io.modelcontextprotocol.server.McpServerFeatures;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpTransport;
import io.modelcontextprotocol.spec.ServerMcpTransport;
import io.modelcontextprotocol.spec.McpSchema.CallToolResult;
import io.modelcontextprotocol.spec.McpSchema.GetPromptResult;
import io.modelcontextprotocol.spec.McpSchema.Prompt;
import io.modelcontextprotocol.spec.McpSchema.PromptMessage;
import io.modelcontextprotocol.spec.McpSchema.ReadResourceResult;
import io.modelcontextprotocol.spec.McpSchema.Resource;
import io.modelcontextprotocol.spec.McpSchema.ServerCapabilities;
import io.modelcontextprotocol.spec.McpSchema.Tool;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Test suite for the {@link McpAsyncServer} that can be used with different
* {@link McpTransport} implementations.
*
* @author Christian Tzolov
*/
public abstract class AbstractMcpAsyncServerTests {
private static final String TEST_TOOL_NAME = "test-tool";
private static final String TEST_RESOURCE_URI = "test://resource";
private static final String TEST_PROMPT_NAME = "test-prompt";
abstract protected ServerMcpTransport createMcpTransport();
protected void onStart() {
}
protected void onClose() {
}
@BeforeEach
void setUp() {
}
@AfterEach
void tearDown() {
onClose();
}
// ---------------------------------------
// Server Lifecycle Tests
// ---------------------------------------
@Test
void testConstructorWithInvalidArguments() {
assertThatThrownBy(() -> McpServer.async(null)).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Transport must not be null");
assertThatThrownBy(() -> McpServer.async(createMcpTransport()).serverInfo((McpSchema.Implementation) null))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Server info must not be null");
}
@Test
void testGracefulShutdown() {
var mcpAsyncServer = McpServer.async(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
StepVerifier.create(mcpAsyncServer.closeGracefully()).verifyComplete();
}
@Test
void testImmediateClose() {
var mcpAsyncServer = McpServer.async(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
assertThatCode(() -> mcpAsyncServer.close()).doesNotThrowAnyException();
}
// ---------------------------------------
// Tools Tests
// ---------------------------------------
String emptyJsonSchema = """
{
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"properties": {}
}
""";
@Test
void testAddTool() {
Tool newTool = new McpSchema.Tool("new-tool", "New test tool", emptyJsonSchema);
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().tools(true).build())
.build();
StepVerifier.create(mcpAsyncServer.addTool(new McpServerFeatures.AsyncToolRegistration(newTool,
args -> Mono.just(new CallToolResult(List.of(), false)))))
.verifyComplete();
assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
@Test
void testAddDuplicateTool() {
Tool duplicateTool = new McpSchema.Tool(TEST_TOOL_NAME, "Duplicate tool", emptyJsonSchema);
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().tools(true).build())
.tool(duplicateTool, args -> Mono.just(new CallToolResult(List.of(), false)))
.build();
StepVerifier.create(mcpAsyncServer.addTool(new McpServerFeatures.AsyncToolRegistration(duplicateTool,
args -> Mono.just(new CallToolResult(List.of(), false)))))
.verifyErrorSatisfies(error -> {
assertThat(error).isInstanceOf(McpError.class)
.hasMessage("Tool with name '" + TEST_TOOL_NAME + "' already exists");
});
assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
@Test
void testRemoveTool() {
Tool too = new McpSchema.Tool(TEST_TOOL_NAME, "Duplicate tool", emptyJsonSchema);
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().tools(true).build())
.tool(too, args -> Mono.just(new CallToolResult(List.of(), false)))
.build();
StepVerifier.create(mcpAsyncServer.removeTool(TEST_TOOL_NAME)).verifyComplete();
assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
@Test
void testRemoveNonexistentTool() {
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().tools(true).build())
.build();
StepVerifier.create(mcpAsyncServer.removeTool("nonexistent-tool")).verifyErrorSatisfies(error -> {
assertThat(error).isInstanceOf(McpError.class).hasMessage("Tool with name 'nonexistent-tool' not found");
});
assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
@Test
void testNotifyToolsListChanged() {
Tool too = new McpSchema.Tool(TEST_TOOL_NAME, "Duplicate tool", emptyJsonSchema);
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().tools(true).build())
.tool(too, args -> Mono.just(new CallToolResult(List.of(), false)))
.build();
StepVerifier.create(mcpAsyncServer.notifyToolsListChanged()).verifyComplete();
assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
// ---------------------------------------
// Resources Tests
// ---------------------------------------
@Test
void testNotifyResourcesListChanged() {
var mcpAsyncServer = McpServer.async(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
StepVerifier.create(mcpAsyncServer.notifyResourcesListChanged()).verifyComplete();
assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
@Test
void testAddResource() {
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().resources(true, false).build())
.build();
Resource resource = new Resource(TEST_RESOURCE_URI, "Test Resource", "text/plain", "Test resource description",
null);
McpServerFeatures.AsyncResourceRegistration registration = new McpServerFeatures.AsyncResourceRegistration(
resource, req -> Mono.just(new ReadResourceResult(List.of())));
StepVerifier.create(mcpAsyncServer.addResource(registration)).verifyComplete();
assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
@Test
void testAddResourceWithNullRegistration() {
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().resources(true, false).build())
.build();
StepVerifier.create(mcpAsyncServer.addResource((McpServerFeatures.AsyncResourceRegistration) null))
.verifyErrorSatisfies(error -> {
assertThat(error).isInstanceOf(McpError.class).hasMessage("Resource must not be null");
});
assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
@Test
void testAddResourceWithoutCapability() {
// Create a server without resource capabilities
McpAsyncServer serverWithoutResources = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.build();
Resource resource = new Resource(TEST_RESOURCE_URI, "Test Resource", "text/plain", "Test resource description",
null);
McpServerFeatures.AsyncResourceRegistration registration = new McpServerFeatures.AsyncResourceRegistration(
resource, req -> Mono.just(new ReadResourceResult(List.of())));
StepVerifier.create(serverWithoutResources.addResource(registration)).verifyErrorSatisfies(error -> {
assertThat(error).isInstanceOf(McpError.class)
.hasMessage("Server must be configured with resource capabilities");
});
}
@Test
void testRemoveResourceWithoutCapability() {
// Create a server without resource capabilities
McpAsyncServer serverWithoutResources = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.build();
StepVerifier.create(serverWithoutResources.removeResource(TEST_RESOURCE_URI)).verifyErrorSatisfies(error -> {
assertThat(error).isInstanceOf(McpError.class)
.hasMessage("Server must be configured with resource capabilities");
});
}
// ---------------------------------------
// Prompts Tests
// ---------------------------------------
@Test
void testNotifyPromptsListChanged() {
var mcpAsyncServer = McpServer.async(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
StepVerifier.create(mcpAsyncServer.notifyPromptsListChanged()).verifyComplete();
assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
@Test
void testAddPromptWithNullRegistration() {
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().prompts(false).build())
.build();
StepVerifier.create(mcpAsyncServer.addPrompt((McpServerFeatures.AsyncPromptRegistration) null))
.verifyErrorSatisfies(error -> {
assertThat(error).isInstanceOf(McpError.class).hasMessage("Prompt registration must not be null");
});
}
@Test
void testAddPromptWithoutCapability() {
// Create a server without prompt capabilities
McpAsyncServer serverWithoutPrompts = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.build();
Prompt prompt = new Prompt(TEST_PROMPT_NAME, "Test Prompt", List.of());
McpServerFeatures.AsyncPromptRegistration registration = new McpServerFeatures.AsyncPromptRegistration(prompt,
req -> Mono.just(new GetPromptResult("Test prompt description", List
.of(new PromptMessage(McpSchema.Role.ASSISTANT, new McpSchema.TextContent("Test content"))))));
StepVerifier.create(serverWithoutPrompts.addPrompt(registration)).verifyErrorSatisfies(error -> {
assertThat(error).isInstanceOf(McpError.class)
.hasMessage("Server must be configured with prompt capabilities");
});
}
@Test
void testRemovePromptWithoutCapability() {
// Create a server without prompt capabilities
McpAsyncServer serverWithoutPrompts = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.build();
StepVerifier.create(serverWithoutPrompts.removePrompt(TEST_PROMPT_NAME)).verifyErrorSatisfies(error -> {
assertThat(error).isInstanceOf(McpError.class)
.hasMessage("Server must be configured with prompt capabilities");
});
}
@Test
void testRemovePrompt() {
String TEST_PROMPT_NAME_TO_REMOVE = "TEST_PROMPT_NAME678";
Prompt prompt = new Prompt(TEST_PROMPT_NAME_TO_REMOVE, "Test Prompt", List.of());
McpServerFeatures.AsyncPromptRegistration registration = new McpServerFeatures.AsyncPromptRegistration(prompt,
req -> Mono.just(new GetPromptResult("Test prompt description", List
.of(new PromptMessage(McpSchema.Role.ASSISTANT, new McpSchema.TextContent("Test content"))))));
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().prompts(true).build())
.prompts(registration)
.build();
StepVerifier.create(mcpAsyncServer.removePrompt(TEST_PROMPT_NAME_TO_REMOVE)).verifyComplete();
assertThatCode(() -> mcpAsyncServer.closeGracefully().block(Duration.ofSeconds(10))).doesNotThrowAnyException();
}
@Test
void testRemoveNonexistentPrompt() {
var mcpAsyncServer2 = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().prompts(true).build())
.build();
StepVerifier.create(mcpAsyncServer2.removePrompt("nonexistent-prompt")).verifyErrorSatisfies(error -> {
assertThat(error).isInstanceOf(McpError.class)
.hasMessage("Prompt with name 'nonexistent-prompt' not found");
});
assertThatCode(() -> mcpAsyncServer2.closeGracefully().block(Duration.ofSeconds(10)))
.doesNotThrowAnyException();
}
// ---------------------------------------
// Roots Tests
// ---------------------------------------
@Test
void testRootsChangeConsumers() {
// Test with single consumer
var rootsReceived = new McpSchema.Root[1];
var consumerCalled = new boolean[1];
var singleConsumerServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.rootsChangeConsumers(List.of(roots -> Mono.fromRunnable(() -> {
consumerCalled[0] = true;
if (!roots.isEmpty()) {
rootsReceived[0] = roots.get(0);
}
})))
.build();
assertThat(singleConsumerServer).isNotNull();
assertThatCode(() -> singleConsumerServer.closeGracefully().block(Duration.ofSeconds(10)))
.doesNotThrowAnyException();
onClose();
// Test with multiple consumers
var consumer1Called = new boolean[1];
var consumer2Called = new boolean[1];
var rootsContent = new List[1];
var multipleConsumersServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.rootsChangeConsumers(List.of(roots -> Mono.fromRunnable(() -> {
consumer1Called[0] = true;
rootsContent[0] = roots;
}), roots -> Mono.fromRunnable(() -> consumer2Called[0] = true)))
.build();
assertThat(multipleConsumersServer).isNotNull();
assertThatCode(() -> multipleConsumersServer.closeGracefully().block(Duration.ofSeconds(10)))
.doesNotThrowAnyException();
onClose();
// Test error handling
var errorHandlingServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.rootsChangeConsumers(List.of(roots -> {
throw new RuntimeException("Test error");
}))
.build();
assertThat(errorHandlingServer).isNotNull();
assertThatCode(() -> errorHandlingServer.closeGracefully().block(Duration.ofSeconds(10)))
.doesNotThrowAnyException();
onClose();
// Test without consumers
var noConsumersServer = McpServer.async(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
assertThat(noConsumersServer).isNotNull();
assertThatCode(() -> noConsumersServer.closeGracefully().block(Duration.ofSeconds(10)))
.doesNotThrowAnyException();
}
// ---------------------------------------
// Logging Tests
// ---------------------------------------
@Test
void testLoggingLevels() {
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().logging().build())
.build();
// Test all logging levels
for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
var notification = McpSchema.LoggingMessageNotification.builder()
.level(level)
.logger("test-logger")
.data("Test message with level " + level)
.build();
StepVerifier.create(mcpAsyncServer.loggingNotification(notification)).verifyComplete();
}
}
@Test
void testLoggingWithoutCapability() {
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().build()) // No logging capability
.build();
var notification = McpSchema.LoggingMessageNotification.builder()
.level(McpSchema.LoggingLevel.INFO)
.logger("test-logger")
.data("Test log message")
.build();
StepVerifier.create(mcpAsyncServer.loggingNotification(notification)).verifyComplete();
}
@Test
void testLoggingWithNullNotification() {
var mcpAsyncServer = McpServer.async(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().logging().build())
.build();
StepVerifier.create(mcpAsyncServer.loggingNotification(null)).verifyError(McpError.class);
}
}

434
mcp-test/src/main/java/io/modelcontextprotocol/server/AbstractMcpSyncServerTests.java Normal file
View file

@ -0,0 +1,434 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.server;
import java.util.List;
import io.modelcontextprotocol.server.McpServer;
import io.modelcontextprotocol.server.McpServerFeatures;
import io.modelcontextprotocol.server.McpSyncServer;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpTransport;
import io.modelcontextprotocol.spec.ServerMcpTransport;
import io.modelcontextprotocol.spec.McpSchema.CallToolResult;
import io.modelcontextprotocol.spec.McpSchema.GetPromptResult;
import io.modelcontextprotocol.spec.McpSchema.Prompt;
import io.modelcontextprotocol.spec.McpSchema.PromptMessage;
import io.modelcontextprotocol.spec.McpSchema.ReadResourceResult;
import io.modelcontextprotocol.spec.McpSchema.Resource;
import io.modelcontextprotocol.spec.McpSchema.ServerCapabilities;
import io.modelcontextprotocol.spec.McpSchema.Tool;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Test suite for the {@link McpSyncServer} that can be used with different
* {@link McpTransport} implementations.
*
* @author Christian Tzolov
*/
public abstract class AbstractMcpSyncServerTests {
private static final String TEST_TOOL_NAME = "test-tool";
private static final String TEST_RESOURCE_URI = "test://resource";
private static final String TEST_PROMPT_NAME = "test-prompt";
abstract protected ServerMcpTransport createMcpTransport();
protected void onStart() {
}
protected void onClose() {
}
@BeforeEach
void setUp() {
// onStart();
}
@AfterEach
void tearDown() {
onClose();
}
// ---------------------------------------
// Server Lifecycle Tests
// ---------------------------------------
@Test
void testConstructorWithInvalidArguments() {
assertThatThrownBy(() -> McpServer.sync(null)).isInstanceOf(IllegalArgumentException.class)
.hasMessage("Transport must not be null");
assertThatThrownBy(() -> McpServer.sync(createMcpTransport()).serverInfo(null))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Server info must not be null");
}
@Test
void testGracefulShutdown() {
var mcpSyncServer = McpServer.sync(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
@Test
void testImmediateClose() {
var mcpSyncServer = McpServer.sync(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
assertThatCode(() -> mcpSyncServer.close()).doesNotThrowAnyException();
}
@Test
void testGetAsyncServer() {
var mcpSyncServer = McpServer.sync(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
assertThat(mcpSyncServer.getAsyncServer()).isNotNull();
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
// ---------------------------------------
// Tools Tests
// ---------------------------------------
String emptyJsonSchema = """
{
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"properties": {}
}
""";
@Test
void testAddTool() {
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().tools(true).build())
.build();
Tool newTool = new McpSchema.Tool("new-tool", "New test tool", emptyJsonSchema);
assertThatCode(() -> mcpSyncServer
.addTool(new McpServerFeatures.SyncToolRegistration(newTool, args -> new CallToolResult(List.of(), false))))
.doesNotThrowAnyException();
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
@Test
void testAddDuplicateTool() {
Tool duplicateTool = new McpSchema.Tool(TEST_TOOL_NAME, "Duplicate tool", emptyJsonSchema);
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().tools(true).build())
.tool(duplicateTool, args -> new CallToolResult(List.of(), false))
.build();
assertThatThrownBy(() -> mcpSyncServer.addTool(new McpServerFeatures.SyncToolRegistration(duplicateTool,
args -> new CallToolResult(List.of(), false))))
.isInstanceOf(McpError.class)
.hasMessage("Tool with name '" + TEST_TOOL_NAME + "' already exists");
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
@Test
void testRemoveTool() {
Tool tool = new McpSchema.Tool(TEST_TOOL_NAME, "Test tool", emptyJsonSchema);
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().tools(true).build())
.tool(tool, args -> new CallToolResult(List.of(), false))
.build();
assertThatCode(() -> mcpSyncServer.removeTool(TEST_TOOL_NAME)).doesNotThrowAnyException();
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
@Test
void testRemoveNonexistentTool() {
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().tools(true).build())
.build();
assertThatThrownBy(() -> mcpSyncServer.removeTool("nonexistent-tool")).isInstanceOf(McpError.class)
.hasMessage("Tool with name 'nonexistent-tool' not found");
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
@Test
void testNotifyToolsListChanged() {
var mcpSyncServer = McpServer.sync(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
assertThatCode(() -> mcpSyncServer.notifyToolsListChanged()).doesNotThrowAnyException();
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
// ---------------------------------------
// Resources Tests
// ---------------------------------------
@Test
void testNotifyResourcesListChanged() {
var mcpSyncServer = McpServer.sync(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
assertThatCode(() -> mcpSyncServer.notifyResourcesListChanged()).doesNotThrowAnyException();
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
@Test
void testAddResource() {
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().resources(true, false).build())
.build();
Resource resource = new Resource(TEST_RESOURCE_URI, "Test Resource", "text/plain", "Test resource description",
null);
McpServerFeatures.SyncResourceRegistration registration = new McpServerFeatures.SyncResourceRegistration(
resource, req -> new ReadResourceResult(List.of()));
assertThatCode(() -> mcpSyncServer.addResource(registration)).doesNotThrowAnyException();
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
@Test
void testAddResourceWithNullRegistration() {
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().resources(true, false).build())
.build();
assertThatThrownBy(() -> mcpSyncServer.addResource((McpServerFeatures.SyncResourceRegistration) null))
.isInstanceOf(McpError.class)
.hasMessage("Resource must not be null");
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
@Test
void testAddResourceWithoutCapability() {
var serverWithoutResources = McpServer.sync(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
Resource resource = new Resource(TEST_RESOURCE_URI, "Test Resource", "text/plain", "Test resource description",
null);
McpServerFeatures.SyncResourceRegistration registration = new McpServerFeatures.SyncResourceRegistration(
resource, req -> new ReadResourceResult(List.of()));
assertThatThrownBy(() -> serverWithoutResources.addResource(registration)).isInstanceOf(McpError.class)
.hasMessage("Server must be configured with resource capabilities");
}
@Test
void testRemoveResourceWithoutCapability() {
var serverWithoutResources = McpServer.sync(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
assertThatThrownBy(() -> serverWithoutResources.removeResource(TEST_RESOURCE_URI)).isInstanceOf(McpError.class)
.hasMessage("Server must be configured with resource capabilities");
}
// ---------------------------------------
// Prompts Tests
// ---------------------------------------
@Test
void testNotifyPromptsListChanged() {
var mcpSyncServer = McpServer.sync(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
assertThatCode(() -> mcpSyncServer.notifyPromptsListChanged()).doesNotThrowAnyException();
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
@Test
void testAddPromptWithNullRegistration() {
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().prompts(false).build())
.build();
assertThatThrownBy(() -> mcpSyncServer.addPrompt((McpServerFeatures.SyncPromptRegistration) null))
.isInstanceOf(McpError.class)
.hasMessage("Prompt registration must not be null");
}
@Test
void testAddPromptWithoutCapability() {
var serverWithoutPrompts = McpServer.sync(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
Prompt prompt = new Prompt(TEST_PROMPT_NAME, "Test Prompt", List.of());
McpServerFeatures.SyncPromptRegistration registration = new McpServerFeatures.SyncPromptRegistration(prompt,
req -> new GetPromptResult("Test prompt description", List
.of(new PromptMessage(McpSchema.Role.ASSISTANT, new McpSchema.TextContent("Test content")))));
assertThatThrownBy(() -> serverWithoutPrompts.addPrompt(registration)).isInstanceOf(McpError.class)
.hasMessage("Server must be configured with prompt capabilities");
}
@Test
void testRemovePromptWithoutCapability() {
var serverWithoutPrompts = McpServer.sync(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
assertThatThrownBy(() -> serverWithoutPrompts.removePrompt(TEST_PROMPT_NAME)).isInstanceOf(McpError.class)
.hasMessage("Server must be configured with prompt capabilities");
}
@Test
void testRemovePrompt() {
Prompt prompt = new Prompt(TEST_PROMPT_NAME, "Test Prompt", List.of());
McpServerFeatures.SyncPromptRegistration registration = new McpServerFeatures.SyncPromptRegistration(prompt,
req -> new GetPromptResult("Test prompt description", List
.of(new PromptMessage(McpSchema.Role.ASSISTANT, new McpSchema.TextContent("Test content")))));
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().prompts(true).build())
.prompts(registration)
.build();
assertThatCode(() -> mcpSyncServer.removePrompt(TEST_PROMPT_NAME)).doesNotThrowAnyException();
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
@Test
void testRemoveNonexistentPrompt() {
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().prompts(true).build())
.build();
assertThatThrownBy(() -> mcpSyncServer.removePrompt("nonexistent-prompt")).isInstanceOf(McpError.class)
.hasMessage("Prompt with name 'nonexistent-prompt' not found");
assertThatCode(() -> mcpSyncServer.closeGracefully()).doesNotThrowAnyException();
}
// ---------------------------------------
// Roots Tests
// ---------------------------------------
@Test
void testRootsChangeConsumers() {
// Test with single consumer
var rootsReceived = new McpSchema.Root[1];
var consumerCalled = new boolean[1];
var singleConsumerServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.rootsChangeConsumers(List.of(roots -> {
consumerCalled[0] = true;
if (!roots.isEmpty()) {
rootsReceived[0] = roots.get(0);
}
}))
.build();
assertThat(singleConsumerServer).isNotNull();
assertThatCode(() -> singleConsumerServer.closeGracefully()).doesNotThrowAnyException();
onClose();
// Test with multiple consumers
var consumer1Called = new boolean[1];
var consumer2Called = new boolean[1];
var rootsContent = new List[1];
var multipleConsumersServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.rootsChangeConsumers(List.of(roots -> {
consumer1Called[0] = true;
rootsContent[0] = roots;
}, roots -> consumer2Called[0] = true))
.build();
assertThat(multipleConsumersServer).isNotNull();
assertThatCode(() -> multipleConsumersServer.closeGracefully()).doesNotThrowAnyException();
onClose();
// Test error handling
var errorHandlingServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.rootsChangeConsumers(List.of(roots -> {
throw new RuntimeException("Test error");
}))
.build();
assertThat(errorHandlingServer).isNotNull();
assertThatCode(() -> errorHandlingServer.closeGracefully()).doesNotThrowAnyException();
onClose();
// Test without consumers
var noConsumersServer = McpServer.sync(createMcpTransport()).serverInfo("test-server", "1.0.0").build();
assertThat(noConsumersServer).isNotNull();
assertThatCode(() -> noConsumersServer.closeGracefully()).doesNotThrowAnyException();
}
// ---------------------------------------
// Logging Tests
// ---------------------------------------
@Test
void testLoggingLevels() {
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().logging().build())
.build();
// Test all logging levels
for (McpSchema.LoggingLevel level : McpSchema.LoggingLevel.values()) {
var notification = McpSchema.LoggingMessageNotification.builder()
.level(level)
.logger("test-logger")
.data("Test message with level " + level)
.build();
assertThatCode(() -> mcpSyncServer.loggingNotification(notification)).doesNotThrowAnyException();
}
}
@Test
void testLoggingWithoutCapability() {
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().build()) // No logging capability
.build();
var notification = McpSchema.LoggingMessageNotification.builder()
.level(McpSchema.LoggingLevel.INFO)
.logger("test-logger")
.data("Test log message")
.build();
assertThatCode(() -> mcpSyncServer.loggingNotification(notification)).doesNotThrowAnyException();
}
@Test
void testLoggingWithNullNotification() {
var mcpSyncServer = McpServer.sync(createMcpTransport())
.serverInfo("test-server", "1.0.0")
.capabilities(ServerCapabilities.builder().logging().build())
.build();
assertThatThrownBy(() -> mcpSyncServer.loggingNotification(null)).isInstanceOf(McpError.class);
}
}

2
mcp-transport/mcp-webflux-sse-transport/README.md
View file

@ -2,7 +2,7 @@
```xml
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-webflux-sse-transport</artifactId>
</dependency>
```

6
mcp-transport/mcp-webflux-sse-transport/pom.xml
View file

@ -4,7 +4,7 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-parent</artifactId>
<version>0.7.0-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath>
@ -23,13 +23,13 @@
<dependencies>
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp</artifactId>
<version>0.7.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.modelcontextprotocol.sdk</groupId>
<groupId>io.modelcontextprotocol.sdk</groupId>
<artifactId>mcp-test</artifactId>
<version>0.7.0-SNAPSHOT</version>
<scope>test</scope>

324
mcp-transport/mcp-webflux-sse-transport/src/main/java/io/modelcontextprotocol/client/transport/WebFluxSseClientTransport.java Normal file
View file

@ -0,0 +1,324 @@
/*
* Copyright 2024 - 2024 the original author or authors.
*/
package io.modelcontextprotocol.client.transport;
import java.io.IOException;
import java.util.function.BiConsumer;
import java.util.function.Function;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage;
import io.modelcontextprotocol.util.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.publisher.SynchronousSink;
import reactor.core.scheduler.Schedulers;
import reactor.util.retry.Retry;
import reactor.util.retry.Retry.RetrySignal;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.reactive.function.client.WebClient;
/**
* Server-Sent Events (SSE) implementation of the
* {@link io.modelcontextprotocol.spec.McpTransport} that follows the MCP HTTP with SSE
* transport specification.
*
* <p>
* This transport establishes a bidirectional communication channel where:
* <ul>
* <li>Inbound messages are received through an SSE connection from the server</li>
* <li>Outbound messages are sent via HTTP POST requests to a server-provided
* endpoint</li>
* </ul>
*
* <p>
* The message flow follows these steps:
* <ol>
* <li>The client establishes an SSE connection to the server's /sse endpoint</li>
* <li>The server sends an 'endpoint' event containing the URI for sending messages</li>
* </ol>
*
* This implementation uses {@link WebClient} for HTTP communications and supports JSON
* serialization/deserialization of messages.
*
* @author Christian Tzolov
* @see <a href=
* "https://spec.modelcontextprotocol.io/specification/basic/transports/#http-with-sse">MCP
* HTTP with SSE Transport Specification</a>
*/
public class WebFluxSseClientTransport implements ClientMcpTransport {
private static final Logger logger = LoggerFactory.getLogger(WebFluxSseClientTransport.class);
/**
* Event type for JSON-RPC messages received through the SSE connection. The server
* sends messages with this event type to transmit JSON-RPC protocol data.
*/
private static final String MESSAGE_EVENT_TYPE = "message";
/**
* Event type for receiving the message endpoint URI from the server. The server MUST
* send this event when a client connects, providing the URI where the client should
* send its messages via HTTP POST.
*/
private static final String ENDPOINT_EVENT_TYPE = "endpoint";
/**
* Default SSE endpoint path as specified by the MCP transport specification. This
* endpoint is used to establish the SSE connection with the server.
*/
private static final String SSE_ENDPOINT = "/sse";
/**
* Type reference for parsing SSE events containing string data.
*/
private static final ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE = new ParameterizedTypeReference<>() {
};
/**
* WebClient instance for handling both SSE connections and HTTP POST requests. Used
* for establishing the SSE connection and sending outbound messages.
*/
private final WebClient webClient;
/**
* ObjectMapper for serializing outbound messages and deserializing inbound messages.
* Handles conversion between JSON-RPC messages and their string representation.
*/
protected ObjectMapper objectMapper;
/**
* Subscription for the SSE connection handling inbound messages. Used for cleanup
* during transport shutdown.
*/
private Disposable inboundSubscription;
/**
* Flag indicating if the transport is in the process of shutting down. Used to
* prevent new operations during shutdown and handle cleanup gracefully.
*/
private volatile boolean isClosing = false;
/**
* Sink for managing the message endpoint URI provided by the server. Stores the most
* recent endpoint URI and makes it available for outbound message processing.
*/
protected final Sinks.One<String> messageEndpointSink = Sinks.one();
/**
* Constructs a new SseClientTransport with the specified WebClient builder. Uses a
* default ObjectMapper instance for JSON processing.
* @param webClientBuilder the WebClient.Builder to use for creating the WebClient
* instance
* @throws IllegalArgumentException if webClientBuilder is null
*/
public WebFluxSseClientTransport(WebClient.Builder webClientBuilder) {
this(webClientBuilder, new ObjectMapper());
}
/**
* Constructs a new SseClientTransport with the specified WebClient builder and
* ObjectMapper. Initializes both inbound and outbound message processing pipelines.
* @param webClientBuilder the WebClient.Builder to use for creating the WebClient
* instance
* @param objectMapper the ObjectMapper to use for JSON processing
* @throws IllegalArgumentException if either parameter is null
*/
public WebFluxSseClientTransport(WebClient.Builder webClientBuilder, ObjectMapper objectMapper) {
Assert.notNull(objectMapper, "ObjectMapper must not be null");
Assert.notNull(webClientBuilder, "WebClient.Builder must not be null");
this.objectMapper = objectMapper;
this.webClient = webClientBuilder.build();
}
/**
* Establishes a connection to the MCP server using Server-Sent Events (SSE). This
* method initiates the SSE connection and sets up the message processing pipeline.
*
* <p>
* The connection process follows these steps:
* <ol>
* <li>Establishes an SSE connection to the server's /sse endpoint</li>
* <li>Waits for the server to send an 'endpoint' event with the message posting
* URI</li>
* <li>Sets up message handling for incoming JSON-RPC messages</li>
* </ol>
*
* <p>
* The connection is considered established only after receiving the endpoint event
* from the server.
* @param handler a function that processes incoming JSON-RPC messages and returns
* responses
* @return a Mono that completes when the connection is fully established
* @throws McpError if there's an error processing SSE events or if an unrecognized
* event type is received
*/
@Override
public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> handler) {
Flux<ServerSentEvent<String>> events = eventStream();
this.inboundSubscription = events.concatMap(event -> Mono.just(event).<JSONRPCMessage>handle((e, s) -> {
if (ENDPOINT_EVENT_TYPE.equals(event.event())) {
String messageEndpointUri = event.data();
if (messageEndpointSink.tryEmitValue(messageEndpointUri).isSuccess()) {
s.complete();
}
else {
// TODO: clarify with the spec if multiple events can be
// received
s.error(new McpError("Failed to handle SSE endpoint event"));
}
}
else if (MESSAGE_EVENT_TYPE.equals(event.event())) {
try {
JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.objectMapper, event.data());
s.next(message);
}
catch (IOException ioException) {
s.error(ioException);
}
}
else {
s.error(new McpError("Received unrecognized SSE event type: " + event.event()));
}
}).transform(handler)).subscribe();
// The connection is established once the server sends the endpoint event
return messageEndpointSink.asMono().then();
}
/**
* Sends a JSON-RPC message to the server using the endpoint provided during
* connection.
*
* <p>
* Messages are sent via HTTP POST requests to the server-provided endpoint URI. The
* message is serialized to JSON before transmission. If the transport is in the
* process of closing, the message send operation is skipped gracefully.
* @param message the JSON-RPC message to send
* @return a Mono that completes when the message has been sent successfully
* @throws RuntimeException if message serialization fails
*/
@Override
public Mono<Void> sendMessage(JSONRPCMessage message) {
// The messageEndpoint is the endpoint URI to send the messages
// It is provided by the server as part of the endpoint event
return messageEndpointSink.asMono().flatMap(messageEndpointUri -> {
if (isClosing) {
return Mono.empty();
}
try {
String jsonText = this.objectMapper.writeValueAsString(message);
return webClient.post()
.uri(messageEndpointUri)
.contentType(MediaType.APPLICATION_JSON)
.bodyValue(jsonText)
.retrieve()
.toBodilessEntity()
.doOnSuccess(response -> {
logger.debug("Message sent successfully");
})
.doOnError(error -> {
if (!isClosing) {
logger.error("Error sending message: {}", error.getMessage());
}
});
}
catch (IOException e) {
if (!isClosing) {
return Mono.error(new RuntimeException("Failed to serialize message", e));
}
return Mono.empty();
}
}).then(); // TODO: Consider non-200-ok response
}
/**
* Initializes and starts the inbound SSE event processing. Establishes the SSE
* connection and sets up event handling for both message and endpoint events.
* Includes automatic retry logic for handling transient connection failures.
*/
// visible for tests
protected Flux<ServerSentEvent<String>> eventStream() {// @formatter:off
return this.webClient
.get()
.uri(SSE_ENDPOINT)
.accept(MediaType.TEXT_EVENT_STREAM)
.retrieve()
.bodyToFlux(SSE_TYPE)
.retryWhen(Retry.from(retrySignal -> retrySignal.handle(inboundRetryHandler)));
} // @formatter:on
/**
* Retry handler for the inbound SSE stream. Implements the retry logic for handling
* connection failures and other errors.
*/
private BiConsumer<RetrySignal, SynchronousSink<Object>> inboundRetryHandler = (retrySpec, sink) -> {
if (isClosing) {
logger.debug("SSE connection closed during shutdown");
sink.error(retrySpec.failure());
return;
}
if (retrySpec.failure() instanceof IOException) {
logger.debug("Retrying SSE connection after IO error");
sink.next(retrySpec);
return;
}
logger.error("Fatal SSE error, not retrying: {}", retrySpec.failure().getMessage());
sink.error(retrySpec.failure());
};
/**
* Implements graceful shutdown of the transport. Cleans up all resources including
* subscriptions and schedulers. Ensures orderly shutdown of both inbound and outbound
* message processing.
* @return a Mono that completes when shutdown is finished
*/
@Override
public Mono<Void> closeGracefully() { // @formatter:off
return Mono.fromRunnable(() -> {
isClosing = true;
// Dispose of subscriptions
if (inboundSubscription != null) {
inboundSubscription.dispose();
}
})
.then()
.subscribeOn(Schedulers.boundedElastic());
} // @formatter:on
/**
* Unmarshals data from a generic Object into the specified type using the configured
* ObjectMapper.
*
* <p>
* This method is particularly useful when working with JSON-RPC parameters or result
* objects that need to be converted to specific Java types. It leverages Jackson's
* type conversion capabilities to handle complex object structures.
* @param <T> the target type to convert the data into
* @param data the source object to convert
* @param typeRef the TypeReference describing the target type
* @return the unmarshaled object of type T
* @throws IllegalArgumentException if the conversion cannot be performed
*/
@Override
public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
return this.objectMapper.convertValue(data, typeRef);
}
}

410
mcp-transport/mcp-webflux-sse-transport/src/main/java/io/modelcontextprotocol/server/transport/WebFluxSseServerTransport.java Normal file
View file

@ -0,0 +1,410 @@
package io.modelcontextprotocol.server.transport;
import java.io.IOException;
import java.time.Duration;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.ServerMcpTransport;
import io.modelcontextprotocol.util.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
/**
* Server-side implementation of the MCP (Model Context Protocol) HTTP transport using
* Server-Sent Events (SSE). This implementation provides a bidirectional communication
* channel between MCP clients and servers using HTTP POST for client-to-server messages
* and SSE for server-to-client messages.
*
* <p>
* Key features:
* <ul>
* <li>Implements the {@link ServerMcpTransport} interface for MCP server transport
* functionality</li>
* <li>Uses WebFlux for non-blocking request handling and SSE support</li>
* <li>Maintains client sessions for reliable message delivery</li>
* <li>Supports graceful shutdown with session cleanup</li>
* <li>Thread-safe message broadcasting to multiple clients</li>
* </ul>
*
* <p>
* The transport sets up two main endpoints:
* <ul>
* <li>SSE endpoint (/sse) - For establishing SSE connections with clients</li>
* <li>Message endpoint (configurable) - For receiving JSON-RPC messages from clients</li>
* </ul>
*
* <p>
* This implementation is thread-safe and can handle multiple concurrent client
* connections. It uses {@link ConcurrentHashMap} for session management and Reactor's
* {@link Sinks} for thread-safe message broadcasting.
*
* @author Christian Tzolov
* @author Alexandros Pappas
* @see ServerMcpTransport
* @see ServerSentEvent
*/
public class WebFluxSseServerTransport implements ServerMcpTransport {
private static final Logger logger = LoggerFactory.getLogger(WebFluxSseServerTransport.class);
/**
* Event type for JSON-RPC messages sent through the SSE connection.
*/
public static final String MESSAGE_EVENT_TYPE = "message";
/**
* Event type for sending the message endpoint URI to clients.
*/
public static final String ENDPOINT_EVENT_TYPE = "endpoint";
/**
* Default SSE endpoint path as specified by the MCP transport specification.
*/
public static final String DEFAULT_SSE_ENDPOINT = "/sse";
private final ObjectMapper objectMapper;
private final String messageEndpoint;
private final String sseEndpoint;
private final RouterFunction<?> routerFunction;
/**
* Map of active client sessions, keyed by session ID.
*/
private final ConcurrentHashMap<String, ClientSession> sessions = new ConcurrentHashMap<>();
/**
* Flag indicating if the transport is shutting down.
*/
private volatile boolean isClosing = false;
private Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> connectHandler;
/**
* Constructs a new WebFlux SSE server transport instance.
* @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
* of MCP messages. Must not be null.
* @param messageEndpoint The endpoint URI where clients should send their JSON-RPC
* messages. This endpoint will be communicated to clients during SSE connection
* setup. Must not be null.
* @throws IllegalArgumentException if either parameter is null
*/
public WebFluxSseServerTransport(ObjectMapper objectMapper, String messageEndpoint, String sseEndpoint) {
Assert.notNull(objectMapper, "ObjectMapper must not be null");
Assert.notNull(messageEndpoint, "Message endpoint must not be null");
Assert.notNull(sseEndpoint, "SSE endpoint must not be null");
this.objectMapper = objectMapper;
this.messageEndpoint = messageEndpoint;
this.sseEndpoint = sseEndpoint;
this.routerFunction = RouterFunctions.route()
.GET(this.sseEndpoint, this::handleSseConnection)
.POST(this.messageEndpoint, this::handleMessage)
.build();
}
/**
* Constructs a new WebFlux SSE server transport instance with the default SSE
* endpoint.
* @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
* of MCP messages. Must not be null.
* @param messageEndpoint The endpoint URI where clients should send their JSON-RPC
* messages. This endpoint will be communicated to clients during SSE connection
* setup. Must not be null.
* @throws IllegalArgumentException if either parameter is null
*/
public WebFluxSseServerTransport(ObjectMapper objectMapper, String messageEndpoint) {
this(objectMapper, messageEndpoint, DEFAULT_SSE_ENDPOINT);
}
/**
* Configures the message handler for this transport. In the WebFlux SSE
* implementation, this method stores the handler for processing incoming messages but
* doesn't establish any connections since the server accepts connections rather than
* initiating them.
* @param handler A function that processes incoming JSON-RPC messages and returns
* responses. This handler will be called for each message received through the
* message endpoint.
* @return An empty Mono since the server doesn't initiate connections
*/
@Override
public Mono<Void> connect(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
this.connectHandler = handler;
// Server-side transport doesn't initiate connections
return Mono.empty().then();
}
/**
* Broadcasts a JSON-RPC message to all connected clients through their SSE
* connections. The message is serialized to JSON and sent as a server-sent event to
* each active session.
*
* <p>
* The method:
* <ul>
* <li>Serializes the message to JSON</li>
* <li>Creates a server-sent event with the message data</li>
* <li>Attempts to send the event to all active sessions</li>
* <li>Tracks and reports any delivery failures</li>
* </ul>
* @param message The JSON-RPC message to broadcast
* @return A Mono that completes when the message has been sent to all sessions, or
* errors if any session fails to receive the message
*/
@Override
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
if (sessions.isEmpty()) {
logger.debug("No active sessions to broadcast message to");
return Mono.empty();
}
return Mono.<Void>create(sink -> {
try {// @formatter:off
String jsonText = objectMapper.writeValueAsString(message);
ServerSentEvent<Object> event = ServerSentEvent.builder()
.event(MESSAGE_EVENT_TYPE)
.data(jsonText)
.build();
logger.debug("Attempting to broadcast message to {} active sessions", sessions.size());
List<String> failedSessions = sessions.values().stream()
.filter(session -> session.messageSink.tryEmitNext(event).isFailure())
.map(session -> session.id)
.toList();
if (failedSessions.isEmpty()) {
logger.debug("Successfully broadcast message to all sessions");
sink.success();
}
else {
String error = "Failed to broadcast message to sessions: " + String.join(", ", failedSessions);
logger.error(error);
sink.error(new RuntimeException(error));
} // @formatter:on
}
catch (IOException e) {
logger.error("Failed to serialize message: {}", e.getMessage());
sink.error(e);
}
});
}
/**
* Converts data from one type to another using the configured ObjectMapper. This
* method is primarily used for converting between different representations of
* JSON-RPC message data.
* @param <T> The target type to convert to
* @param data The source data to convert
* @param typeRef Type reference describing the target type
* @return The converted data
* @throws IllegalArgumentException if the conversion fails
*/
@Override
public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
return this.objectMapper.convertValue(data, typeRef);
}
/**
* Initiates a graceful shutdown of the transport. This method ensures all active
* sessions are properly closed and cleaned up.
*
* <p>
* The shutdown process:
* <ul>
* <li>Marks the transport as closing to prevent new connections</li>
* <li>Closes each active session</li>
* <li>Removes closed sessions from the sessions map</li>
* <li>Times out after 5 seconds if shutdown takes too long</li>
* </ul>
* @return A Mono that completes when all sessions have been closed
*/
@Override
public Mono<Void> closeGracefully() {
return Mono.fromRunnable(() -> {
isClosing = true;
logger.debug("Initiating graceful shutdown with {} active sessions", sessions.size());
}).then(Mono.when(sessions.values().stream().map(session -> {
String sessionId = session.id;
return Mono.fromRunnable(() -> session.close())
.then(Mono.delay(Duration.ofMillis(100)))
.then(Mono.fromRunnable(() -> sessions.remove(sessionId)));
}).toList()))
.timeout(Duration.ofSeconds(5))
.doOnSuccess(v -> logger.info("Graceful shutdown completed"))
.doOnError(e -> logger.error("Error during graceful shutdown: {}", e.getMessage()));
}
/**
* Returns the WebFlux router function that defines the transport's HTTP endpoints.
* This router function should be integrated into the application's web configuration.
*
* <p>
* The router function defines two endpoints:
* <ul>
* <li>GET {sseEndpoint} - For establishing SSE connections</li>
* <li>POST {messageEndpoint} - For receiving client messages</li>
* </ul>
* @return The configured {@link RouterFunction} for handling HTTP requests
*/
public RouterFunction<?> getRouterFunction() {
return this.routerFunction;
}
/**
* Handles new SSE connection requests from clients. Creates a new session for each
* connection and sets up the SSE event stream.
*
* <p>
* The handler performs the following steps:
* <ul>
* <li>Generates a unique session ID</li>
* <li>Creates a new ClientSession instance</li>
* <li>Sends the message endpoint URI as an initial event</li>
* <li>Sets up message forwarding for the session</li>
* <li>Handles connection cleanup on completion or errors</li>
* </ul>
* @param request The incoming server request
* @return A response with the SSE event stream
*/
private Mono<ServerResponse> handleSseConnection(ServerRequest request) {
if (isClosing) {
return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).bodyValue("Server is shutting down");
}
String sessionId = UUID.randomUUID().toString();
logger.debug("Creating new SSE connection for session: {}", sessionId);
ClientSession session = new ClientSession(sessionId);
this.sessions.put(sessionId, session);
return ServerResponse.ok()
.contentType(MediaType.TEXT_EVENT_STREAM)
.body(Flux.<ServerSentEvent<?>>create(sink -> {
// Send initial endpoint event
logger.debug("Sending initial endpoint event to session: {}", sessionId);
sink.next(ServerSentEvent.builder().event(ENDPOINT_EVENT_TYPE).data(messageEndpoint).build());
// Subscribe to session messages
session.messageSink.asFlux()
.doOnSubscribe(s -> logger.debug("Session {} subscribed to message sink", sessionId))
.doOnComplete(() -> {
logger.debug("Session {} completed", sessionId);
sessions.remove(sessionId);
})
.doOnError(error -> {
logger.error("Error in session {}: {}", sessionId, error.getMessage());
sessions.remove(sessionId);
})
.doOnCancel(() -> {
logger.debug("Session {} cancelled", sessionId);
sessions.remove(sessionId);
})
.subscribe(event -> {
logger.debug("Forwarding event to session {}: {}", sessionId, event);
sink.next(event);
}, sink::error, sink::complete);
sink.onCancel(() -> {
logger.debug("Session {} cancelled", sessionId);
sessions.remove(sessionId);
});
}), ServerSentEvent.class);
}
/**
* Handles incoming JSON-RPC messages from clients. Deserializes the message and
* processes it through the configured message handler.
*
* <p>
* The handler:
* <ul>
* <li>Deserializes the incoming JSON-RPC message</li>
* <li>Passes it through the message handler chain</li>
* <li>Returns appropriate HTTP responses based on processing results</li>
* <li>Handles various error conditions with appropriate error responses</li>
* </ul>
* @param request The incoming server request containing the JSON-RPC message
* @return A response indicating the message processing result
*/
private Mono<ServerResponse> handleMessage(ServerRequest request) {
if (isClosing) {
return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).bodyValue("Server is shutting down");
}
return request.bodyToMono(String.class).flatMap(body -> {
try {
McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(objectMapper, body);
return Mono.just(message)
.transform(this.connectHandler)
.flatMap(response -> ServerResponse.ok().build())
.onErrorResume(error -> {
logger.error("Error processing message: {}", error.getMessage());
return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR)
.bodyValue(new McpError(error.getMessage()));
});
}
catch (IllegalArgumentException | IOException e) {
logger.error("Failed to deserialize message: {}", e.getMessage());
return ServerResponse.badRequest().bodyValue(new McpError("Invalid message format"));
}
});
}
/**
* Represents an active client SSE connection session. Manages the message sink for
* sending events to the client and handles session lifecycle.
*
* <p>
* Each session:
* <ul>
* <li>Has a unique identifier</li>
* <li>Maintains its own message sink for event broadcasting</li>
* <li>Supports clean shutdown through the close method</li>
* </ul>
*/
private static class ClientSession {
private final String id;
private final Sinks.Many<ServerSentEvent<?>> messageSink;
ClientSession(String id) {
this.id = id;
logger.debug("Creating new session: {}", id);
this.messageSink = Sinks.many().replay().latest();
logger.debug("Session {} initialized with replay sink", id);
}
void close() {
logger.debug("Closing session: {}", id);
Sinks.EmitResult result = messageSink.tryEmitComplete();
if (result.isFailure()) {
logger.warn("Failed to complete message sink for session {}: {}", id, result);
}
else {
logger.debug("Successfully completed message sink for session {}", id);
}
}
}
}

462
mcp-transport/mcp-webflux-sse-transport/src/test/java/io/modelcontextprotocol/WebFluxSseIntegrationTests.java Normal file
View file

@ -0,0 +1,462 @@
/*
* Copyright 2024 - 2024 the original author or authors.
*/
package io.modelcontextprotocol;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.client.McpClient;
import io.modelcontextprotocol.client.transport.HttpClientSseClientTransport;
import io.modelcontextprotocol.client.transport.WebFluxSseClientTransport;
import io.modelcontextprotocol.server.McpServer;
import io.modelcontextprotocol.server.McpServerFeatures;
import io.modelcontextprotocol.server.transport.WebFluxSseServerTransport;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.CallToolResult;
import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
import io.modelcontextprotocol.spec.McpSchema.CreateMessageRequest;
import io.modelcontextprotocol.spec.McpSchema.CreateMessageResult;
import io.modelcontextprotocol.spec.McpSchema.InitializeResult;
import io.modelcontextprotocol.spec.McpSchema.Role;
import io.modelcontextprotocol.spec.McpSchema.Root;
import io.modelcontextprotocol.spec.McpSchema.ServerCapabilities;
import io.modelcontextprotocol.spec.McpSchema.Tool;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import reactor.netty.DisposableServer;
import reactor.netty.http.server.HttpServer;
import reactor.test.StepVerifier;
import org.springframework.http.server.reactive.HttpHandler;
import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
import org.springframework.web.client.RestClient;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.server.RouterFunctions;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.awaitility.Awaitility.await;
public class WebFluxSseIntegrationTests {
private static final int PORT = 8182;
private static final String MESSAGE_ENDPOINT = "/mcp/message";
private DisposableServer httpServer;
private WebFluxSseServerTransport mcpServerTransport;
ConcurrentHashMap<String, McpClient.SyncSpec> clientBulders = new ConcurrentHashMap<>();
@BeforeEach
public void before() {
this.mcpServerTransport = new WebFluxSseServerTransport(new ObjectMapper(), MESSAGE_ENDPOINT);
HttpHandler httpHandler = RouterFunctions.toHttpHandler(mcpServerTransport.getRouterFunction());
ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(httpHandler);
this.httpServer = HttpServer.create().port(PORT).handle(adapter).bindNow();
clientBulders.put("httpclient", McpClient.sync(new HttpClientSseClientTransport("http://localhost:" + PORT)));
clientBulders.put("webflux",
McpClient.sync(new WebFluxSseClientTransport(WebClient.builder().baseUrl("http://localhost:" + PORT))));
}
@AfterEach
public void after() {
if (httpServer != null) {
httpServer.disposeNow();
}
}
// ---------------------------------------
// Sampling Tests
// ---------------------------------------
@Test
void testCreateMessageWithoutInitialization() {
var mcpAsyncServer = McpServer.async(mcpServerTransport).serverInfo("test-server", "1.0.0").build();
var messages = List
.of(new McpSchema.SamplingMessage(McpSchema.Role.USER, new McpSchema.TextContent("Test message")));
var modelPrefs = new McpSchema.ModelPreferences(List.of(), 1.0, 1.0, 1.0);
var request = new McpSchema.CreateMessageRequest(messages, modelPrefs, null,
McpSchema.CreateMessageRequest.ContextInclusionStrategy.NONE, null, 100, List.of(), Map.of());
StepVerifier.create(mcpAsyncServer.createMessage(request)).verifyErrorSatisfies(error -> {
assertThat(error).isInstanceOf(McpError.class)
.hasMessage("Client must be initialized. Call the initialize method first!");
});
}
@ParameterizedTest(name = "{0} : {displayName} ")
@ValueSource(strings = { "httpclient", "webflux" })
void testCreateMessageWithoutSamplingCapabilities(String clientType) {
var mcpAsyncServer = McpServer.async(mcpServerTransport).serverInfo("test-server", "1.0.0").build();
var clientBuilder = clientBulders.get(clientType);
var client = clientBuilder.clientInfo(new McpSchema.Implementation("Sample client", "0.0.0")).build();
InitializeResult initResult = client.initialize();
assertThat(initResult).isNotNull();
var messages = List
.of(new McpSchema.SamplingMessage(McpSchema.Role.USER, new McpSchema.TextContent("Test message")));
var modelPrefs = new McpSchema.ModelPreferences(List.of(), 1.0, 1.0, 1.0);
var request = new McpSchema.CreateMessageRequest(messages, modelPrefs, null,
McpSchema.CreateMessageRequest.ContextInclusionStrategy.NONE, null, 100, List.of(), Map.of());
StepVerifier.create(mcpAsyncServer.createMessage(request)).verifyErrorSatisfies(error -> {
assertThat(error).isInstanceOf(McpError.class)
.hasMessage("Client must be configured with sampling capabilities");
});
}
@ParameterizedTest(name = "{0} : {displayName} ")
@ValueSource(strings = { "httpclient", "webflux" })
void testCreateMessageSuccess(String clientType) throws InterruptedException {
var clientBuilder = clientBulders.get(clientType);
var mcpAsyncServer = McpServer.async(mcpServerTransport).serverInfo("test-server", "1.0.0").build();
Function<CreateMessageRequest, CreateMessageResult> samplingHandler = request -> {
assertThat(request.messages()).hasSize(1);
assertThat(request.messages().get(0).content()).isInstanceOf(McpSchema.TextContent.class);
return new CreateMessageResult(Role.USER, new McpSchema.TextContent("Test message"), "MockModelName",
CreateMessageResult.StopReason.STOP_SEQUENCE);
};
var client = clientBuilder.clientInfo(new McpSchema.Implementation("Sample client", "0.0.0"))
.capabilities(ClientCapabilities.builder().sampling().build())
.sampling(samplingHandler)
.build();
InitializeResult initResult = client.initialize();
assertThat(initResult).isNotNull();
var messages = List
.of(new McpSchema.SamplingMessage(McpSchema.Role.USER, new McpSchema.TextContent("Test message")));
var modelPrefs = new McpSchema.ModelPreferences(List.of(), 1.0, 1.0, 1.0);
var request = new McpSchema.CreateMessageRequest(messages, modelPrefs, null,
McpSchema.CreateMessageRequest.ContextInclusionStrategy.NONE, null, 100, List.of(), Map.of());
StepVerifier.create(mcpAsyncServer.createMessage(request)).consumeNextWith(result -> {
assertThat(result).isNotNull();
assertThat(result.role()).isEqualTo(Role.USER);
assertThat(result.content()).isInstanceOf(McpSchema.TextContent.class);
assertThat(((McpSchema.TextContent) result.content()).text()).isEqualTo("Test message");
assertThat(result.model()).isEqualTo("MockModelName");
assertThat(result.stopReason()).isEqualTo(CreateMessageResult.StopReason.STOP_SEQUENCE);
}).verifyComplete();
}
// ---------------------------------------
// Roots Tests
// ---------------------------------------
@ParameterizedTest(name = "{0} : {displayName} ")
@ValueSource(strings = { "httpclient", "webflux" })
void testRootsSuccess(String clientType) {
var clientBuilder = clientBulders.get(clientType);
List<Root> roots = List.of(new Root("uri1://", "root1"), new Root("uri2://", "root2"));
AtomicReference<List<Root>> rootsRef = new AtomicReference<>();
var mcpServer = McpServer.sync(mcpServerTransport)
.rootsChangeConsumer(rootsUpdate -> rootsRef.set(rootsUpdate))
.build();
var mcpClient = clientBuilder.capabilities(ClientCapabilities.builder().roots(true).build())
.roots(roots)
.build();
InitializeResult initResult = mcpClient.initialize();
assertThat(initResult).isNotNull();
assertThat(rootsRef.get()).isNull();
assertThat(mcpServer.listRoots().roots()).containsAll(roots);
mcpClient.rootsListChangedNotification();
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
assertThat(rootsRef.get()).containsAll(roots);
});
// Remove a root
mcpClient.removeRoot(roots.get(0).uri());
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
assertThat(rootsRef.get()).containsAll(List.of(roots.get(1)));
});
// Add a new root
var root3 = new Root("uri3://", "root3");
mcpClient.addRoot(root3);
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
assertThat(rootsRef.get()).containsAll(List.of(roots.get(1), root3));
});
mcpClient.close();
mcpServer.close();
}
@ParameterizedTest(name = "{0} : {displayName} ")
@ValueSource(strings = { "httpclient", "webflux" })
void testRootsWithoutCapability(String clientType) {
var clientBuilder = clientBulders.get(clientType);
var mcpServer = McpServer.sync(mcpServerTransport).rootsChangeConsumer(rootsUpdate -> {
}).build();
// Create client without roots capability
var mcpClient = clientBuilder.capabilities(ClientCapabilities.builder().build()) // No
// roots
// capability
.build();
InitializeResult initResult = mcpClient.initialize();
assertThat(initResult).isNotNull();
// Attempt to list roots should fail
assertThatThrownBy(() -> mcpServer.listRoots().roots()).isInstanceOf(McpError.class)
.hasMessage("Roots not supported");
mcpClient.close();
mcpServer.close();
}
@ParameterizedTest(name = "{0} : {displayName} ")
@ValueSource(strings = { "httpclient", "webflux" })
void testRootsWithEmptyRootsList(String clientType) {
var clientBuilder = clientBulders.get(clientType);
AtomicReference<List<Root>> rootsRef = new AtomicReference<>();
var mcpServer = McpServer.sync(mcpServerTransport)
.rootsChangeConsumer(rootsUpdate -> rootsRef.set(rootsUpdate))
.build();
var mcpClient = clientBuilder.capabilities(ClientCapabilities.builder().roots(true).build())
.roots(List.of()) // Empty roots list
.build();
InitializeResult initResult = mcpClient.initialize();
assertThat(initResult).isNotNull();
mcpClient.rootsListChangedNotification();
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
assertThat(rootsRef.get()).isEmpty();
});
mcpClient.close();
mcpServer.close();
}
@ParameterizedTest(name = "{0} : {displayName} ")
@ValueSource(strings = { "httpclient", "webflux" })
void testRootsWithMultipleConsumers(String clientType) {
var clientBuilder = clientBulders.get(clientType);
List<Root> roots = List.of(new Root("uri1://", "root1"));
AtomicReference<List<Root>> rootsRef1 = new AtomicReference<>();
AtomicReference<List<Root>> rootsRef2 = new AtomicReference<>();
var mcpServer = McpServer.sync(mcpServerTransport)
.rootsChangeConsumer(rootsUpdate -> rootsRef1.set(rootsUpdate))
.rootsChangeConsumer(rootsUpdate -> rootsRef2.set(rootsUpdate))
.build();
var mcpClient = clientBuilder.capabilities(ClientCapabilities.builder().roots(true).build())
.roots(roots)
.build();
InitializeResult initResult = mcpClient.initialize();
assertThat(initResult).isNotNull();
mcpClient.rootsListChangedNotification();
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
assertThat(rootsRef1.get()).containsAll(roots);
assertThat(rootsRef2.get()).containsAll(roots);
});
mcpClient.close();
mcpServer.close();
}
@ParameterizedTest(name = "{0} : {displayName} ")
@ValueSource(strings = { "httpclient", "webflux" })
void testRootsServerCloseWithActiveSubscription(String clientType) {
var clientBuilder = clientBulders.get(clientType);
List<Root> roots = List.of(new Root("uri1://", "root1"));
AtomicReference<List<Root>> rootsRef = new AtomicReference<>();
var mcpServer = McpServer.sync(mcpServerTransport)
.rootsChangeConsumer(rootsUpdate -> rootsRef.set(rootsUpdate))
.build();
var mcpClient = clientBuilder.capabilities(ClientCapabilities.builder().roots(true).build())
.roots(roots)
.build();
InitializeResult initResult = mcpClient.initialize();
assertThat(initResult).isNotNull();
mcpClient.rootsListChangedNotification();
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
assertThat(rootsRef.get()).containsAll(roots);
});
// Close server while subscription is active
mcpServer.close();
// Verify client can handle server closure gracefully
mcpClient.close();
}
// ---------------------------------------
// Tools Tests
// ---------------------------------------
String emptyJsonSchema = """
{
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"properties": {}
}
""";
@ParameterizedTest(name = "{0} : {displayName} ")
@ValueSource(strings = { "httpclient", "webflux" })
void testToolCallSuccess(String clientType) {
var clientBuilder = clientBulders.get(clientType);
var callResponse = new McpSchema.CallToolResult(List.of(new McpSchema.TextContent("CALL RESPONSE")), null);
McpServerFeatures.SyncToolRegistration tool1 = new McpServerFeatures.SyncToolRegistration(
new McpSchema.Tool("tool1", "tool1 description", emptyJsonSchema), request -> {
// perform a blocking call to a remote service
String response = RestClient.create()
.get()
.uri("https://github.com/modelcontextprotocol/specification/blob/main/README.md")
.retrieve()
.body(String.class);
assertThat(response).isNotBlank();
return callResponse;
});
var mcpServer = McpServer.sync(mcpServerTransport)
.capabilities(ServerCapabilities.builder().tools(true).build())
.tools(tool1)
.build();
var mcpClient = clientBuilder.build();
InitializeResult initResult = mcpClient.initialize();
assertThat(initResult).isNotNull();
assertThat(mcpClient.listTools().tools()).contains(tool1.tool());
CallToolResult response = mcpClient.callTool(new McpSchema.CallToolRequest("tool1", Map.of()));
assertThat(response).isNotNull();
assertThat(response).isEqualTo(callResponse);
mcpClient.close();
mcpServer.close();
}
@ParameterizedTest(name = "{0} : {displayName} ")
@ValueSource(strings = { "httpclient", "webflux" })
void testToolListChangeHandlingSuccess(String clientType) {
var clientBuilder = clientBulders.get(clientType);
var callResponse = new McpSchema.CallToolResult(List.of(new McpSchema.TextContent("CALL RESPONSE")), null);
McpServerFeatures.SyncToolRegistration tool1 = new McpServerFeatures.SyncToolRegistration(
new McpSchema.Tool("tool1", "tool1 description", emptyJsonSchema), request -> {
// perform a blocking call to a remote service
String response = RestClient.create()
.get()
.uri("https://github.com/modelcontextprotocol/specification/blob/main/README.md")
.retrieve()
.body(String.class);
assertThat(response).isNotBlank();
return callResponse;
});
var mcpServer = McpServer.sync(mcpServerTransport)
.capabilities(ServerCapabilities.builder().tools(true).build())
.tools(tool1)
.build();
AtomicReference<List<Tool>> rootsRef = new AtomicReference<>();
var mcpClient = clientBuilder.toolsChangeConsumer(toolsUpdate -> {
// perform a blocking call to a remote service
String response = RestClient.create()
.get()
.uri("https://github.com/modelcontextprotocol/specification/blob/main/README.md")
.retrieve()
.body(String.class);
assertThat(response).isNotBlank();
rootsRef.set(toolsUpdate);
}).build();
InitializeResult initResult = mcpClient.initialize();
assertThat(initResult).isNotNull();
assertThat(rootsRef.get()).isNull();
assertThat(mcpClient.listTools().tools()).contains(tool1.tool());
mcpServer.notifyToolsListChanged();
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
assertThat(rootsRef.get()).containsAll(List.of(tool1.tool()));
});
// Remove a tool
mcpServer.removeTool("tool1");
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
assertThat(rootsRef.get()).isEmpty();
});
// Add a new tool
McpServerFeatures.SyncToolRegistration tool2 = new McpServerFeatures.SyncToolRegistration(
new McpSchema.Tool("tool2", "tool2 description", emptyJsonSchema), request -> callResponse);
mcpServer.addTool(tool2);
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
assertThat(rootsRef.get()).containsAll(List.of(tool2.tool()));
});
mcpClient.close();
mcpServer.close();
}
}

51
mcp-transport/mcp-webflux-sse-transport/src/test/java/io/modelcontextprotocol/client/WebFluxSseMcpAsyncClientTests.java Normal file
View file

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

51
mcp-transport/mcp-webflux-sse-transport/src/test/java/io/modelcontextprotocol/client/WebFluxSseMcpSyncClientTests.java Normal file
View file

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

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