feat: Make SSE endpoint configurable in server transports

This commit updates the WebFluxSseServerTransport, WebMvcSseServerTransport, and HttpServletSseServerTransport classes to accept the SSE endpoint path as a constructor argument.

This change provides more flexibility in configuring the SSE endpoint, allowing it to be customized instead of being hardcoded to "/sse".
It also aligns the Java SDK's behavior with the Python SDK's SseServerTransport, which already supports this customization.

A default SSE endpoint of "/sse" is maintained for backward compatibility.

Also, addressed minor code quality issues identified by SonarLint and reordered modifiers for static final fields to conform with the Java Language Specification.

Signed-off-by: Alexandros Pappas <alexandros.pappas@yiluhub.com>
This commit is contained in:
Alexandros Pappas 2025-01-16 15:28:35 +01:00 • committed by Christian Tzolov
commit c2ece20cc2
10 changed files with 111 additions and 58 deletions

View file

@ -72,31 +72,31 @@ import org.springframework.web.reactive.function.client.WebClient;
*/
public class WebFluxSseClientTransport implements ClientMcpTransport {
private final static Logger logger = LoggerFactory.getLogger(WebFluxSseClientTransport.class);
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 final static String MESSAGE_EVENT_TYPE = "message";
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 final static String ENDPOINT_EVENT_TYPE = "endpoint";
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 final static String SSE_ENDPOINT = "/sse";
private static final String SSE_ENDPOINT = "/sse";
/**
* Type reference for parsing SSE events containing string data.
*/
private final static ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE = new ParameterizedTypeReference<>() {
private static final ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE = new ParameterizedTypeReference<>() {
};
/**

View file

@ -57,32 +57,35 @@ import org.springframework.web.reactive.function.server.ServerResponse;
* {@link Sinks} for thread-safe message broadcasting.
*
* @author Christian Tzolov
* @author Alexandros Pappas
* @see ServerMcpTransport
* @see ServerSentEvent
*/
public class WebFluxSseServerTransport implements ServerMcpTransport {
private final static Logger logger = LoggerFactory.getLogger(WebFluxSseServerTransport.class);
private static final Logger logger = LoggerFactory.getLogger(WebFluxSseServerTransport.class);
/**
* Event type for JSON-RPC messages sent through the SSE connection.
*/
public final static String MESSAGE_EVENT_TYPE = "message";
public static final String MESSAGE_EVENT_TYPE = "message";
/**
* Event type for sending the message endpoint URI to clients.
*/
public final static String ENDPOINT_EVENT_TYPE = "endpoint";
public static final String ENDPOINT_EVENT_TYPE = "endpoint";
/**
* Default SSE endpoint path as specified by the MCP transport specification.
*/
public final static String SSE_ENDPOINT = "/sse";
public static final String DEFAULT_SSE_ENDPOINT = "/sse";
private final ObjectMapper objectMapper;
private final String messageEndpoint;
private final String sseEndpoint;
private final RouterFunction<?> routerFunction;
/**
@ -106,18 +109,34 @@ public class WebFluxSseServerTransport implements ServerMcpTransport {
* setup. Must not be null.
* @throws IllegalArgumentException if either parameter is null
*/
public WebFluxSseServerTransport(ObjectMapper objectMapper, String messageEndpoint) {
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(SSE_ENDPOINT, this::handleSseConnection)
.POST(messageEndpoint, this::handleMessage)
.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
@ -243,7 +262,7 @@ public class WebFluxSseServerTransport implements ServerMcpTransport {
* <p>
* The router function defines two endpoints:
* <ul>
* <li>GET {SSE_ENDPOINT} - For establishing SSE connections</li>
* <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
@ -343,11 +362,7 @@ public class WebFluxSseServerTransport implements ServerMcpTransport {
.bodyValue(new McpError(error.getMessage()));
});
}
catch (IllegalArgumentException e) {
logger.error("Failed to deserialize message: {}", e.getMessage());
return ServerResponse.badRequest().bodyValue(new McpError("Invalid message format"));
}
catch (IOException e) {
catch (IllegalArgumentException | IOException e) {
logger.error("Failed to deserialize message: {}", e.getMessage());
return ServerResponse.badRequest().bodyValue(new McpError("Invalid message format"));
}

View file

@ -70,32 +70,35 @@ import org.springframework.web.servlet.function.ServerResponse.SseBuilder;
* maintains its own SSE connection.
*
* @author Christian Tzolov
* @author Alexandros Pappas
* @see ServerMcpTransport
* @see RouterFunction
*/
public class WebMvcSseServerTransport implements ServerMcpTransport {
private final static Logger logger = LoggerFactory.getLogger(WebMvcSseServerTransport.class);
private static final Logger logger = LoggerFactory.getLogger(WebMvcSseServerTransport.class);
/**
* Event type for JSON-RPC messages sent through the SSE connection.
*/
public final static String MESSAGE_EVENT_TYPE = "message";
public static final String MESSAGE_EVENT_TYPE = "message";
/**
* Event type for sending the message endpoint URI to clients.
*/
public final static String ENDPOINT_EVENT_TYPE = "endpoint";
public static final String ENDPOINT_EVENT_TYPE = "endpoint";
/**
* Default SSE endpoint path as specified by the MCP transport specification.
*/
public final static String SSE_ENDPOINT = "/sse";
public static final String DEFAULT_SSE_ENDPOINT = "/sse";
private final ObjectMapper objectMapper;
private final String messageEndpoint;
private final String sseEndpoint;
private final RouterFunction<ServerResponse> routerFunction;
/**
@ -122,18 +125,33 @@ public class WebMvcSseServerTransport implements ServerMcpTransport {
* SSE connection's initial endpoint event.
* @throws IllegalArgumentException if either objectMapper or messageEndpoint is null
*/
public WebMvcSseServerTransport(ObjectMapper objectMapper, String messageEndpoint) {
public WebMvcSseServerTransport(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(SSE_ENDPOINT, this::handleSseConnection)
.POST(messageEndpoint, this::handleMessage)
.GET(this.sseEndpoint, this::handleSseConnection)
.POST(this.messageEndpoint, this::handleMessage)
.build();
}
/**
* Constructs a new WebMvcSseServerTransport instance with the default SSE endpoint.
* @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
* of messages.
* @param messageEndpoint The endpoint URI where clients should send their JSON-RPC
* messages via HTTP POST. This endpoint will be communicated to clients through the
* SSE connection's initial endpoint event.
* @throws IllegalArgumentException if either objectMapper or messageEndpoint is null
*/
public WebMvcSseServerTransport(ObjectMapper objectMapper, String messageEndpoint) {
this(objectMapper, messageEndpoint, DEFAULT_SSE_ENDPOINT);
}
/**
* Sets up the message handler for this transport. In the WebMVC SSE implementation,
* this method only stores the handler for later use, as connections are initiated by

View file

@ -62,7 +62,7 @@ import org.springframework.ai.mcp.util.Utils;
*/
public class McpAsyncClient {
private final static Logger logger = LoggerFactory.getLogger(McpAsyncClient.class);
private static final Logger logger = LoggerFactory.getLogger(McpAsyncClient.class);
private static TypeReference<Void> VOID_TYPE_REFERENCE = new TypeReference<>() {
};
@ -498,7 +498,7 @@ public class McpAsyncClient {
logger.error("Error handling tools list change notification", error);
return Mono.empty();
}).then(); // Convert to Mono<Void>
};
}
// --------------------------
// Resources
@ -619,7 +619,7 @@ public class McpAsyncClient {
logger.error("Error handling resources list change notification", error);
return Mono.empty();
}).then();
};
}
// --------------------------
// Prompts
@ -680,7 +680,7 @@ public class McpAsyncClient {
return Mono.empty();
}).then(); // Convert to Mono<Void>
}; // @formatter:on
};
}
// --------------------------
// Logging

