+ * The transport handles two types of endpoints: + *
+ * Features: + *
+ * This method sets up a new SSE connection when a client connects to the SSE + * endpoint. It configures the response headers for SSE, creates a new session, and + * sends the initial endpoint information to the client. + * @param request The HTTP servlet request + * @param response The HTTP servlet response + * @throws ServletException If a servlet-specific error occurs + * @throws IOException If an I/O error occurs + */ @Override protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException { @@ -102,12 +146,23 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server PrintWriter writer = response.getWriter(); ClientSession session = new ClientSession(sessionId, asyncContext, writer); - sessions.put(sessionId, session); + this.sessions.put(sessionId, session); // Send initial endpoint event - sendEvent(writer, ENDPOINT_EVENT_TYPE, messageEndpoint); + this.sendEvent(writer, ENDPOINT_EVENT_TYPE, messageEndpoint); } + /** + * Handles POST requests for client messages. + *
+ * This method processes incoming messages from clients, routes them through the
+ * connect handler if configured, and sends back the appropriate response. It handles
+ * error cases and formats error responses according to the MCP specification.
+ * @param request The HTTP servlet request
+ * @param response The HTTP servlet response
+ * @throws ServletException If a servlet-specific error occurs
+ * @throws IOException If an I/O error occurs
+ */
@Override
protected void doPost(HttpServletRequest request, HttpServletResponse response)
throws ServletException, IOException {
@@ -200,12 +255,25 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
}
}
+ /**
+ * Sets up the message handler for processing client requests.
+ * @param handler The function to process incoming messages and produce responses
+ * @return A Mono that completes when the handler is set up
+ */
@Override
public Mono
+ * This method serializes the message and sends it to all active client sessions. If a
+ * client is disconnected, its session is removed.
+ * @param message The message to broadcast
+ * @return A Mono that completes when the message has been sent to all clients
+ */
@Override
public Mono
+ * This implementation delegates to the super class's close method.
+ */
@Override
public void close() {
ServerMcpTransport.super.close();
}
+ /**
+ * Unmarshals data from one type to another using the object mapper.
+ * @param
+ * This method marks the transport as closing and closes all active client sessions.
+ * New connection attempts will be rejected during shutdown.
+ * @return A Mono that completes when all sessions have been closed
+ */
@Override
public Mono
+ * This class holds the necessary information about a client's SSE connection,
+ * including its ID, async context, and output writer.
+ */
private static class ClientSession {
private final String id;
@@ -288,6 +392,12 @@ public class HttpServletSseServerTransport extends HttpServlet implements Server
}
+ /**
+ * Cleans up resources when the servlet is being destroyed.
+ *
+ * This method ensures a graceful shutdown by closing all client connections before
+ * calling the parent's destroy method.
+ */
@Override
public void destroy() {
closeGracefully().block();
diff --git a/mcp/src/test/java/org/springframework/ai/mcp/server/_SseAsyncIntegrationTests.java_ b/mcp/src/test/java/org/springframework/ai/mcp/server/transport/HttpServletSseServerTransportIntegrationTests.java
similarity index 62%
rename from mcp/src/test/java/org/springframework/ai/mcp/server/_SseAsyncIntegrationTests.java_
rename to mcp/src/test/java/org/springframework/ai/mcp/server/transport/HttpServletSseServerTransportIntegrationTests.java
index babba8d..71338a9 100644
--- a/mcp/src/test/java/org/springframework/ai/mcp/server/_SseAsyncIntegrationTests.java_
+++ b/mcp/src/test/java/org/springframework/ai/mcp/server/transport/HttpServletSseServerTransportIntegrationTests.java
@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.springframework.ai.mcp.server;
+package org.springframework.ai.mcp.server.transport;
import java.time.Duration;
import java.util.List;
@@ -22,17 +22,21 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.catalina.Context;
+import org.apache.catalina.LifecycleException;
+import org.apache.catalina.LifecycleState;
+import org.apache.catalina.startup.Tomcat;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import reactor.netty.DisposableServer;
-import reactor.netty.http.server.HttpServer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import reactor.test.StepVerifier;
import org.springframework.ai.mcp.client.McpClient;
-import org.springframework.ai.mcp.client.transport.SseClientTransport;
+import org.springframework.ai.mcp.client.transport.HttpClientSseClientTransport;
+import org.springframework.ai.mcp.server.McpServer;
import org.springframework.ai.mcp.server.McpServer.ToolRegistration;
-import org.springframework.ai.mcp.server.transport.SseServerTransport;
import org.springframework.ai.mcp.spec.McpError;
import org.springframework.ai.mcp.spec.McpSchema;
import org.springframework.ai.mcp.spec.McpSchema.CallToolResult;
@@ -44,50 +48,76 @@ import org.springframework.ai.mcp.spec.McpSchema.Role;
import org.springframework.ai.mcp.spec.McpSchema.Root;
import org.springframework.ai.mcp.spec.McpSchema.ServerCapabilities;
import org.springframework.ai.mcp.spec.McpSchema.Tool;
-import org.springframework.http.server.reactive.HttpHandler;
-import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
import org.springframework.web.client.RestClient;
-import org.springframework.web.reactive.function.client.WebClient;
-import org.springframework.web.reactive.function.server.RouterFunctions;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.awaitility.Awaitility.await;
-public class SseAsyncIntegrationTests {
+public class HttpServletSseServerTransportIntegrationTests {
- private static final int PORT = 8181;
+ private static final Logger logger = LoggerFactory.getLogger(HttpServletSseServerTransportIntegrationTests.class);
+
+ private static final int PORT = 8184;
private static final String MESSAGE_ENDPOINT = "/mcp/message";
- private DisposableServer httpServer;
-
- private SseServerTransport mcpServerTransport;
+ private HttpServletSseServerTransport mcpServerTransport;
McpClient.Builder clientBuilder;
+ private Tomcat tomcat;
+
@BeforeEach
public void before() {
- this.mcpServerTransport = new SseServerTransport(new ObjectMapper(), MESSAGE_ENDPOINT);
+ tomcat = new Tomcat();
+ tomcat.setPort(PORT);
- HttpHandler httpHandler = RouterFunctions.toHttpHandler(mcpServerTransport.getRouterFunction());
- ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(httpHandler);
- this.httpServer = HttpServer.create().port(PORT).handle(adapter).bindNow();
+ String baseDir = System.getProperty("java.io.tmpdir");
+ tomcat.setBaseDir(baseDir);
- this.clientBuilder = McpClient
- .using(new SseClientTransport(WebClient.builder().baseUrl("http://localhost:" + PORT)));
+ Context context = tomcat.addContext("", baseDir);
+
+ // Create and configure the transport
+ mcpServerTransport = new HttpServletSseServerTransport(new ObjectMapper(), MESSAGE_ENDPOINT);
+
+ // Add transport servlet to Tomcat
+ org.apache.catalina.Wrapper wrapper = context.createWrapper();
+ wrapper.setName("mcpServlet");
+ wrapper.setServlet(mcpServerTransport);
+ wrapper.setLoadOnStartup(1);
+ wrapper.setAsyncSupported(true);
+ context.addChild(wrapper);
+ context.addServletMappingDecoded("/*", "mcpServlet");
+
+ try {
+ var connector = tomcat.getConnector();
+ connector.setAsyncTimeout(3000);
+ tomcat.start();
+ assertThat(tomcat.getServer().getState() == LifecycleState.STARTED);
+ }
+ catch (Exception e) {
+ throw new RuntimeException("Failed to start Tomcat", e);
+ }
+
+ this.clientBuilder = McpClient.using(new HttpClientSseClientTransport("http://localhost:" + PORT));
}
@AfterEach
public void after() {
- if (httpServer != null) {
- httpServer.disposeNow();
+ if (mcpServerTransport != null) {
+ mcpServerTransport.closeGracefully().block();
+ }
+ if (tomcat != null) {
+ try {
+ tomcat.stop();
+ tomcat.destroy();
+ }
+ catch (LifecycleException e) {
+ throw new RuntimeException("Failed to stop Tomcat", e);
+ }
}
}
- // ---------------------------------------
- // Sampling Tests
- // ---------------------------------------
@Test
void testCreateMessageWithoutInitialization() {
var mcpAsyncServer = McpServer.using(mcpServerTransport).serverInfo("test-server", "1.0.0").async();
@@ -107,7 +137,6 @@ public class SseAsyncIntegrationTests {
@Test
void testCreateMessageWithoutSamplingCapabilities() {
-
var mcpAsyncServer = McpServer.using(mcpServerTransport).serverInfo("test-server", "1.0.0").async();
var client = clientBuilder.clientInfo(new McpSchema.Implementation("Sample client", "0.0.0")).sync();
@@ -129,8 +158,7 @@ public class SseAsyncIntegrationTests {
}
@Test
- void testCreateMessageSuccess() throws InterruptedException {
-
+ void testCreateMessageSuccess() {
var mcpAsyncServer = McpServer.using(mcpServerTransport).serverInfo("test-server", "1.0.0").async();
Function