Listening stream already exists for this session and will be closed to make way for the new listening SSE stream

Author: taobaorun

mcp-spring/mcp-spring-webflux/src/main/java/io/modelcontextprotocol/server/transport/WebFluxStreamableServerTransportProvider.java

@@ -190,16 +190,17 @@ private Mono<ServerResponse> handleGet(ServerRequest request) {
 			}
 
 			McpLoggableSession listenedStream = session.getListeningStream();
-			boolean replayRequest = request.headers().asHttpHeaders().containsKey(HttpHeaders.LAST_EVENT_ID);
-			if (replayRequest) {
+			if (request.headers().asHttpHeaders().containsKey(HttpHeaders.LAST_EVENT_ID)) {
 				String lastId = request.headers().asHttpHeaders().getFirst(HttpHeaders.LAST_EVENT_ID);
 				return ServerResponse.ok()
 					.contentType(MediaType.TEXT_EVENT_STREAM)
 					.body(session.replay(lastId), ServerSentEvent.class);
 			}
 			if (listenedStream instanceof McpStreamableServerSessionStream) {
-				logger.debug("Listening stream for session: {} exists.", sessionId);
-				return ServerResponse.ok().build();
+				logger.debug(
+						"Listening stream already exists for this session:{} and will be closed to make way for the new listening SSE stream",
+						sessionId);
+				listenedStream.close();
 			}
 
 			return ServerResponse.ok()

mcp-spring/mcp-spring-webmvc/src/main/java/io/modelcontextprotocol/server/transport/WebMvcStreamableServerTransportProvider.java

@@ -254,12 +254,6 @@ private ServerResponse handleGet(ServerRequest request) {
 		}
 
 		logger.debug("Handling GET request for session: {}", sessionId);
-		McpLoggableSession listenedStream = session.getListeningStream();
-		boolean replayRequest = request.headers().asHttpHeaders().containsKey(HttpHeaders.LAST_EVENT_ID);
-		if (!replayRequest && listenedStream instanceof McpStreamableServerSessionStream) {
-			logger.debug("Listening stream for session: {} exists.", sessionId);
-			return ServerResponse.ok().build();
-		}
 		try {
 			return ServerResponse.sse(sseBuilder -> {
 				sseBuilder.onTimeout(() -> {
@@ -270,7 +264,7 @@ private ServerResponse handleGet(ServerRequest request) {
 						sessionId, sseBuilder);
 
 				// Check if this is a replay request
-				if (replayRequest) {
+				if (request.headers().asHttpHeaders().containsKey(HttpHeaders.LAST_EVENT_ID)) {
 					String lastId = request.headers().asHttpHeaders().getFirst(HttpHeaders.LAST_EVENT_ID);
 					try {
 						session.replay(lastId)
@@ -294,6 +288,13 @@ private ServerResponse handleGet(ServerRequest request) {
 					}
 				}
 				else {
+					McpLoggableSession listenedStream = session.getListeningStream();
+					if (listenedStream instanceof McpStreamableServerSessionStream) {
+						logger.debug(
+								"Listening stream already exists for this session:{} and will be closed to make way for the new listening SSE stream",
+								sessionId);
+						listenedStream.close();
+					}
 					// Establish new listening stream
 					McpStreamableServerSession.McpStreamableServerSessionStream listeningStream = session
 						.listeningStream(sessionTransport);

mcp/src/main/java/io/modelcontextprotocol/server/transport/HttpServletStreamableServerTransportProvider.java

@@ -275,14 +275,6 @@ protected void doGet(HttpServletRequest request, HttpServletResponse response)
 		}
 
 		logger.debug("Handling GET request for session: {}", sessionId);
-		McpLoggableSession listenedStream = session.getListeningStream();
-		boolean replayRequest = request.getHeader(HttpHeaders.LAST_EVENT_ID) != null;
-		if (!replayRequest && listenedStream instanceof McpStreamableServerSessionStream) {
-			logger.debug("Listening stream for session: {} exists.", sessionId);
-			response.setStatus(HttpServletResponse.SC_OK);
-			return;
-		}
-
 		McpTransportContext transportContext = this.contextExtractor.extract(request, new DefaultMcpTransportContext());
 
 		try {
@@ -299,7 +291,7 @@ protected void doGet(HttpServletRequest request, HttpServletResponse response)
 					sessionId, asyncContext, response.getWriter());
 
 			// Check if this is a replay request
-			if (replayRequest) {
+			if (request.getHeader(HttpHeaders.LAST_EVENT_ID) != null) {
 				String lastId = request.getHeader(HttpHeaders.LAST_EVENT_ID);
 
 				try {
@@ -324,6 +316,13 @@ protected void doGet(HttpServletRequest request, HttpServletResponse response)
 				}
 			}
 			else {
+				McpLoggableSession listenedStream = session.getListeningStream();
+				if (listenedStream instanceof McpStreamableServerSessionStream) {
+					logger.debug(
+							"Listening stream already exists for this session:{} and will be closed to make way for the new listening SSE stream",
+							sessionId);
+					listenedStream.close();
+				}
 				// Establish new listening stream
 				McpStreamableServerSession.McpStreamableServerSessionStream listeningStream = session
 					.listeningStream(sessionTransport);