View file

@ -70,16 +70,16 @@ import java.util.function.Function;
*/
public class HttpClientSseClientTransport implements ClientMcpTransport {
private final static Logger logger = LoggerFactory.getLogger(HttpClientSseClientTransport.class);
private static final Logger logger = LoggerFactory.getLogger(HttpClientSseClientTransport.class);
/** SSE event type for JSON-RPC messages */
private final static String MESSAGE_EVENT_TYPE = "message";
private static final String MESSAGE_EVENT_TYPE = "message";
/** SSE event type for endpoint discovery */
private final static String ENDPOINT_EVENT_TYPE = "endpoint";
private static final String ENDPOINT_EVENT_TYPE = "endpoint";
/** Default SSE endpoint path */
private final static String SSE_ENDPOINT = "/sse";
private static final String SSE_ENDPOINT = "/sse";
/** Base URI for the MCP server */
private final String baseUri;
@ -170,7 +170,7 @@ public class HttpClientSseClientTransport implements ClientMcpTransport {
handler.apply(Mono.just(message)).subscribe();
}
else {
logger.error("Received unrecognized SSE event type: " + event.type());
logger.error("Received unrecognized SSE event type: {}", event.type());
}
}
catch (IOException e) {

View file

@ -54,7 +54,7 @@ import org.springframework.ai.mcp.util.Utils;
*/
public class McpAsyncServer {
private final static Logger logger = LoggerFactory.getLogger(McpAsyncServer.class);
private static final Logger logger = LoggerFactory.getLogger(McpAsyncServer.class);
/**
* The MCP session implementation that manages bidirectional JSON-RPC communication
@ -279,7 +279,7 @@ public class McpAsyncServer {
return Mono.empty();
}).then();
};
};
}
// ---------------------------------------
// Tool Management
@ -352,9 +352,7 @@ public class McpAsyncServer {
private DefaultMcpSession.RequestHandler toolsListRequestHandler() {
return params -> {
List<Tool> tools = this.tools.stream().map(toolRegistration -> {
return toolRegistration.tool();
}).toList();
List<Tool> tools = this.tools.stream().map(ToolRegistration::tool).toList();
return Mono.just(new McpSchema.ListToolsResult(tools, null));
};

View file

@ -208,7 +208,7 @@ public interface McpServer {
*/
public static class Builder {
private final static McpSchema.Implementation DEFAULT_SERVER_INFO = new McpSchema.Implementation("mcp-server",
private static final McpSchema.Implementation DEFAULT_SERVER_INFO = new McpSchema.Implementation("mcp-server",
"1.0.0");
private final ServerMcpTransport transport;

View file

@ -63,6 +63,7 @@ import org.springframework.ai.mcp.spec.ServerMcpTransport;
* </ul>
*
* @author Christian Tzolov
* @author Alexandros Pappas
* @see ServerMcpTransport
* @see HttpServlet
*/
@ -73,8 +74,14 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
/** Logger for this class */
private static final Logger logger = LoggerFactory.getLogger(HttpServletSseServerTransport.class);
/** The endpoint path for SSE connections */
public static final String SSE_ENDPOINT = "/sse";
public static final String UTF_8 = "UTF-8";
public static final String APPLICATION_JSON = "application/json";
public static final String FAILED_TO_SEND_ERROR_RESPONSE = "Failed to send error response: {}";
/** Default endpoint path for SSE connections */
public static final String DEFAULT_SSE_ENDPOINT = "/sse";
/** Event type for regular messages */
public static final String MESSAGE_EVENT_TYPE = "message";
@ -88,6 +95,9 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
/** The endpoint path for handling client messages */
private final String messageEndpoint;
/** The endpoint path for handling SSE connections */
private final String sseEndpoint;
/** Map of active client sessions, keyed by session ID */
private final Map<String, ClientSession> sessions = new ConcurrentHashMap<>();
@ -98,14 +108,26 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
private Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> connectHandler;
/**
* Creates a new HttpServletSseServerTransport instance.
* Creates a new HttpServletSseServerTransport instance with a custom SSE endpoint.
* @param objectMapper The JSON object mapper to use for message
* serialization/deserialization
* @param messageEndpoint The endpoint path where clients will send their messages
* @param sseEndpoint The endpoint path where clients will establish SSE connections
*/
public HttpServletSseServerTransport(ObjectMapper objectMapper, String messageEndpoint, String sseEndpoint) {
this.objectMapper = objectMapper;
this.messageEndpoint = messageEndpoint;
this.sseEndpoint = sseEndpoint;
}
/**
* Creates a new HttpServletSseServerTransport instance with the default SSE endpoint.
* @param objectMapper The JSON object mapper to use for message
* serialization/deserialization
* @param messageEndpoint The endpoint path where clients will send their messages
*/
public HttpServletSseServerTransport(ObjectMapper objectMapper, String messageEndpoint) {
this.objectMapper = objectMapper;
this.messageEndpoint = messageEndpoint;
this(objectMapper, messageEndpoint, DEFAULT_SSE_ENDPOINT);
}
/**
@ -124,7 +146,7 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
throws ServletException, IOException {
String pathInfo = request.getPathInfo();
if (!SSE_ENDPOINT.equals(pathInfo)) {
if (!sseEndpoint.equals(pathInfo)) {
response.sendError(HttpServletResponse.SC_NOT_FOUND);
return;
}
@ -135,7 +157,7 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
}
response.setContentType("text/event-stream");
response.setCharacterEncoding("UTF-8");
response.setCharacterEncoding(UTF_8);
response.setHeader("Cache-Control", "no-cache");
response.setHeader("Connection", "keep-alive");
response.setHeader("Access-Control-Allow-Origin", "*");
@ -191,8 +213,8 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
if (connectHandler != null) {
connectHandler.apply(Mono.just(message)).subscribe(responseMessage -> {
try {
response.setContentType("application/json");
response.setCharacterEncoding("UTF-8");
response.setContentType(APPLICATION_JSON);
response.setCharacterEncoding(UTF_8);
String jsonResponse = objectMapper.writeValueAsString(responseMessage);
PrintWriter writer = response.getWriter();
writer.write(jsonResponse);
@ -205,15 +227,15 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
"Error processing response: " + e.getMessage());
}
catch (IOException ex) {
logger.error("Failed to send error response: {}", ex.getMessage());
logger.error(FAILED_TO_SEND_ERROR_RESPONSE, ex.getMessage());
}
}
}, error -> {
try {
logger.error("Error processing message: {}", error.getMessage());
McpError mcpError = new McpError(error.getMessage());
response.setContentType("application/json");
response.setCharacterEncoding("UTF-8");
response.setContentType(APPLICATION_JSON);
response.setCharacterEncoding(UTF_8);
response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
String jsonError = objectMapper.writeValueAsString(mcpError);
PrintWriter writer = response.getWriter();
@ -221,13 +243,13 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
writer.flush();
}
catch (IOException e) {
logger.error("Failed to send error response: {}", e.getMessage());
logger.error(FAILED_TO_SEND_ERROR_RESPONSE, e.getMessage());
try {
response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,
"Error sending error response: " + e.getMessage());
}
catch (IOException ex) {
logger.error("Failed to send error response: {}", ex.getMessage());
logger.error(FAILED_TO_SEND_ERROR_RESPONSE, ex.getMessage());
}
}
});
@ -240,8 +262,8 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
logger.error("Invalid message format: {}", e.getMessage());
try {
McpError mcpError = new McpError("Invalid message format: " + e.getMessage());
response.setContentType("application/json");
response.setCharacterEncoding("UTF-8");
response.setContentType(APPLICATION_JSON);
response.setCharacterEncoding(UTF_8);
response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
String jsonError = objectMapper.writeValueAsString(mcpError);
PrintWriter writer = response.getWriter();
@ -249,7 +271,7 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
writer.flush();
}
catch (IOException ex) {
logger.error("Failed to send error response: {}", ex.getMessage());
logger.error(FAILED_TO_SEND_ERROR_RESPONSE, ex.getMessage());
response.sendError(HttpServletResponse.SC_BAD_REQUEST, "Invalid message format");
}
}

View file

@ -140,7 +140,7 @@ public final class McpSchema {
}
private final static TypeReference<HashMap<String, Object>> MAP_TYPE_REF = new TypeReference<>() {
private static final TypeReference<HashMap<String, Object>> MAP_TYPE_REF = new TypeReference<>() {
};
/**

View file

@ -42,7 +42,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
*/
class DefaultMcpSessionTests {
private final static Logger logger = LoggerFactory.getLogger(DefaultMcpSessionTests.class);
private static final Logger logger = LoggerFactory.getLogger(DefaultMcpSessionTests.class);
private static final Duration TIMEOUT = Duration.ofSeconds(5);