feat: implement MCP-compliant keep-alive functionality for server transports

- Add KeepAliveScheduler utility class for configurable periodic session pings
- Integrate keep-alive support in WebFlux, WebMVC, and HttpServlet SSE transport providers
- Add keepAliveInterval configuration option to all transport provider builders
- Deprecate existing constructors in favor of builder pattern with enhanced configuration
- Update graceful shutdown to properly clean up keep-alive schedulers
- Add unit tests for KeepAliveScheduler functionality

Implements MCP specification recommendations for connection health detection:
- Configurable ping frequency to suit different network environments
- Optional keep-alive (disabled by default) to avoid excessive network overhead
- Proper resource cleanup to prevent connection leaks

https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/ping#implementation-considerations

Signed-off-by: Christian Tzolov <[email protected]>

Author: tzolov

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

@@ -1,6 +1,7 @@
 package io.modelcontextprotocol.server.transport;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.concurrent.ConcurrentHashMap;
 
 import com.fasterxml.jackson.core.type.TypeReference;
@@ -11,6 +12,8 @@
 import io.modelcontextprotocol.spec.McpServerTransport;
 import io.modelcontextprotocol.spec.McpServerTransportProvider;
 import io.modelcontextprotocol.util.Assert;
+import io.modelcontextprotocol.util.KeepAliveScheduler;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.Exceptions;
@@ -109,6 +112,12 @@ public class WebFluxSseServerTransportProvider implements McpServerTransportProv
 	 */
 	private volatile boolean isClosing = false;
 
+	/**
+	 * Keep-alive scheduler for managing session pings. Activated if keepAliveInterval is
+	 * set. Disabled by default.
+	 */
+	private KeepAliveScheduler keepAliveScheduler;
+
 	/**
 	 * Constructs a new WebFlux SSE server transport provider instance with the default
 	 * SSE endpoint.
@@ -118,7 +127,10 @@ public class WebFluxSseServerTransportProvider implements McpServerTransportProv
 	 * messages. This endpoint will be communicated to clients during SSE connection
 	 * setup. Must not be null.
 	 * @throws IllegalArgumentException if either parameter is null
+	 * @deprecated Use the builder {@link #builder()} instead for better configuration
+	 * options.
 	 */
+	@Deprecated
 	public WebFluxSseServerTransportProvider(ObjectMapper objectMapper, String messageEndpoint) {
 		this(objectMapper, messageEndpoint, DEFAULT_SSE_ENDPOINT);
 	}
