@@ -36,6 +36,7 @@
 
 import java.io.IOException;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -44,6 +45,7 @@
  * Implementation of a WebFlux based {@link McpStreamableServerTransportProvider}.
  *
  * @author Dariusz Jędrzejczyk
+ * @author Dimitar Proynov
  */
 public class WebFluxStreamableServerTransportProvider implements McpStreamableServerTransportProvider {
 
@@ -69,14 +71,34 @@ public class WebFluxStreamableServerTransportProvider implements McpStreamableSe
 
 	private KeepAliveScheduler keepAliveScheduler;
 
+	/**
+	 * Duration of inactivity after which an idle session is evicted. If {@code null},
+	 * idle sessions are never evicted.
+	 */
+	private final Duration sessionIdleTimeout;
+
+	/**
+	 * Tracks the last access time per session id, used to evict idle sessions.
+	 */
+	private final ConcurrentHashMap<String, Instant> sessionLastAccessTimes = new ConcurrentHashMap<>();
+
+	private Disposable idleSessionScheduler;
+
+	/**
+	 * Max number of session this provider supports before discarding new initialization
+	 * requests.
+	 */
+	private final long maxSessions;
+
 	/**
 	 * Security validator for validating HTTP requests.
 	 */
 	private final ServerTransportSecurityValidator securityValidator;
 
 	private WebFluxStreamableServerTransportProvider(McpJsonMapper jsonMapper, String mcpEndpoint,
 			McpTransportContextExtractor<ServerRequest> contextExtractor, boolean disallowDelete,
-			Duration keepAliveInterval, ServerTransportSecurityValidator securityValidator) {
+			Duration keepAliveInterval, Duration sessionIdleTimeout, ServerTransportSecurityValidator securityValidator,
+			long maxSessions) {
 		Assert.notNull(jsonMapper, "JsonMapper must not be null");
 		Assert.notNull(mcpEndpoint, "Message endpoint must not be null");
 		Assert.notNull(contextExtractor, "Context extractor must not be null");
@@ -87,6 +109,8 @@ private WebFluxStreamableServerTransportProvider(McpJsonMapper jsonMapper, Strin
 		this.contextExtractor = contextExtractor;
 		this.disallowDelete = disallowDelete;
 		this.securityValidator = securityValidator;
+		this.maxSessions = maxSessions;
+		this.sessionIdleTimeout = sessionIdleTimeout;
 		this.routerFunction = RouterFunctions.route()
 			.GET(this.mcpEndpoint, this::handleGet)
 			.POST(this.mcpEndpoint, this::handlePost)
@@ -102,6 +126,11 @@ private WebFluxStreamableServerTransportProvider(McpJsonMapper jsonMapper, Strin
 
 			this.keepAliveScheduler.start();
 		}
+
+		if (sessionIdleTimeout != null) {
+			this.idleSessionScheduler = Flux.interval(sessionIdleTimeout, sessionIdleTimeout)
+				.subscribe(tick -> this.evictIdleSessions());
+		}
 	}
 
 	@Override
@@ -142,9 +171,13 @@ public Mono<Void> closeGracefully() {
 				.then();
 		}).then().doOnSuccess(v -> {
 			sessions.clear();
+			this.sessionLastAccessTimes.clear();
 			if (this.keepAliveScheduler != null) {
 				this.keepAliveScheduler.shutdown();
 			}
+			if (this.idleSessionScheduler != null) {
+				this.idleSessionScheduler.dispose();
+			}
 		});
 	}
 
@@ -191,7 +224,7 @@ private Mono<ServerResponse> handleGet(ServerRequest request) {
 				return ServerResponse.badRequest().build();
 			}
 
