[
https://issues.apache.org/jira/browse/CAMEL-23078?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Federico Mariani resolved CAMEL-23078.
--------------------------------------
Resolution: Fixed
> 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)
