This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 87050476ec1c CAMEL-24327: Add MCP session keep-alive and idle TTL
eviction
87050476ec1c is described below
commit 87050476ec1cf656bbb6018b519973b38699e21c
Author: Omar Atie <[email protected]>
AuthorDate: Thu Aug 6 13:02:50 2026 -0700
CAMEL-24327: Add MCP session keep-alive and idle TTL eviction
Evict orphaned MCP streamable HTTP sessions on the Vert.x transport using
periodic keep-alive pings (default 30s, evict after 3 consecutive failures)
and a configurable idle TTL (default 5 minutes). POST-only sessions without
an SSE stream are not pinged; idle TTL handles those. Configuration via
camel.server.mcp-session-keep-alive-interval and
camel.server.mcp-session-idle-ttl properties (set to 0 to disable).
Closes #25397
Co-Authored-By: Cursor <[email protected]>
---
.../org/apache/camel/catalog/docs/main.adoc | 4 +-
.../org/apache/camel/catalog/docs/mcp-server.adoc | 13 ++
.../main/camel-main-configuration-metadata.json | 2 +
.../component/mcp/server/McpServerBridge.java | 8 +-
.../mcp/server/McpServerConfiguration.java | 27 +++
.../component/mcp/server/McpServerConstants.java | 12 ++
.../camel/component/mcp/server/McpServerInfo.java | 34 ++-
.../camel-mcp-server/src/main/docs/mcp-server.adoc | 13 ++
.../mcp/server/main/DefaultMcpServerFactory.java | 2 +
.../mcp/server/vertx/VertxMcpServerEngine.java | 8 +-
.../VertxMcpStreamableServerTransportProvider.java | 230 ++++++++++++++++++---
.../McpServerSessionEvictionPropertiesTest.java | 90 ++++++++
.../server/vertx/VertxMcpSessionEvictionTest.java | 158 ++++++++++++++
...ttpServerConfigurationPropertiesConfigurer.java | 14 ++
.../camel-main-configuration-metadata.json | 2 +
core/camel-main/src/main/docs/main.adoc | 4 +-
.../main/HttpServerConfigurationProperties.java | 44 ++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 8 +
18 files changed, 641 insertions(+), 32 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/main.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/main.adoc
index e3d33c3295b0..db7bd0eb0a19 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/main.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/main.adoc
@@ -195,7 +195,7 @@ The camel.routecontroller supports 12 options, which are
listed below.
=== Camel Embedded HTTP Server (only for standalone; not Spring Boot or
Quarkus) configurations
-The camel.server supports 26 options, which are listed below.
+The camel.server supports 28 options, which are listed below.
[width="100%",cols="2,5,^1,2",options="header"]
|===
@@ -218,6 +218,8 @@ The camel.server supports 26 options, which are listed
below.
| *camel.server.mcpEnabled* | Whether to expose ai-tool routes as MCP tools
over streamable HTTP. Requires camel-mcp-server on the classpath. By default,
the MCP server is not enabled. | false | boolean
| *camel.server.mcpPath* | HTTP path where the MCP endpoint is served. | /mcp
| String
| *camel.server.mcpServerName* | MCP server name advertised to clients.
Defaults to the CamelContext name. | | String
+| *camel.server.mcpSessionIdleTtl* | Idle TTL in milliseconds for MCP sessions
on the Vert.x streamable transport. Sessions with no activity for longer than
this interval are evicted. 0 disables idle eviction. | 300000 | long
+| *camel.server.mcpSessionKeepAliveInterval* | Keep-alive ping interval in
milliseconds for MCP sessions on the Vert.x streamable transport. Dead sessions
are evicted after consecutive ping failures. 0 disables keep-alive pings. |
30000 | long
| *camel.server.mcpTags* | Comma-separated list of ai-tool tags to expose as
MCP tools. Only tools registered under one of these tags are exposed; the
untagged default pool is never exposed. When not set, no tools are exposed. |
| String
| *camel.server.mcpToolTimeout* | Per-call MCP tool execution timeout in
milliseconds. A call exceeding the timeout returns an error result to the MCP
client; the underlying route keeps running until it completes on its own. |
20000 | long
| *camel.server.path* | Context-path to use for embedded HTTP server | / |
String
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mcp-server.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mcp-server.adoc
index ffe21a708a75..fdc8fdef6e3a 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mcp-server.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/mcp-server.adoc
@@ -169,6 +169,14 @@ JBang (see the xref:main.adoc[camel-main] options) or on
`/mcp` | engine
| `camel.server.mcp-server-name` | MCP server name advertised to clients. |
CamelContext name | engine
+| `camel.server.mcp-session-keep-alive-interval` | Keep-alive ping interval in
+ milliseconds for MCP sessions on the Vert.x streamable transport. Dead
sessions
+ are evicted after consecutive ping failures. `0` disables keep-alive
+ pings. | `30000` | engine
+| `camel.server.mcp-session-idle-ttl` | Idle TTL in milliseconds for MCP
+ sessions on the Vert.x streamable transport. Sessions with no activity for
+ longer than this interval are evicted. `0` disables idle eviction. |
+ `300000` | engine
|===
Bridge-owned options are honored identically on every runtime. Engine-owned
@@ -192,6 +200,11 @@ The Vert.x engine implements the MCP streamable HTTP
transport:
`Last-Event-ID` replay,
* session management via the `Mcp-Session-Id` header and `DELETE /mcp` for
session termination.
+* active session eviction: keep-alive pings (default every 30 seconds) remove
+ sessions whose ping fails repeatedly, and an idle TTL (default 5 minutes)
+ removes sessions with no traffic. Configure with
+ `camel.server.mcp-session-keep-alive-interval` and
+ `camel.server.mcp-session-idle-ttl` (set either to `0` to disable).
Tools appearing or disappearing (routes starting and stopping) emit
`notifications/tools/list_changed` to connected clients.
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
index 3d8ec1317988..c03dd1557f00 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/main/camel-main-configuration-metadata.json
@@ -442,6 +442,8 @@
{ "name": "camel.server.mcpEnabled", "required": false, "description":
"Whether to expose ai-tool routes as MCP tools over streamable HTTP. Requires
camel-mcp-server on the classpath. By default, the MCP server is not enabled.",
"sourceType": "org.apache.camel.main.HttpServerConfigurationProperties",
"type": "boolean", "javaType": "boolean", "defaultValue": false, "secret":
false },
{ "name": "camel.server.mcpPath", "required": false, "description": "HTTP
path where the MCP endpoint is served.", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "string",
"javaType": "java.lang.String", "defaultValue": "\/mcp", "secret": false },
{ "name": "camel.server.mcpServerName", "required": false, "description":
"MCP server name advertised to clients. Defaults to the CamelContext name.",
"sourceType": "org.apache.camel.main.HttpServerConfigurationProperties",
"type": "string", "javaType": "java.lang.String", "secret": false },
+ { "name": "camel.server.mcpSessionIdleTtl", "required": false,
"description": "Idle TTL in milliseconds for MCP sessions on the Vert.x
streamable transport. Sessions with no activity for longer than this interval
are evicted. 0 disables idle eviction.", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "integer",
"javaType": "long", "defaultValue": 300000, "secret": false },
+ { "name": "camel.server.mcpSessionKeepAliveInterval", "required": false,
"description": "Keep-alive ping interval in milliseconds for MCP sessions on
the Vert.x streamable transport. Dead sessions are evicted after consecutive
ping failures. 0 disables keep-alive pings.", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "integer",
"javaType": "long", "defaultValue": 30000, "secret": false },
{ "name": "camel.server.mcpTags", "required": false, "description":
"Comma-separated list of ai-tool tags to expose as MCP tools. Only tools
registered under one of these tags are exposed; the untagged default pool is
never exposed. When not set, no tools are exposed.", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "string",
"javaType": "java.lang.String", "secret": false },
{ "name": "camel.server.mcpToolTimeout", "required": false, "description":
"Per-call MCP tool execution timeout in milliseconds. A call exceeding the
timeout returns an error result to the MCP client; the underlying route keeps
running until it completes on its own.", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "integer",
"javaType": "long", "defaultValue": 20000, "secret": false },
{ "name": "camel.server.path", "required": false, "description":
"Context-path to use for embedded HTTP server", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "string",
"javaType": "java.lang.String", "defaultValue": "\/", "secret": false },
diff --git
a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerBridge.java
b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerBridge.java
index cb94a38b520a..1c478aed0b1c 100644
---
a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerBridge.java
+++
b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerBridge.java
@@ -117,7 +117,13 @@ public class McpServerBridge extends ServiceSupport
implements CamelContextAware
CamelContextAware.trySetCamelContext(engine, camelContext);
String serverName = configuration.getServerName() != null ?
configuration.getServerName() : camelContext.getName();
- engine.initialize(new McpServerInfo(serverName,
camelContext.getVersion(), configuration.getPath()));
+ String version = camelContext.getVersion();
+ if (version == null || version.isBlank()) {
+ version = "1.0";
+ }
+ engine.initialize(new McpServerInfo(
+ serverName, version, configuration.getPath(),
+ configuration.getSessionKeepAliveInterval(),
configuration.getSessionIdleTtl()));
if (!engine.consumesServingConfiguration()) {
if
(!McpServerConstants.DEFAULT_PATH.equals(configuration.getPath())) {
diff --git
a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConfiguration.java
b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConfiguration.java
index 3439bf7db8de..376b4c37637a 100644
---
a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConfiguration.java
+++
b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConfiguration.java
@@ -31,6 +31,8 @@ public class McpServerConfiguration {
private long toolTimeout = McpServerConstants.DEFAULT_TOOL_TIMEOUT;
private String path = McpServerConstants.DEFAULT_PATH;
private String serverName;
+ private long sessionKeepAliveInterval =
McpServerConstants.DEFAULT_SESSION_KEEP_ALIVE_INTERVAL;
+ private long sessionIdleTtl = McpServerConstants.DEFAULT_SESSION_IDLE_TTL;
/**
* Comma-separated list of ai-tool tags to expose as MCP tools. Only tools
registered under one of these tags are
@@ -78,4 +80,29 @@ public class McpServerConfiguration {
public void setServerName(String serverName) {
this.serverName = serverName;
}
+
+ /**
+ * Keep-alive ping interval in milliseconds for the Vert.x streamable
transport. Dead sessions are evicted after
+ * consecutive ping failures. {@code 0} disables keep-alive pings.
Engine-owned: ignored by native engines.
+ */
+ public long getSessionKeepAliveInterval() {
+ return sessionKeepAliveInterval;
+ }
+
+ public void setSessionKeepAliveInterval(long sessionKeepAliveInterval) {
+ this.sessionKeepAliveInterval = sessionKeepAliveInterval;
+ }
+
+ /**
+ * Idle TTL in milliseconds for MCP sessions managed by the Vert.x
streamable transport. Sessions with no activity
+ * for longer than this interval are evicted. {@code 0} disables idle
eviction. Engine-owned: ignored by native
+ * engines.
+ */
+ public long getSessionIdleTtl() {
+ return sessionIdleTtl;
+ }
+
+ public void setSessionIdleTtl(long sessionIdleTtl) {
+ this.sessionIdleTtl = sessionIdleTtl;
+ }
}
diff --git
a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConstants.java
b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConstants.java
index 59ffd9b8b717..59750cce6e1c 100644
---
a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConstants.java
+++
b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerConstants.java
@@ -39,6 +39,18 @@ public final class McpServerConstants {
*/
public static final long DEFAULT_TOOL_TIMEOUT = 20_000;
+ /**
+ * Default MCP session keep-alive ping interval in milliseconds for the
Vert.x streamable transport. {@code 0}
+ * disables keep-alive pings.
+ */
+ public static final long DEFAULT_SESSION_KEEP_ALIVE_INTERVAL = 30_000;
+
+ /**
+ * Default MCP session idle TTL in milliseconds for the Vert.x streamable
transport. Sessions with no activity for
+ * longer than this interval are evicted. {@code 0} disables idle eviction.
+ */
+ public static final long DEFAULT_SESSION_IDLE_TTL = 300_000;
+
private McpServerConstants() {
}
}
diff --git
a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerInfo.java
b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerInfo.java
index 1893a4789c86..9e13bfb4baa0 100644
---
a/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerInfo.java
+++
b/components/camel-ai/camel-mcp-server-api/src/main/java/org/apache/camel/component/mcp/server/McpServerInfo.java
@@ -22,11 +22,35 @@ package org.apache.camel.component.mcp.server;
* Engines backed by a native runtime MCP server (Quarkus, Spring Boot) MAY
ignore the serving hints ({@code path}) —
* their own runtime configuration decides how the server is exposed.
*
- * @param serverName the MCP server name advertised to clients (defaults to
the CamelContext name)
- * @param version the MCP server version advertised to clients
- * @param path the HTTP path where the MCP endpoint should be served
+ * @param serverName the MCP server name advertised to clients
(defaults to the CamelContext name)
+ * @param version the MCP server version advertised to
clients
+ * @param path the HTTP path where the MCP endpoint
should be served
+ * @param sessionKeepAliveIntervalMs keep-alive ping interval in milliseconds
for the Vert.x streamable transport;
+ * {@code 0} disables keep-alive pings
+ * @param sessionIdleTtlMs idle TTL in milliseconds; sessions with
no activity for longer are evicted;
+ * {@code 0} disables idle eviction
*
- * @since 4.22
+ * @since 4.22
*/
-public record McpServerInfo(String serverName, String version, String path) {
+public record McpServerInfo(
+ String serverName,
+ String version,
+ String path,
+ long sessionKeepAliveIntervalMs,
+ long sessionIdleTtlMs) {
+
+ public McpServerInfo(String serverName, String version, String path) {
+ this(serverName, version, path,
+ McpServerConstants.DEFAULT_SESSION_KEEP_ALIVE_INTERVAL,
+ McpServerConstants.DEFAULT_SESSION_IDLE_TTL);
+ }
+
+ public McpServerInfo {
+ if (sessionKeepAliveIntervalMs < 0) {
+ throw new IllegalArgumentException("sessionKeepAliveIntervalMs
must be >= 0");
+ }
+ if (sessionIdleTtlMs < 0) {
+ throw new IllegalArgumentException("sessionIdleTtlMs must be >=
0");
+ }
+ }
}
diff --git a/components/camel-ai/camel-mcp-server/src/main/docs/mcp-server.adoc
b/components/camel-ai/camel-mcp-server/src/main/docs/mcp-server.adoc
index ffe21a708a75..fdc8fdef6e3a 100644
--- a/components/camel-ai/camel-mcp-server/src/main/docs/mcp-server.adoc
+++ b/components/camel-ai/camel-mcp-server/src/main/docs/mcp-server.adoc
@@ -169,6 +169,14 @@ JBang (see the xref:main.adoc[camel-main] options) or on
`/mcp` | engine
| `camel.server.mcp-server-name` | MCP server name advertised to clients. |
CamelContext name | engine
+| `camel.server.mcp-session-keep-alive-interval` | Keep-alive ping interval in
+ milliseconds for MCP sessions on the Vert.x streamable transport. Dead
sessions
+ are evicted after consecutive ping failures. `0` disables keep-alive
+ pings. | `30000` | engine
+| `camel.server.mcp-session-idle-ttl` | Idle TTL in milliseconds for MCP
+ sessions on the Vert.x streamable transport. Sessions with no activity for
+ longer than this interval are evicted. `0` disables idle eviction. |
+ `300000` | engine
|===
Bridge-owned options are honored identically on every runtime. Engine-owned
@@ -192,6 +200,11 @@ The Vert.x engine implements the MCP streamable HTTP
transport:
`Last-Event-ID` replay,
* session management via the `Mcp-Session-Id` header and `DELETE /mcp` for
session termination.
+* active session eviction: keep-alive pings (default every 30 seconds) remove
+ sessions whose ping fails repeatedly, and an idle TTL (default 5 minutes)
+ removes sessions with no traffic. Configure with
+ `camel.server.mcp-session-keep-alive-interval` and
+ `camel.server.mcp-session-idle-ttl` (set either to `0` to disable).
Tools appearing or disappearing (routes starting and stopping) emit
`notifications/tools/list_changed` to connected clients.
diff --git
a/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/main/DefaultMcpServerFactory.java
b/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/main/DefaultMcpServerFactory.java
index d524fa46fa20..e080fdf2c559 100644
---
a/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/main/DefaultMcpServerFactory.java
+++
b/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/main/DefaultMcpServerFactory.java
@@ -39,6 +39,8 @@ public class DefaultMcpServerFactory implements
McpServerFactory {
mcpConfiguration.setToolTimeout(configuration.getMcpToolTimeout());
mcpConfiguration.setPath(configuration.getMcpPath());
mcpConfiguration.setServerName(configuration.getMcpServerName());
+
mcpConfiguration.setSessionKeepAliveInterval(configuration.getMcpSessionKeepAliveInterval());
+
mcpConfiguration.setSessionIdleTtl(configuration.getMcpSessionIdleTtl());
return new McpServerBridge(mcpConfiguration);
}
}
diff --git
a/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpServerEngine.java
b/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpServerEngine.java
index a8381639dbd3..a0435525722c 100644
---
a/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpServerEngine.java
+++
b/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpServerEngine.java
@@ -104,7 +104,9 @@ public class VertxMcpServerEngine extends ServiceSupport
implements McpServerEng
protected void doStart() throws Exception {
VertxPlatformHttpRouter router = lookupRouter();
jsonMapper = McpJsonDefaults.getMapper();
- transport = new VertxMcpStreamableServerTransportProvider(jsonMapper,
info.path());
+ transport = new VertxMcpStreamableServerTransportProvider(
+ jsonMapper, info.path(),
+ info.sessionKeepAliveIntervalMs(), info.sessionIdleTtlMs());
server = McpServer.sync(transport)
.serverInfo(info.serverName(), info.version())
.capabilities(McpSchema.ServerCapabilities.builder().tools(true).build())
@@ -167,6 +169,10 @@ public class VertxMcpServerEngine extends ServiceSupport
implements McpServerEng
}
}
+ int sessionCount() {
+ return transport != null ? transport.sessionCount() : 0;
+ }
+
private McpSchema.Tool buildMcpTool(McpServerTool tool) {
String schema = tool.inputSchemaJson() != null ?
tool.inputSchemaJson() : EMPTY_OBJECT_SCHEMA;
McpSchema.Tool.Builder builder = McpSchema.Tool.builder(tool.name(),
jsonMapper, schema)
diff --git
a/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpStreamableServerTransportProvider.java
b/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpStreamableServerTransportProvider.java
index 12c8e5ed90a2..7ef1cf0a2ffe 100644
---
a/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpStreamableServerTransportProvider.java
+++
b/components/camel-ai/camel-mcp-server/src/main/java/org/apache/camel/component/mcp/server/vertx/VertxMcpStreamableServerTransportProvider.java
@@ -20,6 +20,8 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import io.modelcontextprotocol.json.McpJsonMapper;
import io.modelcontextprotocol.json.TypeRef;
@@ -59,6 +61,8 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
private static final Duration NOTIFICATION_TIMEOUT = Duration.ofSeconds(5);
private static final Duration INITIALIZATION_TIMEOUT =
Duration.ofSeconds(30);
+ private static final Duration KEEP_ALIVE_PING_TIMEOUT =
Duration.ofSeconds(5);
+ private static final int MAX_CONSECUTIVE_PING_FAILURES = 3;
private static final String ACCEPT = "Accept";
private static final String APPLICATION_JSON = "application/json";
@@ -66,15 +70,30 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
private final McpJsonMapper jsonMapper;
private final String path;
- private final ConcurrentHashMap<String, McpStreamableServerSession>
sessions = new ConcurrentHashMap<>();
+ private final long sessionKeepAliveIntervalMs;
+ private final long sessionIdleTtlMs;
+ private final ConcurrentHashMap<String, ManagedSession> sessions = new
ConcurrentHashMap<>();
private final List<Route> routes = new ArrayList<>();
private McpStreamableServerSession.Factory sessionFactory;
private volatile boolean closing;
+ private Vertx vertx;
+ private Long keepAliveTimerId;
+ private Long idleTimerId;
+ private final AtomicBoolean keepAliveInProgress = new AtomicBoolean();
+ private final AtomicBoolean idleEvictionInProgress = new AtomicBoolean();
public VertxMcpStreamableServerTransportProvider(McpJsonMapper jsonMapper,
String path) {
+ this(jsonMapper, path, 0, 0);
+ }
+
+ public VertxMcpStreamableServerTransportProvider(
+ McpJsonMapper jsonMapper,
String path, long sessionKeepAliveIntervalMs,
+ long sessionIdleTtlMs) {
this.jsonMapper = jsonMapper;
this.path = path;
+ this.sessionKeepAliveIntervalMs = sessionKeepAliveIntervalMs;
+ this.sessionIdleTtlMs = sessionIdleTtlMs;
}
@Override
@@ -87,12 +106,12 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
if (sessions.isEmpty()) {
return Mono.empty();
}
- return Mono.fromRunnable(() -> sessions.values().forEach(session -> {
+ return Mono.fromRunnable(() -> sessions.values().forEach(managed -> {
try {
// bounded so a single stalled session cannot starve
notifications to healthy sessions
- session.sendNotification(method,
params).block(NOTIFICATION_TIMEOUT);
+ managed.session.sendNotification(method,
params).block(NOTIFICATION_TIMEOUT);
} catch (Exception e) {
- LOG.debug("Failed to send notification to MCP session {}: {}",
session.getId(), e.getMessage());
+ LOG.debug("Failed to send notification to MCP session {}: {}",
managed.session.getId(), e.getMessage());
}
}));
}
@@ -101,11 +120,12 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
public Mono<Void> closeGracefully() {
return Mono.fromRunnable(() -> {
closing = true;
- sessions.values().forEach(session -> {
+ stopSessionMaintenance();
+ sessions.values().forEach(managed -> {
try {
- session.closeGracefully().block(NOTIFICATION_TIMEOUT);
+
managed.session.closeGracefully().block(NOTIFICATION_TIMEOUT);
} catch (Exception e) {
- LOG.debug("Failed to close MCP session {}: {}",
session.getId(), e.getMessage());
+ LOG.debug("Failed to close MCP session {}: {}",
managed.session.getId(), e.getMessage());
}
});
sessions.clear();
@@ -117,7 +137,7 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
* (the server sets the session factory on construction).
*/
public void registerRoutes(VertxPlatformHttpRouter router) {
- Vertx vertx = router.vertx();
+ vertx = router.vertx();
Route post = router.route(path).method(HttpMethod.POST);
post.handler(BodyHandler.create(false));
post.handler(ctx -> dispatch(vertx, ctx, this::handlePost));
@@ -128,13 +148,19 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
Route delete = router.route(path).method(HttpMethod.DELETE);
delete.handler(ctx -> dispatch(vertx, ctx, this::handleDelete));
routes.add(delete);
+ startSessionMaintenance();
}
public void unregisterRoutes() {
+ stopSessionMaintenance();
routes.forEach(Route::remove);
routes.clear();
}
+ int sessionCount() {
+ return sessions.size();
+ }
+
@FunctionalInterface
private interface BlockingRequestHandler {
void handle(RoutingContext ctx, Context connection) throws Exception;
@@ -200,23 +226,24 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
if (respondBadRequest(connection, ctx, badRequestErrors)) {
return;
}
- McpStreamableServerSession session = sessions.get(sessionId);
- if (session == null) {
+ ManagedSession managed = sessions.get(sessionId);
+ if (managed == null) {
respondError(connection, ctx, 404,
McpError.builder(McpSchema.ErrorCodes.INTERNAL_ERROR)
.message("Session not found: " + sessionId).build());
return;
}
+ managed.touch();
if (message instanceof McpSchema.JSONRPCResponse response) {
- session.accept(response).block();
+ managed.session.accept(response).block();
endWithStatus(connection, ctx, 202);
} else if (message instanceof McpSchema.JSONRPCNotification
notification) {
- session.accept(notification).block();
+ managed.session.accept(notification).block();
endWithStatus(connection, ctx, 202);
} else if (message instanceof McpSchema.JSONRPCRequest request) {
VertxMcpSessionTransport transport = startSseResponse(ctx,
connection, sessionId);
try {
- session.responseStream(request, transport).block();
+ managed.session.responseStream(request, transport).block();
} catch (Exception e) {
LOG.warn("Failed to handle MCP request stream: {}",
e.getMessage());
transport.close();
@@ -233,7 +260,7 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
= jsonMapper.convertValue(request.params(), new
TypeRef<McpSchema.InitializeRequest>() {
});
McpStreamableServerSession.McpStreamableServerSessionInit init =
sessionFactory.startSession(initializeRequest);
- sessions.put(init.session().getId(), init.session());
+ sessions.put(init.session().getId(), new
ManagedSession(init.session()));
McpSchema.InitializeResult initResult =
init.initResult().block(INITIALIZATION_TIMEOUT);
String json =
jsonMapper.writeValueAsString(McpSchema.JSONRPCResponse.result(request.id(),
initResult));
connection.runOnContext(v -> ctx.response()
@@ -260,24 +287,25 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
if (respondBadRequest(connection, ctx, badRequestErrors)) {
return;
}
- McpStreamableServerSession session = sessions.get(sessionId);
- if (session == null) {
+ ManagedSession managed = sessions.get(sessionId);
+ if (managed == null) {
endWithStatus(connection, ctx, 404);
return;
}
+ managed.touch();
VertxMcpSessionTransport transport = startSseResponse(ctx, connection,
sessionId);
String lastEventId =
ctx.request().getHeader(HttpHeaders.LAST_EVENT_ID);
if (lastEventId != null) {
try {
- session.replay(lastEventId).toIterable().forEach(message ->
transport.sendMessage(message).block());
+
managed.session.replay(lastEventId).toIterable().forEach(message ->
transport.sendMessage(message).block());
} catch (Exception e) {
LOG.warn("Failed to replay MCP messages: {}", e.getMessage());
transport.close();
}
} else {
McpStreamableServerSession.McpStreamableServerSessionStream
listeningStream
- = session.listeningStream(transport);
+ = managed.session.listeningStream(transport);
connection.runOnContext(v -> ctx.response().closeHandler(x ->
listeningStream.close()));
}
}
@@ -293,16 +321,141 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
.message("Session ID required in " +
HttpHeaders.MCP_SESSION_ID + " header").build());
return;
}
- McpStreamableServerSession session = sessions.get(sessionId);
- if (session == null) {
+ ManagedSession managed = sessions.get(sessionId);
+ if (managed == null) {
endWithStatus(connection, ctx, 404);
return;
}
- session.delete().block();
+ managed.session.delete().block();
sessions.remove(sessionId);
endWithStatus(connection, ctx, 200);
}
+ private void startSessionMaintenance() {
+ if (vertx == null) {
+ return;
+ }
+ if (sessionKeepAliveIntervalMs > 0) {
+ keepAliveTimerId = vertx.setPeriodic(sessionKeepAliveIntervalMs,
id -> {
+ if (!keepAliveInProgress.compareAndSet(false, true)) {
+ return;
+ }
+ vertx.executeBlocking(() -> {
+ try {
+ pingSessions();
+ } finally {
+ keepAliveInProgress.set(false);
+ }
+ return null;
+ }, false);
+ });
+ }
+ if (sessionIdleTtlMs > 0) {
+ long scanInterval = Math.max(100L, Math.min(sessionIdleTtlMs / 2,
1_000L));
+ idleTimerId = vertx.setPeriodic(scanInterval, id -> {
+ if (!idleEvictionInProgress.compareAndSet(false, true)) {
+ return;
+ }
+ vertx.executeBlocking(() -> {
+ try {
+ evictIdleSessions();
+ } finally {
+ idleEvictionInProgress.set(false);
+ }
+ return null;
+ }, false);
+ });
+ }
+ }
+
+ private void stopSessionMaintenance() {
+ if (vertx == null) {
+ return;
+ }
+ if (keepAliveTimerId != null) {
+ vertx.cancelTimer(keepAliveTimerId);
+ keepAliveTimerId = null;
+ }
+ if (idleTimerId != null) {
+ vertx.cancelTimer(idleTimerId);
+ idleTimerId = null;
+ }
+ }
+
+ private void pingSessions() {
+ if (closing || sessions.isEmpty()) {
+ return;
+ }
+ for (ManagedSession managed : sessions.values()) {
+ try {
+ managed.session.sendRequest(McpSchema.METHOD_PING, null, new
TypeRef<Object>() {
+ }).block(KEEP_ALIVE_PING_TIMEOUT);
+ managed.resetPingFailures();
+ managed.touch();
+ } catch (Exception e) {
+ if (isMissingTransport(e)) {
+ // POST-only clients without an open GET stream cannot
receive pings; idle TTL handles orphans
+ continue;
+ }
+ if (managed.recordPingFailure() >=
MAX_CONSECUTIVE_PING_FAILURES) {
+ evictSession(managed.session.getId(), "keep-alive ping
failed");
+ }
+ }
+ }
+ }
+
+ private static boolean isMissingTransport(Throwable error) {
+ Throwable current = error;
+ while (current != null) {
+ if (current instanceof IllegalStateException &&
current.getMessage() != null
+ && current.getMessage().contains("Stream unavailable")) {
+ return true;
+ }
+ current = current.getCause();
+ }
+ return false;
+ }
+
+ private void evictIdleSessions() {
+ if (closing || sessions.isEmpty()) {
+ return;
+ }
+ long cutoff = System.nanoTime() -
Duration.ofMillis(sessionIdleTtlMs).toNanos();
+ sessions.forEach((sessionId, managed) -> {
+ if (managed.lastActivityNanos() < cutoff) {
+ evictSession(sessionId, "idle TTL exceeded");
+ }
+ });
+ }
+
+ private void evictSession(String sessionId, String reason) {
+ ManagedSession managed = sessions.remove(sessionId);
+ if (managed == null) {
+ return;
+ }
+ LOG.debug("Evicting MCP session {}: {}", sessionId, reason);
+ closeSessionQuietly(managed.session, sessionId);
+ }
+
+ private void scheduleAsyncSessionClose(String sessionId,
McpStreamableServerSession session) {
+ if (vertx == null) {
+ closeSessionQuietly(session, sessionId);
+ return;
+ }
+ vertx.executeBlocking(() -> {
+ closeSessionQuietly(session, sessionId);
+ return null;
+ }, false);
+ }
+
+ private void closeSessionQuietly(McpStreamableServerSession session,
String sessionId) {
+ try {
+ session.closeGracefully().block(NOTIFICATION_TIMEOUT);
+ } catch (Exception e) {
+ LOG.debug("Failed to close MCP session {}: {}", sessionId,
e.getMessage());
+ }
+ }
+
private VertxMcpSessionTransport startSseResponse(RoutingContext ctx,
Context connection, String sessionId) {
connection.runOnContext(v -> ctx.response()
.setChunked(true)
@@ -338,6 +491,35 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
connection.runOnContext(v ->
ctx.response().setStatusCode(status).end());
}
+ private static final class ManagedSession {
+
+ private final McpStreamableServerSession session;
+ private volatile long lastActivityNanos;
+ private final AtomicInteger consecutivePingFailures = new
AtomicInteger();
+
+ private ManagedSession(McpStreamableServerSession session) {
+ this.session = session;
+ touch();
+ }
+
+ private void touch() {
+ lastActivityNanos = System.nanoTime();
+ consecutivePingFailures.set(0);
+ }
+
+ private long lastActivityNanos() {
+ return lastActivityNanos;
+ }
+
+ private void resetPingFailures() {
+ consecutivePingFailures.set(0);
+ }
+
+ private int recordPingFailure() {
+ return consecutivePingFailures.incrementAndGet();
+ }
+ }
+
/**
* Per-connection transport writing SSE frames on the connection's
event-loop context. The SDK session awaits each
* write, providing natural backpressure.
@@ -377,8 +559,10 @@ public class VertxMcpStreamableServerTransportProvider
implements McpStreamableS
LOG.debug("Failed to write to MCP session {}: {}",
sessionId,
result.cause() != null ?
result.cause().getMessage() : "unknown");
closed = true;
- // the client is gone: drop the session like the
SDK servlet transport does
- sessions.remove(sessionId);
+ ManagedSession managed =
sessions.remove(sessionId);
+ if (managed != null) {
+ scheduleAsyncSessionClose(sessionId,
managed.session);
+ }
}
sink.success();
});
diff --git
a/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/McpServerSessionEvictionPropertiesTest.java
b/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/McpServerSessionEvictionPropertiesTest.java
new file mode 100644
index 000000000000..0eb27a4e10bf
--- /dev/null
+++
b/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/McpServerSessionEvictionPropertiesTest.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mcp.server.vertx;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mcp.server.McpServerBridge;
+import org.apache.camel.main.Main;
+import org.apache.camel.test.AvailablePortFinder;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Verifies MCP session idle TTL wiring from {@code
camel.server.mcp-session-*} properties (CAMEL-24327).
+ */
+class McpServerSessionEvictionPropertiesTest {
+
+ private static final HttpClient HTTP = HttpClient.newHttpClient();
+
+ private static final String INITIALIZE_REQUEST = """
+
{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05",
+ "capabilities":{},"clientInfo":{"name":"test","version":"1"}}}""";
+
+ @Test
+ void testSessionIdleTtlPropertyEvictsOrphanSessions() throws Exception {
+ int port = AvailablePortFinder.getNextAvailable();
+
+ Main main = new Main();
+ main.configure().addRoutesBuilder(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("ai-tool:ping?tags=crm&description=Ping tool")
+ .setBody(constant("pong"));
+ }
+ });
+ main.addInitialProperty("camel.server.enabled", "true");
+ main.addInitialProperty("camel.server.port", String.valueOf(port));
+ main.addInitialProperty("camel.server.mcp-enabled", "true");
+ main.addInitialProperty("camel.server.mcp-tags", "crm");
+
main.addInitialProperty("camel.server.mcp-session-keep-alive-interval", "0");
+ main.addInitialProperty("camel.server.mcp-session-idle-ttl", "500");
+ main.start();
+
+ try {
+ initializeSession(port);
+ McpServerBridge bridge =
main.getCamelContext().hasService(McpServerBridge.class);
+ assertThat(bridge).isNotNull();
+
assertThat(bridge.getEngine()).isInstanceOf(VertxMcpServerEngine.class);
+ VertxMcpServerEngine engine = (VertxMcpServerEngine)
bridge.getEngine();
+ assertThat(engine.sessionCount()).isEqualTo(1);
+
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
assertThat(engine.sessionCount()).isZero());
+ } finally {
+ main.stop();
+ }
+ }
+
+ private static void initializeSession(int port) throws Exception {
+ HttpResponse<String> response = HTTP.send(HttpRequest.newBuilder()
+ .uri(URI.create("http://localhost:"; + port + "/mcp"))
+ .timeout(Duration.ofSeconds(10))
+ .header("Content-Type", "application/json")
+ .header("Accept", "application/json, text/event-stream")
+ .POST(HttpRequest.BodyPublishers.ofString(INITIALIZE_REQUEST))
+ .build(), HttpResponse.BodyHandlers.ofString());
+ assertThat(response.statusCode()).isEqualTo(200);
+ }
+}
diff --git
a/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/VertxMcpSessionEvictionTest.java
b/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/VertxMcpSessionEvictionTest.java
new file mode 100644
index 000000000000..d8232a6db1a7
--- /dev/null
+++
b/components/camel-ai/camel-mcp-server/src/test/java/org/apache/camel/component/mcp/server/vertx/VertxMcpSessionEvictionTest.java
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mcp.server.vertx;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.mcp.server.McpServerInfo;
+import org.apache.camel.component.platform.http.main.MainHttpServer;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.AvailablePortFinder;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Verifies active MCP session eviction on the Vert.x streamable transport
(CAMEL-24327).
+ */
+class VertxMcpSessionEvictionTest {
+
+ private static final HttpClient HTTP = HttpClient.newHttpClient();
+
+ private static final String INITIALIZE_REQUEST = """
+
{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05",
+ "capabilities":{},"clientInfo":{"name":"test","version":"1"}}}""";
+
+ private static final String INITIALIZED_NOTIFICATION = """
+
{"jsonrpc":"2.0","method":"notifications/initialized","params":{}}""";
+
+ @Test
+ void evictsIdleSessionWithoutDelete() throws Exception {
+ int port = AvailablePortFinder.getNextAvailable();
+ VertxMcpServerEngine engine = startEngine(port, new
McpServerInfo("eviction", "1.0", "/mcp", 0, 500));
+ try {
+ initializeSession(port);
+ assertThat(engine.sessionCount()).isEqualTo(1);
+
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
assertThat(engine.sessionCount()).isZero());
+ } finally {
+ engine.stop();
+ }
+ }
+
+ @Test
+ void postOnlySessionSurvivesKeepAliveUntilIdle() throws Exception {
+ int port = AvailablePortFinder.getNextAvailable();
+ VertxMcpServerEngine engine = startEngine(port, new
McpServerInfo("eviction", "1.0", "/mcp", 300, 10_000));
+ try {
+ String sessionId = initializeSession(port);
+
+ await().during(2, TimeUnit.SECONDS).atMost(3,
TimeUnit.SECONDS).untilAsserted(() -> {
+ assertThat(postWithSession(port, sessionId,
INITIALIZED_NOTIFICATION)).isEqualTo(202);
+ assertThat(engine.sessionCount()).isEqualTo(1);
+ });
+ } finally {
+ engine.stop();
+ }
+ }
+
+ @Test
+ void activeSessionSurvivesIdleEviction() throws Exception {
+ int port = AvailablePortFinder.getNextAvailable();
+ VertxMcpServerEngine engine = startEngine(port, new
McpServerInfo("eviction", "1.0", "/mcp", 0, 2_000));
+ try {
+ String sessionId = initializeSession(port);
+
+ await().during(2, TimeUnit.SECONDS).atMost(3,
TimeUnit.SECONDS).untilAsserted(() -> {
+ assertThat(postWithSession(port, sessionId,
INITIALIZED_NOTIFICATION)).isEqualTo(202);
+ assertThat(engine.sessionCount()).isEqualTo(1);
+ });
+ } finally {
+ engine.stop();
+ }
+ }
+
+ @Test
+ void explicitDeleteRemovesSessionImmediately() throws Exception {
+ int port = AvailablePortFinder.getNextAvailable();
+ VertxMcpServerEngine engine = startEngine(port, new
McpServerInfo("eviction", "1.0", "/mcp", 0, 60_000));
+ try {
+ String sessionId = initializeSession(port);
+ assertThat(deleteSession(port, sessionId)).isEqualTo(200);
+ assertThat(engine.sessionCount()).isZero();
+ } finally {
+ engine.stop();
+ }
+ }
+
+ private static VertxMcpServerEngine startEngine(int port, McpServerInfo
info) throws Exception {
+ CamelContext camelContext = new DefaultCamelContext();
+ MainHttpServer main = new MainHttpServer();
+ main.setCamelContext(camelContext);
+ main.setHost("0.0.0.0");
+ main.setPort(port);
+ camelContext.addService(main);
+ camelContext.start();
+
+ VertxMcpServerEngine engine = new VertxMcpServerEngine();
+ engine.setCamelContext(camelContext);
+ engine.initialize(info);
+ engine.start();
+ return engine;
+ }
+
+ private static String initializeSession(int port) throws Exception {
+ HttpResponse<String> response = HTTP.send(HttpRequest.newBuilder()
+ .uri(URI.create("http://localhost:"; + port + "/mcp"))
+ .timeout(Duration.ofSeconds(10))
+ .header("Content-Type", "application/json")
+ .header("Accept", "application/json, text/event-stream")
+ .POST(HttpRequest.BodyPublishers.ofString(INITIALIZE_REQUEST))
+ .build(), HttpResponse.BodyHandlers.ofString());
+ assertThat(response.statusCode()).isEqualTo(200);
+ return response.headers().firstValue("mcp-session-id").orElseThrow();
+ }
+
+ private static int postWithSession(int port, String sessionId, String
body) throws Exception {
+ HttpResponse<Void> response = HTTP.send(HttpRequest.newBuilder()
+ .uri(URI.create("http://localhost:"; + port + "/mcp"))
+ .timeout(Duration.ofSeconds(10))
+ .header("Content-Type", "application/json")
+ .header("Accept", "application/json, text/event-stream")
+ .header("Mcp-Session-Id", sessionId)
+ .POST(HttpRequest.BodyPublishers.ofString(body))
+ .build(), HttpResponse.BodyHandlers.discarding());
+ return response.statusCode();
+ }
+
+ private static int deleteSession(int port, String sessionId) throws
Exception {
+ HttpResponse<Void> response = HTTP.send(HttpRequest.newBuilder()
+ .uri(URI.create("http://localhost:"; + port + "/mcp"))
+ .timeout(Duration.ofSeconds(10))
+ .header("Mcp-Session-Id", sessionId)
+ .DELETE()
+ .build(), HttpResponse.BodyHandlers.discarding());
+ return response.statusCode();
+ }
+}
diff --git
a/core/camel-main/src/generated/java/org/apache/camel/main/HttpServerConfigurationPropertiesConfigurer.java
b/core/camel-main/src/generated/java/org/apache/camel/main/HttpServerConfigurationPropertiesConfigurer.java
index 9129b2b45a81..42f63d53a06d 100644
---
a/core/camel-main/src/generated/java/org/apache/camel/main/HttpServerConfigurationPropertiesConfigurer.java
+++
b/core/camel-main/src/generated/java/org/apache/camel/main/HttpServerConfigurationPropertiesConfigurer.java
@@ -40,6 +40,8 @@ public class HttpServerConfigurationPropertiesConfigurer
extends org.apache.came
map.put("McpEnabled", boolean.class);
map.put("McpPath", java.lang.String.class);
map.put("McpServerName", java.lang.String.class);
+ map.put("McpSessionIdleTtl", long.class);
+ map.put("McpSessionKeepAliveInterval", long.class);
map.put("McpTags", java.lang.String.class);
map.put("McpToolTimeout", long.class);
map.put("Path", java.lang.String.class);
@@ -89,6 +91,10 @@ public class HttpServerConfigurationPropertiesConfigurer
extends org.apache.came
case "mcpPath": target.setMcpPath(property(camelContext,
java.lang.String.class, value)); return true;
case "mcpservername":
case "mcpServerName": target.setMcpServerName(property(camelContext,
java.lang.String.class, value)); return true;
+ case "mcpsessionidlettl":
+ case "mcpSessionIdleTtl":
target.setMcpSessionIdleTtl(property(camelContext, long.class, value)); return
true;
+ case "mcpsessionkeepaliveinterval":
+ case "mcpSessionKeepAliveInterval":
target.setMcpSessionKeepAliveInterval(property(camelContext, long.class,
value)); return true;
case "mcptags":
case "mcpTags": target.setMcpTags(property(camelContext,
java.lang.String.class, value)); return true;
case "mcptooltimeout":
@@ -149,6 +155,10 @@ public class HttpServerConfigurationPropertiesConfigurer
extends org.apache.came
case "mcpPath": return java.lang.String.class;
case "mcpservername":
case "mcpServerName": return java.lang.String.class;
+ case "mcpsessionidlettl":
+ case "mcpSessionIdleTtl": return long.class;
+ case "mcpsessionkeepaliveinterval":
+ case "mcpSessionKeepAliveInterval": return long.class;
case "mcptags":
case "mcpTags": return java.lang.String.class;
case "mcptooltimeout":
@@ -205,6 +215,10 @@ public class HttpServerConfigurationPropertiesConfigurer
extends org.apache.came
case "mcpPath": return target.getMcpPath();
case "mcpservername":
case "mcpServerName": return target.getMcpServerName();
+ case "mcpsessionidlettl":
+ case "mcpSessionIdleTtl": return target.getMcpSessionIdleTtl();
+ case "mcpsessionkeepaliveinterval":
+ case "mcpSessionKeepAliveInterval": return
target.getMcpSessionKeepAliveInterval();
case "mcptags":
case "mcpTags": return target.getMcpTags();
case "mcptooltimeout":
diff --git
a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
index 3d8ec1317988..c03dd1557f00 100644
---
a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
+++
b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
@@ -442,6 +442,8 @@
{ "name": "camel.server.mcpEnabled", "required": false, "description":
"Whether to expose ai-tool routes as MCP tools over streamable HTTP. Requires
camel-mcp-server on the classpath. By default, the MCP server is not enabled.",
"sourceType": "org.apache.camel.main.HttpServerConfigurationProperties",
"type": "boolean", "javaType": "boolean", "defaultValue": false, "secret":
false },
{ "name": "camel.server.mcpPath", "required": false, "description": "HTTP
path where the MCP endpoint is served.", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "string",
"javaType": "java.lang.String", "defaultValue": "\/mcp", "secret": false },
{ "name": "camel.server.mcpServerName", "required": false, "description":
"MCP server name advertised to clients. Defaults to the CamelContext name.",
"sourceType": "org.apache.camel.main.HttpServerConfigurationProperties",
"type": "string", "javaType": "java.lang.String", "secret": false },
+ { "name": "camel.server.mcpSessionIdleTtl", "required": false,
"description": "Idle TTL in milliseconds for MCP sessions on the Vert.x
streamable transport. Sessions with no activity for longer than this interval
are evicted. 0 disables idle eviction.", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "integer",
"javaType": "long", "defaultValue": 300000, "secret": false },
+ { "name": "camel.server.mcpSessionKeepAliveInterval", "required": false,
"description": "Keep-alive ping interval in milliseconds for MCP sessions on
the Vert.x streamable transport. Dead sessions are evicted after consecutive
ping failures. 0 disables keep-alive pings.", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "integer",
"javaType": "long", "defaultValue": 30000, "secret": false },
{ "name": "camel.server.mcpTags", "required": false, "description":
"Comma-separated list of ai-tool tags to expose as MCP tools. Only tools
registered under one of these tags are exposed; the untagged default pool is
never exposed. When not set, no tools are exposed.", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "string",
"javaType": "java.lang.String", "secret": false },
{ "name": "camel.server.mcpToolTimeout", "required": false, "description":
"Per-call MCP tool execution timeout in milliseconds. A call exceeding the
timeout returns an error result to the MCP client; the underlying route keeps
running until it completes on its own.", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "integer",
"javaType": "long", "defaultValue": 20000, "secret": false },
{ "name": "camel.server.path", "required": false, "description":
"Context-path to use for embedded HTTP server", "sourceType":
"org.apache.camel.main.HttpServerConfigurationProperties", "type": "string",
"javaType": "java.lang.String", "defaultValue": "\/", "secret": false },
diff --git a/core/camel-main/src/main/docs/main.adoc
b/core/camel-main/src/main/docs/main.adoc
index e3d33c3295b0..db7bd0eb0a19 100644
--- a/core/camel-main/src/main/docs/main.adoc
+++ b/core/camel-main/src/main/docs/main.adoc
@@ -195,7 +195,7 @@ The camel.routecontroller supports 12 options, which are
listed below.
=== Camel Embedded HTTP Server (only for standalone; not Spring Boot or
Quarkus) configurations
-The camel.server supports 26 options, which are listed below.
+The camel.server supports 28 options, which are listed below.
[width="100%",cols="2,5,^1,2",options="header"]
|===
@@ -218,6 +218,8 @@ The camel.server supports 26 options, which are listed
below.
| *camel.server.mcpEnabled* | Whether to expose ai-tool routes as MCP tools
over streamable HTTP. Requires camel-mcp-server on the classpath. By default,
the MCP server is not enabled. | false | boolean
| *camel.server.mcpPath* | HTTP path where the MCP endpoint is served. | /mcp
| String
| *camel.server.mcpServerName* | MCP server name advertised to clients.
Defaults to the CamelContext name. | | String
+| *camel.server.mcpSessionIdleTtl* | Idle TTL in milliseconds for MCP sessions
on the Vert.x streamable transport. Sessions with no activity for longer than
this interval are evicted. 0 disables idle eviction. | 300000 | long
+| *camel.server.mcpSessionKeepAliveInterval* | Keep-alive ping interval in
milliseconds for MCP sessions on the Vert.x streamable transport. Dead sessions
are evicted after consecutive ping failures. 0 disables keep-alive pings. |
30000 | long
| *camel.server.mcpTags* | Comma-separated list of ai-tool tags to expose as
MCP tools. Only tools registered under one of these tags are exposed; the
untagged default pool is never exposed. When not set, no tools are exposed. |
| String
| *camel.server.mcpToolTimeout* | Per-call MCP tool execution timeout in
milliseconds. A call exceeding the timeout returns an error result to the MCP
client; the underlying route keeps running until it completes on its own. |
20000 | long
| *camel.server.path* | Context-path to use for embedded HTTP server | / |
String
diff --git
a/core/camel-main/src/main/java/org/apache/camel/main/HttpServerConfigurationProperties.java
b/core/camel-main/src/main/java/org/apache/camel/main/HttpServerConfigurationProperties.java
index da2603c3cce3..5285a09a12af 100644
---
a/core/camel-main/src/main/java/org/apache/camel/main/HttpServerConfigurationProperties.java
+++
b/core/camel-main/src/main/java/org/apache/camel/main/HttpServerConfigurationProperties.java
@@ -79,6 +79,10 @@ public class HttpServerConfigurationProperties implements
BootstrapCloseable {
private long mcpToolTimeout = 20000;
@Metadata(defaultValue = "/mcp")
private String mcpPath = "/mcp";
+ @Metadata(defaultValue = "30000")
+ private long mcpSessionKeepAliveInterval = 30000;
+ @Metadata(defaultValue = "300000")
+ private long mcpSessionIdleTtl = 300000;
@Metadata
private String mcpServerName;
@@ -395,6 +399,30 @@ public class HttpServerConfigurationProperties implements
BootstrapCloseable {
this.mcpServerName = mcpServerName;
}
+ public long getMcpSessionKeepAliveInterval() {
+ return mcpSessionKeepAliveInterval;
+ }
+
+ /**
+ * Keep-alive ping interval in milliseconds for MCP sessions on the Vert.x
streamable transport. Dead sessions are
+ * evicted after consecutive ping failures. {@code 0} disables keep-alive
pings.
+ */
+ public void setMcpSessionKeepAliveInterval(long
mcpSessionKeepAliveInterval) {
+ this.mcpSessionKeepAliveInterval = mcpSessionKeepAliveInterval;
+ }
+
+ public long getMcpSessionIdleTtl() {
+ return mcpSessionIdleTtl;
+ }
+
+ /**
+ * Idle TTL in milliseconds for MCP sessions on the Vert.x streamable
transport. Sessions with no activity for
+ * longer than this interval are evicted. {@code 0} disables idle eviction.
+ */
+ public void setMcpSessionIdleTtl(long mcpSessionIdleTtl) {
+ this.mcpSessionIdleTtl = mcpSessionIdleTtl;
+ }
+
/**
* Whether embedded HTTP server is enabled. By default, the server is not
enabled.
*/
@@ -618,4 +646,20 @@ public class HttpServerConfigurationProperties implements
BootstrapCloseable {
return this;
}
+ /**
+ * Keep-alive ping interval in milliseconds for MCP sessions on the Vert.x
streamable transport.
+ */
+ public HttpServerConfigurationProperties
withMcpSessionKeepAliveInterval(long mcpSessionKeepAliveInterval) {
+ this.mcpSessionKeepAliveInterval = mcpSessionKeepAliveInterval;
+ return this;
+ }
+
+ /**
+ * Idle TTL in milliseconds for MCP sessions on the Vert.x streamable
transport.
+ */
+ public HttpServerConfigurationProperties withMcpSessionIdleTtl(long
mcpSessionIdleTtl) {
+ this.mcpSessionIdleTtl = mcpSessionIdleTtl;
+ return this;
+ }
+
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 1fcc875dc9b4..8a05e7b290b1 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -23,6 +23,14 @@ this executor is functionally identical to the built-in
`DefaultReactiveExecutor
To migrate, remove the `camel-reactive-executor-tomcat` dependency from your
project. Camel will
automatically use the default reactive executor.
+=== camel-mcp-server
+
+The Vert.x streamable transport now evicts orphaned MCP sessions
automatically: keep-alive pings
+every 30 seconds (evict after consecutive failures on sessions with an open
SSE stream) and an idle
+TTL of 5 minutes (evict sessions with no POST/GET activity). Configure with
+`camel.server.mcp-session-keep-alive-interval` and
`camel.server.mcp-session-idle-ttl`; set either
+option to `0` to disable that mechanism.
+
=== camel-core
==== Property placeholders in toD and enrich dynamic endpoint URIs
