davsclaus commented on code in PR #27958:
URL: https://github.com/apache/camel/pull/27958#discussion_r4297379127
##########
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:
Same as in `RuntimeTools`: the command is `camel cmd receive`.
```suggestion
`camel cmd receive` is receiving from the same process.
```
##########
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:
`receive` is a subcommand of `camel cmd` (`CamelJBangMain.java:179`), so
`camel receive --action=stop` is not a command, and an agent runs this text as
it is. The same name is in `camel-jbang-mcp.adoc` (line 1016), in the upgrade
guide (line 1269) and in two comments of `RuntimeService` (lines 134 and 157).
```suggestion
+ " (camel cmd receive); stop that
first with: camel cmd receive --action=stop",
```
##########
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:
`camel.main.tracing-standby` puts the log `Tracer` on standby
(`camelContext.setTracingStandby`). The history comes from the `BacklogTracer`:
`MessageHistoryDevConsole` reads `getLatestMessageHistory()`, which
`BacklogTracer.shouldTrace` records when it is enabled or on standby and
message history is on. Its standby is `camel.trace.standby`, which is what the
dev profile sets (`ProfileConfigurer.configureMain`:
`tracerConfig().withStandby(true)`). With `tracing-standby` the history stays
empty, and the agent has followed the tool's advice.
```suggestion
camel.main.message-history=true and
camel.trace.standby=true.""")
```
##########
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:
The command is `camel cmd receive`, as in the error message above.
```suggestion
refuses to run while `camel cmd receive` is receiving from the same
process.
```
##########
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:
Same as the tool description: the history needs the backlog tracer on
standby (`camel.trace.standby`), not the log tracer.
```suggestion
`camel.trace.standby=true`.
```
##########
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);
+ return answer;
+ }
+ if (watch.taken() >= timeoutMs) {
+ answer.put("received", 0);
+ answer.put("note", "No message arrived within " +
(timeoutMs / 1000) + " seconds");
+ return answer;
+ }
+ try {
+ TimeUnit.MILLISECONDS.sleep(RECEIVE_POLL_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ToolCallException("Interrupted while waiting for
a message", null);
+ }
+ }
+ } finally {
+ // an action without an endpoint stops the consumers of the
receive console
+ runtimeService.executeAction(pid, "receive", null);
Review Comment:
Nit: if this call fails (`executeAction` throws a `ToolCallException` on a
timeout), its exception replaces the answer with the received messages, and the
consumer keeps taking messages from the routes, which is the leak this PR
fixes. Catching it and returning the answer with a warning that names the stop
command (`camel cmd receive --action=stop`) keeps both.
##########
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:
Nit: since #27931, `ToolContext.executeAction` throws a
`ToolExecutionException` when the integration does not answer in time, and
`delegateToRegistry` (and `AiTraceTools.executeRegistryTool`) already turn it
into an error. No registry tool returns the timeout text any more (only
`RuntimeService` uses the `RuntimeHelper.executeAction` text API, and it checks
the text itself), so this check cannot trigger, and `aTimeoutIsAnError` tests
it with a made-up string. Could it go, together with `TIMEOUT_PREFIX`?
Otherwise a comment saying it is only a guard would help, as the PR description
presents it as what turns timeouts into errors.
##########
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:
Every message of the first poll that finds any is returned, each with its
headers and its body (the receive console keeps up to 100 messages, with bodies
of up to 32 KB). On a busy Kafka or JMS endpoint the file transport can write
many messages in one tick, so one call can answer with megabytes, while #27839
caps the catalog answers at 16 000 characters. The description says the tool
returns *a* message: could it return the first one (or a `limit`, default 1)
and say how many more were taken? The others are gone from the routes either
way, so the count tells the agent that.
##########
dsl/camel-jbang/camel-jbang-mcp/src/main/java/org/apache/camel/dsl/jbang/core/commands/mcp/RuntimeService.java:
##########
@@ -123,6 +129,58 @@ public String stopApplication(long pid) {
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 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.
+ */
+ public List<JsonObject> readReceivedMessages(long pid, long offset) {
+ List<JsonObject> messages = new ArrayList<>();
+ Path file = receiveFile(pid);
+ try {
+ if (!Files.exists(file)) {
+ return messages;
+ }
+ byte[] data = Files.readAllBytes(file);
Review Comment:
Nit: the receive file is only appended to (and kept across `camel cmd
receive` sessions), and this reads all of it every 200 ms for up to 60 seconds.
Reading from `offset` (a `SeekableByteChannel` positioned there, or a
`RandomAccessFile`) keeps the poll cheap when the file has grown.
--
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]
