This is an automated email from the ASF dual-hosted git repository.
Croway 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 68f43491f3cd CAMEL-25642: camel-jbang-mcp - runtime tools wiring:
receive in one call, routeId honoured, prompts name real tools (#27958)
68f43491f3cd is described below
commit 68f43491f3cdce96318261f3da0579a3bd21bb5d
Author: Federico Mariani <[email protected]>
AuthorDate: Sat Oct 17 15:00:53 2026 +0200
CAMEL-25642: camel-jbang-mcp - runtime tools wiring: receive in one call,
routeId honoured, prompts name real tools (#27958)
* CAMEL-25642: camel-jbang-mcp - runtime tools send the routeId as the
filter the route-dump and route-structure actions read
The registry put the route pattern under "id" while the CLI connector reads
"filter" (as camel cmd route-dump
sends it), so get_route_dump, get_route_structure and get_route_analysis
always returned every route.
get_route_structure also takes brief, write_heap_dump refuses a name with a
directory, and send_message is
destructive as in the MCP server.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25642: camel-jbang-mcp - the diagnose and optimize prompts name the
MCP runtime tools
camel_diagnose_route and camel_optimize_route told the model to call the
registry names (get_context,
get_top_processors, ...), which the MCP server does not list. A test checks
that every tool a prompt names
is one the server lists, with the @Tool scan shared with
McpSecuredToolClassesTest.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25642: camel-jbang-mcp - camel_runtime_receive receives one
message, and the runtime tools report failures
* camel_runtime_receive started a consumer that kept running and returned
no message: it now enables the
receive console for the endpoint, waits for a message in the receive file
the CLI connector writes,
returns it and disables the console, and is marked destructive.
* A process that does not answer in time is an error of every runtime tool;
camel_runtime_ai_trace no longer
turns a failed query into "No AI components detected", and its nameOrPid
is optional.
* camel_runtime_memory_leak starts one recording by default and returns at
once (dual blocked ~90 s); dual is
opt-in, compare is a command, and a bad duration or mode is refused.
* camel_runtime_http_request passes timeoutSeconds,
camel_runtime_route_structure brief,
camel_runtime_trace_dump reads the traces for read-only clients,
heap_dump goes through the registry's name
check, and the history description says when the history is recorded.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25642: camel-jbang-mcp - camel cmd receive naming,
camel.trace.standby hint, bounded camel_runtime_receive, and review nits
- Name the command camel cmd receive (not camel receive) in the receive
error, RuntimeService comments, the MCP page
and the upgrade guide.
- The message-history hint names camel.trace.standby (the BacklogTracer, as
the dev profile sets it), not
camel.main.tracing-standby (the log Tracer).
- camel_runtime_receive takes limit (default 1, at least 1): it returns the
first limit messages of the poll and
counts the others (notReturned), which were taken from the endpoint all
the same.
- If the receive console cannot be stopped, return the messages with a
warning naming
camel cmd receive --action=stop (or add it to the error) instead of
losing the answer.
- Read the receive file from the offset with a SeekableByteChannel instead
of reading the whole file each poll.
- Remove the timeout-text check (failOnTimeout,
RuntimeHelper.TIMEOUT_PREFIX) and its test: ToolContext.executeAction
throws on timeout since CAMEL-25398, so no registry tool returns that
text; drop the upgrade-guide bullet that
claimed it as a change of this branch.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25642: camel-jbang-mcp - camel_runtime_receive reports the messages
its own consumer took
- The console numbers every message it takes (uid) and the status reports
the count (total): the call reads the
count before enabling the console and keeps only the messages above it,
so the leftovers of an earlier session
are not returned as its own.
- After the stop, the call waits (3 seconds at most) until the status shows
the console disabled: the count then
says how many messages this call took, including the ones the connector
dumps only after the stop, which are
returned too (up to limit) and otherwise counted as notReturned. The
answer says when the process did not
confirm the stop or had not written every message it took.
- One camel_runtime_receive call at a time per process: a second call waits
up to its own timeout, then fails
with a message naming the first call, instead of reading its messages and
stopping its consumer.
- The messages of exceptions without a message are their names, not "null".
Co-Authored-By: Claude Fable 5.1 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 19 +
.../modules/ROOT/pages/camel-jbang-mcp.adoc | 46 ++-
.../dsl/jbang/core/commands/ai/ToolRegistry.java | 46 ++-
.../jbang/core/commands/ai/RuntimeToolsTest.java | 6 +-
.../ai/ToolRegistryRuntimeActionsTest.java | 102 +++++
.../dsl/jbang/core/commands/mcp/AiTraceTools.java | 8 +-
.../jbang/core/commands/mcp/PromptDefinitions.java | 56 +--
.../jbang/core/commands/mcp/RuntimeService.java | 71 ++++
.../dsl/jbang/core/commands/mcp/RuntimeTools.java | 393 +++++++++++++++++---
.../commands/mcp/McpSecuredToolClassesTest.java | 36 +-
...redToolClassesTest.java => McpToolClasses.java} | 39 +-
.../core/commands/mcp/PromptDefinitionsTest.java | 33 ++
.../jbang/core/commands/mcp/RuntimeToolsTest.java | 6 +-
.../core/commands/mcp/RuntimeToolsWiringTest.java | 411 +++++++++++++++++++++
14 files changed, 1102 insertions(+), 170 deletions(-)
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index e9117fe96d6d..22dac2264492 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1537,3 +1537,22 @@ When a protobuf message is converted to a `Map` (for
example `getBody(Map.class)
a `bytes` field is now a `byte[]` with the content of the field. Previously it
was the `String` that
`ByteString.toString()` returns, a description such as `<ByteString@1b6d3586
size=4 contents="...">`, which does not
contain the whole value. When a `Map` is marshalled, a `byte[]` value is
accepted for a `bytes` field.
+
+=== camel-jbang-mcp - runtime tools that changed behaviour
+
+Some runtime tools of the Camel MCP server now do what their description says,
which changes what a client sees:
+
+* `camel_runtime_receive` receives a message in one call: it starts a consumer
on the endpoint, waits up to
+ `timeoutSeconds` (default 10, at most 60) for a message, returns it and
stops the consumer. It returns one message
+ by default; `limit` returns up to that many, and `received` counts every
message the consumer took, returned or
+ not. Before, it started a consumer that kept running, took the messages from
the routes, and returned none of
+ them. It is now marked destructive, so the `read-write` access level
(`camel.mcp.security.access-level`) no longer
+ offers it, only `admin` does. It runs one call at a time per process, and
refuses to run while `camel cmd receive`
+ is receiving from the same process.
+* `camel_runtime_memory_leak` with `command=start` and no `mode` makes one
recording (60 seconds by default) and
+ returns at once; get the results with `query` once it is done. Before, it
made two recordings and blocked for about
+ 90 seconds, longer than many MCP clients wait. Pass `mode=dual` for the
previous behaviour, or make the two
+ recordings yourself and use the new `compare` command. A `duration` that is
not a whole number of seconds is
+ refused.
+* `camel_runtime_route_dump`, `camel_runtime_route_structure` and
`camel_runtime_route_analysis` return the routes
+ that `routeId` names; they returned every route whatever `routeId` was.
diff --git a/docs/user-manual/modules/ROOT/pages/camel-jbang-mcp.adoc
b/docs/user-manual/modules/ROOT/pages/camel-jbang-mcp.adoc
index 765691486c2e..db1b82730279 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-jbang-mcp.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-jbang-mcp.adoc
@@ -819,10 +819,12 @@ process is running).
| Get the source code of routes in the running Camel application. Supports
wildcard filtering.
| `camel_runtime_route_dump`
-| Dump route definitions in XML or YAML format.
+| Dump route definitions in XML, YAML or Java DSL format. `routeId` takes a
route id or a pattern such as `orders*`
+ (all routes when omitted).
| `camel_runtime_route_structure`
-| Show the route structure as a tree of processors.
+| Show the route structure as a tree of processors, for the routes `routeId`
names (a route id or a pattern, all
+ routes when omitted). `brief=true` leaves out the options and expressions.
| `camel_runtime_processor_detail`
| Show the configured options of every processor in a route: each processor's
type, id, and configured options
@@ -830,7 +832,8 @@ process is running).
with its documentation from the Camel catalog.
| `camel_runtime_route_control`
-| Control a route: start, stop, suspend, or resume it by route ID.
+| Control a route: start, stop, suspend, or resume it by route ID. Marked
destructive for all four commands, as an
+ MCP hint applies to a tool, not to one of its arguments, and stop and
suspend take the route's consumer offline.
| `camel_runtime_route_topology`
| Get the inter-route topology showing how routes connect to each other and to
external endpoints.
@@ -867,16 +870,21 @@ process is running).
| `camel_runtime_heap_dump`
| Write a heap dump (`.hprof`) from a running integration for deep memory
analysis with tools such as Eclipse MAT,
- VisualVM or jhat. The dump is written to the process working directory.
+ VisualVM or jhat. The dump is written to the process working directory; a
`name` with a directory is refused.
| `camel_runtime_memory_leak`
| Diagnose memory leaks in a running Camel integration using Java Flight
Recorder (JFR).
- Use command 'start' to begin recording, 'stop' to get results, 'status' to
check state,
- and 'query' to retrieve cached results. Supports dual-recording mode for
trend comparison.
+ `start` begins a recording of `duration` seconds (default 60) and returns at
once; `status` checks it, `query`
+ (or `stop`, to end it early) returns the results, and `compare` compares the
last two recordings (growth and
+ trend per class). For a trend, record X seconds, then 2X seconds, then
compare. `mode=dual` does all of this in one
+ call that blocks for about three times the duration (90 seconds with its
default of 30), longer than many MCP
+ clients wait: use it only with a client that waits that long.
| `camel_runtime_history`
| Get the message history trace of the last completed exchange. Shows the
route path, processors visited,
- headers, body, and timing.
+ headers, body, and timing. It is recorded when message history is on and the
tracer is enabled or on standby,
+ which is the default of `camel run` (dev profile); elsewhere set
`camel.main.message-history=true` and
+ `camel.trace.standby=true`.
| `camel_runtime_thread_dump`
| Get a JVM thread dump showing thread names, states, and stack traces.
@@ -901,7 +909,7 @@ process is running).
| `camel_runtime_http_request`
| Sends an HTTP request (method, path, headers, body) to the application's own
server on localhost and its port,
and returns the status, the headers and the body (cut at 6000 characters,
the answer says when). Any other host
- is refused, and redirects are not followed.
+ is refused, and redirects are not followed. `timeoutSeconds` bounds the wait
(default 30, at most 120).
| `camel_runtime_tool_groups`
| Which runtime tool groups (`sql`, `tracing`, `resilience`, `http`) the
application needs, from what its status shows.
@@ -1000,23 +1008,37 @@
xref:camel-jbang-tui-local-models.adoc#_tool_groups_from_the_selected_integratio
| Tool | Description
| `camel_runtime_send`
-| Send a test message to a Camel endpoint in the running application.
+| Send a test message to a Camel endpoint in the running application. Marked
destructive: the message runs through
+ the routes and whatever they call.
| `camel_runtime_trace`
| Enable, disable, or dump message tracing for the running Camel application.
+| `camel_runtime_trace_dump`
+| Dump the traced messages, the same as `camel_runtime_trace` with
`action=dump`, as a read-only tool, so a client
+ at the `read-only` access level can read the traces.
+
| `camel_runtime_ai_trace`
| Trace the AI-specific exchange flow of a running application: token usage,
model ids, guardrail outcomes,
completion reasons, streaming chunk counts and per-component latency for the
AI components (Bedrock,
LangChain4j, Docling, Textract, OpenAI, KServe, DJL, TensorFlow Serving,
HuggingFace). It combines the message
- history with the processor statistics, filtered to the AI steps.
+ history with the processor statistics, filtered to the AI steps. A process
that does not answer is reported as
+ an error.
| `camel_runtime_browse`
| Browse messages in a Camel endpoint (e.g., messages queued in a SEDA
endpoint).
| `camel_runtime_receive`
-| Receive (poll) a message from a Camel endpoint in the running application.
Consumes one message
- from the endpoint.
+| Receive a message from an endpoint of the running application, the
complement of `camel_runtime_send`, in one
+ call: it starts a consumer on the endpoint (an endpoint uri, or a pattern
matching an endpoint the application
+ sends to), waits up to `timeoutSeconds` (default 10, at most 60) for a
message, returns it and stops the consumer.
+ It returns one message by default: `limit` (at least 1) returns up to that
many. `received` counts every message
+ the consumer took, including the ones taken after the first that arrived and
before the consumer stopped, and the
+ answer says how many of them it does not return (`notReturned`). The
consumer competes with the routes that
+ consume the endpoint, so the messages it takes do not reach them: the tool
is marked destructive. One call at a
+ time per process: a second call waits for the first, up to its own
`timeoutSeconds`. Use `camel_runtime_browse`
+ to look at queued messages without taking them. It refuses to run while
`camel cmd receive` is receiving from the
+ same process.
|===
== Available Prompts
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java
index 308b9779f734..6f1bc9a97fd5 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistry.java
@@ -254,25 +254,31 @@ public final class ToolRegistry {
}));
register(tool("get_route_dump",
- "Dump route definitions in XML or YAML format.")
- .param("routeId", "string", "Route ID to dump (use * for all
routes)", false)
- .param("format", "string", "Output format: xml or yaml
(default: yaml)", false)
+ "Dump route definitions in XML, YAML or Java DSL format.")
+ .param("routeId", "string", "Route ID or pattern such as
orders* to dump (use * for all routes)", false)
+ .param("format", "string", "Output format: xml, yaml or java
(default: yaml)", false)
.executor((ctx, args) -> {
String routeId = args.get("routeId");
String format = args.get("format");
+ // the route-dump action reads the routes from "filter",
as camel cmd route-dump sends it
return ctx.executeAction("route-dump", root -> {
- root.put("id", routeId != null ? routeId : "*");
+ root.put("filter", routeFilter(routeId));
root.put("format", format != null ? format : "yaml");
});
}));
register(tool("get_route_structure",
"Show the route structure as a tree of processors.")
- .param("routeId", "string", "Route ID to inspect (use * for
all routes)", false)
+ .param("routeId", "string", "Route ID or pattern such as
orders* to inspect (use * for all routes)",
+ false)
+ .param("brief", "boolean", "Only the overall structure,
without the options and expressions", false)
.executor((ctx, args) -> {
String routeId = args.get("routeId");
- return ctx.executeAction("route-structure",
- root -> root.put("id", routeId != null ? routeId :
"*"));
+ boolean brief = "true".equalsIgnoreCase(args.get("brief"));
+ return ctx.executeAction("route-structure", root -> {
+ root.put("filter", routeFilter(routeId));
+ root.put("brief", String.valueOf(brief));
+ });
}));
register(tool("get_processor_detail",
@@ -343,7 +349,8 @@ public final class ToolRegistry {
.param("body", "string", "Message body to send", false)
.param("headers", "string",
"Message headers as key=value pairs separated by
newlines", false)
- .readOnly(false).destructive(false)
+ // the message runs through the routes and whatever they call,
so it can change data
+ .readOnly(false).destructive(true)
.core(true)
.executor((ctx, args) -> {
String endpoint = args.get("endpoint");
@@ -421,6 +428,11 @@ public final class ToolRegistry {
.executor((ctx, args) -> {
String name = args.get("name");
String live = args.get("live");
+ if (name != null && !isPlainFileName(name)) {
+ throw new ToolExecutionException(
+ "name must be a file name without a directory
(the dump is written to the process"
+ + " working
directory): " + name);
+ }
return ctx.executeAction("heap-dump", root -> {
if (name != null && !name.isBlank()) {
root.put("name", name);
@@ -714,10 +726,10 @@ public final class ToolRegistry {
String routeId = args.get("routeId");
// Get route structure for analysis
String structureJson = ctx.executeAction("route-structure",
- root -> root.put("id", routeId != null ? routeId :
"*"));
+ root -> root.put("filter", routeFilter(routeId)));
// Get route dump for error handler info
String dumpJson = ctx.executeAction("route-dump", root -> {
- root.put("id", routeId != null ? routeId : "*");
+ root.put("filter", routeFilter(routeId));
root.put("format", "yaml");
});
JsonObject response = new JsonObject();
@@ -737,7 +749,7 @@ public final class ToolRegistry {
String topJson = ctx.executeAction("top-processors", null);
// Get route structure for EIP type counting
String structureJson = ctx.executeAction("route-structure",
- root -> root.put("id", "*"));
+ root -> root.put("filter", "*"));
JsonObject response = new JsonObject();
response.put("processorStats", json(topJson));
response.put("routeStructures", json(structureJson));
@@ -758,7 +770,7 @@ public final class ToolRegistry {
// Get running route definitions
String dumpFormat = format;
String runningRoutes = ctx.executeAction("route-dump",
root -> {
- root.put("id", "*");
+ root.put("filter", "*");
root.put("format", dumpFormat);
});
// Get source files
@@ -784,6 +796,16 @@ public final class ToolRegistry {
return parsed != null ? parsed : answer;
}
+ /** The route pattern of the route-dump and route-structure actions: all
routes when none is given. */
+ private static String routeFilter(String routeId) {
+ return routeId != null && !routeId.isBlank() ? routeId : "*";
+ }
+
+ /** Whether a heap dump name is only a file name, so the dump stays in the
process working directory. */
+ static boolean isPlainFileName(String name) {
+ return !name.contains("/") && !name.contains("\\") &&
!".".equals(name) && !"..".equals(name);
+ }
+
private static JsonObject buildAnalysisHints() {
JsonObject hints = new JsonObject();
hints.put("errorHandling",
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/ai/RuntimeToolsTest.java
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/ai/RuntimeToolsTest.java
index af84fc01157d..b7981f07e904 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/ai/RuntimeToolsTest.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/ai/RuntimeToolsTest.java
@@ -82,11 +82,11 @@ class RuntimeToolsTest {
answers.put("route-dump", Map.of("routes", List.of()));
call("get_route_dump", Map.of());
- assertThat(integration.lastAction().getString("id")).isEqualTo("*");
+
assertThat(integration.lastAction().getString("filter")).isEqualTo("*");
assertThat(integration.lastAction().getString("format")).isEqualTo("yaml");
call("get_route_dump", Map.of("routeId", "orders", "format", "xml"));
-
assertThat(integration.lastAction().getString("id")).isEqualTo("orders");
+
assertThat(integration.lastAction().getString("filter")).isEqualTo("orders");
assertThat(integration.lastAction().getString("format")).isEqualTo("xml");
}
@@ -97,7 +97,7 @@ class RuntimeToolsTest {
answers.put("route-topology", Map.of("nodes", List.of()));
call("get_route_structure", Map.of());
- assertThat(integration.lastAction().getString("id")).isEqualTo("*");
+
assertThat(integration.lastAction().getString("filter")).isEqualTo("*");
call("get_top_processors", Map.of());
assertThat(integration.lastAction().getString("action")).isEqualTo("top-processors");
call("get_route_topology", Map.of());
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistryRuntimeActionsTest.java
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistryRuntimeActionsTest.java
new file mode 100644
index 000000000000..acab240d1ffa
--- /dev/null
+++
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/ai/ToolRegistryRuntimeActionsTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.dsl.jbang.core.commands.ai;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.camel.util.json.JsonObject;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * The requests the runtime tools send to the CLI connector, recorded instead
of sent: the connector's route-dump and
+ * route-structure actions read the routes from {@code filter} (as {@code
camel cmd route-dump} sends it).
+ */
+class ToolRegistryRuntimeActionsTest {
+
+ private final List<JsonObject> requests = new ArrayList<>();
+
+ private final ToolContext ctx = new ToolContext() {
+ @Override
+ public String executeAction(String action, Consumer<JsonObject>
configure) {
+ JsonObject request = new JsonObject();
+ request.put("action", action);
+ if (configure != null) {
+ configure.accept(request);
+ }
+ requests.add(request);
+ return "{}";
+ }
+ };
+
+ ToolRegistryRuntimeActionsTest() {
+ ctx.selectProcess(4711);
+ }
+
+ @Test
+ void routeDumpSendsTheRouteIdAsFilter() {
+ ToolRegistry.execute("get_route_dump", ctx, Map.of("routeId",
"orders"));
+
+ assertThat(requests).singleElement().satisfies(r -> {
+ assertThat(r.getString("action")).isEqualTo("route-dump");
+ assertThat(r.getString("filter")).isEqualTo("orders");
+ assertThat(r).doesNotContainKey("id");
+ });
+ }
+
+ @Test
+ void routeStructureSendsTheRouteIdAsFilterAndBrief() {
+ ToolRegistry.execute("get_route_structure", ctx, Map.of("routeId",
"orders", "brief", "true"));
+ ToolRegistry.execute("get_route_structure", ctx, Map.of());
+
+ assertThat(requests).extracting(r ->
r.getString("filter")).containsExactly("orders", "*");
+ assertThat(requests).extracting(r ->
r.getString("brief")).containsExactly("true", "false");
+ }
+
+ @Test
+ void routeAnalysisFiltersBothActions() {
+ ToolRegistry.execute("get_route_analysis", ctx, Map.of("routeId",
"orders"));
+
+ assertThat(requests).extracting(r ->
r.getString("action")).containsExactly("route-structure", "route-dump");
+ assertThat(requests).extracting(r ->
r.getString("filter")).containsOnly("orders");
+ }
+
+ @Test
+ void heapDumpRefusesANameWithADirectory() {
+ assertThatThrownBy(() -> ToolRegistry.execute("write_heap_dump", ctx,
Map.of("name", "../../tmp/x")))
+ .isInstanceOf(ToolExecutionException.class)
+ .hasMessageContaining("without a directory");
+ assertThatThrownBy(() -> ToolRegistry.execute("write_heap_dump", ctx,
Map.of("name", "..")))
+ .isInstanceOf(ToolExecutionException.class);
+ assertThat(requests).isEmpty();
+
+ ToolRegistry.execute("write_heap_dump", ctx, Map.of("name",
"before-fix"));
+ assertThat(requests).singleElement().satisfies(r ->
assertThat(r.getString("name")).isEqualTo("before-fix"));
+ }
+
+ @Test
+ void sendingAMessageIsDestructive() {
+ ToolDescriptor send = ToolRegistry.findTool("send_message");
+ assertThat(send.isReadOnly()).isFalse();
+ assertThat(send.isDestructive()).isTrue();
+ }
+}
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/AiTraceTools.java
b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/AiTraceTools.java
index 7edb72a3c062..968cc390b7b6 100644
---
a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/AiTraceTools.java
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/AiTraceTools.java
@@ -72,7 +72,7 @@ public class AiTraceTools {
+ "TensorFlow Serving, HuggingFace). "
+ "Combines message history with processor statistics
filtered to AI steps.")
public AiTraceResult camel_runtime_ai_trace(
- @ToolArg(description = NAME_OR_PID_DESC) String nameOrPid,
+ @ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid,
@ToolArg(description = "Route ID filter (exact match or * for all,
default: *)",
required = false) String routeFilter) {
@@ -107,7 +107,11 @@ public class AiTraceTools {
}
}
- private JsonObject executeRegistryTool(String toolName, ToolContext ctx,
Map<String, String> args) {
+ /**
+ * Runs one registry query. A failure (the process is gone, or did not
answer in time) is an error of the whole
+ * call: answering with what the other queries returned would read as "no
AI components".
+ */
+ JsonObject executeRegistryTool(String toolName, ToolContext ctx,
Map<String, String> args) {
try {
Object result = ToolRegistry.execute(toolName, ctx, args);
return RuntimeTools.toJsonObject(result);
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/PromptDefinitions.java
b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/PromptDefinitions.java
index 92383d9696b3..2cb4a1371637 100644
---
a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/PromptDefinitions.java
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/PromptDefinitions.java
@@ -242,32 +242,32 @@ public class PromptDefinitions {
Follow these steps in order:
### Step 1: Gather context
- Call `get_context` to understand the application state
(version, uptime, total exchanges).
- Call `get_routes` to see all routes, their states, and error
counts.
+ Call `camel_runtime_context` to understand the application
state (version, uptime, total exchanges).
+ Call `camel_runtime_routes` to see all routes, their states,
and error counts.
### Step 2: Check health
- Call `get_health` to see if any health checks are failing.
+ Call `camel_runtime_health` to see if any health checks are
failing.
Note any DOWN or UNKNOWN health statuses.
### Step 3: Collect errors
- Call `get_errors` to get captured routing errors with stack
traces.
- Call `get_history` to see the message history trace of the
last completed exchange.
+ Call `camel_get_errors` to get the failed exchanges with their
stack traces.
+ Call `camel_runtime_history` to see the message history trace
of the last completed exchange.
### Step 4: Inspect route details
- Call `get_route_source` with filter="%s" to see the route
definition.
- Call `get_top_processors` to identify slow or failing
processors.
- Call `get_inflight` to check for stuck exchanges.
- Call `get_blocked` to check for blocked exchanges.
+ Call `camel_runtime_route_dump` with routeId="%s" to see the
route definition.
+ Call `camel_runtime_top` to identify slow or failing
processors.
+ Call `camel_runtime_inflight` to check for stuck exchanges.
+ Call `camel_runtime_blocked` to check for blocked exchanges.
### Step 5: Check infrastructure
- Call `get_memory` to check for memory pressure or excessive GC.
- Call `get_endpoints` to verify all endpoints are reachable.
+ Call `camel_runtime_memory` to check for memory pressure or
excessive GC.
+ Call `camel_runtime_endpoints` to verify all endpoints are
reachable.
### Step 6: Advanced diagnostics (if needed)
- If circuit breakers are involved, call `get_circuit_breakers`.
- If datasources are involved, call `get_datasources`.
- If the issue involves timing, call `get_spans` for trace data.
- If the issue involves metrics, call `get_metrics`.
+ If circuit breakers are involved, call
`camel_runtime_circuit_breakers`.
+ If datasources are involved, call `camel_runtime_datasources`.
+ If the issue involves timing, call `camel_runtime_spans` for
trace data.
+ If the issue involves metrics, call `camel_runtime_metrics`.
### Step 7: Root cause analysis
Synthesize the gathered data into a structured diagnosis:
@@ -322,27 +322,27 @@ public class PromptDefinitions {
Follow these steps in order:
### Step 1: Baseline performance
- Call `get_context` to get overall exchange statistics (total,
completed, failed, mean time).
- Call `get_routes` to get per-route throughput and timing data.
- Call `get_top_processors` to identify the slowest processors.
+ Call `camel_runtime_context` to get overall exchange
statistics (total, completed, failed, mean time).
+ Call `camel_runtime_routes` to get per-route throughput and
timing data.
+ Call `camel_runtime_top` to identify the slowest processors.
### Step 2: Resource analysis
- Call `get_memory` to assess JVM memory usage, GC frequency,
and thread counts.
- Call `get_consumers` to check consumer configuration and
polling strategies.
- Call `get_endpoints` to review endpoint usage patterns.
+ Call `camel_runtime_memory` to assess JVM memory usage, GC
frequency, and thread counts.
+ Call `camel_runtime_consumers` to check consumer configuration
and polling strategies.
+ Call `camel_runtime_endpoints` to review endpoint usage
patterns.
### Step 3: Route structure analysis
- Call `get_route_analysis` to get route structure and
anti-pattern hints.
- Call `get_eip_stats` to understand EIP usage and processor
performance distribution.
+ Call `camel_runtime_route_analysis` with routeId="%s" to get
route structure and anti-pattern hints.
+ Call `camel_runtime_eip_stats` to understand EIP usage and
processor performance distribution.
### Step 4: Infrastructure checks
- Call `get_metrics` for detailed Micrometer metrics (if
available).
- Call `get_datasources` if database access is involved.
- Call `get_circuit_breakers` if resilience patterns are in use.
+ Call `camel_runtime_metrics` for detailed Micrometer metrics
(if available).
+ Call `camel_runtime_datasources` if database access is
involved.
+ Call `camel_runtime_circuit_breakers` if resilience patterns
are in use.
### Step 5: Route inspection
- Call `get_route_source` with filter="%s" to review the route
code.
- Call `get_route_dump` with routeId="%s" and format="yaml" to
see the full definition.
+ Call `camel_runtime_route_dump` with routeId="%s" and
format="yaml" to see the full definition.
+ Call `camel_runtime_route_source` to read the source file it
was loaded from.
### Step 6: Optimization report
Produce a structured optimization report:
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeService.java
b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeService.java
index a75f5e63191b..aeafaeca701c 100644
---
a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeService.java
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeService.java
@@ -16,12 +16,21 @@
*/
package org.apache.camel.dsl.jbang.core.commands.mcp;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import jakarta.enterprise.context.ApplicationScoped;
import io.quarkiverse.mcp.server.ToolCallException;
+import org.apache.camel.dsl.jbang.core.common.CommandLineHelper;
import org.apache.camel.dsl.jbang.core.common.RuntimeHelper;
import org.apache.camel.util.json.JsonObject;
import org.apache.camel.util.json.Jsoner;
@@ -123,6 +132,68 @@ public class RuntimeService {
return RuntimeHelper.stopApplication(pid);
}
+ /**
+ * The size of the file the CLI connector appends the messages of the
receive console to
+ * ({@code {pid}-receive.json}, read by {@code camel cmd receive} too), to
read only what arrives after this point.
+ */
+ public long receiveFileSize(long pid) {
+ try {
+ Path file = receiveFile(pid);
+ return Files.exists(file) ? Files.size(file) : 0;
+ } catch (IOException e) {
+ return 0;
+ }
+ }
+
+ /**
+ * The messages the CLI connector appended to the receive file after the
given offset: each complete line is a
+ * {@code {"messages": [...]}} dump of the receive console. Only what was
appended after the offset is read, as the
+ * file is append-only and kept across sessions.
+ */
+ public List<JsonObject> readReceivedMessages(long pid, long offset) {
+ List<JsonObject> messages = new ArrayList<>();
+ Path file = receiveFile(pid);
+ try {
+ if (!Files.exists(file)) {
+ return messages;
+ }
+ String text;
+ try (SeekableByteChannel channel = Files.newByteChannel(file,
StandardOpenOption.READ)) {
+ long size = channel.size();
+ // camel cmd receive --action=clear rewrites the file: read it
from the start then
+ long from = offset <= size ? offset : 0;
+ ByteBuffer buffer = ByteBuffer.allocate(Math.toIntExact(size -
from));
+ channel.position(from);
+ int read = 0;
+ while (buffer.hasRemaining() && read >= 0) {
+ read = channel.read(buffer);
+ }
+ text = new String(buffer.array(), 0, buffer.position(),
StandardCharsets.UTF_8);
+ }
+ int end = text.lastIndexOf('\n');
+ if (end < 0) {
+ return messages;
+ }
+ for (String line : text.substring(0, end).split("\n")) {
+ Object parsed = Jsoner.deserialize(line.trim(), (JsonObject)
null);
+ if (parsed instanceof JsonObject dump &&
dump.getCollection("messages") != null) {
+ for (Object m : dump.getCollection("messages")) {
+ if (m instanceof JsonObject jo) {
+ messages.add(jo);
+ }
+ }
+ }
+ }
+ } catch (IOException e) {
+ // not readable yet: the caller polls again
+ }
+ return messages;
+ }
+
+ private static Path receiveFile(long pid) {
+ return CommandLineHelper.getCamelDir().resolve(pid + "-receive.json");
+ }
+
public JsonObject executeAction(long pid, String action,
Consumer<JsonObject> configure) {
String result = RuntimeHelper.executeAction(pid, action, configure);
if (result != null && result.startsWith("Timeout")) {
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java
b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java
index 35afe7ebd77c..e05ca885f891 100644
---
a/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java
@@ -16,9 +16,17 @@
*/
package org.apache.camel.dsl.jbang.core.commands.mcp;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
@@ -29,6 +37,7 @@ import io.quarkiverse.mcp.server.ToolCallException;
import org.apache.camel.dsl.jbang.core.commands.ai.ToolContext;
import org.apache.camel.dsl.jbang.core.commands.ai.ToolExecutionException;
import org.apache.camel.dsl.jbang.core.commands.ai.ToolRegistry;
+import org.apache.camel.util.StopWatch;
import org.apache.camel.util.json.JsonObject;
import org.apache.camel.util.json.Jsoner;
@@ -49,6 +58,21 @@ public class RuntimeTools {
private static final String NAME_OR_PID_DESC
= "Name or PID of the Camel process. Leave empty to auto-detect
(works when exactly one Camel process is running)";
+ private static final Set<String> MEMORY_LEAK_COMMANDS = Set.of("start",
"stop", "status", "query", "compare");
+ private static final int SINGLE_DEFAULT_DURATION = 60;
+ private static final int DUAL_DEFAULT_DURATION = 30;
+
+ private static final int RECEIVE_DEFAULT_TIMEOUT = 10;
+ private static final int RECEIVE_MAX_TIMEOUT = 60;
+ private static final long RECEIVE_POLL_MS = 200;
+ // the connector applies the stop and writes the status in one tick, and
dumps the console every second tick
+ // (one tick is a second by default)
+ private static final long RECEIVE_SETTLE_DEFAULT_MS = 3000;
+
+ private final ConcurrentMap<Long, ReentrantLock> receiveLocks = new
ConcurrentHashMap<>();
+ /** How long to wait after the stop for the process to confirm it and to
write the last messages. */
+ long receiveSettleMs = RECEIVE_SETTLE_DEFAULT_MS;
+
@Inject
RuntimeService runtimeService;
@@ -204,7 +228,8 @@ public class RuntimeTools {
description = "Dump route definitions in XML, YAML, or Java DSL
format.")
public JsonObject camel_runtime_route_dump(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid,
- @ToolArg(description = "Route ID to dump (use * for all routes)",
required = false) String routeId,
+ @ToolArg(description = "Route ID or pattern such as orders* to
dump (default * for all routes)",
+ required = false) String routeId,
@ToolArg(description = "Output format: xml, yaml, or java
(default: yaml)", required = false) String format) {
Map<String, String> args = new HashMap<>();
args.put("routeId", routeId != null ? routeId : "*");
@@ -216,9 +241,16 @@ public class RuntimeTools {
description = "Show the route structure as a tree of processors.")
public JsonObject camel_runtime_route_structure(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid,
- @ToolArg(description = "Route ID to inspect (use * for all
routes)", required = false) String routeId) {
- return delegateToRegistry("get_route_structure", nameOrPid,
- Map.of("routeId", routeId != null ? routeId : "*"));
+ @ToolArg(description = "Route ID or pattern such as orders* to
inspect (default * for all routes)",
+ required = false) String routeId,
+ @ToolArg(description = "Only the overall structure, without the
options and expressions (default false)",
+ required = false) Boolean brief) {
+ Map<String, String> args = new HashMap<>();
+ args.put("routeId", routeId != null ? routeId : "*");
+ if (Boolean.TRUE.equals(brief)) {
+ args.put("brief", "true");
+ }
+ return delegateToRegistry("get_route_structure", nameOrPid, args);
}
@Tool(annotations = @Tool.Annotations(readOnlyHint = true, destructiveHint
= false, openWorldHint = false),
@@ -240,6 +272,8 @@ public class RuntimeTools {
return delegateToRegistry("get_processor_detail", nameOrPid, args);
}
+ // one tool for the four commands, and the hint is per tool: destructive
because stop and suspend take the route's
+ // consumer offline (the shared registry, with one tool per command, marks
only stop_route destructive)
@Tool(annotations = @Tool.Annotations(readOnlyHint = false,
destructiveHint = true, openWorldHint = false),
description = "Control a route: start, stop, suspend, or resume it.")
public JsonObject camel_runtime_route_control(
@@ -326,7 +360,9 @@ public class RuntimeTools {
@ToolArg(description = "The path with query, e.g.
/api/orders?status=open") String path,
@ToolArg(description = "Headers as a JSON object or one 'name:
value' per line",
required = false) String headers,
- @ToolArg(description = "The request body", required = false)
String body) {
+ @ToolArg(description = "The request body", required = false)
String body,
+ @ToolArg(description = "How long to wait for the answer in seconds
(default 30, at most 120)",
+ required = false) Integer timeoutSeconds) {
if (path == null || path.isBlank()) {
throw new ToolCallException("path is required, e.g.
/api/orders/1", null);
}
@@ -337,6 +373,9 @@ public class RuntimeTools {
if (body != null) {
args.put("body", body);
}
+ if (timeoutSeconds != null) {
+ args.put("timeoutSeconds", timeoutSeconds.toString());
+ }
return delegateToRegistry("http_request", nameOrPid, args);
}
@@ -460,7 +499,9 @@ public class RuntimeTools {
}
@Tool(annotations = @Tool.Annotations(readOnlyHint = false,
destructiveHint = false, openWorldHint = false),
- description = "Enable, disable, or dump message tracing for the
running Camel application.")
+ description = """
+ Enable, disable, or dump message tracing for the running
Camel application. \
+ camel_runtime_trace_dump returns the traced messages too,
and is offered to read-only clients.""")
public JsonObject camel_runtime_trace(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid,
@ToolArg(description = "Action: enable, disable, or dump") String
action) {
@@ -470,6 +511,15 @@ public class RuntimeTools {
return delegateToRegistry("trace_control", nameOrPid, Map.of("action",
action));
}
+ @Tool(annotations = @Tool.Annotations(readOnlyHint = true, destructiveHint
= false, openWorldHint = false),
+ description = """
+ Dump the messages traced in the running Camel application
(tracing must be enabled, \
+ see camel_runtime_trace). Same answer as camel_runtime_trace
with action dump.""")
+ public JsonObject camel_runtime_trace_dump(
+ @ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid) {
+ return delegateToRegistry("trace_control", nameOrPid, Map.of("action",
"dump"));
+ }
+
@Tool(annotations = @Tool.Annotations(readOnlyHint = true, destructiveHint
= false, openWorldHint = false),
description = "Show top processor statistics: which processors are
slowest and most active.")
public JsonObject camel_runtime_top(
@@ -515,19 +565,15 @@ public class RuntimeTools {
with tools like Eclipse MAT, VisualVM, or jhat. The dump is
written to the process working directory.""")
public JsonObject camel_runtime_heap_dump(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid,
- @ToolArg(description = "File name for the heap dump (without
.hprof extension). Defaults to heap-dump-<timestamp>",
+ @ToolArg(description = "File name for the heap dump, without a
directory and without the .hprof extension. "
+ + "Defaults to heap-dump-<timestamp>",
required = false) String name,
@ToolArg(description = "Whether to dump only live objects (default
true). Live dumps trigger a GC first",
required = false) String live) {
- RuntimeService.ProcessInfo p =
runtimeService.findSingleProcess(nameOrPid);
- return runtimeService.executeAction(p.pid(), "heap-dump", root -> {
- if (name != null && !name.isBlank()) {
- root.put("name", name);
- }
- if (live != null && !live.isBlank()) {
- root.put("live", live);
- }
- });
+ Map<String, String> args = new HashMap<>();
+ putIfNotBlank(args, "name", name);
+ putIfNotBlank(args, "live", live);
+ return delegateToRegistry("write_heap_dump", nameOrPid, args);
}
@Tool(annotations = @Tool.Annotations(readOnlyHint = false,
destructiveHint = false, openWorldHint = false),
@@ -536,39 +582,48 @@ public class RuntimeTools {
JFR samples objects that survive multiple GC cycles and
captures their reference chains back to GC roots. \
This is lightweight and safe for production, but shows
sampled sizes not total heap — \
use values to compare classes relative to each other and
spot trends, not as absolute numbers. \
- For deep analysis, use a full heap dump (jmap -dump:live)
with tools like Eclipse MAT or jhat. \
- Use command 'start' to begin recording, 'stop' to stop and
get results, \
- 'status' to check recording state, and 'query' to retrieve
cached results from the last recording. \
- Use mode 'dual' with 'start' to run two sequential
recordings (Xs then 2Xs) and automatically \
- compare trends — returns growth ratios and trend
classifications (growing, stable, shrinking, new, gone). \
+ For deep analysis, use a full heap dump
(camel_runtime_heap_dump) with tools like Eclipse MAT. \
+ Use command 'start' to begin a recording (it stops by itself
after the duration), 'status' to check \
+ it, 'query' (or 'stop' to end it early) to get the results,
and 'compare' to compare the last two \
+ recordings: growth ratios and trends (growing, stable,
shrinking, new, gone). \
+ For a trend, start a recording of X seconds, then one of 2X
seconds, then compare. \
+ mode 'dual' with 'start' does all of this in one call, which
blocks for about three times the \
+ duration (90 s with the default 30): use it only when the
client waits that long. \
Entries with lowConfidence=true have unreliable growth
percentages due to low sample counts or \
sample counts that diverge significantly between runs —
recommend a longer recording duration.""")
public JsonObject camel_runtime_memory_leak(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid,
- @ToolArg(description = "Command: start, stop, status, or query")
String command,
- @ToolArg(description = "Recording duration in seconds (only for
start command, default 60, use 0 for manual stop)",
+ @ToolArg(description = "Command: start, stop, status, query, or
compare") String command,
+ @ToolArg(description = "Recording duration in seconds, for start
(default 60, use 0 to stop it manually; "
+ + "in dual mode default 30, the second
recording takes twice as long)",
required = false) String duration,
- @ToolArg(description = "Recording mode: dual (default, two
recordings at Xs and 2Xs with trend comparison) or single (one recording)",
+ @ToolArg(description = "Recording mode: single (default, start
returns at once) or dual "
+ + "(two recordings at Xs and 2Xs with trend
comparison, the call blocks until done)",
required = false) String mode,
@ToolArg(description = "Include allocation stack traces in results
(default false, set true for detailed analysis)",
required = false) String stacktrace,
@ToolArg(description = "Minimum total size in bytes to include a
sample (e.g. 1024 for 1KB). Filters out small allocations to reduce noise.
Default 1024 (1KB) in dual mode",
required = false) String minSize) {
- if (command == null || command.isBlank()) {
- throw new ToolCallException("command is required (start, stop,
status, or query)", null);
+ String cmd = command != null ? command.trim().toLowerCase(Locale.ROOT)
: "";
+ if (!MEMORY_LEAK_COMMANDS.contains(cmd)) {
+ throw new ToolCallException("command is required: start, stop,
status, query, or compare", null);
+ }
+ boolean dual = mode != null && "dual".equalsIgnoreCase(mode.trim());
+ if (!dual && mode != null && !mode.isBlank() &&
!"single".equalsIgnoreCase(mode.trim())) {
+ throw new ToolCallException("mode must be single or dual, got: " +
mode, null);
}
+ Integer seconds = parseDuration(duration);
RuntimeService.ProcessInfo p =
runtimeService.findSingleProcess(nameOrPid);
- // dual is the documented default: an omitted or blank mode records
twice too
- boolean dual = mode == null || mode.isBlank() ||
"dual".equalsIgnoreCase(mode);
- if ("start".equals(command) && dual) {
- return doDualJfrRecording(p.pid(), duration, stacktrace, minSize);
+ if ("start".equals(cmd) && dual) {
+ return doDualJfrRecording(p.pid(), seconds == null || seconds <= 0
? DUAL_DEFAULT_DURATION : seconds,
+ stacktrace, minSize);
}
return runtimeService.executeAction(p.pid(), "jfr-memory-leak", root
-> {
- root.put("command", command);
- if ("start".equals(command) && duration != null &&
!duration.isBlank()) {
- root.put("duration", duration);
+ root.put("command", cmd);
+ if ("start".equals(cmd)) {
+ root.put("duration", String.valueOf(seconds != null ? seconds
: SINGLE_DEFAULT_DURATION));
}
if (stacktrace != null) {
root.put("stacktrace", stacktrace);
@@ -579,16 +634,25 @@ public class RuntimeTools {
});
}
- private JsonObject doDualJfrRecording(long pid, String duration, String
stacktrace, String minSize) {
- int dur = 30;
- if (duration != null && !duration.isBlank()) {
- dur = Integer.parseInt(duration);
+ static Integer parseDuration(String duration) {
+ if (duration == null || duration.isBlank()) {
+ return null;
}
- if (dur <= 0) {
- dur = 30;
+ try {
+ int seconds = Integer.parseInt(duration.trim());
+ if (seconds < 0) {
+ throw new NumberFormatException();
+ }
+ return seconds;
+ } catch (NumberFormatException e) {
+ throw new ToolCallException(
+ "duration must be a whole number of seconds (0 or more),
got: " + duration,
+ null);
}
- int dur1 = dur;
- int dur2 = dur * 2;
+ }
+
+ private JsonObject doDualJfrRecording(long pid, int dur1, String
stacktrace, String minSize) {
+ int dur2 = dur1 * 2;
// run 1
JsonObject r1 = runtimeService.executeAction(pid, "jfr-memory-leak",
root -> {
@@ -662,50 +726,265 @@ public class RuntimeTools {
});
}
+ /**
+ * Waits until the recording stopped by itself: sleeps for the time the
console says is left, then checks again (a
+ * recording ends a little after its duration, as it is parsed).
+ */
private void waitForJfrRecordingComplete(long pid, int dur) {
long deadline = System.currentTimeMillis() + (dur + 30) * 1000L;
while (System.currentTimeMillis() < deadline) {
+ JsonObject status = runtimeService.executeAction(pid,
"jfr-memory-leak", root -> {
+ root.put("command", "status");
+ });
+ if (status == null ||
!"recording".equals(status.getString("status"))) {
+ return;
+ }
+ long remaining = status.getLongOrDefault("remainingMs", 1000L);
+ long wait = Math.min(Math.max(remaining, 500L), Math.max(1L,
deadline - System.currentTimeMillis()));
try {
- Thread.sleep(3000);
+ TimeUnit.MILLISECONDS.sleep(wait);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new ToolCallException("Interrupted while waiting for
recording", null);
}
- JsonObject status = runtimeService.executeAction(pid,
"jfr-memory-leak", root -> {
- root.put("command", "status");
- });
- if (status != null &&
!"recording".equals(status.getString("status"))) {
- break;
- }
}
}
@Tool(annotations = @Tool.Annotations(readOnlyHint = true, destructiveHint
= false, openWorldHint = false),
description = """
- Get the message history trace of the last completed
exchange. \
- This is always captured (no need to enable tracing) and
shows the single most recent exchange \
- with its route path, processors visited, headers, body, and
timing.""")
+ Get the message history trace of the last completed
exchange, with its route path, processors \
+ visited, headers, body, and timing. Recorded when message
history is on and the tracer is \
+ enabled or on standby: the default of camel run (dev
profile); otherwise set \
+ camel.main.message-history=true and
camel.trace.standby=true.""")
public JsonObject camel_runtime_history(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid) {
return delegateToRegistry("get_history", nameOrPid, Map.of());
}
- @Tool(annotations = @Tool.Annotations(readOnlyHint = false,
destructiveHint = false, openWorldHint = false),
+ @Tool(annotations = @Tool.Annotations(readOnlyHint = false,
destructiveHint = true, openWorldHint = false),
description = """
- Receive (poll) a message from a Camel endpoint in the
running application. \
- This is the complement to camel_runtime_send — it consumes
one message from the endpoint.""")
+ Receive (consume) a message from an endpoint of the running
application, the complement of \
+ camel_runtime_send: starts a consumer on the endpoint, waits
for a message, returns it and stops \
+ the consumer again. Returns one message by default; limit
returns up to that many, and received \
+ counts every message the consumer took, returned or not. The
consumer competes with the routes \
+ that consume the endpoint, so the messages it takes do not
reach them (a queue or topic message \
+ is gone); one call at a time per process. To look at queued
messages without taking them use \
+ camel_runtime_browse.""")
public JsonObject camel_runtime_receive(
@ToolArg(description = NAME_OR_PID_DESC, required = false) String
nameOrPid,
- @ToolArg(description = "Endpoint URI to receive from") String
endpoint) {
+ @ToolArg(description = "Endpoint URI to receive from, or a pattern
matching an endpoint the application "
+ + "sends to, e.g. seda:orders or kafka*")
String endpoint,
+ @ToolArg(description = "How long to wait for a message in seconds
(default 10, at most 60)",
+ required = false) Integer timeoutSeconds,
+ @ToolArg(description = "How many of the messages the consumer
takes to return (default 1, at least 1)",
+ required = false) Integer limit) {
if (endpoint == null || endpoint.isBlank()) {
throw new ToolCallException("endpoint is required", null);
}
- // receive is MCP-only (no ToolRegistry equivalent) — call
RuntimeService directly
+ if (limit != null && limit < 1) {
+ throw new ToolCallException("limit must be at least 1: " + limit,
null);
+ }
+ int timeout = timeoutSeconds == null || timeoutSeconds <= 0
+ ? RECEIVE_DEFAULT_TIMEOUT : Math.min(timeoutSeconds,
RECEIVE_MAX_TIMEOUT);
RuntimeService.ProcessInfo p =
runtimeService.findSingleProcess(nameOrPid);
- return runtimeService.executeAction(p.pid(), "receive", root -> {
+ return receive(p.pid(), endpoint, timeout * 1000L, limit != null ?
limit : 1);
+ }
+
+ /**
+ * Enables the receive console of the process for the endpoint, waits for
the messages the CLI connector writes to
+ * the receive file, and disables it again whatever happens. One call at a
time per process: the console is shared,
+ * and disabling it stops every consumer it runs.
+ */
+ JsonObject receive(long pid, String endpoint, long timeoutMs, int limit) {
+ ReentrantLock lock = receiveLocks.computeIfAbsent(pid, k -> new
ReentrantLock());
+ boolean locked;
+ try {
+ locked = lock.tryLock(timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ToolCallException("Interrupted while waiting for a
message", null);
+ }
+ if (!locked) {
+ throw new ToolCallException(
+ "Another camel_runtime_receive call is receiving from this
process (the receive console is shared,"
+ + " and stopping it would stop that
call's consumer): call again when it has finished",
+ null);
+ }
+ try {
+ return receiveExclusively(pid, endpoint, timeoutMs, limit);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private JsonObject receiveExclusively(long pid, String endpoint, long
timeoutMs, int limit) {
+ JsonObject state = runtimeService.readStatusSection(pid, "receive");
+ Collection<?> active = state.getCollection("endpoints");
+ if (Boolean.TRUE.equals(state.getBoolean("enabled")) && active != null
&& !active.isEmpty()) {
+ // disabling would stop those consumers too
+ throw new ToolCallException(
+ "The process is already receiving from " + active
+ + " (camel cmd receive); stop that
first with: camel cmd receive --action=stop",
+ null);
+ }
+ // the console numbers every message it takes (uid) from one counter,
which the status reports as total:
+ // the messages above the count before this call are the ones its
consumer takes
+ long startTotal = longValue(state, "total", 0);
+ long offset = runtimeService.receiveFileSize(pid);
+ JsonObject started = runtimeService.executeAction(pid, "receive", root
-> {
root.put("enabled", "true");
root.put("endpoint", endpoint);
});
+ List<JsonObject> messages;
+ try {
+ messages = awaitMessages(pid, endpoint, started, offset,
startTotal, timeoutMs);
+ } catch (RuntimeException e) {
+ try {
+ stopReceiving(pid);
+ } catch (RuntimeException stop) {
+ throw new ToolCallException(
+ message(e) + ". The consumer could not be stopped (" +
message(stop)
+ + "): stop it with: camel cmd
receive --action=stop",
+ e);
+ }
+ throw e;
+ }
+ JsonObject answer = new JsonObject();
+ answer.put("endpoint", started.getString("url"));
+ Long taken = null;
+ try {
+ stopReceiving(pid);
+ // the connector writes the queue of the console every second
tick, so the messages taken between the dump
+ // read above and the stop come later: the status after the stop
says how many there are
+ taken = settle(pid, offset, startTotal, messages);
+ } catch (RuntimeException e) {
+ // the messages are taken from the endpoint already: return them,
and say the consumer may still run
+ answer.put("warning", "The consumer could not be stopped (" +
message(e)
+ + ") and may still take messages from the
endpoint: stop it with: "
+ + "camel cmd receive --action=stop");
+ }
+ long received = taken != null ? taken : messages.size();
+ int returned = (int) Math.min(limit, messages.size());
+ answer.put("received", received);
+ answer.put("messages", messages.subList(0, returned));
+ if (received == 0) {
+ answer.put("note", "No message arrived within " + (timeoutMs /
1000) + " seconds");
+ return answer;
+ }
+ long notReturned = received - returned;
+ StringBuilder note = new StringBuilder();
+ if (notReturned > 0) {
+ answer.put("notReturned", notReturned);
+ note.append(notReturned).append(" more message(s) were taken from
the endpoint by this call and are not")
+ .append(" returned");
+ if (messages.size() > returned) {
+ note.append(" (raise limit to see them)");
+ }
+ if (received > messages.size()) {
+ note.append("; ").append(received - messages.size())
+ .append(" of them were not written by the process
within the wait after the stop");
+ }
+ note.append('.');
+ }
+ if (taken == null) {
+ note.append(" The process did not confirm the stop, so received
may be short of what the consumer took.");
+ }
+ if (!note.isEmpty()) {
+ answer.put("note", note.toString().trim());
+ }
+ return answer;
+ }
+
+ /**
+ * Waits for the first dump of the receive file that holds a message of
this call, or for the timeout; returns the
+ * messages found (none on timeout).
+ */
+ private List<JsonObject> awaitMessages(
+ long pid, String endpoint, JsonObject started, long offset, long
startTotal, long timeoutMs) {
+ if (started.getString("error") != null) {
+ throw new ToolCallException("Cannot receive from " + endpoint + ":
" + started.getString("error"), null);
+ }
+ if (started.getString("url") == null) {
+ throw new ToolCallException(
+ "No endpoint matches " + endpoint
+ + " (a pattern matches the endpoints
the application sends to; "
+ + "camel_runtime_endpoints lists
them)",
+ null);
+ }
+ StopWatch watch = new StopWatch();
+ while (true) {
+ List<JsonObject> messages = ownMessages(pid, offset, startTotal,
null);
+ if (!messages.isEmpty() || watch.taken() >= timeoutMs) {
+ return messages;
+ }
+ sleep();
+ }
+ }
+
+ /**
+ * After the stop: waits until the status of the process shows the console
disabled, so its total is the number of
+ * messages this call took, and until the receive file holds them (the
connector dumps the queue every second tick).
+ * Returns the number taken, or null when the process did not confirm the
stop within the wait; the list of messages
+ * is completed with the ones dumped after the stop.
+ */
+ private Long settle(long pid, long offset, long startTotal,
List<JsonObject> messages) {
+ StopWatch watch = new StopWatch();
+ Long taken = null;
+ while (true) {
+ JsonObject state = runtimeService.readStatusSection(pid,
"receive");
+ if (Boolean.FALSE.equals(state.getBoolean("enabled")) &&
state.get("total") != null) {
+ long endTotal = longValue(state, "total", startTotal);
+ taken = Math.max(endTotal - startTotal, messages.size());
+ List<JsonObject> all = ownMessages(pid, offset, startTotal,
endTotal);
+ if (all.size() >= messages.size()) {
+ messages.clear();
+ messages.addAll(all);
+ }
+ if (messages.size() >= taken) {
+ return taken;
+ }
+ }
+ if (watch.taken() >= receiveSettleMs) {
+ return taken;
+ }
+ sleep();
+ }
+ }
+
+ /** The messages of the receive file after the offset that this call's
consumer took: uid above the start count. */
+ private List<JsonObject> ownMessages(long pid, long offset, long
startTotal, Long endTotal) {
+ List<JsonObject> messages = new
ArrayList<>(runtimeService.readReceivedMessages(pid, offset));
+ messages.removeIf(m -> {
+ Object uid = m.get("uid");
+ if (!(uid instanceof Number n)) {
+ return false;
+ }
+ return n.longValue() <= startTotal || endTotal != null &&
n.longValue() > endTotal;
+ });
+ return messages;
+ }
+
+ /** An action without an endpoint stops the consumers of the receive
console. */
+ private void stopReceiving(long pid) {
+ runtimeService.executeAction(pid, "receive", null);
+ }
+
+ private static void sleep() {
+ try {
+ TimeUnit.MILLISECONDS.sleep(RECEIVE_POLL_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ToolCallException("Interrupted while waiting for a
message", null);
+ }
+ }
+
+ private static long longValue(JsonObject json, String key, long
defaultValue) {
+ return json.get(key) instanceof Number n ? n.longValue() :
defaultValue;
+ }
+
+ /** The message of an exception, or its name when it has none. */
+ private static String message(Throwable e) {
+ return e.getMessage() != null ? e.getMessage() : e.toString();
}
@Tool(annotations = @Tool.Annotations(readOnlyHint = true, destructiveHint
= false, openWorldHint = false),
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/McpSecuredToolClassesTest.java
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/McpSecuredToolClassesTest.java
index 04df44cd56a0..4f26090fb8d8 100644
---
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/McpSecuredToolClassesTest.java
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/McpSecuredToolClassesTest.java
@@ -16,14 +16,7 @@
*/
package org.apache.camel.dsl.jbang.core.commands.mcp;
-import java.io.IOException;
-import java.net.URISyntaxException;
-import java.net.URL;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.Arrays;
import java.util.List;
-import java.util.stream.Stream;
import io.quarkiverse.mcp.server.Tool;
import org.junit.jupiter.api.Test;
@@ -39,7 +32,7 @@ class McpSecuredToolClassesTest {
@Test
void everyToolClassIsMcpSecured() throws Exception {
- List<Class<?>> toolClasses = toolClasses();
+ List<Class<?>> toolClasses = McpToolClasses.toolClasses();
// guard against a scan that finds nothing and makes the check below
vacuous
assertThat(toolClasses).hasSizeGreaterThan(20).contains(AuthoringTools.class,
CatalogTools.class);
@@ -48,31 +41,4 @@ class McpSecuredToolClassesTest {
.as("tool classes without @McpSecured")
.isEmpty();
}
-
- private static List<Class<?>> toolClasses() throws IOException,
URISyntaxException {
- URL url =
McpSecured.class.getResource(McpSecured.class.getSimpleName() + ".class");
- assertThat(url).isNotNull();
- assertThat(url.getProtocol()).as("the main classes are scanned as a
directory").isEqualTo("file");
- Path dir = Path.of(url.toURI()).getParent();
- String pkg = McpSecured.class.getPackageName();
- try (Stream<Path> files = Files.list(dir)) {
- return files.map(p -> p.getFileName().toString())
- .filter(n -> n.endsWith(".class") && !n.contains("$"))
- .<Class<?>> map(n -> load(pkg + "." + n.substring(0,
n.length() - ".class".length())))
- .filter(McpSecuredToolClassesTest::declaresTool)
- .toList();
- }
- }
-
- private static boolean declaresTool(Class<?> clazz) {
- return Arrays.stream(clazz.getDeclaredMethods()).anyMatch(m ->
m.isAnnotationPresent(Tool.class));
- }
-
- private static Class<?> load(String name) {
- try {
- return Class.forName(name, false,
McpSecuredToolClassesTest.class.getClassLoader());
- } catch (ClassNotFoundException e) {
- throw new IllegalStateException(e);
- }
- }
}
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/McpSecuredToolClassesTest.java
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/McpToolClasses.java
similarity index 67%
copy from
dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/McpSecuredToolClassesTest.java
copy to
dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/McpToolClasses.java
index 04df44cd56a0..a1c5d138d1df 100644
---
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/McpSecuredToolClassesTest.java
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/McpToolClasses.java
@@ -23,33 +23,24 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Arrays;
import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
import io.quarkiverse.mcp.server.Tool;
-import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/**
- * Every class of the MCP module that exposes a {@link Tool} must carry {@link
McpSecured}: the security interceptor
- * (audit, argument sanitisation, secret redaction) is bound by that
annotation only, so a tool class without it
- * silently skips all of it.
+ * The classes of the MCP module that expose a {@link Tool}, found by scanning
the main classes directory, and the names
+ * of their tools: what the server lists.
*/
-class McpSecuredToolClassesTest {
+final class McpToolClasses {
- @Test
- void everyToolClassIsMcpSecured() throws Exception {
- List<Class<?>> toolClasses = toolClasses();
-
- // guard against a scan that finds nothing and makes the check below
vacuous
-
assertThat(toolClasses).hasSizeGreaterThan(20).contains(AuthoringTools.class,
CatalogTools.class);
- assertThat(toolClasses)
- .filteredOn(c -> !c.isAnnotationPresent(McpSecured.class))
- .as("tool classes without @McpSecured")
- .isEmpty();
+ private McpToolClasses() {
}
- private static List<Class<?>> toolClasses() throws IOException,
URISyntaxException {
+ static List<Class<?>> toolClasses() throws IOException, URISyntaxException
{
URL url =
McpSecured.class.getResource(McpSecured.class.getSimpleName() + ".class");
assertThat(url).isNotNull();
assertThat(url.getProtocol()).as("the main classes are scanned as a
directory").isEqualTo("file");
@@ -59,18 +50,30 @@ class McpSecuredToolClassesTest {
return files.map(p -> p.getFileName().toString())
.filter(n -> n.endsWith(".class") && !n.contains("$"))
.<Class<?>> map(n -> load(pkg + "." + n.substring(0,
n.length() - ".class".length())))
- .filter(McpSecuredToolClassesTest::declaresTool)
+ .filter(McpToolClasses::declaresTool)
.toList();
}
}
+ /** The tool names: the {@link Tool#name()} when set, else the method
name. */
+ static Set<String> toolNames() throws IOException, URISyntaxException {
+ return toolClasses().stream()
+ .flatMap(c -> Arrays.stream(c.getDeclaredMethods()))
+ .filter(m -> m.isAnnotationPresent(Tool.class))
+ .map(m -> {
+ String name = m.getAnnotation(Tool.class).name();
+ return Tool.ELEMENT_NAME.equals(name) ? m.getName() : name;
+ })
+ .collect(Collectors.toSet());
+ }
+
private static boolean declaresTool(Class<?> clazz) {
return Arrays.stream(clazz.getDeclaredMethods()).anyMatch(m ->
m.isAnnotationPresent(Tool.class));
}
private static Class<?> load(String name) {
try {
- return Class.forName(name, false,
McpSecuredToolClassesTest.class.getClassLoader());
+ return Class.forName(name, false,
McpToolClasses.class.getClassLoader());
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e);
}
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/PromptDefinitionsTest.java
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/PromptDefinitionsTest.java
index 405e9b0407a7..5ae648d67beb 100644
---
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/PromptDefinitionsTest.java
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/PromptDefinitionsTest.java
@@ -16,8 +16,15 @@
*/
package org.apache.camel.dsl.jbang.core.commands.mcp;
+import java.lang.reflect.Method;
+import java.util.Arrays;
import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import io.quarkiverse.mcp.server.Prompt;
import io.quarkiverse.mcp.server.PromptMessage;
import org.junit.jupiter.api.Test;
@@ -232,6 +239,32 @@ class PromptDefinitionsTest {
assertThat(text).contains("Recommendations");
}
+ // ---- every prompt ----
+
+ @Test
+ void everyToolAPromptNamesExists() throws Exception {
+ // a prompt names its tools as `snake_case`: each must be a tool the
server lists
+ Set<String> tools = McpToolClasses.toolNames();
+ Pattern named = Pattern.compile("`([a-z][a-z0-9]*(?:_[a-z0-9]+)+)`");
+ Set<String> mentioned = new TreeSet<>();
+ for (Method m : PromptDefinitions.class.getDeclaredMethods()) {
+ if (!m.isAnnotationPresent(Prompt.class)) {
+ continue;
+ }
+ Object[] args = new Object[m.getParameterCount()];
+ Arrays.fill(args, "myRoute");
+ @SuppressWarnings("unchecked")
+ List<PromptMessage> messages = (List<PromptMessage>)
m.invoke(prompts, args);
+ Matcher matcher = named.matcher(extractText(messages));
+ while (matcher.find()) {
+ mentioned.add(matcher.group(1));
+ }
+ }
+
+ assertThat(mentioned).contains("camel_runtime_context",
"camel_runtime_top", "camel_catalog_find");
+ assertThat(mentioned).as("tools named by the
prompts").isSubsetOf(tools);
+ }
+
// ---- helper ----
private String extractText(List<PromptMessage> messages) {
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsTest.java
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsTest.java
index f95a01a2d991..66bc21f02de8 100644
---
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsTest.java
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsTest.java
@@ -89,7 +89,7 @@ class RuntimeToolsTest {
@Test
void httpRequestRequiresPath() {
RuntimeTools tools = createTools();
- assertThatThrownBy(() -> tools.camel_runtime_http_request(null, "GET",
" ", null, null))
+ assertThatThrownBy(() -> tools.camel_runtime_http_request(null, "GET",
" ", null, null, null))
.isInstanceOf(ToolCallException.class)
.hasMessageContaining("path is required");
}
@@ -100,7 +100,7 @@ class RuntimeToolsTest {
RuntimeTools tools = createTools();
assertThatThrownBy(() ->
tools.camel_runtime_http_endpoints("no-such-app-25307", null))
.isInstanceOf(ToolCallException.class);
- assertThatThrownBy(() ->
tools.camel_runtime_http_request("no-such-app-25307", "GET", "/hello", null,
null))
+ assertThatThrownBy(() ->
tools.camel_runtime_http_request("no-such-app-25307", "GET", "/hello", null,
null, null))
.isInstanceOf(ToolCallException.class);
}
@@ -112,7 +112,7 @@ class RuntimeToolsTest {
assertThat(McpSecurityConfig.AccessLevel.READ_ONLY.permits(
endpoints.annotations().readOnlyHint(),
endpoints.annotations().destructiveHint())).isTrue();
Tool request =
RuntimeTools.class.getDeclaredMethod("camel_runtime_http_request",
String.class, String.class,
- String.class, String.class,
String.class).getAnnotation(Tool.class);
+ String.class, String.class, String.class,
Integer.class).getAnnotation(Tool.class);
assertThat(request.annotations().readOnlyHint()).isFalse();
assertThat(request.annotations().destructiveHint()).isFalse();
assertThat(request.annotations().openWorldHint()).isFalse();
diff --git
a/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsWiringTest.java
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsWiringTest.java
new file mode 100644
index 000000000000..d664bb379cec
--- /dev/null
+++
b/dsl/camel-jbang/camel-jbang-mcp/src/test/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeToolsWiringTest.java
@@ -0,0 +1,411 @@
+/*
+ * 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.dsl.jbang.core.commands.mcp;
+
+import java.lang.reflect.Method;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import io.quarkiverse.mcp.server.Tool;
+import io.quarkiverse.mcp.server.ToolArg;
+import io.quarkiverse.mcp.server.ToolCallException;
+import org.apache.camel.dsl.jbang.core.commands.ai.ToolContext;
+import org.apache.camel.dsl.jbang.core.common.CommandLineHelper;
+import org.apache.camel.util.json.JsonArray;
+import org.apache.camel.util.json.JsonObject;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * The requests the MCP runtime tools send to a process, against a recorded
runtime: what the tools ask the CLI
+ * connector, and how they read its answers.
+ */
+class RuntimeToolsWiringTest {
+
+ private static final long PID = 4711;
+
+ /**
+ * Records the actions and answers them from {@link #answers}; the
received messages come from {@link #inbox}, and
+ * the receive status follows the actions the way the CLI connector writes
it: enabled after the start, disabled
+ * after the stop, with the console's count of taken messages as total.
+ */
+ static class RecordingRuntime extends RuntimeService {
+ final List<JsonObject> actions = new CopyOnWriteArrayList<>();
+ final Map<String, JsonObject> answers = new HashMap<>();
+ JsonObject receiveState = new JsonObject();
+ final List<JsonObject> inbox = new CopyOnWriteArrayList<>();
+ RuntimeException stopFailure;
+ /** The console's count before the call: messages of earlier sessions,
which may still be in the inbox. */
+ long totalBefore;
+ /** The count the status reports once stopped; by default the highest
uid of the inbox. */
+ Long totalAfterStop;
+ /** The status keeps saying enabled after the stop. */
+ boolean stopIgnored;
+ /** What the process writes when it applies the stop. */
+ Runnable onStop;
+ /** Counts down when the start action is applied. */
+ final CountDownLatch startApplied = new CountDownLatch(1);
+ private boolean started;
+ private boolean receiving;
+
+ @Override
+ public ProcessInfo findSingleProcess(String nameOrPid) {
+ return new ProcessInfo(PID, "shop", "shop");
+ }
+
+ @Override
+ public JsonObject readStatusSection(long pid, String section) {
+ if (!receiveState.isEmpty()) {
+ return receiveState;
+ }
+ JsonObject state = new JsonObject();
+ state.put("enabled", receiving || started && stopIgnored);
+ long highest = inbox.stream().mapToLong(m ->
m.getLong("uid")).max().orElse(0);
+ state.put("total",
+ !started ? totalBefore : totalAfterStop != null ?
totalAfterStop : Math.max(totalBefore, highest));
+ return state;
+ }
+
+ @Override
+ public long receiveFileSize(long pid) {
+ return 0;
+ }
+
+ @Override
+ public List<JsonObject> readReceivedMessages(long pid, long offset) {
+ return new ArrayList<>(inbox);
+ }
+
+ @Override
+ public JsonObject executeAction(long pid, String action,
Consumer<JsonObject> configure) {
+ JsonObject request = new JsonObject();
+ request.put("action", action);
+ if (configure != null) {
+ configure.accept(request);
+ }
+ actions.add(request);
+ if (configure == null && stopFailure != null) {
+ throw stopFailure;
+ }
+ if (request.containsKey("endpoint")) {
+ started = true;
+ receiving = true;
+ startApplied.countDown();
+ JsonObject answer = answers.get("receive-start");
+ return answer != null ? answer : new JsonObject();
+ }
+ receiving = false;
+ if (onStop != null) {
+ onStop.run();
+ }
+ return new JsonObject();
+ }
+ }
+
+ private final RecordingRuntime runtime = new RecordingRuntime();
+ private final RuntimeTools tools = toolsOf(runtime);
+
+ private static RuntimeTools toolsOf(RecordingRuntime runtime) {
+ RuntimeTools tools = new RuntimeTools();
+ tools.runtimeService = runtime;
+ tools.receiveSettleMs = 300;
+ return tools;
+ }
+
+ private void endpointMatches(String uri) {
+ JsonObject started = new JsonObject();
+ started.put("enabled", true);
+ started.put("url", uri);
+ runtime.answers.put("receive-start", started);
+ }
+
+ private static JsonObject message(long uid) {
+ JsonObject message = new JsonObject();
+ message.put("uid", uid);
+ message.put("endpointUri", "seda://orders");
+ return message;
+ }
+
+ private static List<Long> uids(JsonObject answer) {
+ return answer.<List<JsonObject>>
getCollection("messages").stream().map(m -> m.getLong("uid")).toList();
+ }
+
+ @Test
+ void receiveReturnsTheMessageAndStopsTheConsumer() {
+ endpointMatches("seda://orders");
+ runtime.inbox.add(message(1));
+
+ JsonObject answer = tools.receive(PID, "seda:orders", 5000, 1);
+
+ assertThat(answer.getString("endpoint")).isEqualTo("seda://orders");
+ assertThat(answer.getLong("received")).isEqualTo(1);
+ assertThat(answer.<List<Object>>
getCollection("messages")).containsExactly(message(1));
+ assertThat(answer).doesNotContainKeys("notReturned", "note",
"warning");
+ assertThat(runtime.actions).hasSize(2);
+ assertThat(runtime.actions.get(0)).containsEntry("enabled",
"true").containsEntry("endpoint", "seda:orders");
+
assertThat(runtime.actions.get(1)).containsOnlyKeys("action").containsEntry("action",
"receive");
+ }
+
+ @Test
+ void receiveReturnsTheFirstMessagesUpToTheLimitAndCountsTheOthers() {
+ endpointMatches("seda://orders");
+ for (long uid = 1; uid <= 4; uid++) {
+ runtime.inbox.add(message(uid));
+ }
+
+ JsonObject one = tools.receive(PID, "seda:orders", 5000, 1);
+ assertThat(one.getLong("received")).isEqualTo(4);
+ assertThat(uids(one)).containsExactly(1L);
+ assertThat(one.getLong("notReturned")).isEqualTo(3);
+ assertThat(one.getString("note")).contains("3 more", "taken from the
endpoint", "raise limit");
+
+ // the same four messages taken by a fresh call with a higher limit
+ RecordingRuntime again = new RecordingRuntime();
+ again.answers.putAll(runtime.answers);
+ again.inbox.addAll(runtime.inbox);
+ JsonObject all = toolsOf(again).receive(PID, "seda:orders", 5000, 10);
+ assertThat(uids(all)).containsExactly(1L, 2L, 3L, 4L);
+ assertThat(all).doesNotContainKeys("notReturned", "note");
+ }
+
+ @Test
+ void receiveRefusesALimitBelowOne() {
+ assertThatThrownBy(() -> tools.camel_runtime_receive(null,
"seda:orders", null, 0))
+ .isInstanceOf(ToolCallException.class)
+ .hasMessageContaining("limit must be at least 1");
+ assertThat(runtime.actions).isEmpty();
+ }
+
+ @Test
+ void receiveLeavesTheMessagesOfAnEarlierSessionOut() {
+ endpointMatches("seda://orders");
+ // the file still holds two messages an earlier call took after the
dump it read
+ runtime.totalBefore = 2;
+ runtime.inbox.add(message(1));
+ runtime.inbox.add(message(2));
+ runtime.inbox.add(message(3));
+
+ JsonObject answer = tools.receive(PID, "seda:orders", 5000, 5);
+
+ assertThat(answer.getLong("received")).isEqualTo(1);
+ assertThat(uids(answer)).containsExactly(3L);
+ }
+
+ @Test
+ void receiveCountsAndReturnsTheMessagesWrittenAfterTheStop() {
+ endpointMatches("seda://orders");
+ runtime.inbox.add(message(1));
+ // taken between the dump the call read and the stop: the connector
writes it on its next dump
+ runtime.onStop = () -> runtime.inbox.add(message(2));
+
+ JsonObject both = tools.receive(PID, "seda:orders", 5000, 2);
+ assertThat(both.getLong("received")).isEqualTo(2);
+ assertThat(uids(both)).containsExactly(1L, 2L);
+ assertThat(both).doesNotContainKey("note");
+ }
+
+ @Test
+ void receiveSaysWhenTheProcessHasNotWrittenEveryMessageItTook() {
+ endpointMatches("seda://orders");
+ runtime.inbox.add(message(1));
+ runtime.totalAfterStop = 3L;
+
+ JsonObject answer = tools.receive(PID, "seda:orders", 5000, 1);
+
+ assertThat(answer.getLong("received")).isEqualTo(3);
+ assertThat(answer.getLong("notReturned")).isEqualTo(2);
+ assertThat(uids(answer)).containsExactly(1L);
+ assertThat(answer.getString("note")).contains("2 more", "2 of them
were not written").doesNotContain("raise limit");
+ }
+
+ @Test
+ void receiveSaysWhenTheProcessDoesNotConfirmTheStop() {
+ endpointMatches("seda://orders");
+ runtime.inbox.add(message(1));
+ runtime.stopIgnored = true;
+
+ JsonObject answer = tools.receive(PID, "seda:orders", 5000, 1);
+
+ assertThat(answer.getLong("received")).isEqualTo(1);
+ assertThat(answer.getString("note")).contains("did not confirm the
stop");
+ }
+
+ @Test
+ void receiveRunsOneCallAtATimePerProcess() throws Exception {
+ endpointMatches("seda://orders");
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<JsonObject> first = executor.submit(() ->
tools.receive(PID, "seda:orders", 3000, 1));
+ assertThat(runtime.startApplied.await(5,
TimeUnit.SECONDS)).isTrue();
+
+ assertThatThrownBy(() -> tools.receive(PID, "seda:orders", 300, 1))
+ .isInstanceOf(ToolCallException.class)
+ .hasMessageContaining("Another camel_runtime_receive
call");
+
+ runtime.inbox.add(message(1));
+ assertThat(first.get(5,
TimeUnit.SECONDS).getLong("received")).isEqualTo(1);
+ assertThat(runtime.actions).as("the second call did not touch the
console").hasSize(2);
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void receiveReturnsTheMessageWithAWarningWhenTheConsumerCannotBeStopped() {
+ endpointMatches("seda://orders");
+ runtime.inbox.add(message(1));
+ runtime.stopFailure = new ToolCallException("Timeout waiting for
response", null);
+
+ JsonObject answer = tools.receive(PID, "seda:orders", 5000, 1);
+
+ assertThat(answer.getLong("received")).isEqualTo(1);
+ assertThat(answer.<List<Object>> getCollection("messages")).hasSize(1);
+ assertThat(answer.getString("warning")).contains("Timeout waiting",
"camel cmd receive --action=stop");
+ }
+
+ @Test
+ void receiveNamesTheStopCommandWhenItFailsAndCannotStopTheConsumer() {
+ runtime.answers.put("receive-start", new JsonObject());
+ runtime.stopFailure = new ToolCallException("Timeout waiting for
response", null);
+
+ assertThatThrownBy(() -> tools.receive(PID, "kafka*", 300, 1))
+ .isInstanceOf(ToolCallException.class)
+ .hasMessageContaining("No endpoint matches kafka*")
+ .hasMessageContaining("camel cmd receive --action=stop");
+ }
+
+ @Test
+ void receiveStopsTheConsumerWhenNoMessageArrives() {
+ endpointMatches("seda://orders");
+
+ JsonObject answer = tools.receive(PID, "seda:orders", 300, 1);
+
+ assertThat(answer.getLong("received")).isZero();
+ assertThat(answer.getString("note")).contains("No message arrived");
+ assertThat(runtime.actions).last().satisfies(r ->
assertThat(r).doesNotContainKey("endpoint"));
+ }
+
+ @Test
+ void receiveReportsAnEndpointNoneMatchesAndStops() {
+ runtime.answers.put("receive-start", new JsonObject());
+
+ assertThatThrownBy(() -> tools.receive(PID, "kafka*", 300, 1))
+ .isInstanceOf(ToolCallException.class)
+ .hasMessageContaining("No endpoint matches kafka*");
+ assertThat(runtime.actions).hasSize(2);
+ }
+
+ @Test
+ void receiveLeavesAReceiveModeStartedByTheCliAlone() {
+ runtime.receiveState.put("enabled", true);
+ runtime.receiveState.put("endpoints", new
JsonArray(List.of(Map.of("uri", "seda://audit"))));
+
+ assertThatThrownBy(() -> tools.receive(PID, "seda:orders", 300, 1))
+ .isInstanceOf(ToolCallException.class)
+ .hasMessageContaining("already receiving");
+ assertThat(runtime.actions).isEmpty();
+ }
+
+ @Test
+ void readReceivedMessagesReadsTheLinesAfterTheOffset(@TempDir Path home)
throws Exception {
+ Path previous = CommandLineHelper.getHomeDir();
+ CommandLineHelper.useHomeDir(home.toString());
+ try {
+ Path camelDir = CommandLineHelper.getCamelDir();
+ Files.createDirectories(camelDir);
+ Path file = camelDir.resolve(PID + "-receive.json");
+ Files.writeString(file, "{\"messages\":[{\"uid\":1}]}\n");
+ RuntimeService service = new RuntimeService();
+ long offset = service.receiveFileSize(PID);
+
+ Files.writeString(file,
"{\"messages\":[{\"uid\":2},{\"uid\":3}]}\n{\"messages\":[{\"uid\":4}",
+ StandardOpenOption.APPEND);
+
+ assertThat(service.readReceivedMessages(PID, offset)).extracting(m
-> m.getLong("uid"))
+ .as("the half written last line is read on the next poll")
+ .containsExactly(2L, 3L);
+
+ // camel cmd receive --action=clear rewrites the file shorter than
the offset: read it from the start
+ Files.writeString(file, "{\"messages\":[{\"uid\":5}]}\n");
+ assertThat(service.readReceivedMessages(PID, offset +
100)).extracting(m -> m.getLong("uid"))
+ .containsExactly(5L);
+ } finally {
+ CommandLineHelper.useHomeDir(previous.toString());
+ }
+ }
+
+ @Test
+ void memoryLeakStartsOneRecordingByDefault() {
+ tools.camel_runtime_memory_leak(null, "start", null, null, null, null);
+ tools.camel_runtime_memory_leak(null, "compare", null, null, null,
null);
+
+ assertThat(runtime.actions).extracting(r ->
r.getString("command")).containsExactly("start", "compare");
+
assertThat(runtime.actions.get(0).getString("duration")).isEqualTo("60");
+ assertThat(runtime.actions.get(1)).doesNotContainKey("duration");
+ }
+
+ @Test
+ void memoryLeakRefusesABadDurationOrMode() {
+ assertThatThrownBy(() -> tools.camel_runtime_memory_leak(null,
"start", "1m", null, null, null))
+ .isInstanceOf(ToolCallException.class)
+ .hasMessageContaining("whole number of seconds");
+ assertThatThrownBy(() -> tools.camel_runtime_memory_leak(null,
"start", "30", "triple", null, null))
+ .isInstanceOf(ToolCallException.class)
+ .hasMessageContaining("single or dual");
+ assertThat(runtime.actions).isEmpty();
+ }
+
+ @Test
+ void aiTraceReportsAFailedQueryAsAnError() {
+ // no process selected: the registry refuses, which must not read as
"no AI components"
+ assertThatThrownBy(() -> new
AiTraceTools().executeRegistryTool("get_history", new ToolContext(), Map.of()))
+ .isInstanceOf(ToolCallException.class)
+ .hasMessageContaining("get_history");
+ }
+
+ @Test
+ void theHintsMatchWhatTheToolsDo() throws Exception {
+ assertThat(hints("camel_runtime_receive", String.class, String.class,
Integer.class, Integer.class))
+ .as("receive takes the message from the
routes").containsExactly(false, true);
+ assertThat(hints("camel_runtime_send", String.class, String.class,
String.class, String.class))
+ .containsExactly(false, true);
+ assertThat(hints("camel_runtime_trace_dump",
String.class)).containsExactly(true, false);
+
+ Method aiTrace =
AiTraceTools.class.getDeclaredMethod("camel_runtime_ai_trace", String.class,
String.class);
+
assertThat(aiTrace.getParameters()[0].getAnnotation(ToolArg.class).required()).isFalse();
+ }
+
+ private static List<Boolean> hints(String tool, Class<?>... parameters)
throws Exception {
+ Tool.Annotations a = RuntimeTools.class.getDeclaredMethod(tool,
parameters).getAnnotation(Tool.class).annotations();
+ return List.of(a.readOnlyHint(), a.destructiveHint());
+ }
+}
