fix: keep connection open for keep-alive ping in responseStream

After sending a response in responseStream(), if no listening stream
exists yet (i.e., the client hasn't established a GET SSE connection),
promote the current response stream to a listening stream instead of
closing it. This allows KeepAliveScheduler to send periodic ping
messages through the transport.

Clients like Cursor that don't establish a separate GET listening
stream would otherwise have the connection closed immediately after
each response, causing the MCP server to appear as disconnected
after the idle-timeout period.

Fixes #681

Author: JGoP-L

mcp-core/src/main/java/io/modelcontextprotocol/spec/McpStreamableServerSession.java

@@ -171,7 +171,17 @@ public Mono<Void> responseStream(McpSchema.JSONRPCRequest jsonrpcRequest, McpStr
 				return transport
 					.sendMessage(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, jsonrpcRequest.id(), null,
 							new McpSchema.JSONRPCResponse.JSONRPCError(McpSchema.ErrorCodes.METHOD_NOT_FOUND,
-									error.message(), error.data())));
+									error.message(), error.data())))
+					.then(Mono.defer(() -> {
+						McpLoggableSession currentListeningStream = this.listeningStreamRef.get();
+						if (currentListeningStream == this.missingMcpTransportSession) {
+							if (this.listeningStreamRef.compareAndSet(this.missingMcpTransportSession, stream)) {
+								logger.debug("Converted response stream to listening stream for session {}", this.id);
+								return Mono.empty();
+							}
+						}
+						return transport.closeGracefully();
+					}));
 			}
 			return requestHandler
 				.handle(new McpAsyncServerExchange(this.id, stream, clientCapabilities.get(), clientInfo.get(),
@@ -189,7 +199,16 @@ public Mono<Void> responseStream(McpSchema.JSONRPCRequest jsonrpcRequest, McpStr
 					return Mono.just(errorResponse);
 				})
 				.flatMap(transport::sendMessage)
-				.then(transport.closeGracefully());
+				.then(Mono.defer(() -> {
+					McpLoggableSession currentListeningStream = this.listeningStreamRef.get();
+					if (currentListeningStream == this.missingMcpTransportSession) {
+						if (this.listeningStreamRef.compareAndSet(this.missingMcpTransportSession, stream)) {
+							logger.debug("Converted response stream to listening stream for session {}", this.id);
+							return Mono.empty();
+						}
+					}
+					return transport.closeGracefully();
+				}));
 		});
 	}