[
https://issues.apache.org/jira/browse/CAMEL-23078?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18100670#comment-18100670
]
Federico Mariani commented on CAMEL-23078:
------------------------------------------
Still applicable on 4.22.0-SNAPSHOT: tool calls are executed sequentially in
{{OpenAIProducer.processNonStreamingAgentic}} and
{{OpenAIToolExecutionProducer.process}}, and {{tools/list_changed}}
notifications are ignored (tools are only re-listed on transport reconnect).
Verified that {{McpClient.SyncSpec.toolsChangeConsumer}} exists in MCP Java SDK
2.0.0 and that the SDK already re-lists tools itself before invoking the
consumer, on a {{boundedElastic}} thread.
Note that the original description is partly stale: CAMEL-23957 replaced the
individual {{cachedMcpTools}} / {{toolClientMap}} / {{returnDirectTools}}
fields with the immutable {{McpToolState}} record guarded by a global lock, so
the refresh publishes a new snapshot rather than mutating maps.
Planned implementation (one PR, two commits):
h4. 1. Parallel tool execution
* Extract the duplicated per-batch tool execution logic out of both producers
into a new package-private {{McpToolCallExecutor}}, so the agentic loop and
{{openai:tool-execution}} share one code path.
* New options: {{parallelToolExecution}} (boolean, default {{false}}) and
{{parallelToolTimeout}} (millis, default {{0}} = disabled; {{mcpTimeout}}
already bounds each MCP request).
* Thread pool obtained from Camel's
{{ExecutorServiceManager.newDefaultThreadPool}}, created lazily on producer
start only when the feature is enabled and shut down on stop. This is
deliberately not a hand-rolled {{Executors.newFixedThreadPool}}: going through
{{ExecutorServiceManager}} means the pool automatically becomes a
thread-per-task virtual executor when
{{camel.main.virtualThreadsEnabled=true}}, and is JMX-managed and tied to
context shutdown.
* Results are collected positionally so the {{tool}} messages stay in tool-call
order; {{returnDirect}} becomes a per-result flag AND-ed after the batch drains
rather than a loop variable.
* Batches of one run inline, with no thread hand-off.
* Behaviour change to note in the upgrade guide: with
{{parallelToolExecution=true}} and {{toolExecutionErrorStrategy=failExchange}},
the sibling tool calls already dispatched complete before the exchange fails,
instead of being abandoned mid-loop.
* MDC is propagated onto the worker threads so tool-call logs stay correlated.
h4. 2. Runtime tool refresh
* Extract the per-server state-rebuild block currently inlined in
{{doReconnectMcpServer}} into {{republishServerTools(serverName, client,
tools)}}, and drive both reconnect and refresh through it, so duplicate-name
skipping (CAMEL-23958), the per-server {{toolNames}} include list (CAMEL-23964)
and {{returnDirect}} detection cannot drift between the two paths.
* Register a {{toolsChangeConsumer}} per MCP client when the new
{{mcpToolRefresh}} option is enabled (default {{true}}, since ignoring
{{tools/list_changed}} is not spec-conformant and the MCP server is trusted
configuration rather than untrusted input). Default change goes in the upgrade
guide.
* The tool list handed to the consumer is unfiltered, so the per-server
{{toolNames}} filter is re-applied before republishing.
* Manual {{addReturnDirectTool}} / {{removeReturnDirectTool}} overrides are
tracked separately so a refresh (or a reconnect) no longer discards them.
Tests use the existing {{camel-test-infra-openai-mock}} ({{andInvokeTool}}
emits several tool calls in a single response) plus mocked {{McpSyncClient}}s,
covering ordering, per-strategy error handling, timeout, {{returnDirect}},
refresh add/remove/filter/duplicate, and the refresh-versus-reconnect race.
_Claude Code on behalf of Croway_
> camel-openai: MCP improvements — parallel tool execution and runtime tool
> refresh
> ---------------------------------------------------------------------------------
>
> Key: CAMEL-23078
> URL: https://issues.apache.org/jira/browse/CAMEL-23078
> Project: Camel
> Issue Type: Improvement
> Components: camel-openai
> Affects Versions: 4.18.0
> Reporter: Federico Mariani
> Assignee: Federico Mariani
> Priority: Minor
>
> Implementing https://issues.apache.org/jira/browse/CAMEL-23076 I noticed the
> following improvements:
> * Parallel tool execution - mcpToolExecutorThreads config, CompletableFuture
> for batch calls
> When the model returns multiple tool calls in a single response (e.g., "get
> weather for London" and "get weather for Paris" simultaneously), they are
> executed sequentially in {{OpenAIProducer.processNonStreamingAgentic()}}.
> Since tool calls in a batch are independent by design (the model produces
> them all from the same context, before seeing any result), they can safely
> run in parallel.
> *Impact:* If the model requests 5 tool calls that each take 2 seconds,
> sequential execution takes 10 seconds while parallel execution takes ~2
> seconds.
> *Proposed changes:*
> - Add a {{parallelToolExecution}} boolean parameter (default: {{false}}).
> - When enabled, use Camel's {{ExecutorServiceManager}} to obtain a managed
> thread pool (not a raw {{Executors.newFixedThreadPool}}) so it participates
> in Camel's lifecycle and graceful shutdown.
> - Collect results and feed them back to the model in the *same order* as the
> original tool calls — the OpenAI API requires tool result messages to match
> the tool call IDs.
> - Add a {{parallelToolTimeout}} parameter (default: equals {{mcpTimeout}}) to
> prevent a single slow tool from blocking the entire batch. Tools that exceed
> the timeout should return an error message to the model rather than failing
> the exchange.
> *Edge cases to handle:*
> - *returnDirect*: the current loop short-circuits when all tools in a batch
> have {{returnDirect=true}}. This logic must work identically in parallel
> mode, wait for all tools to complete, then check {{allReturnDirect}}.
> - *Error handling*: currently tool errors are wrapped as {{"Error: ..."}}
> strings and fed back to the model. In parallel mode, one tool's failure must
> not cancel other in-flight tools.
> - *Interaction with toolCallFilter*: if a tool call filter/interceptor is
> added in the future, filtering should happen *before* dispatching to the
> thread pool.
> * Tool list refresh - mcpToolRefreshInterval or SDK toolsChangeConsumer
> callback
> Tools are listed once during {{OpenAIEndpoint.initializeMcpServers()}}
> (called from {{doStart()}}) and cached in {{cachedMcpTools}} for the lifetime
> of the route. If an MCP server adds, removes, or updates tools at runtime,
> the cached list becomes stale.
> The MCP Java SDK provides a {{toolsChangeConsumer}} callback on the
> {{McpClient}} builder. When an MCP server's tool set changes, the server
> sends a {{notifications/tools/list_changed}} JSON-RPC notification through
> the existing transport connection. The SDK then automatically calls
> {{listTools()}} to fetch the updated list and invokes the registered consumer
> with the complete {{List<McpSchema.Tool>}}. This is event-driven, requires no
> polling, and works with all transport types (stdio, SSE, streamableHttp)
> since the notification handling sits at the {{McpClientSession}} layer above
> the transport.
> *Proposed changes:*
> - Register a {{toolsChangeConsumer}} callback during MCP client
> initialization in {{initializeMcpServers()}}. When fired, update
> {{cachedMcpTools}}, {{toolClientMap}}, and {{returnDirectTools}} for the
> affected server.
> {code:java}
> McpSyncClient mcpClient = McpClient.sync(transport)
> .requestTimeout(timeout)
> .initializationTimeout(timeout)
> .toolsChangeConsumer(updatedTools -> {
> // updatedTools is the full current List<McpSchema.Tool> for this
> server
> // rebuild cachedMcpTools, toolClientMap, returnDirectTools
> })
> .build();
> {code}
> - Add a {{mcpToolRefresh}} boolean parameter (default: {{true}}) to
> enable/disable dynamic refresh. Some deployments may prefer the deterministic
> behavior of a fixed tool set.
> - Ensure thread safety: the agentic loop reads {{cachedMcpTools}} and
> {{toolClientMap}} during execution. Updates from the callback must not
> corrupt in-flight iterations. Use {{CopyOnWriteArrayList}} /
> {{ConcurrentHashMap}} or swap references atomically.
> - The reconnection logic already in {{OpenAIEndpoint.reconnectMcpServer()}}
> re-lists tools after reconnecting. The refresh callback should reuse this
> same update logic to avoid duplication.
> *Related:* if the MCP server management is extracted to a dedicated class
> (see McpServerManager extraction), the refresh callback and reconnection
> logic would live together in that class.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