-			if (request.headers().header(HttpHeaders.MCP_SESSION_ID).isEmpty()) {
+			if (!hasMcpSessionIdHeader(request)) {
 				return ServerResponse.badRequest().build(); // TODO: say we need a session
 															// id
 			}
@@ -204,6 +237,8 @@ private Mono<ServerResponse> handleGet(ServerRequest request) {
 				return ServerResponse.notFound().build();
 			}
 
+			touchSession(sessionId);
+
 			if (!request.headers().header(HttpHeaders.LAST_EVENT_ID).isEmpty()) {
 				String lastId = request.headers().asHttpHeaders().getFirst(HttpHeaders.LAST_EVENT_ID);
 				return ServerResponse.ok()
@@ -266,31 +301,10 @@ private Mono<ServerResponse> handlePost(ServerRequest request) {
 				McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(jsonMapper, body);
 				if (message instanceof McpSchema.JSONRPCRequest jsonrpcRequest
 						&& jsonrpcRequest.method().equals(McpSchema.METHOD_INITIALIZE)) {
-					var typeReference = new TypeRef<McpSchema.InitializeRequest>() {
-					};
-					McpSchema.InitializeRequest initializeRequest = jsonMapper.convertValue(jsonrpcRequest.params(),
-							typeReference);
-					McpStreamableServerSession.McpStreamableServerSessionInit init = this.sessionFactory
-						.startSession(initializeRequest);
-					sessions.put(init.session().getId(), init.session());
-					return init.initResult().map(initializeResult -> {
-						McpSchema.JSONRPCResponse jsonrpcResponse = new McpSchema.JSONRPCResponse(
-								McpSchema.JSONRPC_VERSION, jsonrpcRequest.id(), initializeResult, null);
-						try {
-							return this.jsonMapper.writeValueAsString(jsonrpcResponse);
-						}
-						catch (IOException e) {
-							logger.warn("Failed to serialize initResponse", e);
-							throw Exceptions.propagate(e);
-						}
-					})
-						.flatMap(initResult -> ServerResponse.ok()
-							.contentType(MediaType.APPLICATION_JSON)
-							.header(HttpHeaders.MCP_SESSION_ID, init.session().getId())
-							.bodyValue(initResult));
+					return handleInitRequest(jsonrpcRequest, hasMcpSessionIdHeader(request));
 				}
 
-				if (request.headers().header(HttpHeaders.MCP_SESSION_ID).isEmpty()) {
+				if (!hasMcpSessionIdHeader(request)) {
 					return ServerResponse.badRequest().bodyValue(new McpError("Session ID missing"));
 				}
 
@@ -302,6 +316,8 @@ private Mono<ServerResponse> handlePost(ServerRequest request) {
 						.bodyValue(new McpError("Session not found: " + sessionId));
 				}
 
+				touchSession(sessionId);
+
 				if (message instanceof McpSchema.JSONRPCResponse jsonrpcResponse) {
 					return session.accept(jsonrpcResponse).then(ServerResponse.accepted().build());
 				}
@@ -353,7 +369,7 @@ private Mono<ServerResponse> handleDelete(ServerRequest request) {
 		McpTransportContext transportContext = this.contextExtractor.extract(request);
 
 		return Mono.defer(() -> {
-			if (request.headers().header(HttpHeaders.MCP_SESSION_ID).isEmpty()) {
+			if (!hasMcpSessionIdHeader(request)) {
 				return ServerResponse.badRequest().build(); // TODO: say we need a session
 															// id
 			}
@@ -370,10 +386,99 @@ private Mono<ServerResponse> handleDelete(ServerRequest request) {
 				return ServerResponse.notFound().build();
 			}
 
-			return session.delete().then(ServerResponse.ok().build());
+			return session.delete().doOnSuccess(unused -> removeSession(sessionId)).then(ServerResponse.ok().build());
 		}).contextWrite(ctx -> ctx.put(McpTransportContext.KEY, transportContext));
 	}
 
+	private Mono<ServerResponse> handleInitRequest(McpSchema.JSONRPCRequest jsonrpcRequest,
+			boolean hasMcpSessionIdHeader) {
+		if (hasMcpSessionIdHeader) {
+			return ServerResponse.status(HttpStatus.BAD_REQUEST)
+				.bodyValue(McpError.builder(McpSchema.ErrorCodes.INTERNAL_ERROR)
+					.message("Session already initialized")
+					.build());
+		}
+		if (this.sessions.size() >= this.maxSessions) {
+			return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE)
+				.bodyValue(McpError.builder(McpSchema.ErrorCodes.INTERNAL_ERROR)
+					.message("Max number of sessions reached")
+					.build());
+		}
+		var typeReference = new TypeRef<McpSchema.InitializeRequest>() {
+		};
+		McpSchema.InitializeRequest initializeRequest = jsonMapper.convertValue(jsonrpcRequest.params(), typeReference);
+		McpStreamableServerSession.McpStreamableServerSessionInit init = this.sessionFactory
+			.startSession(initializeRequest);
+		sessions.put(init.session().getId(), init.session());
+		if (this.sessionIdleTimeout != null) {
+			this.sessionLastAccessTimes.put(init.session().getId(), Instant.now());
+		}
+		return init.initResult().map(initializeResult -> {
+			McpSchema.JSONRPCResponse jsonrpcResponse = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION,
+					jsonrpcRequest.id(), initializeResult, null);
+			try {
+				return this.jsonMapper.writeValueAsString(jsonrpcResponse);
+			}
+			catch (IOException e) {
+				logger.warn("Failed to serialize initResponse", e);
+				throw Exceptions.propagate(e);
+			}
+		})
+			.flatMap(initResult -> ServerResponse.ok()
+				.contentType(MediaType.APPLICATION_JSON)
+				.header(HttpHeaders.MCP_SESSION_ID, init.session().getId())
+				.bodyValue(initResult));
+	}
+
+	/**
+	 * Records the current time as the last access time for the given session, so that an
+	 * active session is not evicted as idle. No-op when idle eviction is disabled or the
+	 * session is no longer tracked.
+	 * @param sessionId the id of the session that was just accessed
+	 */
+	private void touchSession(String sessionId) {
+		if (this.sessionIdleTimeout != null) {
+			this.sessionLastAccessTimes.computeIfPresent(sessionId, (id, lastAccess) -> Instant.now());
+		}
+	}
+
+	/**
+	 * Evicts and closes sessions that have been idle for longer than the configured
+	 * {@code sessionIdleTimeout}. This reclaims session slots held by clients that
+	 * initialized a session but never interacted with it again, mitigating resource
+	 * exhaustion from abandoned sessions.
+	 */
+	private void evictIdleSessions() {
+		if (this.sessionIdleTimeout == null || this.isClosing) {
+			return;
+		}
+		Instant now = Instant.now();
+		for (var entry : this.sessionLastAccessTimes.entrySet()) {
+			if (Duration.between(entry.getValue(), now).compareTo(this.sessionIdleTimeout) > 0) {
+				String sessionId = entry.getKey();
+				McpStreamableServerSession session = removeSession(sessionId);
+				logger.debug("Evicting idle session: {}", sessionId);
+				if (session != null) {
+					session.closeGracefully().onErrorComplete().subscribe();
+				}
+			}
+		}
+	}
+
+	/**
+	 * Checks whether the given request carries an {@code Mcp-Session-Id} header.
+	 * @param request the incoming server request
+	 * @return {@code true} if the header is present, regardless of its value
+	 */
+	private static boolean hasMcpSessionIdHeader(ServerRequest request) {
+		return !request.headers().header(HttpHeaders.MCP_SESSION_ID).isEmpty();
+	}
+
+	private McpStreamableServerSession removeSession(String sessionId) {
+		this.sessionLastAccessTimes.remove(sessionId);
+		return this.sessions.remove(sessionId);
+	}
+
 	private class WebFluxStreamableMcpSessionTransport implements McpStreamableServerTransport {
 
 		private final FluxSink<ServerSentEvent<?>> sink;
@@ -450,8 +555,12 @@ public static class Builder {
 
 		private Duration keepAliveInterval;
 
+		private long maxSessions = 100_000L;
+
 		private ServerTransportSecurityValidator securityValidator = ServerTransportSecurityValidator.NOOP;
 
+		private Duration sessionIdleTimeout;
+
 		private Builder() {
 			// used by a static method
 		}
@@ -531,6 +640,41 @@ public Builder securityValidator(ServerTransportSecurityValidator securityValida
 			return this;
 		}
 
+		/**
+		 * Sets the maximum supported sessions for this provider. If not set defaults to
+		 * 100_000.
+		 * @param maxSessions - maximum allowed sessions for this provider
+		 * @return this builder instance
+		 * @throws IllegalArgumentException if maximum sessions is not positive number
+		 */
+		public Builder maxSessions(long maxSessions) {
+			Assert.isTrue(maxSessions >= 0, "Max sessions must be greater than 0");
+			this.maxSessions = maxSessions;
+			return this;
+		}
+
+		/**
+		 * Sets the idle timeout after which sessions with no client activity are evicted
+		 * and closed. Activity is any inbound request (GET, POST or DELETE) resolving to
+		 * the session. This reclaims session slots held by clients that initialized a
+		 * session but never interacted with it again, mitigating resource exhaustion from
+		 * abandoned sessions.
+		 * <p>
+		 * Keep-alive and idle eviction are complementary and may be enabled together:
+		 * keep-alive pings preserve connected clients (a ping response counts as
+		 * activity), while idle eviction reclaims abandoned sessions. When both are set,
+		 * {@code sessionIdleTimeout} must be greater than the keep-alive interval so that
+		 * a connected client answering pings is not evicted before it can respond;
+		 * otherwise {@link #build()} throws.
+		 * @param sessionIdleTimeout the maximum idle duration before eviction, or null to
+		 * disable idle eviction (the default)
+		 * @return this builder instance
+		 */
+		public Builder sessionIdleTimeout(Duration sessionIdleTimeout) {
+			this.sessionIdleTimeout = sessionIdleTimeout;
+			return this;
+		}
+
 		/**
 		 * Builds a new instance of {@link WebFluxStreamableServerTransportProvider} with
 		 * the configured settings.
@@ -539,9 +683,14 @@ public Builder securityValidator(ServerTransportSecurityValidator securityValida
 		 */
 		public WebFluxStreamableServerTransportProvider build() {
 			Assert.notNull(mcpEndpoint, "Message endpoint must be set");
+			if (this.sessionIdleTimeout != null && this.keepAliveInterval != null) {
+				Assert.isTrue(this.sessionIdleTimeout.compareTo(this.keepAliveInterval) > 0,
+						"sessionIdleTimeout must be greater than keepAliveInterval, otherwise a connected client "
+								+ "may be evicted before it can answer a keep-alive ping");
+			}
 			return new WebFluxStreamableServerTransportProvider(
 					jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper, mcpEndpoint, contextExtractor,
-					disallowDelete, keepAliveInterval, securityValidator);
+					disallowDelete, keepAliveInterval, sessionIdleTimeout, securityValidator, maxSessions);
 		}
 
 	}