feat: Add Spring WebMVC streamable server transport provider (#425)
- Add WebMvcStreamableServerTransportProvider with SSE support for streamable sessions - Support GET, POST, DELETE endpoints for MCP protocol operations - Implement thread-safe SSE operations using ReentrantLock in WebMvcSseServerTransportProvider - Add test infrastructure with AbstractMcpClientServerIntegrationTests - Refactor WebMvcStreamableIntegrationTests to use parameterized tests - Support testing with both HttpClient and WebFlux transports - Add streamable transport tests for both async and sync server modes - Refactor existing WebMVC SSE integration tests to use shared test base - Add error handling improvements in McpStreamableServerSession - Update dependencies: add json-unit-assertj for enhanced JSON testing - Reorganize POM dependencies and add mcp-spring-webflux test dependency - Wrap handler invocations with Mono.defer for lazy evaluation - Wrap consumer, tool, resource, prompt, and completion handler calls with Mono.defer() - Ensures proper lazy evaluation and error handling in reactive streams Related to #72 Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com> Signed-off-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com> Co-authored-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>
This commit is contained in:
Christian Tzolov
GitHub
parent
a8f5a3fb4f
commit
73515346ca
11 changed files with 2410 additions and 1159 deletions
|
|
@ -28,6 +28,12 @@
|
|||
<version>0.11.0-SNAPSHOT</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework</groupId>
|
||||
<artifactId>spring-webmvc</artifactId>
|
||||
<version>${springframework.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>io.modelcontextprotocol.sdk</groupId>
|
||||
<artifactId>mcp-test</artifactId>
|
||||
|
|
@ -35,10 +41,11 @@
|
|||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework</groupId>
|
||||
<artifactId>spring-webmvc</artifactId>
|
||||
<version>${springframework.version}</version>
|
||||
<dependency>
|
||||
<groupId>io.modelcontextprotocol.sdk</groupId>
|
||||
<artifactId>mcp-spring-webflux</artifactId>
|
||||
<version>0.11.0-SNAPSHOT</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import java.io.IOException;
|
|||
import java.time.Duration;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
|
@ -339,6 +340,12 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
|
|||
|
||||
private final SseBuilder sseBuilder;
|
||||
|
||||
/**
|
||||
* Lock to ensure thread-safe access to the SSE builder when sending messages.
|
||||
* This prevents concurrent modifications that could lead to corrupted SSE events.
|
||||
*/
|
||||
private final ReentrantLock sseBuilderLock = new ReentrantLock();
|
||||
|
||||
/**
|
||||
* Creates a new session transport with the specified ID and SSE builder.
|
||||
* @param sessionId The unique identifier for this session
|
||||
|
|
@ -358,6 +365,7 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
|
|||
@Override
|
||||
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
|
||||
return Mono.fromRunnable(() -> {
|
||||
sseBuilderLock.lock();
|
||||
try {
|
||||
String jsonText = objectMapper.writeValueAsString(message);
|
||||
sseBuilder.id(sessionId).event(MESSAGE_EVENT_TYPE).data(jsonText);
|
||||
|
|
@ -367,6 +375,9 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
|
|||
logger.error("Failed to send message to session {}: {}", sessionId, e.getMessage());
|
||||
sseBuilder.error(e);
|
||||
}
|
||||
finally {
|
||||
sseBuilderLock.unlock();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -390,6 +401,7 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
|
|||
public Mono<Void> closeGracefully() {
|
||||
return Mono.fromRunnable(() -> {
|
||||
logger.debug("Closing session transport: {}", sessionId);
|
||||
sseBuilderLock.lock();
|
||||
try {
|
||||
sseBuilder.complete();
|
||||
logger.debug("Successfully completed SSE builder for session {}", sessionId);
|
||||
|
|
@ -397,6 +409,9 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
|
|||
catch (Exception e) {
|
||||
logger.warn("Failed to complete SSE builder for session {}: {}", sessionId, e.getMessage());
|
||||
}
|
||||
finally {
|
||||
sseBuilderLock.unlock();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -405,6 +420,7 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
|
|||
*/
|
||||
@Override
|
||||
public void close() {
|
||||
sseBuilderLock.lock();
|
||||
try {
|
||||
sseBuilder.complete();
|
||||
logger.debug("Successfully completed SSE builder for session {}", sessionId);
|
||||
|
|
@ -412,6 +428,9 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
|
|||
catch (Exception e) {
|
||||
logger.warn("Failed to complete SSE builder for session {}: {}", sessionId, e.getMessage());
|
||||
}
|
||||
finally {
|
||||
sseBuilderLock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,654 @@
|
|||
/*
|
||||
* Copyright 2024-2024 the original author or authors.
|
||||
*/
|
||||
|
||||
package io.modelcontextprotocol.server.transport;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.web.servlet.function.RouterFunction;
|
||||
import org.springframework.web.servlet.function.RouterFunctions;
|
||||
import org.springframework.web.servlet.function.ServerRequest;
|
||||
import org.springframework.web.servlet.function.ServerResponse;
|
||||
import org.springframework.web.servlet.function.ServerResponse.SseBuilder;
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
import io.modelcontextprotocol.server.DefaultMcpTransportContext;
|
||||
import io.modelcontextprotocol.server.McpTransportContext;
|
||||
import io.modelcontextprotocol.server.McpTransportContextExtractor;
|
||||
import io.modelcontextprotocol.spec.HttpHeaders;
|
||||
import io.modelcontextprotocol.spec.McpError;
|
||||
import io.modelcontextprotocol.spec.McpSchema;
|
||||
import io.modelcontextprotocol.spec.McpStreamableServerSession;
|
||||
import io.modelcontextprotocol.spec.McpStreamableServerTransport;
|
||||
import io.modelcontextprotocol.spec.McpStreamableServerTransportProvider;
|
||||
import io.modelcontextprotocol.util.Assert;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
/**
|
||||
* Server-side implementation of the Model Context Protocol (MCP) streamable transport
|
||||
* layer using HTTP with Server-Sent Events (SSE) through Spring WebMVC. This
|
||||
* implementation provides a bridge between synchronous WebMVC operations and reactive
|
||||
* programming patterns to maintain compatibility with the reactive transport interface.
|
||||
*
|
||||
* <p>
|
||||
* This is the non-reactive version of
|
||||
* {@link io.modelcontextprotocol.server.transport.WebFluxStreamableServerTransportProvider}
|
||||
*
|
||||
* @author Christian Tzolov
|
||||
* @author Dariusz Jędrzejczyk
|
||||
* @see McpStreamableServerTransportProvider
|
||||
* @see RouterFunction
|
||||
*/
|
||||
public class WebMvcStreamableServerTransportProvider implements McpStreamableServerTransportProvider {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(WebMvcStreamableServerTransportProvider.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 base URL for the message endpoint.
|
||||
*/
|
||||
public static final String DEFAULT_BASE_URL = "";
|
||||
|
||||
/**
|
||||
* The endpoint URI where clients should send their JSON-RPC messages. Defaults to
|
||||
* "/mcp".
|
||||
*/
|
||||
private final String mcpEndpoint;
|
||||
|
||||
/**
|
||||
* Flag indicating whether DELETE requests are disallowed on the endpoint.
|
||||
*/
|
||||
private final boolean disallowDelete;
|
||||
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
private final RouterFunction<ServerResponse> routerFunction;
|
||||
|
||||
private McpStreamableServerSession.Factory sessionFactory;
|
||||
|
||||
/**
|
||||
* Map of active client sessions, keyed by mcp-session-id.
|
||||
*/
|
||||
private final ConcurrentHashMap<String, McpStreamableServerSession> sessions = new ConcurrentHashMap<>();
|
||||
|
||||
private McpTransportContextExtractor<ServerRequest> contextExtractor;
|
||||
|
||||
// private Function<ServerRequest, McpTransportContext> contextExtractor = req -> new
|
||||
// DefaultMcpTransportContext();
|
||||
|
||||
/**
|
||||
* Flag indicating if the transport is shutting down.
|
||||
*/
|
||||
private volatile boolean isClosing = false;
|
||||
|
||||
/**
|
||||
* Constructs a new WebMvcStreamableServerTransportProvider instance.
|
||||
* @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
|
||||
* of messages.
|
||||
* @param baseUrl The base URL for the message endpoint, used to construct the full
|
||||
* endpoint URL for clients.
|
||||
* @param mcpEndpoint The endpoint URI where clients should send their JSON-RPC
|
||||
* messages via HTTP. This endpoint will handle GET, POST, and DELETE requests.
|
||||
* @param disallowDelete Whether to disallow DELETE requests on the endpoint.
|
||||
* @throws IllegalArgumentException if any parameter is null
|
||||
*/
|
||||
private WebMvcStreamableServerTransportProvider(ObjectMapper objectMapper, String mcpEndpoint,
|
||||
boolean disallowDelete, McpTransportContextExtractor<ServerRequest> contextExtractor) {
|
||||
Assert.notNull(objectMapper, "ObjectMapper must not be null");
|
||||
Assert.notNull(mcpEndpoint, "MCP endpoint must not be null");
|
||||
Assert.notNull(contextExtractor, "McpTransportContextExtractor must not be null");
|
||||
|
||||
this.objectMapper = objectMapper;
|
||||
this.mcpEndpoint = mcpEndpoint;
|
||||
this.disallowDelete = disallowDelete;
|
||||
this.contextExtractor = contextExtractor;
|
||||
this.routerFunction = RouterFunctions.route()
|
||||
.GET(this.mcpEndpoint, this::handleGet)
|
||||
.POST(this.mcpEndpoint, this::handlePost)
|
||||
.DELETE(this.mcpEndpoint, this::handleDelete)
|
||||
.build();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setSessionFactory(McpStreamableServerSession.Factory sessionFactory) {
|
||||
this.sessionFactory = sessionFactory;
|
||||
}
|
||||
|
||||
/**
|
||||
* Broadcasts a notification to all connected clients through their SSE connections.
|
||||
* If any errors occur during sending to a particular client, they are logged but
|
||||
* don't prevent sending to other clients.
|
||||
* @param method The method name for the notification
|
||||
* @param params The parameters for the notification
|
||||
* @return A Mono that completes when the broadcast attempt is finished
|
||||
*/
|
||||
@Override
|
||||
public Mono<Void> notifyClients(String method, Object params) {
|
||||
if (this.sessions.isEmpty()) {
|
||||
logger.debug("No active sessions to broadcast message to");
|
||||
return Mono.empty();
|
||||
}
|
||||
|
||||
logger.debug("Attempting to broadcast message to {} active sessions", this.sessions.size());
|
||||
|
||||
return Mono.fromRunnable(() -> {
|
||||
this.sessions.values().parallelStream().forEach(session -> {
|
||||
try {
|
||||
session.sendNotification(method, params).block();
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to send message to session {}: {}", session.getId(), e.getMessage());
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Initiates a graceful shutdown of the transport.
|
||||
* @return A Mono that completes when all cleanup operations are finished
|
||||
*/
|
||||
@Override
|
||||
public Mono<Void> closeGracefully() {
|
||||
return Mono.fromRunnable(() -> {
|
||||
this.isClosing = true;
|
||||
logger.debug("Initiating graceful shutdown with {} active sessions", this.sessions.size());
|
||||
|
||||
this.sessions.values().parallelStream().forEach(session -> {
|
||||
try {
|
||||
session.closeGracefully().block();
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to close session {}: {}", session.getId(), e.getMessage());
|
||||
}
|
||||
});
|
||||
|
||||
this.sessions.clear();
|
||||
logger.debug("Graceful shutdown completed");
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the RouterFunction that defines the HTTP endpoints for this transport. The
|
||||
* router function handles three endpoints:
|
||||
* <ul>
|
||||
* <li>GET [mcpEndpoint] - For establishing SSE connections and message replay</li>
|
||||
* <li>POST [mcpEndpoint] - For receiving JSON-RPC messages from clients</li>
|
||||
* <li>DELETE [mcpEndpoint] - For session deletion (if enabled)</li>
|
||||
* </ul>
|
||||
* @return The configured RouterFunction for handling HTTP requests
|
||||
*/
|
||||
public RouterFunction<ServerResponse> getRouterFunction() {
|
||||
return this.routerFunction;
|
||||
}
|
||||
|
||||
/**
|
||||
* Setup the listening SSE connections and message replay.
|
||||
* @param request The incoming server request
|
||||
* @return A ServerResponse configured for SSE communication, or an error response
|
||||
*/
|
||||
private ServerResponse handleGet(ServerRequest request) {
|
||||
if (this.isClosing) {
|
||||
return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).body("Server is shutting down");
|
||||
}
|
||||
|
||||
List<MediaType> acceptHeaders = request.headers().asHttpHeaders().getAccept();
|
||||
if (!acceptHeaders.contains(MediaType.TEXT_EVENT_STREAM)) {
|
||||
return ServerResponse.badRequest().body("Invalid Accept header. Expected TEXT_EVENT_STREAM");
|
||||
}
|
||||
|
||||
McpTransportContext transportContext = this.contextExtractor.extract(request, new DefaultMcpTransportContext());
|
||||
|
||||
if (!request.headers().asHttpHeaders().containsKey(HttpHeaders.MCP_SESSION_ID)) {
|
||||
return ServerResponse.badRequest().body("Session ID required in mcp-session-id header");
|
||||
}
|
||||
|
||||
String sessionId = request.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID);
|
||||
McpStreamableServerSession session = this.sessions.get(sessionId);
|
||||
|
||||
if (session == null) {
|
||||
return ServerResponse.notFound().build();
|
||||
}
|
||||
|
||||
logger.debug("Handling GET request for session: {}", sessionId);
|
||||
|
||||
try {
|
||||
return ServerResponse.sse(sseBuilder -> {
|
||||
sseBuilder.onTimeout(() -> {
|
||||
logger.debug("SSE connection timed out for session: {}", sessionId);
|
||||
});
|
||||
|
||||
WebMvcStreamableMcpSessionTransport sessionTransport = new WebMvcStreamableMcpSessionTransport(
|
||||
sessionId, sseBuilder);
|
||||
|
||||
// Check if this is a replay request
|
||||
if (request.headers().asHttpHeaders().containsKey(HttpHeaders.LAST_EVENT_ID)) {
|
||||
String lastId = request.headers().asHttpHeaders().getFirst(HttpHeaders.LAST_EVENT_ID);
|
||||
|
||||
try {
|
||||
session.replay(lastId)
|
||||
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY, transportContext))
|
||||
.toIterable()
|
||||
.forEach(message -> {
|
||||
try {
|
||||
sessionTransport.sendMessage(message)
|
||||
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY, transportContext))
|
||||
.block();
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to replay message: {}", e.getMessage());
|
||||
sseBuilder.error(e);
|
||||
}
|
||||
});
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to replay messages: {}", e.getMessage());
|
||||
sseBuilder.error(e);
|
||||
}
|
||||
}
|
||||
else {
|
||||
// Establish new listening stream
|
||||
McpStreamableServerSession.McpStreamableServerSessionStream listeningStream = session
|
||||
.listeningStream(sessionTransport);
|
||||
|
||||
sseBuilder.onComplete(() -> {
|
||||
logger.debug("SSE connection completed for session: {}", sessionId);
|
||||
listeningStream.close();
|
||||
});
|
||||
}
|
||||
}, Duration.ZERO);
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to handle GET request for session {}: {}", sessionId, e.getMessage());
|
||||
return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles POST requests for incoming JSON-RPC messages from clients.
|
||||
* @param request The incoming server request containing the JSON-RPC message
|
||||
* @return A ServerResponse indicating success or appropriate error status
|
||||
*/
|
||||
private ServerResponse handlePost(ServerRequest request) {
|
||||
if (this.isClosing) {
|
||||
return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).body("Server is shutting down");
|
||||
}
|
||||
|
||||
List<MediaType> acceptHeaders = request.headers().asHttpHeaders().getAccept();
|
||||
if (!acceptHeaders.contains(MediaType.TEXT_EVENT_STREAM)
|
||||
|| !acceptHeaders.contains(MediaType.APPLICATION_JSON)) {
|
||||
return ServerResponse.badRequest()
|
||||
.body(new McpError("Invalid Accept headers. Expected TEXT_EVENT_STREAM and APPLICATION_JSON"));
|
||||
}
|
||||
|
||||
McpTransportContext transportContext = this.contextExtractor.extract(request, new DefaultMcpTransportContext());
|
||||
|
||||
try {
|
||||
String body = request.body(String.class);
|
||||
McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(objectMapper, body);
|
||||
|
||||
// Handle initialization request
|
||||
if (message instanceof McpSchema.JSONRPCRequest jsonrpcRequest
|
||||
&& jsonrpcRequest.method().equals(McpSchema.METHOD_INITIALIZE)) {
|
||||
McpSchema.InitializeRequest initializeRequest = objectMapper.convertValue(jsonrpcRequest.params(),
|
||||
new TypeReference<McpSchema.InitializeRequest>() {
|
||||
});
|
||||
McpStreamableServerSession.McpStreamableServerSessionInit init = this.sessionFactory
|
||||
.startSession(initializeRequest);
|
||||
this.sessions.put(init.session().getId(), init.session());
|
||||
|
||||
try {
|
||||
McpSchema.InitializeResult initResult = init.initResult().block();
|
||||
|
||||
return ServerResponse.ok()
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.header(HttpHeaders.MCP_SESSION_ID, init.session().getId())
|
||||
.body(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, jsonrpcRequest.id(), initResult,
|
||||
null));
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to initialize session: {}", e.getMessage());
|
||||
return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).body(new McpError(e.getMessage()));
|
||||
}
|
||||
}
|
||||
|
||||
// Handle other messages that require a session
|
||||
if (!request.headers().asHttpHeaders().containsKey(HttpHeaders.MCP_SESSION_ID)) {
|
||||
return ServerResponse.badRequest().body(new McpError("Session ID missing"));
|
||||
}
|
||||
|
||||
String sessionId = request.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID);
|
||||
McpStreamableServerSession session = this.sessions.get(sessionId);
|
||||
|
||||
if (session == null) {
|
||||
return ServerResponse.status(HttpStatus.NOT_FOUND)
|
||||
.body(new McpError("Session not found: " + sessionId));
|
||||
}
|
||||
|
||||
if (message instanceof McpSchema.JSONRPCResponse jsonrpcResponse) {
|
||||
session.accept(jsonrpcResponse)
|
||||
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY, transportContext))
|
||||
.block();
|
||||
return ServerResponse.accepted().build();
|
||||
}
|
||||
else if (message instanceof McpSchema.JSONRPCNotification jsonrpcNotification) {
|
||||
session.accept(jsonrpcNotification)
|
||||
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY, transportContext))
|
||||
.block();
|
||||
return ServerResponse.accepted().build();
|
||||
}
|
||||
else if (message instanceof McpSchema.JSONRPCRequest jsonrpcRequest) {
|
||||
// For streaming responses, we need to return SSE
|
||||
return ServerResponse.sse(sseBuilder -> {
|
||||
sseBuilder.onComplete(() -> {
|
||||
logger.debug("Request response stream completed for session: {}", sessionId);
|
||||
});
|
||||
sseBuilder.onTimeout(() -> {
|
||||
logger.debug("Request response stream timed out for session: {}", sessionId);
|
||||
});
|
||||
|
||||
WebMvcStreamableMcpSessionTransport sessionTransport = new WebMvcStreamableMcpSessionTransport(
|
||||
sessionId, sseBuilder);
|
||||
|
||||
try {
|
||||
session.responseStream(jsonrpcRequest, sessionTransport)
|
||||
.contextWrite(ctx -> ctx.put(McpTransportContext.KEY, transportContext))
|
||||
.block();
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to handle request stream: {}", e.getMessage());
|
||||
sseBuilder.error(e);
|
||||
}
|
||||
}, Duration.ZERO);
|
||||
}
|
||||
else {
|
||||
return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR)
|
||||
.body(new McpError("Unknown message type"));
|
||||
}
|
||||
}
|
||||
catch (IllegalArgumentException | IOException e) {
|
||||
logger.error("Failed to deserialize message: {}", e.getMessage());
|
||||
return ServerResponse.badRequest().body(new McpError("Invalid message format"));
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Error handling message: {}", e.getMessage());
|
||||
return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).body(new McpError(e.getMessage()));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles DELETE requests for session deletion.
|
||||
* @param request The incoming server request
|
||||
* @return A ServerResponse indicating success or appropriate error status
|
||||
*/
|
||||
private ServerResponse handleDelete(ServerRequest request) {
|
||||
if (this.isClosing) {
|
||||
return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).body("Server is shutting down");
|
||||
}
|
||||
|
||||
if (this.disallowDelete) {
|
||||
return ServerResponse.status(HttpStatus.METHOD_NOT_ALLOWED).build();
|
||||
}
|
||||
|
||||
McpTransportContext transportContext = this.contextExtractor.extract(request, new DefaultMcpTransportContext());
|
||||
|
||||
if (!request.headers().asHttpHeaders().containsKey(HttpHeaders.MCP_SESSION_ID)) {
|
||||
return ServerResponse.badRequest().body("Session ID required in mcp-session-id header");
|
||||
}
|
||||
|
||||
String sessionId = request.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID);
|
||||
McpStreamableServerSession session = this.sessions.get(sessionId);
|
||||
|
||||
if (session == null) {
|
||||
return ServerResponse.notFound().build();
|
||||
}
|
||||
|
||||
try {
|
||||
session.delete().contextWrite(ctx -> ctx.put(McpTransportContext.KEY, transportContext)).block();
|
||||
this.sessions.remove(sessionId);
|
||||
return ServerResponse.ok().build();
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to delete session {}: {}", sessionId, e.getMessage());
|
||||
return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).body(new McpError(e.getMessage()));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Implementation of McpStreamableServerTransport for WebMVC SSE sessions. This class
|
||||
* handles the transport-level communication for a specific client session.
|
||||
*
|
||||
* <p>
|
||||
* This class is thread-safe and uses a ReentrantLock to synchronize access to the
|
||||
* underlying SSE builder to prevent race conditions when multiple threads attempt to
|
||||
* send messages concurrently.
|
||||
*/
|
||||
private class WebMvcStreamableMcpSessionTransport implements McpStreamableServerTransport {
|
||||
|
||||
private final String sessionId;
|
||||
|
||||
private final SseBuilder sseBuilder;
|
||||
|
||||
private final ReentrantLock lock = new ReentrantLock();
|
||||
|
||||
private volatile boolean closed = false;
|
||||
|
||||
/**
|
||||
* Creates a new session transport with the specified ID and SSE builder.
|
||||
* @param sessionId The unique identifier for this session
|
||||
* @param sseBuilder The SSE builder for sending server events to the client
|
||||
*/
|
||||
WebMvcStreamableMcpSessionTransport(String sessionId, SseBuilder sseBuilder) {
|
||||
this.sessionId = sessionId;
|
||||
this.sseBuilder = sseBuilder;
|
||||
logger.debug("Streamable session transport {} initialized with SSE builder", sessionId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a JSON-RPC message to the client through the SSE connection.
|
||||
* @param message The JSON-RPC message to send
|
||||
* @return A Mono that completes when the message has been sent
|
||||
*/
|
||||
@Override
|
||||
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
|
||||
return sendMessage(message, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a JSON-RPC message to the client through the SSE connection with a
|
||||
* specific message ID.
|
||||
* @param message The JSON-RPC message to send
|
||||
* @param messageId The message ID for SSE event identification
|
||||
* @return A Mono that completes when the message has been sent
|
||||
*/
|
||||
@Override
|
||||
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message, String messageId) {
|
||||
return Mono.fromRunnable(() -> {
|
||||
if (this.closed) {
|
||||
logger.debug("Attempted to send message to closed session: {}", this.sessionId);
|
||||
return;
|
||||
}
|
||||
|
||||
this.lock.lock();
|
||||
try {
|
||||
if (this.closed) {
|
||||
logger.debug("Session {} was closed during message send attempt", this.sessionId);
|
||||
return;
|
||||
}
|
||||
|
||||
String jsonText = objectMapper.writeValueAsString(message);
|
||||
this.sseBuilder.id(messageId != null ? messageId : this.sessionId)
|
||||
.event(MESSAGE_EVENT_TYPE)
|
||||
.data(jsonText);
|
||||
logger.debug("Message sent to session {} with ID {}", this.sessionId, messageId);
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to send message to session {}: {}", this.sessionId, e.getMessage());
|
||||
try {
|
||||
this.sseBuilder.error(e);
|
||||
}
|
||||
catch (Exception errorException) {
|
||||
logger.error("Failed to send error to SSE builder for session {}: {}", this.sessionId,
|
||||
errorException.getMessage());
|
||||
}
|
||||
}
|
||||
finally {
|
||||
this.lock.unlock();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Converts data from one type to another using the configured ObjectMapper.
|
||||
* @param data The source data object to convert
|
||||
* @param typeRef The target type reference
|
||||
* @return The converted object of type T
|
||||
* @param <T> The target type
|
||||
*/
|
||||
@Override
|
||||
public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
|
||||
return objectMapper.convertValue(data, typeRef);
|
||||
}
|
||||
|
||||
/**
|
||||
* Initiates a graceful shutdown of the transport.
|
||||
* @return A Mono that completes when the shutdown is complete
|
||||
*/
|
||||
@Override
|
||||
public Mono<Void> closeGracefully() {
|
||||
return Mono.fromRunnable(() -> {
|
||||
WebMvcStreamableMcpSessionTransport.this.close();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Closes the transport immediately.
|
||||
*/
|
||||
@Override
|
||||
public void close() {
|
||||
this.lock.lock();
|
||||
try {
|
||||
if (this.closed) {
|
||||
logger.debug("Session transport {} already closed", this.sessionId);
|
||||
return;
|
||||
}
|
||||
|
||||
this.closed = true;
|
||||
|
||||
this.sseBuilder.complete();
|
||||
logger.debug("Successfully completed SSE builder for session {}", sessionId);
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.warn("Failed to complete SSE builder for session {}: {}", sessionId, e.getMessage());
|
||||
}
|
||||
finally {
|
||||
this.lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static Builder builder() {
|
||||
return new Builder();
|
||||
}
|
||||
|
||||
/**
|
||||
* Builder for creating instances of {@link WebMvcStreamableServerTransportProvider}.
|
||||
*/
|
||||
public static class Builder {
|
||||
|
||||
private ObjectMapper objectMapper;
|
||||
|
||||
private String mcpEndpoint = "/mcp";
|
||||
|
||||
private boolean disallowDelete = false;
|
||||
|
||||
private McpTransportContextExtractor<ServerRequest> contextExtractor = (serverRequest, context) -> context;
|
||||
|
||||
/**
|
||||
* Sets the ObjectMapper to use for JSON serialization/deserialization of MCP
|
||||
* messages.
|
||||
* @param objectMapper The ObjectMapper instance. Must not be null.
|
||||
* @return this builder instance
|
||||
* @throws IllegalArgumentException if objectMapper is null
|
||||
*/
|
||||
public Builder objectMapper(ObjectMapper objectMapper) {
|
||||
Assert.notNull(objectMapper, "ObjectMapper must not be null");
|
||||
this.objectMapper = objectMapper;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the endpoint URI where clients should send their JSON-RPC messages.
|
||||
* @param mcpEndpoint The MCP endpoint URI. Must not be null.
|
||||
* @return this builder instance
|
||||
* @throws IllegalArgumentException if mcpEndpoint is null
|
||||
*/
|
||||
public Builder mcpEndpoint(String mcpEndpoint) {
|
||||
Assert.notNull(mcpEndpoint, "MCP endpoint must not be null");
|
||||
this.mcpEndpoint = mcpEndpoint;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets whether to disallow DELETE requests on the endpoint.
|
||||
* @param disallowDelete true to disallow DELETE requests, false otherwise
|
||||
* @return this builder instance
|
||||
*/
|
||||
public Builder disallowDelete(boolean disallowDelete) {
|
||||
this.disallowDelete = disallowDelete;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the context extractor that allows providing the MCP feature
|
||||
* implementations to inspect HTTP transport level metadata that was present at
|
||||
* HTTP request processing time. This allows to extract custom headers and other
|
||||
* useful data for use during execution later on in the process.
|
||||
* @param contextExtractor The contextExtractor to fill in a
|
||||
* {@link McpTransportContext}.
|
||||
* @return this builder instance
|
||||
* @throws IllegalArgumentException if contextExtractor is null
|
||||
*/
|
||||
public Builder contextExtractor(McpTransportContextExtractor<ServerRequest> contextExtractor) {
|
||||
Assert.notNull(contextExtractor, "contextExtractor must not be null");
|
||||
this.contextExtractor = contextExtractor;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds a new instance of {@link WebMvcStreamableServerTransportProvider} with
|
||||
* the configured settings.
|
||||
* @return A new WebMvcStreamableServerTransportProvider instance
|
||||
* @throws IllegalStateException if required parameters are not set
|
||||
*/
|
||||
public WebMvcStreamableServerTransportProvider build() {
|
||||
Assert.notNull(this.objectMapper, "ObjectMapper must be set");
|
||||
Assert.notNull(this.mcpEndpoint, "MCP endpoint must be set");
|
||||
|
||||
return new WebMvcStreamableServerTransportProvider(this.objectMapper, this.mcpEndpoint, this.disallowDelete,
|
||||
this.contextExtractor);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,121 @@
|
|||
/*
|
||||
* Copyright 2024-2024 the original author or authors.
|
||||
*/
|
||||
|
||||
package io.modelcontextprotocol.server;
|
||||
|
||||
import org.apache.catalina.Context;
|
||||
import org.apache.catalina.LifecycleException;
|
||||
import org.apache.catalina.startup.Tomcat;
|
||||
import org.junit.jupiter.api.Timeout;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.web.context.support.AnnotationConfigWebApplicationContext;
|
||||
import org.springframework.web.servlet.DispatcherServlet;
|
||||
import org.springframework.web.servlet.config.annotation.EnableWebMvc;
|
||||
import org.springframework.web.servlet.function.RouterFunction;
|
||||
import org.springframework.web.servlet.function.ServerResponse;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
import io.modelcontextprotocol.server.transport.WebMvcStreamableServerTransportProvider;
|
||||
import io.modelcontextprotocol.spec.McpStreamableServerTransportProvider;
|
||||
import reactor.netty.DisposableServer;
|
||||
|
||||
/**
|
||||
* Tests for {@link McpAsyncServer} using {@link WebFluxSseServerTransportProvider}.
|
||||
*
|
||||
* @author Christian Tzolov
|
||||
*/
|
||||
@Timeout(15) // Giving extra time beyond the client timeout
|
||||
class WebMcpStreamableAsyncServerTransportTests extends AbstractMcpAsyncServerTests {
|
||||
|
||||
private static final int PORT = TestUtil.findAvailablePort();
|
||||
|
||||
private static final String MCP_ENDPOINT = "/mcp";
|
||||
|
||||
private DisposableServer httpServer;
|
||||
|
||||
private AnnotationConfigWebApplicationContext appContext;
|
||||
|
||||
private Tomcat tomcat;
|
||||
|
||||
private McpStreamableServerTransportProvider transportProvider;
|
||||
|
||||
@Configuration
|
||||
@EnableWebMvc
|
||||
static class TestConfig {
|
||||
|
||||
@Bean
|
||||
public WebMvcStreamableServerTransportProvider webMvcSseServerTransportProvider() {
|
||||
return WebMvcStreamableServerTransportProvider.builder()
|
||||
.objectMapper(new ObjectMapper())
|
||||
.mcpEndpoint(MCP_ENDPOINT)
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RouterFunction<ServerResponse> routerFunction(
|
||||
WebMvcStreamableServerTransportProvider transportProvider) {
|
||||
return transportProvider.getRouterFunction();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private McpStreamableServerTransportProvider createMcpTransportProvider() {
|
||||
// Set up Tomcat first
|
||||
tomcat = new Tomcat();
|
||||
tomcat.setPort(PORT);
|
||||
|
||||
// Set Tomcat base directory to java.io.tmpdir to avoid permission issues
|
||||
String baseDir = System.getProperty("java.io.tmpdir");
|
||||
tomcat.setBaseDir(baseDir);
|
||||
|
||||
// Use the same directory for document base
|
||||
Context context = tomcat.addContext("", baseDir);
|
||||
|
||||
// Create and configure Spring WebMvc context
|
||||
appContext = new AnnotationConfigWebApplicationContext();
|
||||
appContext.register(TestConfig.class);
|
||||
appContext.setServletContext(context.getServletContext());
|
||||
appContext.refresh();
|
||||
|
||||
// Get the transport from Spring context
|
||||
transportProvider = appContext.getBean(McpStreamableServerTransportProvider.class);
|
||||
|
||||
// Create DispatcherServlet with our Spring context
|
||||
DispatcherServlet dispatcherServlet = new DispatcherServlet(appContext);
|
||||
|
||||
// Add servlet to Tomcat and get the wrapper
|
||||
var wrapper = Tomcat.addServlet(context, "dispatcherServlet", dispatcherServlet);
|
||||
wrapper.setLoadOnStartup(1);
|
||||
context.addServletMappingDecoded("/*", "dispatcherServlet");
|
||||
|
||||
try {
|
||||
tomcat.start();
|
||||
tomcat.getConnector(); // Create and start the connector
|
||||
}
|
||||
catch (LifecycleException e) {
|
||||
throw new RuntimeException("Failed to start Tomcat", e);
|
||||
}
|
||||
|
||||
return transportProvider;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected McpServer.AsyncSpecification<?> prepareAsyncServerBuilder() {
|
||||
return McpServer.async(createMcpTransportProvider());
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void onStart() {
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void onClose() {
|
||||
if (httpServer != null) {
|
||||
httpServer.disposeNow();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,121 @@
|
|||
/*
|
||||
* Copyright 2024-2024 the original author or authors.
|
||||
*/
|
||||
|
||||
package io.modelcontextprotocol.server;
|
||||
|
||||
import org.apache.catalina.Context;
|
||||
import org.apache.catalina.LifecycleException;
|
||||
import org.apache.catalina.startup.Tomcat;
|
||||
import org.junit.jupiter.api.Timeout;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.web.context.support.AnnotationConfigWebApplicationContext;
|
||||
import org.springframework.web.servlet.DispatcherServlet;
|
||||
import org.springframework.web.servlet.config.annotation.EnableWebMvc;
|
||||
import org.springframework.web.servlet.function.RouterFunction;
|
||||
import org.springframework.web.servlet.function.ServerResponse;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
import io.modelcontextprotocol.server.transport.WebMvcStreamableServerTransportProvider;
|
||||
import io.modelcontextprotocol.spec.McpStreamableServerTransportProvider;
|
||||
import reactor.netty.DisposableServer;
|
||||
|
||||
/**
|
||||
* Tests for {@link McpAsyncServer} using {@link WebFluxSseServerTransportProvider}.
|
||||
*
|
||||
* @author Christian Tzolov
|
||||
*/
|
||||
@Timeout(15) // Giving extra time beyond the client timeout
|
||||
class WebMcpStreamableSyncServerTransportTests extends AbstractMcpSyncServerTests {
|
||||
|
||||
private static final int PORT = TestUtil.findAvailablePort();
|
||||
|
||||
private static final String MCP_ENDPOINT = "/mcp";
|
||||
|
||||
private DisposableServer httpServer;
|
||||
|
||||
private AnnotationConfigWebApplicationContext appContext;
|
||||
|
||||
private Tomcat tomcat;
|
||||
|
||||
private McpStreamableServerTransportProvider transportProvider;
|
||||
|
||||
@Configuration
|
||||
@EnableWebMvc
|
||||
static class TestConfig {
|
||||
|
||||
@Bean
|
||||
public WebMvcStreamableServerTransportProvider webMvcSseServerTransportProvider() {
|
||||
return WebMvcStreamableServerTransportProvider.builder()
|
||||
.objectMapper(new ObjectMapper())
|
||||
.mcpEndpoint(MCP_ENDPOINT)
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RouterFunction<ServerResponse> routerFunction(
|
||||
WebMvcStreamableServerTransportProvider transportProvider) {
|
||||
return transportProvider.getRouterFunction();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private McpStreamableServerTransportProvider createMcpTransportProvider() {
|
||||
// Set up Tomcat first
|
||||
tomcat = new Tomcat();
|
||||
tomcat.setPort(PORT);
|
||||
|
||||
// Set Tomcat base directory to java.io.tmpdir to avoid permission issues
|
||||
String baseDir = System.getProperty("java.io.tmpdir");
|
||||
tomcat.setBaseDir(baseDir);
|
||||
|
||||
// Use the same directory for document base
|
||||
Context context = tomcat.addContext("", baseDir);
|
||||
|
||||
// Create and configure Spring WebMvc context
|
||||
appContext = new AnnotationConfigWebApplicationContext();
|
||||
appContext.register(TestConfig.class);
|
||||
appContext.setServletContext(context.getServletContext());
|
||||
appContext.refresh();
|
||||
|
||||
// Get the transport from Spring context
|
||||
transportProvider = appContext.getBean(McpStreamableServerTransportProvider.class);
|
||||
|
||||
// Create DispatcherServlet with our Spring context
|
||||
DispatcherServlet dispatcherServlet = new DispatcherServlet(appContext);
|
||||
|
||||
// Add servlet to Tomcat and get the wrapper
|
||||
var wrapper = Tomcat.addServlet(context, "dispatcherServlet", dispatcherServlet);
|
||||
wrapper.setLoadOnStartup(1);
|
||||
context.addServletMappingDecoded("/*", "dispatcherServlet");
|
||||
|
||||
try {
|
||||
tomcat.start();
|
||||
tomcat.getConnector(); // Create and start the connector
|
||||
}
|
||||
catch (LifecycleException e) {
|
||||
throw new RuntimeException("Failed to start Tomcat", e);
|
||||
}
|
||||
|
||||
return transportProvider;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected McpServer.SyncSpecification<?> prepareSyncServerBuilder() {
|
||||
return McpServer.sync(createMcpTransportProvider());
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void onStart() {
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void onClose() {
|
||||
if (httpServer != null) {
|
||||
httpServer.disposeNow();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
File diff suppressed because it is too large
Load diff
|
|
@ -0,0 +1,165 @@
|
|||
/*
|
||||
* Copyright 2024 - 2024 the original author or authors.
|
||||
*/
|
||||
package io.modelcontextprotocol.server;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
import org.apache.catalina.LifecycleException;
|
||||
import org.apache.catalina.LifecycleState;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.web.reactive.function.client.WebClient;
|
||||
import org.springframework.web.servlet.config.annotation.EnableWebMvc;
|
||||
import org.springframework.web.servlet.function.RouterFunction;
|
||||
import org.springframework.web.servlet.function.ServerResponse;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
import io.modelcontextprotocol.AbstractMcpClientServerIntegrationTests;
|
||||
import io.modelcontextprotocol.client.McpClient;
|
||||
import io.modelcontextprotocol.client.transport.HttpClientStreamableHttpTransport;
|
||||
import io.modelcontextprotocol.client.transport.WebClientStreamableHttpTransport;
|
||||
import io.modelcontextprotocol.server.McpServer.AsyncSpecification;
|
||||
import io.modelcontextprotocol.server.McpServer.SyncSpecification;
|
||||
import io.modelcontextprotocol.server.transport.WebMvcStreamableServerTransportProvider;
|
||||
import io.modelcontextprotocol.spec.McpSchema;
|
||||
import reactor.core.scheduler.Schedulers;
|
||||
|
||||
class WebMvcStreamableIntegrationTests extends AbstractMcpClientServerIntegrationTests {
|
||||
|
||||
private static final int PORT = TestUtil.findAvailablePort();
|
||||
|
||||
private static final String MESSAGE_ENDPOINT = "/mcp/message";
|
||||
|
||||
private WebMvcStreamableServerTransportProvider mcpServerTransportProvider;
|
||||
|
||||
@Configuration
|
||||
@EnableWebMvc
|
||||
static class TestConfig {
|
||||
|
||||
@Bean
|
||||
public WebMvcStreamableServerTransportProvider webMvcStreamableServerTransportProvider() {
|
||||
return WebMvcStreamableServerTransportProvider.builder()
|
||||
.objectMapper(new ObjectMapper())
|
||||
.mcpEndpoint(MESSAGE_ENDPOINT)
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RouterFunction<ServerResponse> routerFunction(
|
||||
WebMvcStreamableServerTransportProvider transportProvider) {
|
||||
return transportProvider.getRouterFunction();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private TomcatTestUtil.TomcatServer tomcatServer;
|
||||
|
||||
@BeforeEach
|
||||
public void before() {
|
||||
|
||||
tomcatServer = TomcatTestUtil.createTomcatServer("", PORT, TestConfig.class);
|
||||
|
||||
try {
|
||||
tomcatServer.tomcat().start();
|
||||
assertThat(tomcatServer.tomcat().getServer().getState()).isEqualTo(LifecycleState.STARTED);
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new RuntimeException("Failed to start Tomcat", e);
|
||||
}
|
||||
|
||||
clientBuilders
|
||||
.put("httpclient",
|
||||
McpClient.sync(HttpClientStreamableHttpTransport.builder("http://localhost:" + PORT)
|
||||
.endpoint(MESSAGE_ENDPOINT)
|
||||
.build()).initializationTimeout(Duration.ofHours(10)).requestTimeout(Duration.ofHours(10)));
|
||||
|
||||
clientBuilders.put("webflux",
|
||||
McpClient.sync(WebClientStreamableHttpTransport
|
||||
.builder(WebClient.builder().baseUrl("http://localhost:" + PORT))
|
||||
.endpoint(MESSAGE_ENDPOINT)
|
||||
.build()));
|
||||
|
||||
// Get the transport from Spring context
|
||||
this.mcpServerTransportProvider = tomcatServer.appContext()
|
||||
.getBean(WebMvcStreamableServerTransportProvider.class);
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
protected AsyncSpecification<?> prepareAsyncServerBuilder() {
|
||||
return McpServer.async(this.mcpServerTransportProvider);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected SyncSpecification<?> prepareSyncServerBuilder() {
|
||||
return McpServer.sync(this.mcpServerTransportProvider);
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
public void after() {
|
||||
reactor.netty.http.HttpResources.disposeLoopsAndConnections();
|
||||
if (mcpServerTransportProvider != null) {
|
||||
mcpServerTransportProvider.closeGracefully().block();
|
||||
}
|
||||
Schedulers.shutdownNow();
|
||||
if (tomcatServer.appContext() != null) {
|
||||
tomcatServer.appContext().close();
|
||||
}
|
||||
if (tomcatServer.tomcat() != null) {
|
||||
try {
|
||||
tomcatServer.tomcat().stop();
|
||||
tomcatServer.tomcat().destroy();
|
||||
}
|
||||
catch (LifecycleException e) {
|
||||
throw new RuntimeException("Failed to stop Tomcat", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ParameterizedTest(name = "{0} : {displayName} ")
|
||||
@ValueSource(strings = { "httpclient", "webflux" })
|
||||
void simple(String clientType) {
|
||||
|
||||
var clientBuilder = clientBuilders.get(clientType);
|
||||
|
||||
var server = McpServer.async(mcpServerTransportProvider)
|
||||
.serverInfo("test-server", "1.0.0")
|
||||
.requestTimeout(Duration.ofSeconds(1000))
|
||||
.build();
|
||||
|
||||
try (
|
||||
// Create client without sampling capabilities
|
||||
var client = clientBuilder.clientInfo(new McpSchema.Implementation("Sample " + "client", "0.0.0"))
|
||||
.requestTimeout(Duration.ofSeconds(1000))
|
||||
.build()) {
|
||||
|
||||
assertThat(client.initialize()).isNotNull();
|
||||
|
||||
}
|
||||
server.closeGracefully();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void prepareClients(int port, String mcpEndpoint) {
|
||||
|
||||
clientBuilders.put("httpclient", McpClient
|
||||
.sync(HttpClientStreamableHttpTransport.builder("http://localhost:" + port).endpoint(mcpEndpoint).build())
|
||||
.initializationTimeout(Duration.ofHours(10))
|
||||
.requestTimeout(Duration.ofHours(10)));
|
||||
|
||||
clientBuilders.put("webflux",
|
||||
McpClient.sync(WebClientStreamableHttpTransport
|
||||
.builder(WebClient.builder().baseUrl("http://localhost:" + port))
|
||||
.endpoint(mcpEndpoint)
|
||||
.build()));
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -91,6 +91,13 @@
|
|||
<version>${logback.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>net.javacrumbs.json-unit</groupId>
|
||||
<artifactId>json-unit-assertj</artifactId>
|
||||
<version>${json-unit-assertj.version}</version>
|
||||
</dependency>
|
||||
|
||||
|
||||
</dependencies>
|
||||
|
||||
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -296,7 +296,7 @@ public class McpAsyncServer {
|
|||
List<BiFunction<McpAsyncServerExchange, List<McpSchema.Root>, Mono<Void>>> rootsChangeConsumers) {
|
||||
return (exchange, params) -> exchange.listRoots()
|
||||
.flatMap(listRootsResult -> Flux.fromIterable(rootsChangeConsumers)
|
||||
.flatMap(consumer -> consumer.apply(exchange, listRootsResult.roots()))
|
||||
.flatMap(consumer -> Mono.defer(() -> consumer.apply(exchange, listRootsResult.roots())))
|
||||
.onErrorResume(error -> {
|
||||
logger.error("Error handling roots list change notification", error);
|
||||
return Mono.empty();
|
||||
|
|
@ -506,7 +506,7 @@ public class McpAsyncServer {
|
|||
return Mono.error(new McpError("Tool not found: " + callToolRequest.name()));
|
||||
}
|
||||
|
||||
return toolSpecification.map(tool -> tool.callHandler().apply(exchange, callToolRequest))
|
||||
return toolSpecification.map(tool -> Mono.defer(() -> tool.callHandler().apply(exchange, callToolRequest)))
|
||||
.orElse(Mono.error(new McpError("Tool not found: " + callToolRequest.name())));
|
||||
};
|
||||
}
|
||||
|
|
@ -634,7 +634,7 @@ public class McpAsyncServer {
|
|||
.findFirst()
|
||||
.orElseThrow(() -> new McpError("Resource not found: " + resourceUri));
|
||||
|
||||
return specification.readHandler().apply(exchange, resourceRequest);
|
||||
return Mono.defer(() -> specification.readHandler().apply(exchange, resourceRequest));
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -740,7 +740,7 @@ public class McpAsyncServer {
|
|||
return Mono.error(new McpError("Prompt not found: " + promptRequest.name()));
|
||||
}
|
||||
|
||||
return specification.promptHandler().apply(exchange, promptRequest);
|
||||
return Mono.defer(() -> specification.promptHandler().apply(exchange, promptRequest));
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -845,7 +845,7 @@ public class McpAsyncServer {
|
|||
return Mono.error(new McpError("AsyncCompletionSpecification not found: " + request.ref()));
|
||||
}
|
||||
|
||||
return specification.completionHandler().apply(exchange, request);
|
||||
return Mono.defer(() -> specification.completionHandler().apply(exchange, request));
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -4,9 +4,17 @@
|
|||
|
||||
package io.modelcontextprotocol.server;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatCode;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import io.modelcontextprotocol.spec.McpError;
|
||||
import io.modelcontextprotocol.spec.McpSchema;
|
||||
import io.modelcontextprotocol.spec.McpSchema.CallToolResult;
|
||||
|
|
@ -18,18 +26,9 @@ import io.modelcontextprotocol.spec.McpSchema.Resource;
|
|||
import io.modelcontextprotocol.spec.McpSchema.ServerCapabilities;
|
||||
import io.modelcontextprotocol.spec.McpSchema.Tool;
|
||||
import io.modelcontextprotocol.spec.McpServerTransportProvider;
|
||||
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.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 io.modelcontextprotocol.spec.McpServerTransportProvider} implementations.
|
||||
|
|
@ -65,10 +64,7 @@ public abstract class AbstractMcpAsyncServerTests {
|
|||
// ---------------------------------------
|
||||
// Server Lifecycle Tests
|
||||
// ---------------------------------------
|
||||
|
||||
@ParameterizedTest(name = "{0} : {displayName} ")
|
||||
@ValueSource(strings = { "sse", "streamable" })
|
||||
void testConstructorWithInvalidArguments(String serverType) {
|
||||
void testConstructorWithInvalidArguments() {
|
||||
assertThatThrownBy(() -> McpServer.async((McpServerTransportProvider) null))
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessage("Transport provider must not be null");
|
||||
|
|
|
|||
Loading…
Reference in a new issue