@@ -131,7 +143,10 @@ public WebFluxSseServerTransportProvider(ObjectMapper objectMapper, String messa
 	 * messages. This endpoint will be communicated to clients during SSE connection
 	 * setup. Must not be null.
 	 * @throws IllegalArgumentException if either parameter is null
+	 * @deprecated Use the builder {@link #builder()} instead for better configuration
+	 * options.
 	 */
+	@Deprecated
 	public WebFluxSseServerTransportProvider(ObjectMapper objectMapper, String messageEndpoint, String sseEndpoint) {
 		this(objectMapper, DEFAULT_BASE_URL, messageEndpoint, sseEndpoint);
 	}
@@ -145,9 +160,32 @@ public WebFluxSseServerTransportProvider(ObjectMapper objectMapper, String messa
 	 * messages. This endpoint will be communicated to clients during SSE connection
 	 * setup. Must not be null.
 	 * @throws IllegalArgumentException if either parameter is null
+	 * @deprecated Use the builder {@link #builder()} instead for better configuration
+	 * options.
 	 */
+	@Deprecated
 	public WebFluxSseServerTransportProvider(ObjectMapper objectMapper, String baseUrl, String messageEndpoint,
 			String sseEndpoint) {
+		this(objectMapper, baseUrl, messageEndpoint, sseEndpoint, null);
+	}
+
+	/**
+	 * Constructs a new WebFlux SSE server transport provider instance.
+	 * @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
+	 * of MCP messages. Must not be null.
+	 * @param baseUrl webflux message base path
+	 * @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.
+	 * @param sseEndpoint The SSE endpoint path. Must not be null.
+	 * @param keepAliveInterval The interval for sending keep-alive pings to clients.
+	 * @throws IllegalArgumentException if either parameter is null
+	 * @deprecated Use the builder {@link #builder()} instead for better configuration
+	 * options.
+	 */
+	@Deprecated
+	public WebFluxSseServerTransportProvider(ObjectMapper objectMapper, String baseUrl, String messageEndpoint,
+			String sseEndpoint, Duration keepAliveInterval) {
 		Assert.notNull(objectMapper, "ObjectMapper must not be null");
 		Assert.notNull(baseUrl, "Message base path must not be null");
 		Assert.notNull(messageEndpoint, "Message endpoint must not be null");
@@ -161,6 +199,17 @@ public WebFluxSseServerTransportProvider(ObjectMapper objectMapper, String baseU
 			.GET(this.sseEndpoint, this::handleSseConnection)
 			.POST(this.messageEndpoint, this::handleMessage)
 			.build();
+
+		if (keepAliveInterval != null) {
+
+			this.keepAliveScheduler = KeepAliveScheduler
+				.builder(() -> (isClosing) ? Flux.empty() : Flux.fromIterable(sessions.values()))
+				.initialDelay(keepAliveInterval)
+				.interval(keepAliveInterval)
+				.build();
+
+			this.keepAliveScheduler.start();
+		}
 	}
 
 	@Override
@@ -209,23 +258,21 @@ public Mono<Void> notifyClients(String method, Object params) {
 	/**
 	 * Initiates a graceful shutdown of all the sessions. This method ensures all active
 	 * sessions are properly closed and cleaned up.
-	 *
-	 * <p>
-	 * The shutdown process:
-	 * <ul>
-	 * <li>Marks the transport as closing to prevent new connections</li>
-	 * <li>Closes each active session</li>
-	 * <li>Removes closed sessions from the sessions map</li>
-	 * <li>Times out after 5 seconds if shutdown takes too long</li>
-	 * </ul>
 	 * @return A Mono that completes when all sessions have been closed
 	 */
 	@Override
 	public Mono<Void> closeGracefully() {
 		return Flux.fromIterable(sessions.values())
 			.doFirst(() -> logger.debug("Initiating graceful shutdown with {} active sessions", sessions.size()))
 			.flatMap(McpServerSession::closeGracefully)
-			.then();
+			.then()
+			.doOnSuccess(v -> {
+				logger.debug("Graceful shutdown completed");
+				sessions.clear();
+				if (this.keepAliveScheduler != null) {
+					this.keepAliveScheduler.shutdown();
+				}
+			});
 	}
 
 	/**
@@ -396,6 +443,8 @@ public static class Builder {
 
 		private String sseEndpoint = DEFAULT_SSE_ENDPOINT;
 
+		private Duration keepAliveInterval;
+
 		/**
 		 * Sets the ObjectMapper to use for JSON serialization/deserialization of MCP
 		 * messages.
@@ -446,6 +495,17 @@ public Builder sseEndpoint(String sseEndpoint) {
 			return this;
 		}
 
+		/**
+		 * Sets the interval for sending keep-alive pings to clients.
+		 * @param keepAliveInterval The keep-alive interval duration. If null, keep-alive
+		 * is disabled.
+		 * @return this builder instance
+		 */
+		public Builder keepAliveInterval(Duration keepAliveInterval) {
+			this.keepAliveInterval = keepAliveInterval;
+			return this;
+		}
+
 		/**
 		 * Builds a new instance of {@link WebFluxSseServerTransportProvider} with the
 		 * configured settings.
@@ -456,7 +516,8 @@ public WebFluxSseServerTransportProvider build() {
 			Assert.notNull(objectMapper, "ObjectMapper must be set");
 			Assert.notNull(messageEndpoint, "Message endpoint must be set");
 
-			return new WebFluxSseServerTransportProvider(objectMapper, baseUrl, messageEndpoint, sseEndpoint);
+			return new WebFluxSseServerTransportProvider(objectMapper, baseUrl, messageEndpoint, sseEndpoint,
+					keepAliveInterval);
 		}
 
 	}

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

@@ -17,6 +17,8 @@
 import io.modelcontextprotocol.spec.McpServerTransportProvider;
 import io.modelcontextprotocol.spec.McpServerSession;
 import io.modelcontextprotocol.util.Assert;
+import io.modelcontextprotocol.util.KeepAliveScheduler;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
@@ -106,6 +108,8 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
 	 */
 	private volatile boolean isClosing = false;
 
+	private KeepAliveScheduler keepAliveScheduler;
+
 	/**
 	 * Constructs a new WebMvcSseServerTransportProvider instance with the default SSE
 	 * endpoint.
@@ -115,7 +119,10 @@ public class WebMvcSseServerTransportProvider implements McpServerTransportProvi
 	 * 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
+	 * @deprecated Use the builder {@link #builder()} instead for better configuration
+	 * options.
 	 */
+	@Deprecated
 	public WebMvcSseServerTransportProvider(ObjectMapper objectMapper, String messageEndpoint) {
 		this(objectMapper, messageEndpoint, DEFAULT_SSE_ENDPOINT);
 	}
@@ -129,7 +136,10 @@ public WebMvcSseServerTransportProvider(ObjectMapper objectMapper, String messag
 	 * SSE connection's initial endpoint event.
 	 * @param sseEndpoint The endpoint URI where clients establish their SSE connections.
 	 * @throws IllegalArgumentException if any parameter is null
+	 * @deprecated Use the builder {@link #builder()} instead for better configuration
+	 * options.
 	 */
+	@Deprecated
 	public WebMvcSseServerTransportProvider(ObjectMapper objectMapper, String messageEndpoint, String sseEndpoint) {
 		this(objectMapper, "", messageEndpoint, sseEndpoint);
 	}
@@ -145,9 +155,33 @@ public WebMvcSseServerTransportProvider(ObjectMapper objectMapper, String messag
 	 * SSE connection's initial endpoint event.
 	 * @param sseEndpoint The endpoint URI where clients establish their SSE connections.
 	 * @throws IllegalArgumentException if any parameter is null
+	 * @deprecated Use the builder {@link #builder()} instead for better configuration
+	 * options.
 	 */
+	@Deprecated
 	public WebMvcSseServerTransportProvider(ObjectMapper objectMapper, String baseUrl, String messageEndpoint,
 			String sseEndpoint) {
+		this(objectMapper, baseUrl, messageEndpoint, sseEndpoint, null);
+	}
+
+	/**
+	 * Constructs a new WebMvcSseServerTransportProvider instance.
+	 * @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
+	 * of messages.
+	 * @param baseUrl The base URL for the message endpoint, used to construct the full
+	 * endpoint URL for clients.
+	 * @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.
+	 * @param sseEndpoint The endpoint URI where clients establish their SSE connections.
+	 * * @param keepAliveInterval The interval for sending keep-alive messages to
+	 * @throws IllegalArgumentException if any parameter is null
+	 * @deprecated Use the builder {@link #builder()} instead for better configuration
+	 * options.
+	 */
+	@Deprecated
+	public WebMvcSseServerTransportProvider(ObjectMapper objectMapper, String baseUrl, String messageEndpoint,
+			String sseEndpoint, Duration keepAliveInterval) {
 		Assert.notNull(objectMapper, "ObjectMapper must not be null");
 		Assert.notNull(baseUrl, "Message base URL must not be null");
 		Assert.notNull(messageEndpoint, "Message endpoint must not be null");
@@ -161,6 +195,17 @@ public WebMvcSseServerTransportProvider(ObjectMapper objectMapper, String baseUr
 			.GET(this.sseEndpoint, this::handleSseConnection)
 			.POST(this.messageEndpoint, this::handleMessage)
 			.build();
+
+		if (keepAliveInterval != null) {
+
+			this.keepAliveScheduler = KeepAliveScheduler
+				.builder(() -> (isClosing) ? Flux.empty() : Flux.fromIterable(sessions.values()))
+				.initialDelay(keepAliveInterval)
+				.interval(keepAliveInterval)
+				.build();
+
+			this.keepAliveScheduler.start();
+		}
 	}
 
 	@Override
@@ -208,10 +253,13 @@ public Mono<Void> closeGracefully() {
 		return Flux.fromIterable(sessions.values()).doFirst(() -> {
 			this.isClosing = true;
 			logger.debug("Initiating graceful shutdown with {} active sessions", sessions.size());
-		})
-			.flatMap(McpServerSession::closeGracefully)
-			.then()
-			.doOnSuccess(v -> logger.debug("Graceful shutdown completed"));
+		}).flatMap(McpServerSession::closeGracefully).then().doOnSuccess(v -> {
+			logger.debug("Graceful shutdown completed");
+			sessions.clear();
+			if (this.keepAliveScheduler != null) {
+				this.keepAliveScheduler.shutdown();
+			}
+		});
 	}
 
 	/**
@@ -416,4 +464,106 @@ public void close() {
 
 	}
 
+	/**
+	 * Creates a new Builder instance for configuring and creating instances of
+	 * WebMvcSseServerTransportProvider.
+	 * @return A new Builder instance
+	 */
+	public static Builder builder() {
+		return new Builder();
+	}
+
+	/**
+	 * Builder for creating instances of WebMvcSseServerTransportProvider.
+	 * <p>
+	 * This builder provides a fluent API for configuring and creating instances of
+	 * WebMvcSseServerTransportProvider with custom settings.
+	 */
+	public static class Builder {
+
+		private ObjectMapper objectMapper = new ObjectMapper();
+
+		private String baseUrl = "";
+
+		private String messageEndpoint;
+
+		private String sseEndpoint = DEFAULT_SSE_ENDPOINT;
+
+		private Duration keepAliveInterval;
+
+		/**
+		 * Sets the JSON object mapper to use for message serialization/deserialization.
+		 * @param objectMapper The object mapper to use
+		 * @return This builder instance for method chaining
+		 */
+		public Builder objectMapper(ObjectMapper objectMapper) {
+			Assert.notNull(objectMapper, "ObjectMapper must not be null");
+			this.objectMapper = objectMapper;
+			return this;
+		}
+
+		/**
+		 * Sets the base URL for the server transport.
+		 * @param baseUrl The base URL to use
+		 * @return This builder instance for method chaining
+		 */
+		public Builder baseUrl(String baseUrl) {
+			Assert.notNull(baseUrl, "Base URL must not be null");
+			this.baseUrl = baseUrl;
+			return this;
+		}
+
+		/**
+		 * Sets the endpoint path where clients will send their messages.
+		 * @param messageEndpoint The message endpoint path
+		 * @return This builder instance for method chaining
+		 */
+		public Builder messageEndpoint(String messageEndpoint) {
+			Assert.hasText(messageEndpoint, "Message endpoint must not be empty");
+			this.messageEndpoint = messageEndpoint;
+			return this;
+		}
+
+		/**
+		 * Sets the endpoint path where clients will establish SSE connections.
+		 * <p>
+		 * If not specified, the default value of {@link #DEFAULT_SSE_ENDPOINT} will be
+		 * used.
+		 * @param sseEndpoint The SSE endpoint path
+		 * @return This builder instance for method chaining
+		 */
+		public Builder sseEndpoint(String sseEndpoint) {
+			Assert.hasText(sseEndpoint, "SSE endpoint must not be empty");
+			this.sseEndpoint = sseEndpoint;
+			return this;
+		}
+
+		/**
+		 * Sets the interval for keep-alive pings.
+		 * <p>
+		 * If not specified, keep-alive pings will be disabled.
+		 * @param keepAliveInterval The interval duration for keep-alive pings
+		 * @return This builder instance for method chaining
+		 */
+		public Builder keepAliveInterval(Duration keepAliveInterval) {
+			this.keepAliveInterval = keepAliveInterval;
+			return this;
+		}
+
+		/**
+		 * Builds a new instance of WebMvcSseServerTransportProvider with the configured
+		 * settings.
+		 * @return A new WebMvcSseServerTransportProvider instance
+		 * @throws IllegalStateException if objectMapper or messageEndpoint is not set
+		 */
+		public WebMvcSseServerTransportProvider build() {
+			if (messageEndpoint == null) {
+				throw new IllegalStateException("MessageEndpoint must be set");
+			}
+			return new WebMvcSseServerTransportProvider(objectMapper, baseUrl, messageEndpoint, sseEndpoint,
+					keepAliveInterval);
+		}
+
+	}
+
 }