diff --git a/mcp-transport/mcp-webflux-sse-transport/src/main/java/org/springframework/ai/mcp/client/transport/WebFluxSseClientTransport.java b/mcp-transport/mcp-webflux-sse-transport/src/main/java/org/springframework/ai/mcp/client/transport/WebFluxSseClientTransport.java index 0505372..a4605c6 100644 --- a/mcp-transport/mcp-webflux-sse-transport/src/main/java/org/springframework/ai/mcp/client/transport/WebFluxSseClientTransport.java +++ b/mcp-transport/mcp-webflux-sse-transport/src/main/java/org/springframework/ai/mcp/client/transport/WebFluxSseClientTransport.java @@ -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> SSE_TYPE = new ParameterizedTypeReference<>() { + private static final ParameterizedTypeReference> SSE_TYPE = new ParameterizedTypeReference<>() { }; /** diff --git a/mcp-transport/mcp-webflux-sse-transport/src/main/java/org/springframework/ai/mcp/server/transport/WebFluxSseServerTransport.java b/mcp-transport/mcp-webflux-sse-transport/src/main/java/org/springframework/ai/mcp/server/transport/WebFluxSseServerTransport.java index 88812ad..6126eba 100644 --- a/mcp-transport/mcp-webflux-sse-transport/src/main/java/org/springframework/ai/mcp/server/transport/WebFluxSseServerTransport.java +++ b/mcp-transport/mcp-webflux-sse-transport/src/main/java/org/springframework/ai/mcp/server/transport/WebFluxSseServerTransport.java @@ -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 { *

* The router function defines two endpoints: *

* @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")); } diff --git a/mcp-transport/mcp-webmvc-sse-transport/src/main/java/org/springframework/ai/mcp/server/transport/WebMvcSseServerTransport.java b/mcp-transport/mcp-webmvc-sse-transport/src/main/java/org/springframework/ai/mcp/server/transport/WebMvcSseServerTransport.java index cc01f36..cee0b48 100644 --- a/mcp-transport/mcp-webmvc-sse-transport/src/main/java/org/springframework/ai/mcp/server/transport/WebMvcSseServerTransport.java +++ b/mcp-transport/mcp-webmvc-sse-transport/src/main/java/org/springframework/ai/mcp/server/transport/WebMvcSseServerTransport.java @@ -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 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 diff --git a/mcp/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java b/mcp/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java index 004cbc7..7b7958a 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/client/McpAsyncClient.java @@ -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_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 - }; + } // -------------------------- // 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 }; // @formatter:on - }; + } // -------------------------- // Logging diff --git a/mcp/src/main/java/org/springframework/ai/mcp/client/transport/HttpClientSseClientTransport.java b/mcp/src/main/java/org/springframework/ai/mcp/client/transport/HttpClientSseClientTransport.java index 9284806..68f1032 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/client/transport/HttpClientSseClientTransport.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/client/transport/HttpClientSseClientTransport.java @@ -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) { diff --git a/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java b/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java index fa8c512..37d0c65 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java @@ -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 tools = this.tools.stream().map(toolRegistration -> { - return toolRegistration.tool(); - }).toList(); + List tools = this.tools.stream().map(ToolRegistration::tool).toList(); return Mono.just(new McpSchema.ListToolsResult(tools, null)); }; diff --git a/mcp/src/main/java/org/springframework/ai/mcp/server/McpServer.java b/mcp/src/main/java/org/springframework/ai/mcp/server/McpServer.java index 362491b..e5fec07 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/server/McpServer.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/server/McpServer.java @@ -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; diff --git a/mcp/src/main/java/org/springframework/ai/mcp/server/transport/HttpServletSseServerTransport.java b/mcp/src/main/java/org/springframework/ai/mcp/server/transport/HttpServletSseServerTransport.java index 4857a60..d5ae27b 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/server/transport/HttpServletSseServerTransport.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/server/transport/HttpServletSseServerTransport.java @@ -63,6 +63,7 @@ import org.springframework.ai.mcp.spec.ServerMcpTransport; * * * @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 sessions = new ConcurrentHashMap<>(); @@ -98,14 +108,26 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server private Function, Mono> 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"); } } diff --git a/mcp/src/main/java/org/springframework/ai/mcp/spec/McpSchema.java b/mcp/src/main/java/org/springframework/ai/mcp/spec/McpSchema.java index 1a9f3d3..e107fb6 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/spec/McpSchema.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/spec/McpSchema.java @@ -140,7 +140,7 @@ public final class McpSchema { } - private final static TypeReference> MAP_TYPE_REF = new TypeReference<>() { + private static final TypeReference> MAP_TYPE_REF = new TypeReference<>() { }; /** diff --git a/mcp/src/test/java/org/springframework/ai/mcp/spec/DefaultMcpSessionTests.java b/mcp/src/test/java/org/springframework/ai/mcp/spec/DefaultMcpSessionTests.java index 658c49a..acf7c44 100644 --- a/mcp/src/test/java/org/springframework/ai/mcp/spec/DefaultMcpSessionTests.java +++ b/mcp/src/test/java/org/springframework/ai/mcp/spec/DefaultMcpSessionTests.java @@ -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);