Croway commented on code in PR #27958:
URL: https://github.com/apache/camel/pull/27958#discussion_r4300209719
##########
dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java:
##########
@@ -662,50 +727,120 @@ private JsonObject doDualJfrRecording(long pid, String
duration, String stacktra
});
}
+ /**
+ * 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.main.tracing-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. The consumer competes with the routes
that consume the endpoint, so the \
+ message it takes does not reach them (a queue or topic
message is gone). \
+ 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) {
if (endpoint == null || endpoint.isBlank()) {
throw new ToolCallException("endpoint is required", null);
}
- // receive is MCP-only (no ToolRegistry equivalent) — call
RuntimeService directly
+ 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);
+ }
+
+ /**
+ * 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.
+ */
+ JsonObject receive(long pid, String endpoint, long timeoutMs) {
+ 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 receive); stop that first
with: camel receive --action=stop",
Review Comment:
Done in 6676883e9f13: `camel cmd receive` in the error, in both
`RuntimeService` comments, on the MCP page and in the upgrade guide; `camel
receive` no longer appears in the diff.
_Claude Code on behalf of Croway_
##########
docs/user-manual/modules/ROOT/pages/camel-jbang-mcp.adoc:
##########
@@ -979,23 +987,33 @@
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.
+ The consumer competes with the routes that consume the endpoint, so the
message it takes does not reach them:
+ the tool is marked destructive. Use `camel_runtime_browse` to look at queued
messages without taking them. It
+ refuses to run while `camel receive` is receiving from the same process.
Review Comment:
Done in 6676883e9f13.
_Claude Code on behalf of Croway_
##########
docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc:
##########
@@ -1256,3 +1256,23 @@ The consumer now sets the `CamelMiloNodeId` header on
every exchange, including
Stopping a consumer now deletes its monitored items. Previously they stayed on
the server while another endpoint kept
the shared connection open, so a restarted consumer received each value twice.
While the server is unreachable, the
stop waits until the client's next attempt to reconnect has failed.
+
+=== 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. 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 refuses to run while
+ `camel receive` is receiving from the same process.
Review Comment:
Done in 6676883e9f13.
_Claude Code on behalf of Croway_
##########
dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java:
##########
@@ -662,50 +727,120 @@ private JsonObject doDualJfrRecording(long pid, String
duration, String stacktra
});
}
+ /**
+ * 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.main.tracing-standby=true.""")
Review Comment:
Done in 6676883e9f13: the description and the MCP page name
`camel.trace.standby`. Checked on the branch: the dev profile sets
`tracerConfig().withStandby(true)` (`TracerConfigurationProperties.standby`,
key `camel.trace.standby`), `BaseMainSupport` creates the `BacklogTracer` when
it is enabled or on standby, and `MessageHistoryDevConsole` reads
`BacklogTracer.getLatestMessageHistory()`; `camel.main.tracingStandby` goes to
`CamelContext.setTracingStandby`, the log `Tracer`.
_Claude Code on behalf of Croway_
##########
docs/user-manual/modules/ROOT/pages/camel-jbang-mcp.adoc:
##########
@@ -846,16 +849,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.main.tracing-standby=true`.
Review Comment:
Done in 6676883e9f13.
_Claude Code on behalf of Croway_
##########
dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java:
##########
@@ -662,50 +727,120 @@ private JsonObject doDualJfrRecording(long pid, String
duration, String stacktra
});
}
+ /**
+ * 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.main.tracing-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. The consumer competes with the routes
that consume the endpoint, so the \
+ message it takes does not reach them (a queue or topic
message is gone). \
+ 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) {
if (endpoint == null || endpoint.isBlank()) {
throw new ToolCallException("endpoint is required", null);
}
- // receive is MCP-only (no ToolRegistry equivalent) — call
RuntimeService directly
+ 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);
+ }
+
+ /**
+ * 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.
+ */
+ JsonObject receive(long pid, String endpoint, long timeoutMs) {
+ 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 receive); stop that first
with: camel receive --action=stop",
+ null);
+ }
+ long offset = runtimeService.receiveFileSize(pid);
+ JsonObject started = runtimeService.executeAction(pid, "receive", root
-> {
root.put("enabled", "true");
root.put("endpoint", endpoint);
});
+ try {
+ if (started.getString("error") != null) {
+ throw new ToolCallException("Cannot receive from " + endpoint
+ ": " + started.getString("error"), null);
+ }
+ String uri = started.getString("url");
+ if (uri == null) {
+ throw new ToolCallException(
+ "No endpoint matches " + endpoint
+ + " (a pattern matches the
endpoints the application sends to; "
+ + "camel_runtime_endpoints lists
them)",
+ null);
+ }
+ JsonObject answer = new JsonObject();
+ answer.put("endpoint", uri);
+ StopWatch watch = new StopWatch();
+ while (true) {
+ List<JsonObject> messages =
runtimeService.readReceivedMessages(pid, offset);
+ if (!messages.isEmpty()) {
+ answer.put("received", messages.size());
+ answer.put("messages", messages);
Review Comment:
Done in 6676883e9f13: `limit` (default 1, at least 1). The answer keeps the
first `limit` messages of the poll that found them; `received` is the total
taken in that poll, and `notReturned` with a note says how many are left out
and that they were taken from the endpoint all the same. The description says
"receives one message by default". Two tests in `RuntimeToolsWiringTest`
(`receiveReturnsTheFirstMessagesUpToTheLimitAndCountsTheOthers`,
`receiveRefusesALimitBelowOne`).
_Claude Code on behalf of Croway_
##########
dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeTools.java:
##########
@@ -69,12 +83,23 @@ private JsonObject delegateToRegistry(String
registryToolName, String nameOrPid,
ctx.selectProcess(p.pid());
try {
Object result = ToolRegistry.execute(registryToolName, ctx, args);
+ failOnTimeout(result);
return toJsonObject(result);
} catch (ToolExecutionException e) {
throw new ToolCallException(e.getMessage(), e);
}
}
+ /**
+ * The registry answers an action the process did not answer in time (a
dead or hung process) with a text, not an
+ * exception: report it as the failure it is rather than as a result.
+ */
+ static void failOnTimeout(Object result) {
Review Comment:
You are right. On the rebased branch `ToolContext.executeAction` throws
`ToolExecutionException` when the integration does not answer, and
`delegateToRegistry` / `executeRegistryTool` turn it into a
`ToolCallException`; the one text path left in the registry,
`execute_sql_query`, wraps the timeout text as `{"result": ...}`, so the prefix
check never caught it either. Removed in 6676883e9f13: `failOnTimeout`,
`RuntimeHelper.TIMEOUT_PREFIX`, the `aTimeoutIsAnError` test and the
upgrade-guide bullet that presented it as a change; `RuntimeHelper` is back to
the main version. The PR description no longer claims it. `execute_sql_query`
answering a timeout as a result is pre-existing and left out of this PR.
_Claude Code on behalf of Croway_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
