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

WebMvcSseServerTransportProvider: do not include session ID in SSE event

Browse source 
- This aligns messages with WebFluxSseServerTransportProvider

Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>
This commit is contained in:
Daniel Garnier-Moiroux 2025-11-19 11:20:12 +01:00 • committed by Daniel Garnier-Moiroux
commit c561675e74
1 changed files with 29 additions and 47 deletions
Download patch file Download diff file Expand all files Collapse all files

76
mcp-spring/mcp-spring-webmvc/src/main/java/io/modelcontextprotocol/server/transport/WebMvcSseServerTransportProvider.java
View file

@ -7,7 +7,6 @@ package io.modelcontextprotocol.server.transport;
import java.io.IOException; import java.io.IOException;
import java.time.Duration; import java.time.Duration;
import java.util.List; import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantLock;
@ -256,39 +255,31 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).body("Server is shutting down"); return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).body("Server is shutting down");
} }
String sessionId = UUID.randomUUID().toString();
logger.debug("Creating new SSE connection for session: {}", sessionId);
// Send initial endpoint event // Send initial endpoint event
try { return ServerResponse.sse(sseBuilder -> {
return ServerResponse.sse(sseBuilder -> { WebMvcMcpSessionTransport sessionTransport = new WebMvcMcpSessionTransport(sseBuilder);
sseBuilder.onComplete(() -> { McpServerSession session = sessionFactory.create(sessionTransport);
logger.debug("SSE connection completed for session: {}", sessionId); String sessionId = session.getId();
sessions.remove(sessionId); logger.debug("Creating new SSE connection for session: {}", sessionId);
}); sseBuilder.onComplete(() -> {
sseBuilder.onTimeout(() -> { logger.debug("SSE connection completed for session: {}", sessionId);
logger.debug("SSE connection timed out for session: {}", sessionId); sessions.remove(sessionId);
sessions.remove(sessionId); });
}); sseBuilder.onTimeout(() -> {
logger.debug("SSE connection timed out for session: {}", sessionId);
sessions.remove(sessionId);
});
this.sessions.put(sessionId, session);
WebMvcMcpSessionTransport sessionTransport = new WebMvcMcpSessionTransport(sessionId, sseBuilder); try {
McpServerSession session = sessionFactory.create(sessionTransport); sseBuilder.event(ENDPOINT_EVENT_TYPE).data(buildEndpointUrl(sessionId));
this.sessions.put(sessionId, session); }
catch (Exception e) {
try { logger.error("Failed to send initial endpoint event: {}", e.getMessage());
sseBuilder.id(sessionId).event(ENDPOINT_EVENT_TYPE).data(buildEndpointUrl(sessionId)); this.sessions.remove(sessionId);
} sseBuilder.error(e);
catch (Exception e) { }
logger.error("Failed to send initial endpoint event: {}", e.getMessage()); }, Duration.ZERO);
sseBuilder.error(e);
}
}, Duration.ZERO);
}
catch (Exception e) {
logger.error("Failed to send initial endpoint event to session {}: {}", sessionId, e.getMessage());
sessions.remove(sessionId);
return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
}
} }
/** /**
@ -363,8 +354,6 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
*/ */
private class WebMvcMcpSessionTransport implements McpServerTransport { private class WebMvcMcpSessionTransport implements McpServerTransport {
private final String sessionId;
private final SseBuilder sseBuilder; private final SseBuilder sseBuilder;
/** /**
@ -374,14 +363,11 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
private final ReentrantLock sseBuilderLock = new ReentrantLock(); private final ReentrantLock sseBuilderLock = new ReentrantLock();
/** /**
* Creates a new session transport with the specified ID and SSE builder. * Creates a new session transport with the specified SSE builder.
* @param sessionId The unique identifier for this session
* @param sseBuilder The SSE builder for sending server events to the client * @param sseBuilder The SSE builder for sending server events to the client
*/ */
WebMvcMcpSessionTransport(String sessionId, SseBuilder sseBuilder) { WebMvcMcpSessionTransport(SseBuilder sseBuilder) {
this.sessionId = sessionId;
this.sseBuilder = sseBuilder; this.sseBuilder = sseBuilder;
logger.debug("Session transport {} initialized with SSE builder", sessionId);
} }
/** /**
@ -395,11 +381,10 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
sseBuilderLock.lock(); sseBuilderLock.lock();
try { try {
String jsonText = jsonMapper.writeValueAsString(message); String jsonText = jsonMapper.writeValueAsString(message);
sseBuilder.id(sessionId).event(MESSAGE_EVENT_TYPE).data(jsonText); sseBuilder.event(MESSAGE_EVENT_TYPE).data(jsonText);
logger.debug("Message sent to session {}", sessionId);
} }
catch (Exception e) { catch (Exception e) {
logger.error("Failed to send message to session {}: {}", sessionId, e.getMessage()); logger.error("Failed to send message: {}", e.getMessage());
sseBuilder.error(e); sseBuilder.error(e);
} }
finally { finally {
@ -427,14 +412,12 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
@Override @Override
public Mono<Void> closeGracefully() { public Mono<Void> closeGracefully() {
return Mono.fromRunnable(() -> { return Mono.fromRunnable(() -> {
logger.debug("Closing session transport: {}", sessionId);
sseBuilderLock.lock(); sseBuilderLock.lock();
try { try {
sseBuilder.complete(); sseBuilder.complete();
logger.debug("Successfully completed SSE builder for session {}", sessionId);
} }
catch (Exception e) { catch (Exception e) {
logger.warn("Failed to complete SSE builder for session {}: {}", sessionId, e.getMessage()); logger.warn("Failed to complete SSE builder: {}", e.getMessage());
} }
finally { finally {
sseBuilderLock.unlock(); sseBuilderLock.unlock();
@ -450,10 +433,9 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
sseBuilderLock.lock(); sseBuilderLock.lock();
try { try {
sseBuilder.complete(); sseBuilder.complete();
logger.debug("Successfully completed SSE builder for session {}", sessionId);
} }
catch (Exception e) { catch (Exception e) {
logger.warn("Failed to complete SSE builder for session {}: {}", sessionId, e.getMessage()); logger.warn("Failed to complete SSE builder: {}", e.getMessage());
} }
finally { finally {
sseBuilderLock.unlock(); sseBuilderLock.unlock();