[jira] [Resolved] (CAMEL-23078) camel-openai: MCP improvements — parallel tool execution and runtime tool refresh

["Federico Mariani (Jira)" <ji...@apache.org>]

     [ 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)