This is an automated email from the ASF dual-hosted git repository. Croway pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit 029a51c451d82598331c4bf6050656a2857424b4 Author: croway <[email protected]> AuthorDate: Fri Jul 31 09:48:37 2026 +0200 CAMEL-23078: camel-openai - Optional parallel MCP tool execution A model can request several tool calls in a single response, and those calls are independent by design, but they were executed one after another so a batch took as long as the sum of its tools. Add a `parallelToolExecution` option (default false, so existing routes are unaffected) that dispatches a batch concurrently, and a `parallelToolTimeout` option that bounds the batch as a whole so one slow tool cannot block it. Results are collected positionally and always fed back to the model in the order it requested the tools, as the OpenAI API pairs each tool message with its tool_call_id. A batch of a single tool call still runs inline. The thread pool is obtained from Camel's ExecutorServiceManager rather than being hand-rolled, so it follows the configured thread pool profile, is exposed over JMX, is shut down with the CamelContext, and automatically becomes a thread-per-task virtual thread executor when virtual threads are enabled. The per-batch execution logic was duplicated between the agentic loop in OpenAIProducer and the manual loop in OpenAIToolExecutionProducer, so it is extracted into a shared McpToolCallExecutor. That keeps hallucinated tool name handling, argument parsing, the error strategies and returnDirect detection from drifting between the two paths, and gives both the parallel option. The executor also snapshots the immutable McpToolState once per batch, so all calls in a batch see a consistent view of the tools. With toolExecutionErrorStrategy=failExchange the sibling calls already dispatched now complete before the exchange fails, instead of being abandoned mid-loop; this is documented in the upgrade guide. Concurrency is covered deterministically by unit tests that rendezvous the tool calls on a latch which can only be released if they run at the same time. OpenAIMcpParallelToolExecutionIT additionally covers the option end to end against the MCP Everything Server; reaching the parallel path there depends on the model emitting multi-call batches, which test_execution.md documents. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> --- .../apache/camel/catalog/components/openai.json | 60 ++-- .../component/openai/OpenAIEndpointConfigurer.java | 12 + .../component/openai/OpenAIEndpointUriFactory.java | 4 +- .../org/apache/camel/component/openai/openai.json | 60 ++-- .../camel-openai/src/main/docs/openai-mcp.adoc | 69 ++++ .../component/openai/McpToolCallExecutor.java | 272 +++++++++++++++ .../component/openai/OpenAIConfiguration.java | 32 ++ .../camel/component/openai/OpenAIProducer.java | 112 ++----- .../openai/OpenAIToolExecutionProducer.java | 97 ++---- .../component/openai/McpToolCallExecutorTest.java | 364 +++++++++++++++++++++ .../openai/OpenAIParallelToolExecutionTest.java | 168 ++++++++++ .../OpenAIMcpParallelToolExecutionIT.java | 132 ++++++++ components/camel-ai/camel-openai/test_execution.md | 27 ++ .../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 12 + 14 files changed, 1198 insertions(+), 223 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/openai.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/openai.json index 75596ffafff6..c1c640d00adc 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/openai.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/openai.json @@ -122,34 +122,36 @@ "mcpTimeout": { "index": 33, "kind": "parameter", "displayName": "Mcp Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": 20, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Timeout in seconds for MCP tool call requests. Applies to all MCP operations inclu [...] "model": { "index": 34, "kind": "parameter", "displayName": "Model", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The model to use for chat completion" }, "outputClass": { "index": 35, "kind": "parameter", "displayName": "Output Class", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Fully qualified class name for structured output using response format" }, - "previousResponseId": { "index": 36, "kind": "parameter", "displayName": "Previous Response Id", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Previous response id for OpenAI server-side conversation state (Responses [...] - "requestTimeout": { "index": 37, "kind": "parameter", "displayName": "Request Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "HTTP request timeout in milliseconds for the OpenAI SDK client. When 0 or [...] - "speechInstructions": { "index": 38, "kind": "parameter", "displayName": "Speech Instructions", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Optional instructions to control the voice of the generated audio. Does no [...] - "speechModel": { "index": 39, "kind": "parameter", "displayName": "Speech Model", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The model to use for text-to-speech (e.g., gpt-4o-mini-tts, tts-1, tts-1-hd)" }, - "speechResponseFormat": { "index": 40, "kind": "parameter", "displayName": "Speech Response Format", "group": "producer", "label": "", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "mp3", "opus", "aac", "flac", "wav", "pcm" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "mp3", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "descrip [...] - "speechSpeed": { "index": 41, "kind": "parameter", "displayName": "Speech Speed", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The speed of the generated audio, from 0.25 to 4.0 where 1.0 is normal speed" }, - "speechVoice": { "index": 42, "kind": "parameter", "displayName": "Speech Voice", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "alloy", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The voice to use for text-to-speech (e.g., alloy, echo, fable, [...] - "storeFullResponse": { "index": 43, "kind": "parameter", "displayName": "Store Full Response", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Store the full SDK response in non-streaming mode: chat-comp [...] - "streaming": { "index": 44, "kind": "parameter", "displayName": "Streaming", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Enable streaming responses" }, - "stripThinking": { "index": 45, "kind": "parameter", "displayName": "Strip Thinking", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Strip ... blocks from model responses (used by reasoning models like [...] - "systemMessage": { "index": 46, "kind": "parameter", "displayName": "System Message", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "System message to prepend. When set and conversationMemory is enabled, the conversat [...] - "temperature": { "index": 47, "kind": "parameter", "displayName": "Temperature", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Temperature for response generation (0.0 to 2.0)" }, - "toolExecutionErrorStrategy": { "index": 48, "kind": "parameter", "displayName": "Tool Execution Error Strategy", "group": "producer", "label": "", "required": false, "type": "enum", "javaType": "org.apache.camel.component.openai.ToolExecutionErrorStrategy", "enum": [ "failExchange", "repromptModel" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "failExchange", "configurationClass": "org.apache.camel.component.openai.OpenAIConfigur [...] - "topP": { "index": 49, "kind": "parameter", "displayName": "Top P", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Top P for response generation (0.0 to 1.0)" }, - "userMessage": { "index": 50, "kind": "parameter", "displayName": "User Message", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Default user message text to use when no prompt is provided" }, - "lazyStartProducer": { "index": 51, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produ [...] - "oauthProfile": { "index": 52, "kind": "parameter", "displayName": "Oauth Profile", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "OAuth profile name for obtaining an access token via the OAuth 2.0 Client Cred [...] - "sslContextParameters": { "index": 53, "kind": "parameter", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "SSLContextParameters to use [...] - "sslEndpointAlgorithm": { "index": 54, "kind": "parameter", "displayName": "Ssl Endpoint Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "security": "insecure:ssl", "insecureValue": "none", "defaultValue": "https", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", [...] - "sslKeymanagerAlgorithm": { "index": 55, "kind": "parameter", "displayName": "Ssl Keymanager Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "SunX509", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The algorithm used by the key [...] - "sslKeyPassword": { "index": 56, "kind": "parameter", "displayName": "Ssl Key Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The password of the private key in the key store file" }, - "sslKeystoreLocation": { "index": 57, "kind": "parameter", "displayName": "Ssl Keystore Location", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The location of the key store file. This is optional and can be [...] - "sslKeystorePassword": { "index": 58, "kind": "parameter", "displayName": "Ssl Keystore Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The store password for the key store file" }, - "sslKeystoreType": { "index": 59, "kind": "parameter", "displayName": "Ssl Keystore Type", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "JKS", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The file format of the key store file" }, - "sslProtocol": { "index": 60, "kind": "parameter", "displayName": "Ssl Protocol", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "TLSv1.3", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The SSL protocol used to generate the SSLContext" }, - "sslTrustmanagerAlgorithm": { "index": 61, "kind": "parameter", "displayName": "Ssl Trustmanager Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "PKIX", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The algorithm used by the tru [...] - "sslTruststoreLocation": { "index": 62, "kind": "parameter", "displayName": "Ssl Truststore Location", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The location of the trust store file, used to validate the [...] - "sslTruststorePassword": { "index": 63, "kind": "parameter", "displayName": "Ssl Truststore Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The password for the trust store file. [...] - "sslTruststoreType": { "index": 64, "kind": "parameter", "displayName": "Ssl Truststore Type", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "JKS", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The file format of the trust store file" } + "parallelToolExecution": { "index": 36, "kind": "parameter", "displayName": "Parallel Tool Execution", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Execute the tool calls returned by the model in a si [...] + "parallelToolTimeout": { "index": 37, "kind": "parameter", "displayName": "Parallel Tool Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Timeout in milliseconds for a batch of parallel tool calls, so [...] + "previousResponseId": { "index": 38, "kind": "parameter", "displayName": "Previous Response Id", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Previous response id for OpenAI server-side conversation state (Responses [...] + "requestTimeout": { "index": 39, "kind": "parameter", "displayName": "Request Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "HTTP request timeout in milliseconds for the OpenAI SDK client. When 0 or [...] + "speechInstructions": { "index": 40, "kind": "parameter", "displayName": "Speech Instructions", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Optional instructions to control the voice of the generated audio. Does no [...] + "speechModel": { "index": 41, "kind": "parameter", "displayName": "Speech Model", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The model to use for text-to-speech (e.g., gpt-4o-mini-tts, tts-1, tts-1-hd)" }, + "speechResponseFormat": { "index": 42, "kind": "parameter", "displayName": "Speech Response Format", "group": "producer", "label": "", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "mp3", "opus", "aac", "flac", "wav", "pcm" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "mp3", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "descrip [...] + "speechSpeed": { "index": 43, "kind": "parameter", "displayName": "Speech Speed", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The speed of the generated audio, from 0.25 to 4.0 where 1.0 is normal speed" }, + "speechVoice": { "index": 44, "kind": "parameter", "displayName": "Speech Voice", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "alloy", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The voice to use for text-to-speech (e.g., alloy, echo, fable, [...] + "storeFullResponse": { "index": 45, "kind": "parameter", "displayName": "Store Full Response", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Store the full SDK response in non-streaming mode: chat-comp [...] + "streaming": { "index": 46, "kind": "parameter", "displayName": "Streaming", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Enable streaming responses" }, + "stripThinking": { "index": 47, "kind": "parameter", "displayName": "Strip Thinking", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Strip ... blocks from model responses (used by reasoning models like [...] + "systemMessage": { "index": 48, "kind": "parameter", "displayName": "System Message", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "System message to prepend. When set and conversationMemory is enabled, the conversat [...] + "temperature": { "index": 49, "kind": "parameter", "displayName": "Temperature", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Temperature for response generation (0.0 to 2.0)" }, + "toolExecutionErrorStrategy": { "index": 50, "kind": "parameter", "displayName": "Tool Execution Error Strategy", "group": "producer", "label": "", "required": false, "type": "enum", "javaType": "org.apache.camel.component.openai.ToolExecutionErrorStrategy", "enum": [ "failExchange", "repromptModel" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "failExchange", "configurationClass": "org.apache.camel.component.openai.OpenAIConfigur [...] + "topP": { "index": 51, "kind": "parameter", "displayName": "Top P", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Top P for response generation (0.0 to 1.0)" }, + "userMessage": { "index": 52, "kind": "parameter", "displayName": "User Message", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Default user message text to use when no prompt is provided" }, + "lazyStartProducer": { "index": 53, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produ [...] + "oauthProfile": { "index": 54, "kind": "parameter", "displayName": "Oauth Profile", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "OAuth profile name for obtaining an access token via the OAuth 2.0 Client Cred [...] + "sslContextParameters": { "index": 55, "kind": "parameter", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "SSLContextParameters to use [...] + "sslEndpointAlgorithm": { "index": 56, "kind": "parameter", "displayName": "Ssl Endpoint Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "security": "insecure:ssl", "insecureValue": "none", "defaultValue": "https", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", [...] + "sslKeymanagerAlgorithm": { "index": 57, "kind": "parameter", "displayName": "Ssl Keymanager Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "SunX509", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The algorithm used by the key [...] + "sslKeyPassword": { "index": 58, "kind": "parameter", "displayName": "Ssl Key Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The password of the private key in the key store file" }, + "sslKeystoreLocation": { "index": 59, "kind": "parameter", "displayName": "Ssl Keystore Location", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The location of the key store file. This is optional and can be [...] + "sslKeystorePassword": { "index": 60, "kind": "parameter", "displayName": "Ssl Keystore Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The store password for the key store file" }, + "sslKeystoreType": { "index": 61, "kind": "parameter", "displayName": "Ssl Keystore Type", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "JKS", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The file format of the key store file" }, + "sslProtocol": { "index": 62, "kind": "parameter", "displayName": "Ssl Protocol", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "TLSv1.3", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The SSL protocol used to generate the SSLContext" }, + "sslTrustmanagerAlgorithm": { "index": 63, "kind": "parameter", "displayName": "Ssl Trustmanager Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "PKIX", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The algorithm used by the tru [...] + "sslTruststoreLocation": { "index": 64, "kind": "parameter", "displayName": "Ssl Truststore Location", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The location of the trust store file, used to validate the [...] + "sslTruststorePassword": { "index": 65, "kind": "parameter", "displayName": "Ssl Truststore Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The password for the trust store file. [...] + "sslTruststoreType": { "index": 66, "kind": "parameter", "displayName": "Ssl Truststore Type", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "JKS", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The file format of the trust store file" } } } diff --git a/components/camel-ai/camel-openai/src/generated/java/org/apache/camel/component/openai/OpenAIEndpointConfigurer.java b/components/camel-ai/camel-openai/src/generated/java/org/apache/camel/component/openai/OpenAIEndpointConfigurer.java index 825570b9a22b..9519713469f1 100644 --- a/components/camel-ai/camel-openai/src/generated/java/org/apache/camel/component/openai/OpenAIEndpointConfigurer.java +++ b/components/camel-ai/camel-openai/src/generated/java/org/apache/camel/component/openai/OpenAIEndpointConfigurer.java @@ -95,6 +95,10 @@ public class OpenAIEndpointConfigurer extends PropertyConfigurerSupport implemen case "oauthProfile": target.getConfiguration().setOauthProfile(property(camelContext, java.lang.String.class, value)); return true; case "outputclass": case "outputClass": target.getConfiguration().setOutputClass(property(camelContext, java.lang.String.class, value)); return true; + case "paralleltoolexecution": + case "parallelToolExecution": target.getConfiguration().setParallelToolExecution(property(camelContext, boolean.class, value)); return true; + case "paralleltooltimeout": + case "parallelToolTimeout": target.getConfiguration().setParallelToolTimeout(property(camelContext, long.class, value)); return true; case "previousresponseid": case "previousResponseId": target.getConfiguration().setPreviousResponseId(property(camelContext, java.lang.String.class, value)); return true; case "requesttimeout": @@ -226,6 +230,10 @@ public class OpenAIEndpointConfigurer extends PropertyConfigurerSupport implemen case "oauthProfile": return java.lang.String.class; case "outputclass": case "outputClass": return java.lang.String.class; + case "paralleltoolexecution": + case "parallelToolExecution": return boolean.class; + case "paralleltooltimeout": + case "parallelToolTimeout": return long.class; case "previousresponseid": case "previousResponseId": return java.lang.String.class; case "requesttimeout": @@ -358,6 +366,10 @@ public class OpenAIEndpointConfigurer extends PropertyConfigurerSupport implemen case "oauthProfile": return target.getConfiguration().getOauthProfile(); case "outputclass": case "outputClass": return target.getConfiguration().getOutputClass(); + case "paralleltoolexecution": + case "parallelToolExecution": return target.getConfiguration().isParallelToolExecution(); + case "paralleltooltimeout": + case "parallelToolTimeout": return target.getConfiguration().getParallelToolTimeout(); case "previousresponseid": case "previousResponseId": return target.getConfiguration().getPreviousResponseId(); case "requesttimeout": diff --git a/components/camel-ai/camel-openai/src/generated/java/org/apache/camel/component/openai/OpenAIEndpointUriFactory.java b/components/camel-ai/camel-openai/src/generated/java/org/apache/camel/component/openai/OpenAIEndpointUriFactory.java index 60bf0d2c653d..5e751f585b10 100644 --- a/components/camel-ai/camel-openai/src/generated/java/org/apache/camel/component/openai/OpenAIEndpointUriFactory.java +++ b/components/camel-ai/camel-openai/src/generated/java/org/apache/camel/component/openai/OpenAIEndpointUriFactory.java @@ -24,7 +24,7 @@ public class OpenAIEndpointUriFactory extends org.apache.camel.support.component private static final Set<String> ENDPOINT_IDENTITY_PROPERTY_NAMES; private static final Map<String, String> MULTI_VALUE_PREFIXES; static { - Set<String> props = new HashSet<>(65); + Set<String> props = new HashSet<>(67); props.add("additionalBodyProperty"); props.add("additionalHeader"); props.add("additionalResponseHeader"); @@ -63,6 +63,8 @@ public class OpenAIEndpointUriFactory extends org.apache.camel.support.component props.add("oauthProfile"); props.add("operation"); props.add("outputClass"); + props.add("parallelToolExecution"); + props.add("parallelToolTimeout"); props.add("previousResponseId"); props.add("requestTimeout"); props.add("speechInstructions"); diff --git a/components/camel-ai/camel-openai/src/generated/resources/META-INF/org/apache/camel/component/openai/openai.json b/components/camel-ai/camel-openai/src/generated/resources/META-INF/org/apache/camel/component/openai/openai.json index 75596ffafff6..c1c640d00adc 100644 --- a/components/camel-ai/camel-openai/src/generated/resources/META-INF/org/apache/camel/component/openai/openai.json +++ b/components/camel-ai/camel-openai/src/generated/resources/META-INF/org/apache/camel/component/openai/openai.json @@ -122,34 +122,36 @@ "mcpTimeout": { "index": 33, "kind": "parameter", "displayName": "Mcp Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": 20, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Timeout in seconds for MCP tool call requests. Applies to all MCP operations inclu [...] "model": { "index": 34, "kind": "parameter", "displayName": "Model", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The model to use for chat completion" }, "outputClass": { "index": 35, "kind": "parameter", "displayName": "Output Class", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Fully qualified class name for structured output using response format" }, - "previousResponseId": { "index": 36, "kind": "parameter", "displayName": "Previous Response Id", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Previous response id for OpenAI server-side conversation state (Responses [...] - "requestTimeout": { "index": 37, "kind": "parameter", "displayName": "Request Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "HTTP request timeout in milliseconds for the OpenAI SDK client. When 0 or [...] - "speechInstructions": { "index": 38, "kind": "parameter", "displayName": "Speech Instructions", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Optional instructions to control the voice of the generated audio. Does no [...] - "speechModel": { "index": 39, "kind": "parameter", "displayName": "Speech Model", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The model to use for text-to-speech (e.g., gpt-4o-mini-tts, tts-1, tts-1-hd)" }, - "speechResponseFormat": { "index": 40, "kind": "parameter", "displayName": "Speech Response Format", "group": "producer", "label": "", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "mp3", "opus", "aac", "flac", "wav", "pcm" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "mp3", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "descrip [...] - "speechSpeed": { "index": 41, "kind": "parameter", "displayName": "Speech Speed", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The speed of the generated audio, from 0.25 to 4.0 where 1.0 is normal speed" }, - "speechVoice": { "index": 42, "kind": "parameter", "displayName": "Speech Voice", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "alloy", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The voice to use for text-to-speech (e.g., alloy, echo, fable, [...] - "storeFullResponse": { "index": 43, "kind": "parameter", "displayName": "Store Full Response", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Store the full SDK response in non-streaming mode: chat-comp [...] - "streaming": { "index": 44, "kind": "parameter", "displayName": "Streaming", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Enable streaming responses" }, - "stripThinking": { "index": 45, "kind": "parameter", "displayName": "Strip Thinking", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Strip ... blocks from model responses (used by reasoning models like [...] - "systemMessage": { "index": 46, "kind": "parameter", "displayName": "System Message", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "System message to prepend. When set and conversationMemory is enabled, the conversat [...] - "temperature": { "index": 47, "kind": "parameter", "displayName": "Temperature", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Temperature for response generation (0.0 to 2.0)" }, - "toolExecutionErrorStrategy": { "index": 48, "kind": "parameter", "displayName": "Tool Execution Error Strategy", "group": "producer", "label": "", "required": false, "type": "enum", "javaType": "org.apache.camel.component.openai.ToolExecutionErrorStrategy", "enum": [ "failExchange", "repromptModel" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "failExchange", "configurationClass": "org.apache.camel.component.openai.OpenAIConfigur [...] - "topP": { "index": 49, "kind": "parameter", "displayName": "Top P", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Top P for response generation (0.0 to 1.0)" }, - "userMessage": { "index": 50, "kind": "parameter", "displayName": "User Message", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Default user message text to use when no prompt is provided" }, - "lazyStartProducer": { "index": 51, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produ [...] - "oauthProfile": { "index": 52, "kind": "parameter", "displayName": "Oauth Profile", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "OAuth profile name for obtaining an access token via the OAuth 2.0 Client Cred [...] - "sslContextParameters": { "index": 53, "kind": "parameter", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "SSLContextParameters to use [...] - "sslEndpointAlgorithm": { "index": 54, "kind": "parameter", "displayName": "Ssl Endpoint Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "security": "insecure:ssl", "insecureValue": "none", "defaultValue": "https", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", [...] - "sslKeymanagerAlgorithm": { "index": 55, "kind": "parameter", "displayName": "Ssl Keymanager Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "SunX509", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The algorithm used by the key [...] - "sslKeyPassword": { "index": 56, "kind": "parameter", "displayName": "Ssl Key Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The password of the private key in the key store file" }, - "sslKeystoreLocation": { "index": 57, "kind": "parameter", "displayName": "Ssl Keystore Location", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The location of the key store file. This is optional and can be [...] - "sslKeystorePassword": { "index": 58, "kind": "parameter", "displayName": "Ssl Keystore Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The store password for the key store file" }, - "sslKeystoreType": { "index": 59, "kind": "parameter", "displayName": "Ssl Keystore Type", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "JKS", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The file format of the key store file" }, - "sslProtocol": { "index": 60, "kind": "parameter", "displayName": "Ssl Protocol", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "TLSv1.3", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The SSL protocol used to generate the SSLContext" }, - "sslTrustmanagerAlgorithm": { "index": 61, "kind": "parameter", "displayName": "Ssl Trustmanager Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "PKIX", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The algorithm used by the tru [...] - "sslTruststoreLocation": { "index": 62, "kind": "parameter", "displayName": "Ssl Truststore Location", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The location of the trust store file, used to validate the [...] - "sslTruststorePassword": { "index": 63, "kind": "parameter", "displayName": "Ssl Truststore Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The password for the trust store file. [...] - "sslTruststoreType": { "index": 64, "kind": "parameter", "displayName": "Ssl Truststore Type", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "JKS", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The file format of the trust store file" } + "parallelToolExecution": { "index": 36, "kind": "parameter", "displayName": "Parallel Tool Execution", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Execute the tool calls returned by the model in a si [...] + "parallelToolTimeout": { "index": 37, "kind": "parameter", "displayName": "Parallel Tool Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Timeout in milliseconds for a batch of parallel tool calls, so [...] + "previousResponseId": { "index": 38, "kind": "parameter", "displayName": "Previous Response Id", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Previous response id for OpenAI server-side conversation state (Responses [...] + "requestTimeout": { "index": 39, "kind": "parameter", "displayName": "Request Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "HTTP request timeout in milliseconds for the OpenAI SDK client. When 0 or [...] + "speechInstructions": { "index": 40, "kind": "parameter", "displayName": "Speech Instructions", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Optional instructions to control the voice of the generated audio. Does no [...] + "speechModel": { "index": 41, "kind": "parameter", "displayName": "Speech Model", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The model to use for text-to-speech (e.g., gpt-4o-mini-tts, tts-1, tts-1-hd)" }, + "speechResponseFormat": { "index": 42, "kind": "parameter", "displayName": "Speech Response Format", "group": "producer", "label": "", "required": false, "type": "enum", "javaType": "java.lang.String", "enum": [ "mp3", "opus", "aac", "flac", "wav", "pcm" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "mp3", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "descrip [...] + "speechSpeed": { "index": 43, "kind": "parameter", "displayName": "Speech Speed", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The speed of the generated audio, from 0.25 to 4.0 where 1.0 is normal speed" }, + "speechVoice": { "index": 44, "kind": "parameter", "displayName": "Speech Voice", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "alloy", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The voice to use for text-to-speech (e.g., alloy, echo, fable, [...] + "storeFullResponse": { "index": 45, "kind": "parameter", "displayName": "Store Full Response", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Store the full SDK response in non-streaming mode: chat-comp [...] + "streaming": { "index": 46, "kind": "parameter", "displayName": "Streaming", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Enable streaming responses" }, + "stripThinking": { "index": 47, "kind": "parameter", "displayName": "Strip Thinking", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Strip ... blocks from model responses (used by reasoning models like [...] + "systemMessage": { "index": 48, "kind": "parameter", "displayName": "System Message", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "System message to prepend. When set and conversationMemory is enabled, the conversat [...] + "temperature": { "index": 49, "kind": "parameter", "displayName": "Temperature", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Temperature for response generation (0.0 to 2.0)" }, + "toolExecutionErrorStrategy": { "index": 50, "kind": "parameter", "displayName": "Tool Execution Error Strategy", "group": "producer", "label": "", "required": false, "type": "enum", "javaType": "org.apache.camel.component.openai.ToolExecutionErrorStrategy", "enum": [ "failExchange", "repromptModel" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "failExchange", "configurationClass": "org.apache.camel.component.openai.OpenAIConfigur [...] + "topP": { "index": 51, "kind": "parameter", "displayName": "Top P", "group": "producer", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Top P for response generation (0.0 to 1.0)" }, + "userMessage": { "index": 52, "kind": "parameter", "displayName": "User Message", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "Default user message text to use when no prompt is provided" }, + "lazyStartProducer": { "index": 53, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produ [...] + "oauthProfile": { "index": 54, "kind": "parameter", "displayName": "Oauth Profile", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "OAuth profile name for obtaining an access token via the OAuth 2.0 Client Cred [...] + "sslContextParameters": { "index": 55, "kind": "parameter", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "SSLContextParameters to use [...] + "sslEndpointAlgorithm": { "index": 56, "kind": "parameter", "displayName": "Ssl Endpoint Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "security": "insecure:ssl", "insecureValue": "none", "defaultValue": "https", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", [...] + "sslKeymanagerAlgorithm": { "index": 57, "kind": "parameter", "displayName": "Ssl Keymanager Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "SunX509", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The algorithm used by the key [...] + "sslKeyPassword": { "index": 58, "kind": "parameter", "displayName": "Ssl Key Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The password of the private key in the key store file" }, + "sslKeystoreLocation": { "index": 59, "kind": "parameter", "displayName": "Ssl Keystore Location", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The location of the key store file. This is optional and can be [...] + "sslKeystorePassword": { "index": 60, "kind": "parameter", "displayName": "Ssl Keystore Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The store password for the key store file" }, + "sslKeystoreType": { "index": 61, "kind": "parameter", "displayName": "Ssl Keystore Type", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "JKS", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The file format of the key store file" }, + "sslProtocol": { "index": 62, "kind": "parameter", "displayName": "Ssl Protocol", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "TLSv1.3", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The SSL protocol used to generate the SSLContext" }, + "sslTrustmanagerAlgorithm": { "index": 63, "kind": "parameter", "displayName": "Ssl Trustmanager Algorithm", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "PKIX", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The algorithm used by the tru [...] + "sslTruststoreLocation": { "index": 64, "kind": "parameter", "displayName": "Ssl Truststore Location", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The location of the trust store file, used to validate the [...] + "sslTruststorePassword": { "index": 65, "kind": "parameter", "displayName": "Ssl Truststore Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": true, "security": "secret", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The password for the trust store file. [...] + "sslTruststoreType": { "index": 66, "kind": "parameter", "displayName": "Ssl Truststore Type", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "JKS", "configurationClass": "org.apache.camel.component.openai.OpenAIConfiguration", "configurationField": "configuration", "description": "The file format of the trust store file" } } } diff --git a/components/camel-ai/camel-openai/src/main/docs/openai-mcp.adoc b/components/camel-ai/camel-openai/src/main/docs/openai-mcp.adoc index 4b8160b3bf2e..8f5fb2e6accb 100644 --- a/components/camel-ai/camel-openai/src/main/docs/openai-mcp.adoc +++ b/components/camel-ai/camel-openai/src/main/docs/openai-mcp.adoc @@ -300,6 +300,75 @@ YAML:: ---- ==== +=== Parallel Tool Execution + +A model can request several tool calls in a single response. By default, the component executes them one after another, +so the batch takes as long as the sum of its tool calls. Because tool calls emitted in the same assistant message are +independent by design, `parallelToolExecution=true` dispatches them concurrently instead, reducing the batch to the +duration of its slowest tool: + +[tabs] +==== +Java:: ++ +[source,java] +---- +from("direct:chat") + .to("openai:chat-completion?model=gpt-4" + + "¶llelToolExecution=true" + + "¶llelToolTimeout=30000" + + "&mcpServer.api.transportType=streamableHttp" + + "&mcpServer.api.url=http://localhost:9090/mcp";); +---- + +XML:: ++ +[source,xml] +---- +<route> + <from uri="direct:chat"/> + <to uri="openai:chat-completion?model=gpt-4&parallelToolExecution=true&parallelToolTimeout=30000&mcpServer.api.transportType=streamableHttp&mcpServer.api.url=http://localhost:9090/mcp"/> +</route> +---- + +YAML:: ++ +[source,yaml] +---- +- route: + from: + uri: direct:chat + steps: + - to: + uri: openai:chat-completion + parameters: + model: gpt-4 + parallelToolExecution: true + parallelToolTimeout: 30000 + mcpServer.api.transportType: streamableHttp + mcpServer.api.url: http://localhost:9090/mcp +---- +==== + +The option applies to both the agentic loop and the `openai:tool-execution` operation. + +Notes: + +* Results are always fed back to the model in the order the model requested the tools, as required by the OpenAI API, + regardless of the order in which the tool calls complete. +* A batch containing a single tool call is executed inline, without a thread hand-off. +* `parallelToolTimeout` (milliseconds, default `0` = disabled) bounds the batch as a whole, so that one slow tool cannot + block it. A tool call that exceeds it is cancelled and then handled according to `toolExecutionErrorStrategy`. Leaving + it disabled is reasonable, because `mcpTimeout` already bounds each individual MCP request. +* With `toolExecutionErrorStrategy=failExchange`, the sibling tool calls that were already dispatched are allowed to + complete before the exchange fails. Sequential execution instead abandons the remaining calls at the first failure. +* The thread pool comes from Camel's `ExecutorServiceManager` and follows the default thread pool profile, so it is + visible over JMX and shut down with the `CamelContext`. When virtual threads are enabled + (`camel.main.virtualThreadsEnabled=true`), the pool automatically becomes a thread-per-task virtual thread executor and + the profile's pool sizes no longer apply — concurrency is then bounded by the profile's `maxQueueSize` instead. +* `stdio` MCP servers multiplex all calls over a single child process, so the achievable parallelism there depends on + the server implementation. HTTP-based transports (`streamableHttp`, `sse`) benefit the most. + === Manual Tool Loop with `tool-execution` Operation When `autoToolExecution=false`, you can implement your own tool loop using the `openai:tool-execution` operation and Camel's `loopDoWhile` EIP. This gives you full control to add logging, filtering, retry logic, or custom routing between tool calls — without writing any Java code. diff --git a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/McpToolCallExecutor.java b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/McpToolCallExecutor.java new file mode 100644 index 000000000000..bcb5c1f8480f --- /dev/null +++ b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/McpToolCallExecutor.java @@ -0,0 +1,272 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.openai; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.stream.Collectors; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.openai.models.chat.completions.ChatCompletionMessageToolCall; +import io.modelcontextprotocol.client.McpSyncClient; +import io.modelcontextprotocol.spec.McpSchema; +import org.apache.camel.support.service.ServiceSupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.slf4j.MDC; + +/** + * Executes a batch of MCP tool calls returned by the model in a single response. + * + * <p> + * Shared by the agentic loop in {@link OpenAIProducer} and by the manual tool loop in + * {@link OpenAIToolExecutionProducer} so that hallucinated tool name handling, argument parsing, error strategies and + * {@code returnDirect} detection cannot drift between the two. + * + * <p> + * When {@code parallelToolExecution=true} the calls in a batch are dispatched concurrently, since tool calls emitted in + * the same assistant message are independent by design. Results are always returned in the original tool call order, + * because the OpenAI API pairs each {@code tool} message with its {@code tool_call_id}. + */ +class McpToolCallExecutor extends ServiceSupport { + + private static final Logger LOG = LoggerFactory.getLogger(McpToolCallExecutor.class); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + private final OpenAIEndpoint endpoint; + + private ExecutorService executorService; + + McpToolCallExecutor(OpenAIEndpoint endpoint) { + this.endpoint = endpoint; + } + + /** + * The outcome of a single tool call. + * + * @param toolCallId the id the model assigned to the call, used to pair the result back to the request + * @param toolName the name of the tool that was called + * @param content the textual result to feed back to the model + * @param returnDirect whether the call succeeded and the tool is annotated with {@code returnDirect} + */ + record ToolResult(String toolCallId, String toolName, String content, boolean returnDirect) { + } + + @Override + protected void doStart() throws Exception { + if (endpoint.getConfiguration().isParallelToolExecution()) { + // Deliberately goes through Camel's ExecutorServiceManager rather than creating a raw + // ThreadPoolExecutor: the pool then honours the configured thread pool profile, is exposed + // over JMX, is shut down with the CamelContext, and automatically becomes a thread-per-task + // virtual thread executor when virtual threads are enabled. + executorService = endpoint.getCamelContext().getExecutorServiceManager() + .newDefaultThreadPool(this, "OpenAIMcpToolCall"); + } + super.doStart(); + } + + @Override + protected void doStop() throws Exception { + if (executorService != null) { + endpoint.getCamelContext().getExecutorServiceManager().shutdownNow(executorService); + executorService = null; + } + super.doStop(); + } + + /** + * Executes every tool call in the batch and returns the results in the original order. + * + * @param toolCalls the tool calls requested by the model + * @return one result per tool call, in the same order + * @throws Exception when a tool call fails and the configured strategy is to fail the exchange + */ + List<ToolResult> execute(List<ChatCompletionMessageToolCall> toolCalls) throws Exception { + if (toolCalls.isEmpty()) { + return List.of(); + } + + // Snapshot the (immutable) tool state once so the whole batch sees a consistent view even if an + // MCP server publishes a tool list change or a sibling call triggers a reconnect while it runs + McpToolState toolState = endpoint.getMcpToolState(); + + // A single tool call gains nothing from a thread hand-off + if (executorService == null || toolCalls.size() == 1) { + List<ToolResult> results = new ArrayList<>(toolCalls.size()); + for (ChatCompletionMessageToolCall toolCall : toolCalls) { + results.add(executeOne(toolCall, toolState)); + } + return results; + } + + return executeParallel(toolCalls, toolState); + } + + private List<ToolResult> executeParallel(List<ChatCompletionMessageToolCall> toolCalls, McpToolState toolState) + throws Exception { + LOG.debug("Executing {} tool call(s) in parallel", toolCalls.size()); + + // Carry the caller's MDC onto the worker threads so tool call logs stay correlated with the exchange + Map<String, String> mdc = MDC.getCopyOfContextMap(); + + List<Future<ToolResult>> futures = new ArrayList<>(toolCalls.size()); + for (ChatCompletionMessageToolCall toolCall : toolCalls) { + futures.add(executorService.submit(withMdc(mdc, () -> executeOne(toolCall, toolState)))); + } + + long timeout = endpoint.getConfiguration().getParallelToolTimeout(); + long deadline = timeout > 0 ? System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeout) : 0; + + ToolResult[] results = new ToolResult[toolCalls.size()]; + Exception failure = null; + + for (int i = 0; i < futures.size(); i++) { + ChatCompletionMessageToolCall toolCall = toolCalls.get(i); + try { + results[i] = deadline > 0 + ? futures.get(i).get(Math.max(0, deadline - System.nanoTime()), TimeUnit.NANOSECONDS) + : futures.get(i).get(); + } catch (TimeoutException e) { + futures.get(i).cancel(true); + // Do not abandon the siblings: keep collecting so that every dispatched call is accounted for + Exception timedOut = timeoutFailure(toolCall, timeout); + if (timedOut != null) { + failure = failure != null ? failure : timedOut; + } else { + results[i] = errorResult(toolCall, "Error: tool execution timed out after " + timeout + " ms"); + } + } catch (ExecutionException e) { + // executeOne only throws when the configured strategy is to fail the exchange + failure = failure != null ? failure : asException(e.getCause()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + futures.forEach(f -> f.cancel(true)); + throw e; + } + } + + if (failure != null) { + throw failure; + } + return Arrays.asList(results); + } + + /** + * Builds the exception for a timed out call, or returns {@code null} when the configured strategy is to report the + * timeout back to the model instead of failing the exchange. + */ + private Exception timeoutFailure(ChatCompletionMessageToolCall toolCall, long timeout) { + String toolName = toolCall.asFunction().function().name(); + if (endpoint.getConfiguration().getToolExecutionErrorStrategy() == ToolExecutionErrorStrategy.FAIL_EXCHANGE) { + return new TimeoutException( + "MCP tool '" + toolName + "' did not complete within parallelToolTimeout of " + timeout + " ms"); + } + LOG.warn("MCP tool '{}' timed out after {} ms, sending the timeout back to the model", toolName, timeout); + return null; + } + + private static Exception asException(Throwable cause) { + return cause instanceof Exception e ? e : new IllegalStateException(cause); + } + + private static Callable<ToolResult> withMdc(Map<String, String> mdc, Callable<ToolResult> task) { + if (mdc == null || mdc.isEmpty()) { + return task; + } + return () -> { + MDC.setContextMap(mdc); + try { + return task.call(); + } finally { + MDC.clear(); + } + }; + } + + private ToolResult executeOne(ChatCompletionMessageToolCall toolCall, McpToolState toolState) throws Exception { + OpenAIConfiguration config = endpoint.getConfiguration(); + String toolName = toolCall.asFunction().function().name(); + String argsJson = toolCall.asFunction().function().arguments(); + + McpSyncClient mcpClient = toolState.toolClientMap().get(toolName); + if (mcpClient == null) { + if (config.getHallucinatedToolNameStrategy() == HallucinatedToolNameStrategy.FAIL_EXCHANGE) { + throw new IllegalStateException("Tool '" + toolName + "' not found in any configured MCP server"); + } + // repromptModel: send a corrective tool result listing available tools + String available = String.join(", ", toolState.toolClientMap().keySet()); + LOG.warn("Hallucinated tool name '{}', sending corrective result to model", toolName); + return errorResult(toolCall, + "Error: tool '" + toolName + "' does not exist. Available tools: " + available); + } + + LOG.debug("Executing MCP tool '{}' with args: {}", toolName, argsJson); + + try { + Map<String, Object> argsMap = OBJECT_MAPPER.readValue(argsJson, Map.class); + McpSchema.CallToolResult toolResult = endpoint.callTool(mcpClient, toolName, argsMap); + + if (Boolean.TRUE.equals(toolResult.isError())) { + String content = "Error: " + extractTextContent(toolResult.content()); + LOG.warn("MCP tool '{}' returned error: {}", toolName, content); + return errorResult(toolCall, content); + } + + String content = extractTextContent(toolResult.content()); + LOG.debug("Tool '{}' result: {}", toolName, content); + return new ToolResult( + toolCall.asFunction().id(), toolName, content, toolState.returnDirectTools().contains(toolName)); + } catch (JsonProcessingException e) { + if (config.getToolExecutionErrorStrategy() == ToolExecutionErrorStrategy.FAIL_EXCHANGE) { + throw e; + } + LOG.warn("Invalid tool arguments for '{}': {}", toolName, argsJson, e); + return errorResult(toolCall, "Error: invalid tool arguments: " + e.getMessage()); + } catch (Exception e) { + if (config.getToolExecutionErrorStrategy() == ToolExecutionErrorStrategy.FAIL_EXCHANGE) { + throw e; + } + LOG.warn("MCP tool '{}' execution failed: {}", toolName, e.getMessage(), e); + return errorResult(toolCall, "Error: Tool execution failed: " + e.getMessage()); + } + } + + private static ToolResult errorResult(ChatCompletionMessageToolCall toolCall, String content) { + return new ToolResult( + toolCall.asFunction().id(), toolCall.asFunction().function().name(), content, false); + } + + private static String extractTextContent(List<McpSchema.Content> contents) { + if (contents == null || contents.isEmpty()) { + return ""; + } + return contents.stream() + .filter(McpSchema.TextContent.class::isInstance) + .map(McpSchema.TextContent.class::cast) + .map(McpSchema.TextContent::text) + .collect(Collectors.joining()); + } +} diff --git a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIConfiguration.java b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIConfiguration.java index 0c866ce17662..9df4fc72c21a 100644 --- a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIConfiguration.java +++ b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIConfiguration.java @@ -226,6 +226,22 @@ public class OpenAIConfiguration implements Cloneable { + "can self-correct and retry. The maxToolIterations option bounds retries.") private HallucinatedToolNameStrategy hallucinatedToolNameStrategy = HallucinatedToolNameStrategy.FAIL_EXCHANGE; + @UriParam(defaultValue = "false") + @Metadata(description = "Execute the tool calls returned by the model in a single response concurrently instead of " + + "sequentially. Tool calls in the same batch are independent by design, so this reduces the " + + "latency of a batch to that of its slowest tool. Results are always fed back to the model in " + + "the original tool call order. Note that with toolExecutionErrorStrategy=failExchange the " + + "sibling tool calls already dispatched complete before the exchange fails.") + private boolean parallelToolExecution; + + @UriParam(defaultValue = "0") + @Metadata(description = "Timeout in milliseconds for a batch of parallel tool calls, so that one slow tool cannot " + + "block the whole batch. The timeout applies to the batch as a whole, not per tool call. " + + "A tool call that exceeds it is cancelled and handled according to toolExecutionErrorStrategy. " + + "The default of 0 disables the batch timeout and relies on mcpTimeout, which already bounds " + + "each individual MCP request. Only used when parallelToolExecution=true.") + private long parallelToolTimeout; + // ========== EMBEDDINGS CONFIGURATION ========== @UriParam @@ -754,6 +770,22 @@ public class OpenAIConfiguration implements Cloneable { this.hallucinatedToolNameStrategy = hallucinatedToolNameStrategy; } + public boolean isParallelToolExecution() { + return parallelToolExecution; + } + + public void setParallelToolExecution(boolean parallelToolExecution) { + this.parallelToolExecution = parallelToolExecution; + } + + public long getParallelToolTimeout() { + return parallelToolTimeout; + } + + public void setParallelToolTimeout(long parallelToolTimeout) { + this.parallelToolTimeout = parallelToolTimeout; + } + public SSLContextParameters getSslContextParameters() { return sslContextParameters; } diff --git a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIProducer.java b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIProducer.java index d36c7d6e48e2..5820cbfbee7c 100644 --- a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIProducer.java +++ b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIProducer.java @@ -30,7 +30,6 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; import java.util.stream.Collectors; -import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.openai.core.JsonField; import com.openai.core.JsonValue; @@ -52,8 +51,6 @@ import com.openai.models.chat.completions.ChatCompletionSystemMessageParam; import com.openai.models.chat.completions.ChatCompletionToolMessageParam; import com.openai.models.chat.completions.ChatCompletionUserMessageParam; import com.openai.models.completions.CompletionUsage; -import io.modelcontextprotocol.client.McpSyncClient; -import io.modelcontextprotocol.spec.McpSchema; import org.apache.camel.AsyncCallback; import org.apache.camel.CamelExchangeException; import org.apache.camel.Exchange; @@ -62,6 +59,7 @@ import org.apache.camel.WrappedFile; import org.apache.camel.spi.Synchronization; import org.apache.camel.support.DefaultAsyncProducer; import org.apache.camel.support.ResourceHelper; +import org.apache.camel.support.service.ServiceHelper; import org.apache.camel.util.ObjectHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -77,6 +75,7 @@ public class OpenAIProducer extends DefaultAsyncProducer { private static final String PENDING_USER_MESSAGE = "CamelOpenAIPendingUserMessage"; private Class<?> outputClassResolved; + private McpToolCallExecutor toolCallExecutor; public OpenAIProducer(OpenAIEndpoint endpoint) { super(endpoint); @@ -86,6 +85,9 @@ public class OpenAIProducer extends DefaultAsyncProducer { protected void doStart() throws Exception { OpenAIConfiguration config = getEndpoint().getConfiguration(); + toolCallExecutor = new McpToolCallExecutor(getEndpoint()); + ServiceHelper.startService(toolCallExecutor); + if (ObjectHelper.isNotEmpty(config.getOutputClass())) { outputClassResolved = getEndpoint().getCamelContext().getClassResolver() .resolveMandatoryClass(config.getOutputClass()); @@ -104,6 +106,12 @@ public class OpenAIProducer extends DefaultAsyncProducer { super.doStart(); } + @Override + protected void doStop() throws Exception { + ServiceHelper.stopService(toolCallExecutor); + super.doStop(); + } + private String resolveResourceContent(String property) { try (InputStream is = ResourceHelper.resolveResourceAsInputStream(getEndpoint().getCamelContext(), property)) { if (is != null) { @@ -532,93 +540,35 @@ public class OpenAIProducer extends DefaultAsyncProducer { paramsBuilder.addMessage(assistantParam); agenticMessages.add(assistantParam); - // Execute all tool calls in this batch - boolean allReturnDirect = true; - List<ToolResultEntry> batchResults = new ArrayList<>(); - + // Record the requested tools up front so the log keeps the model's ordering regardless of + // whether the batch is executed sequentially or in parallel for (ChatCompletionMessageToolCall toolCall : toolCalls) { - String toolName = toolCall.asFunction().function().name(); - String argsJson = toolCall.asFunction().function().arguments(); - String toolCallId = toolCall.asFunction().id(); - toolCallsLog.add(toolName); - - McpToolState mcpToolState = getEndpoint().getMcpToolState(); - McpSyncClient mcpClient = mcpToolState.toolClientMap().get(toolName); - if (mcpClient == null) { - if (config.getHallucinatedToolNameStrategy() == HallucinatedToolNameStrategy.FAIL_EXCHANGE) { - throw new IllegalStateException( - "Tool '" + toolName + "' not found in any configured MCP server"); - } - // repromptModel: send a corrective tool result listing available tools - String available = String.join(", ", mcpToolState.toolClientMap().keySet()); - String errorMsg = "Error: tool '" + toolName - + "' does not exist. Available tools: " + available; - LOG.warn("Hallucinated tool name '{}', sending corrective result to model", toolName); - batchResults.add(new ToolResultEntry(toolCallId, errorMsg)); - allReturnDirect = false; - continue; - } - - LOG.debug("Executing MCP tool '{}' with args: {}", toolName, argsJson); - String resultContent; - - try { - Map<String, Object> argsMap = OBJECT_MAPPER.readValue(argsJson, Map.class); - McpSchema.CallToolResult toolResult - = getEndpoint().callTool(mcpClient, toolName, argsMap); - - if (Boolean.TRUE.equals(toolResult.isError())) { - resultContent = "Error: " + extractTextContent(toolResult.content()); - allReturnDirect = false; - } else { - resultContent = extractTextContent(toolResult.content()); - if (!mcpToolState.returnDirectTools().contains(toolName)) { - allReturnDirect = false; - } - } - } catch (JsonProcessingException e) { - if (config.getToolExecutionErrorStrategy() == ToolExecutionErrorStrategy.FAIL_EXCHANGE) { - throw e; - } - LOG.warn("Invalid tool arguments for '{}': {}", toolName, argsJson, e); - resultContent = "Error: invalid tool arguments: " + e.getMessage(); - allReturnDirect = false; - } catch (Exception e) { - if (config.getToolExecutionErrorStrategy() == ToolExecutionErrorStrategy.FAIL_EXCHANGE) { - throw e; - } - LOG.warn("MCP tool '{}' execution failed: {}", toolName, e.getMessage(), e); - resultContent = "Error: Tool execution failed: " + e.getMessage(); - allReturnDirect = false; - } - - LOG.debug("Tool '{}' result: {}", toolName, resultContent); - batchResults.add(new ToolResultEntry(toolCallId, resultContent)); + toolCallsLog.add(toolCall.asFunction().function().name()); } + // Execute all tool calls in this batch + List<McpToolCallExecutor.ToolResult> batchResults = toolCallExecutor.execute(toolCalls); + boolean allReturnDirect = batchResults.stream().allMatch(McpToolCallExecutor.ToolResult::returnDirect); + // returnDirect check: if ALL tools in this batch are returnDirect, short-circuit if (allReturnDirect && !batchResults.isEmpty()) { LOG.debug("All tools in batch have returnDirect=true, short-circuiting agentic loop"); - StringBuilder directResult = new StringBuilder(); - for (ToolResultEntry entry : batchResults) { - if (!directResult.isEmpty()) { - directResult.append("\n"); - } - directResult.append(entry.content()); - } + String directResult = batchResults.stream() + .map(McpToolCallExecutor.ToolResult::content) + .collect(Collectors.joining("\n")); - exchange.getMessage().setBody(directResult.toString()); + exchange.getMessage().setBody(directResult); setResponseHeaders(exchange.getMessage(), response); exchange.getMessage().setHeader(OpenAIConstants.TOOL_ITERATIONS, iteration); exchange.getMessage().setHeader(OpenAIConstants.MCP_TOOL_CALLS, toolCallsLog); exchange.getMessage().setHeader(OpenAIConstants.MCP_RETURN_DIRECT, true); - updateConversationHistory(exchange, agenticMessages, directResult.toString()); + updateConversationHistory(exchange, agenticMessages, directResult); return; } // Normal path: feed tool results back to LLM LOG.debug("Feeding {} tool result(s) back to the model", batchResults.size()); - for (ToolResultEntry entry : batchResults) { + for (McpToolCallExecutor.ToolResult entry : batchResults) { ChatCompletionMessageParam toolMsg = ChatCompletionMessageParam.ofTool( ChatCompletionToolMessageParam.builder() .toolCallId(entry.toolCallId()) @@ -651,20 +601,6 @@ public class OpenAIProducer extends DefaultAsyncProducer { tokenTracker.getCompletionTokens(), tokenTracker.getTotalTokens())); } - private String extractTextContent(List<McpSchema.Content> contents) { - if (contents == null || contents.isEmpty()) { - return ""; - } - return contents.stream() - .filter(McpSchema.TextContent.class::isInstance) - .map(McpSchema.TextContent.class::cast) - .map(McpSchema.TextContent::text) - .collect(Collectors.joining()); - } - - private record ToolResultEntry(String toolCallId, String content) { - } - private void processStreaming(Exchange exchange, ChatCompletionCreateParams params) { // NOTE: the stream is going to be closed after the exchange completes. StreamResponse<ChatCompletionChunk> streamResponse = getEndpoint().getClient().chat().completions() // NOSONAR diff --git a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIToolExecutionProducer.java b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIToolExecutionProducer.java index a5313f0d7089..b6021011deaa 100644 --- a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIToolExecutionProducer.java +++ b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIToolExecutionProducer.java @@ -18,21 +18,16 @@ package org.apache.camel.component.openai; import java.util.ArrayList; import java.util.List; -import java.util.Map; -import java.util.stream.Collectors; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; import com.openai.models.chat.completions.ChatCompletion; import com.openai.models.chat.completions.ChatCompletionAssistantMessageParam; import com.openai.models.chat.completions.ChatCompletionMessageParam; import com.openai.models.chat.completions.ChatCompletionMessageToolCall; import com.openai.models.chat.completions.ChatCompletionToolMessageParam; import com.openai.models.chat.completions.ChatCompletionUserMessageParam; -import io.modelcontextprotocol.client.McpSyncClient; -import io.modelcontextprotocol.spec.McpSchema; import org.apache.camel.Exchange; import org.apache.camel.support.DefaultProducer; +import org.apache.camel.support.service.ServiceHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -60,7 +55,8 @@ import org.slf4j.LoggerFactory; public class OpenAIToolExecutionProducer extends DefaultProducer { private static final Logger LOG = LoggerFactory.getLogger(OpenAIToolExecutionProducer.class); - private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + private McpToolCallExecutor toolCallExecutor; public OpenAIToolExecutionProducer(OpenAIEndpoint endpoint) { super(endpoint); @@ -71,6 +67,19 @@ public class OpenAIToolExecutionProducer extends DefaultProducer { return (OpenAIEndpoint) super.getEndpoint(); } + @Override + protected void doStart() throws Exception { + toolCallExecutor = new McpToolCallExecutor(getEndpoint()); + ServiceHelper.startService(toolCallExecutor); + super.doStart(); + } + + @Override + protected void doStop() throws Exception { + ServiceHelper.stopService(toolCallExecutor); + super.doStop(); + } + @Override public void process(Exchange exchange) throws Exception { // Get the full ChatCompletion response (stored by storeFullResponse=true on chat-completion) @@ -130,82 +139,18 @@ public class OpenAIToolExecutionProducer extends DefaultProducer { "No MCP tool clients configured on the endpoint. Configure mcpServer.* parameters."); } - int executedCount = 0; - for (ChatCompletionMessageToolCall toolCall : toolCalls) { - String toolName = toolCall.asFunction().function().name(); - String argsJson = toolCall.asFunction().function().arguments(); - String toolCallId = toolCall.asFunction().id(); - - McpToolState mcpToolState = getEndpoint().getMcpToolState(); - McpSyncClient mcpClient = mcpToolState.toolClientMap().get(toolName); - if (mcpClient == null) { - if (config.getHallucinatedToolNameStrategy() == HallucinatedToolNameStrategy.FAIL_EXCHANGE) { - throw new IllegalStateException( - "Tool '" + toolName + "' not found in any configured MCP server"); - } - // repromptModel: send a corrective tool result listing available tools - String available = String.join(", ", mcpToolState.toolClientMap().keySet()); - String errorMsg = "Error: tool '" + toolName - + "' does not exist. Available tools: " + available; - LOG.warn("Hallucinated tool name '{}', sending corrective result to model", toolName); - history.add(ChatCompletionMessageParam.ofTool( - ChatCompletionToolMessageParam.builder() - .toolCallId(toolCallId) - .content(errorMsg) - .build())); - executedCount++; - continue; - } - - String resultContent; - - try { - Map<String, Object> argsMap = OBJECT_MAPPER.readValue(argsJson, Map.class); - McpSchema.CallToolResult toolResult - = getEndpoint().callTool(mcpClient, toolName, argsMap); - - if (Boolean.TRUE.equals(toolResult.isError())) { - resultContent = "Error: " + extractTextContent(toolResult.content()); - LOG.warn("MCP tool '{}' returned error: {}", toolName, resultContent); - } else { - resultContent = extractTextContent(toolResult.content()); - } - } catch (JsonProcessingException e) { - if (config.getToolExecutionErrorStrategy() == ToolExecutionErrorStrategy.FAIL_EXCHANGE) { - throw e; - } - LOG.warn("Invalid tool arguments for '{}': {}", toolName, argsJson, e); - resultContent = "Error: invalid tool arguments: " + e.getMessage(); - } catch (Exception e) { - if (config.getToolExecutionErrorStrategy() == ToolExecutionErrorStrategy.FAIL_EXCHANGE) { - throw e; - } - LOG.warn("MCP tool '{}' execution failed: {}", toolName, e.getMessage(), e); - resultContent = "Error: Tool execution failed: " + e.getMessage(); - } - + List<McpToolCallExecutor.ToolResult> results = toolCallExecutor.execute(toolCalls); + for (McpToolCallExecutor.ToolResult result : results) { history.add(ChatCompletionMessageParam.ofTool( ChatCompletionToolMessageParam.builder() - .toolCallId(toolCallId) - .content(resultContent) + .toolCallId(result.toolCallId()) + .content(result.content()) .build())); - executedCount++; } // Update conversation history and clear body for the next chat-completion call exchange.setProperty(historyProperty, OpenAIConversationHistoryTrimmer.trim(history, config)); exchange.getMessage().setBody(null); - exchange.getMessage().setHeader(OpenAIConstants.TOOL_ITERATIONS, executedCount); - } - - private String extractTextContent(List<McpSchema.Content> contents) { - if (contents == null || contents.isEmpty()) { - return ""; - } - return contents.stream() - .filter(McpSchema.TextContent.class::isInstance) - .map(McpSchema.TextContent.class::cast) - .map(McpSchema.TextContent::text) - .collect(Collectors.joining()); + exchange.getMessage().setHeader(OpenAIConstants.TOOL_ITERATIONS, results.size()); } } diff --git a/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/McpToolCallExecutorTest.java b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/McpToolCallExecutorTest.java new file mode 100644 index 000000000000..4d9d975dbd13 --- /dev/null +++ b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/McpToolCallExecutorTest.java @@ -0,0 +1,364 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.openai; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import com.openai.models.chat.completions.ChatCompletionMessageFunctionToolCall; +import com.openai.models.chat.completions.ChatCompletionMessageToolCall; +import io.modelcontextprotocol.client.McpSyncClient; +import io.modelcontextprotocol.spec.McpSchema; +import org.apache.camel.impl.DefaultCamelContext; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Tests for CAMEL-23078: the shared MCP tool call batch executor, covering both the sequential and the parallel + * ({@code parallelToolExecution=true}) execution paths. + */ +class McpToolCallExecutorTest { + + private static final long AWAIT_SECONDS = 10; + + private final DefaultCamelContext context = new DefaultCamelContext(); + + private McpToolCallExecutor executor; + + @BeforeEach + void startContext() { + context.start(); + } + + @AfterEach + void tearDown() throws Exception { + if (executor != null) { + executor.stop(); + } + context.stop(); + } + + // ------------------------------------------------------------------ + // Ordering + // ------------------------------------------------------------------ + + @Test + void parallelResultsKeepToolCallOrder() throws Exception { + // Every tool blocks until all of them have started, so the batch can only complete when it runs + // concurrently. The results must still come back in the order the model requested them. + CountDownLatch rendezvous = new CountDownLatch(3); + + OpenAIEndpoint endpoint = newEndpoint(true, 0, Map.of( + "slow_a", rendezvousClient(rendezvous, "A"), + "slow_b", rendezvousClient(rendezvous, "B"), + "slow_c", rendezvousClient(rendezvous, "C")), + Set.of()); + executor = startExecutor(endpoint); + + List<McpToolCallExecutor.ToolResult> results + = executor.execute(List.of(toolCall("id-a", "slow_a"), toolCall("id-b", "slow_b"), + toolCall("id-c", "slow_c"))); + + assertThat(results).extracting(McpToolCallExecutor.ToolResult::toolCallId) + .containsExactly("id-a", "id-b", "id-c"); + assertThat(results).extracting(McpToolCallExecutor.ToolResult::content) + .containsExactly("A", "B", "C"); + } + + @Test + void sequentialExecutionProducesTheSameResults() throws Exception { + OpenAIEndpoint endpoint = newEndpoint(false, 0, Map.of( + "tool_a", staticClient("A"), + "tool_b", staticClient("B")), + Set.of()); + executor = startExecutor(endpoint); + + List<McpToolCallExecutor.ToolResult> results + = executor.execute(List.of(toolCall("id-a", "tool_a"), toolCall("id-b", "tool_b"))); + + assertThat(results).extracting(McpToolCallExecutor.ToolResult::toolCallId) + .containsExactly("id-a", "id-b"); + assertThat(results).extracting(McpToolCallExecutor.ToolResult::content).containsExactly("A", "B"); + } + + @Test + void emptyBatchReturnsNoResults() throws Exception { + OpenAIEndpoint endpoint = newEndpoint(true, 0, Map.of("tool_a", staticClient("A")), Set.of()); + executor = startExecutor(endpoint); + + assertThat(executor.execute(List.of())).isEmpty(); + } + + // ------------------------------------------------------------------ + // returnDirect + // ------------------------------------------------------------------ + + @Test + void returnDirectIsReportedPerToolCall() throws Exception { + OpenAIEndpoint endpoint = newEndpoint(true, 0, Map.of( + "direct_tool", staticClient("A"), + "normal_tool", staticClient("B")), + Set.of("direct_tool")); + executor = startExecutor(endpoint); + + List<McpToolCallExecutor.ToolResult> results + = executor.execute(List.of(toolCall("id-a", "direct_tool"), toolCall("id-b", "normal_tool"))); + + assertThat(results).extracting(McpToolCallExecutor.ToolResult::returnDirect).containsExactly(true, false); + } + + @Test + void returnDirectIsNotReportedForAToolThatReturnedAnError() throws Exception { + McpSyncClient failing = mock(McpSyncClient.class); + when(failing.callTool(any(McpSchema.CallToolRequest.class))).thenReturn( + McpSchema.CallToolResult.builder() + .content(List.of(new McpSchema.TextContent(null, "boom", null))) + .isError(true) + .build()); + + OpenAIEndpoint endpoint + = newEndpoint(true, 0, Map.of("direct_tool", failing), Set.of("direct_tool")); + executor = startExecutor(endpoint); + + List<McpToolCallExecutor.ToolResult> results = executor.execute(List.of(toolCall("id-a", "direct_tool"))); + + assertThat(results).singleElement() + .satisfies(r -> { + assertThat(r.returnDirect()).isFalse(); + assertThat(r.content()).isEqualTo("Error: boom"); + }); + } + + // ------------------------------------------------------------------ + // Error handling + // ------------------------------------------------------------------ + + @Test + void parallelFailureLetsSiblingToolsCompleteBeforeFailingTheExchange() throws Exception { + AtomicInteger siblingCalls = new AtomicInteger(); + McpSyncClient sibling = mock(McpSyncClient.class); + when(sibling.callTool(any(McpSchema.CallToolRequest.class))).thenAnswer(invocation -> { + siblingCalls.incrementAndGet(); + return textResult("ok"); + }); + + McpSyncClient failing = mock(McpSyncClient.class); + when(failing.callTool(any(McpSchema.CallToolRequest.class))) + .thenThrow(new IllegalStateException("tool blew up")); + + // failing tool first, so a mid-loop abort would skip the sibling entirely + OpenAIEndpoint endpoint = newEndpoint(true, 0, Map.of( + "failing_tool", failing, + "sibling_tool", sibling), + Set.of()); + executor = startExecutor(endpoint); + + assertThatThrownBy( + () -> executor.execute(List.of(toolCall("id-a", "failing_tool"), toolCall("id-b", "sibling_tool")))) + .isInstanceOf(IllegalStateException.class) + .hasMessage("tool blew up"); + + assertThat(siblingCalls).hasValue(1); + } + + @Test + void toolFailureIsSentBackToTheModelWhenRepromptingIsConfigured() throws Exception { + McpSyncClient failing = mock(McpSyncClient.class); + when(failing.callTool(any(McpSchema.CallToolRequest.class))) + .thenThrow(new IllegalStateException("tool blew up")); + + OpenAIEndpoint endpoint = newEndpoint(true, 0, Map.of( + "failing_tool", failing, + "ok_tool", staticClient("fine")), + Set.of()); + endpoint.getConfiguration().setToolExecutionErrorStrategy(ToolExecutionErrorStrategy.REPROMPT_MODEL); + executor = startExecutor(endpoint); + + List<McpToolCallExecutor.ToolResult> results + = executor.execute(List.of(toolCall("id-a", "failing_tool"), toolCall("id-b", "ok_tool"))); + + assertThat(results).extracting(McpToolCallExecutor.ToolResult::content) + .containsExactly("Error: Tool execution failed: tool blew up", "fine"); + } + + @Test + void hallucinatedToolNameFailsTheExchangeByDefault() throws Exception { + OpenAIEndpoint endpoint = newEndpoint(true, 0, Map.of("known_tool", staticClient("A")), Set.of()); + executor = startExecutor(endpoint); + + assertThatThrownBy( + () -> executor.execute(List.of(toolCall("id-a", "known_tool"), toolCall("id-b", "made_up_tool")))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("made_up_tool"); + } + + @Test + void hallucinatedToolNameIsCorrectedWhenRepromptingIsConfigured() throws Exception { + OpenAIEndpoint endpoint = newEndpoint(true, 0, Map.of("known_tool", staticClient("A")), Set.of()); + endpoint.getConfiguration().setHallucinatedToolNameStrategy(HallucinatedToolNameStrategy.REPROMPT_MODEL); + executor = startExecutor(endpoint); + + List<McpToolCallExecutor.ToolResult> results + = executor.execute(List.of(toolCall("id-a", "made_up_tool"), toolCall("id-b", "known_tool"))); + + assertThat(results.get(0).content()).contains("made_up_tool", "known_tool"); + assertThat(results.get(0).returnDirect()).isFalse(); + assertThat(results.get(1).content()).isEqualTo("A"); + } + + // ------------------------------------------------------------------ + // Timeout + // ------------------------------------------------------------------ + + @Test + void batchTimeoutFailsTheExchangeButLetsFastToolsFinish() throws Exception { + CountDownLatch release = new CountDownLatch(1); + AtomicInteger fastCalls = new AtomicInteger(); + try { + McpSyncClient fast = mock(McpSyncClient.class); + when(fast.callTool(any(McpSchema.CallToolRequest.class))).thenAnswer(invocation -> { + fastCalls.incrementAndGet(); + return textResult("fast"); + }); + + OpenAIEndpoint endpoint = newEndpoint(true, 200, Map.of( + "blocking_tool", blockingClient(release), + "fast_tool", fast), + Set.of()); + executor = startExecutor(endpoint); + + assertThatThrownBy( + () -> executor.execute(List.of(toolCall("id-a", "blocking_tool"), toolCall("id-b", "fast_tool")))) + .isInstanceOf(TimeoutException.class) + .hasMessageContaining("blocking_tool") + .hasMessageContaining("parallelToolTimeout"); + + // the slow tool must not have cancelled its sibling + assertThat(fastCalls).hasValue(1); + } finally { + release.countDown(); + } + } + + @Test + void batchTimeoutIsSentBackToTheModelWhenRepromptingIsConfigured() throws Exception { + CountDownLatch release = new CountDownLatch(1); + try { + OpenAIEndpoint endpoint = newEndpoint(true, 200, Map.of( + "blocking_tool", blockingClient(release)), + Set.of()); + endpoint.getConfiguration().setToolExecutionErrorStrategy(ToolExecutionErrorStrategy.REPROMPT_MODEL); + executor = startExecutor(endpoint); + + // two calls, so the batch is dispatched in parallel rather than run inline + List<McpToolCallExecutor.ToolResult> results + = executor.execute(List.of(toolCall("id-a", "blocking_tool"), toolCall("id-b", "blocking_tool"))); + + assertThat(results).extracting(McpToolCallExecutor.ToolResult::content) + .allSatisfy(content -> assertThat(content).contains("timed out after 200 ms")); + } finally { + release.countDown(); + } + } + + // ------------------------------------------------------------------ + // Helpers + // ------------------------------------------------------------------ + + private McpToolCallExecutor startExecutor(OpenAIEndpoint endpoint) throws Exception { + McpToolCallExecutor answer = new McpToolCallExecutor(endpoint); + answer.start(); + return answer; + } + + private OpenAIEndpoint newEndpoint( + boolean parallel, long timeout, Map<String, McpSyncClient> toolClients, Set<String> returnDirectTools) { + OpenAIConfiguration configuration = new OpenAIConfiguration(); + configuration.setParallelToolExecution(parallel); + configuration.setParallelToolTimeout(timeout); + + OpenAIComponent component = new OpenAIComponent(); + component.setCamelContext(context); + + OpenAIEndpoint endpoint = new OpenAIEndpoint("openai:chat-completion", component, configuration); + endpoint.setCamelContext(context); + endpoint.setMcpToolState(new McpToolState( + List.of(), toolClients, Map.of(), returnDirectTools)); + return endpoint; + } + + private static ChatCompletionMessageToolCall toolCall(String id, String toolName) { + return ChatCompletionMessageToolCall.ofFunction( + ChatCompletionMessageFunctionToolCall.builder() + .id(id) + .function(ChatCompletionMessageFunctionToolCall.Function.builder() + .name(toolName) + .arguments("{}") + .build()) + .build()); + } + + private static McpSyncClient staticClient(String resultText) { + McpSyncClient client = mock(McpSyncClient.class); + when(client.callTool(any(McpSchema.CallToolRequest.class))).thenReturn(textResult(resultText)); + return client; + } + + /** + * A client that only returns once every tool in the batch has entered its call, so the batch can complete only when + * the calls are dispatched concurrently. + */ + private static McpSyncClient rendezvousClient(CountDownLatch rendezvous, String resultText) { + McpSyncClient client = mock(McpSyncClient.class); + when(client.callTool(any(McpSchema.CallToolRequest.class))).thenAnswer(invocation -> { + rendezvous.countDown(); + if (!rendezvous.await(AWAIT_SECONDS, TimeUnit.SECONDS)) { + throw new IllegalStateException("tool calls were not dispatched concurrently"); + } + return textResult(resultText); + }); + return client; + } + + private static McpSyncClient blockingClient(CountDownLatch release) { + McpSyncClient client = mock(McpSyncClient.class); + when(client.callTool(any(McpSchema.CallToolRequest.class))).thenAnswer(invocation -> { + release.await(AWAIT_SECONDS, TimeUnit.SECONDS); + return textResult("released"); + }); + return client; + } + + private static McpSchema.CallToolResult textResult(String text) { + return McpSchema.CallToolResult.builder() + .content(List.of(new McpSchema.TextContent(null, text, null))) + .isError(false) + .build(); + } +} diff --git a/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIParallelToolExecutionTest.java b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIParallelToolExecutionTest.java new file mode 100644 index 000000000000..9f2bf52dd646 --- /dev/null +++ b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIParallelToolExecutionTest.java @@ -0,0 +1,168 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.openai; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import io.modelcontextprotocol.client.McpSyncClient; +import io.modelcontextprotocol.spec.McpSchema; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.test.infra.openai.mock.OpenAIMock; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * End-to-end tests for CAMEL-23078: {@code parallelToolExecution=true} in the agentic loop. + */ +class OpenAIParallelToolExecutionTest extends CamelTestSupport { + + private static final long AWAIT_SECONDS = 10; + + /** + * Every tool blocks until both have started, so the agentic loop can only make progress when the batch is + * dispatched concurrently. + */ + private final CountDownLatch rendezvous = new CountDownLatch(2); + + @RegisterExtension + public OpenAIMock openAIMock = new OpenAIMock().builder() + .when("call both tools") + .invokeTool("get_weather") + .withParam("city", "London") + .andInvokeTool("get_traffic") + .withParam("city", "London") + .replyWith("London is sunny with light traffic.") + .end() + .build(); + + private String endpointUri(boolean parallel) { + return "openai:chat-completion?model=gpt-5&apiKey=dummy&autoToolExecution=true" + + "¶llelToolExecution=" + parallel + + "&baseUrl=" + openAIMock.getBaseUrl() + "/v1"; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:parallel").to(endpointUri(true)); + from("direct:sequential").to(endpointUri(false)); + } + }; + } + + @Test + void parallelBatchIsDispatchedConcurrently() { + injectMcpTools(endpointUri(true), Set.of()); + + Exchange result = template.request("direct:parallel", e -> e.getIn().setBody("call both tools")); + + assertThat(result.getException()).isNull(); + assertThat(result.getMessage().getBody(String.class)).isEqualTo("London is sunny with light traffic."); + assertThat(result.getMessage().getHeader(OpenAIConstants.TOOL_ITERATIONS, Integer.class)).isEqualTo(1); + assertThat(result.getMessage().getHeader(OpenAIConstants.MCP_TOOL_CALLS, List.class)) + .containsExactly("get_weather", "get_traffic"); + assertThat(rendezvous.getCount()).isZero(); + } + + @Test + void parallelBatchOfReturnDirectToolsShortCircuitsInOrder() { + injectMcpTools(endpointUri(true), Set.of("get_weather", "get_traffic")); + + Exchange result = template.request("direct:parallel", e -> e.getIn().setBody("call both tools")); + + assertThat(result.getException()).isNull(); + assertThat(result.getMessage().getHeader(OpenAIConstants.MCP_RETURN_DIRECT, Boolean.class)).isTrue(); + // results are joined in the order the model requested the tools, not in completion order + assertThat(result.getMessage().getBody(String.class)).isEqualTo("Sunny, 22C\nLight traffic"); + } + + @Test + void sequentialExecutionRemainsTheDefaultBehavior() { + // no rendezvous: the sequential path would deadlock on it, so use independent clients + Map<String, McpSyncClient> toolClients = new LinkedHashMap<>(); + toolClients.put("get_weather", staticClient("Sunny, 22C")); + toolClients.put("get_traffic", staticClient("Light traffic")); + injectMcpTools(endpointUri(false), toolClients, Set.of()); + + Exchange result = template.request("direct:sequential", e -> e.getIn().setBody("call both tools")); + + assertThat(result.getException()).isNull(); + assertThat(result.getMessage().getBody(String.class)).isEqualTo("London is sunny with light traffic."); + assertThat(result.getMessage().getHeader(OpenAIConstants.MCP_TOOL_CALLS, List.class)) + .containsExactly("get_weather", "get_traffic"); + } + + private void injectMcpTools(String endpointKey, Set<String> returnDirectToolNames) { + Map<String, McpSyncClient> toolClients = new LinkedHashMap<>(); + toolClients.put("get_weather", rendezvousClient("Sunny, 22C")); + toolClients.put("get_traffic", rendezvousClient("Light traffic")); + injectMcpTools(endpointKey, toolClients, returnDirectToolNames); + } + + private void injectMcpTools( + String endpointKey, Map<String, McpSyncClient> toolClients, Set<String> returnDirectToolNames) { + OpenAIEndpoint endpoint = context.getEndpoint(endpointKey, OpenAIEndpoint.class); + + List<McpSchema.Tool> mcpTools = toolClients.keySet().stream() + .map(name -> McpSchema.Tool.builder(name, Map.of("type", "object")) + .description("Mock tool: " + name) + .build()) + .toList(); + + endpoint.setMcpToolState(new McpToolState( + McpToolConverter.convert(mcpTools), toolClients, Map.of(), returnDirectToolNames)); + } + + private McpSyncClient rendezvousClient(String resultText) { + McpSyncClient client = mock(McpSyncClient.class); + when(client.callTool(any(McpSchema.CallToolRequest.class))).thenAnswer(invocation -> { + rendezvous.countDown(); + if (!rendezvous.await(AWAIT_SECONDS, TimeUnit.SECONDS)) { + throw new IllegalStateException("tool calls were not dispatched concurrently"); + } + return textResult(resultText); + }); + return client; + } + + private static McpSyncClient staticClient(String resultText) { + McpSyncClient client = mock(McpSyncClient.class); + when(client.callTool(any(McpSchema.CallToolRequest.class))).thenReturn(textResult(resultText)); + return client; + } + + private static McpSchema.CallToolResult textResult(String text) { + return McpSchema.CallToolResult.builder() + .content(List.of(new McpSchema.TextContent(null, text, null))) + .isError(false) + .build(); + } +} diff --git a/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/integration/OpenAIMcpParallelToolExecutionIT.java b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/integration/OpenAIMcpParallelToolExecutionIT.java new file mode 100644 index 000000000000..df82b75800f8 --- /dev/null +++ b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/integration/OpenAIMcpParallelToolExecutionIT.java @@ -0,0 +1,132 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.openai.integration; + +import java.util.List; + +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.openai.OpenAIConstants; +import org.apache.camel.test.infra.mcp.everything.services.McpEverythingService; +import org.apache.camel.test.infra.mcp.everything.services.McpEverythingServiceFactory; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfSystemProperty; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Integration tests for CAMEL-23078: the agentic loop with {@code parallelToolExecution=true} against a real MCP server + * and a real LLM backend. + * + * <p> + * How many tool calls the model emits per response is up to the model, and a batch of one is executed inline rather + * than dispatched to the pool, so reaching the parallel path at all depends on the backend in use. With + * {@code qwen3.5:2b-mlx} the multi-tool prompt below reliably produces two-call batches and does exercise it; smaller + * or older models often serialise the same prompt into one tool call per iteration instead. These tests are therefore + * written to assert that enabling the option yields the same correct outcome as the sequential default whatever the + * model chooses to do, rather than asserting a particular batch size, which would be flaky. The deterministic proof + * that a batch really is dispatched concurrently lives in {@code McpToolCallExecutorTest} and + * {@code OpenAIParallelToolExecutionTest}, which rendezvous the tool calls on a latch that can only be released if they + * run at the same time. + * + * <p> + * Uses the MCP Everything Server (Streamable HTTP) and Ollama as the LLM backend. See {@code test_execution.md} for how + * to run these against a local Ollama. + */ +@DisabledIfSystemProperty(named = "ci.env.name", matches = ".*", disabledReason = "Requires too much network resources") +public class OpenAIMcpParallelToolExecutionIT extends OpenAITestSupport { + + private static final String MCP_PROTOCOL_VERSIONS = "2024-11-05,2025-03-26,2025-06-18"; + + /** + * Models call tools one at a time unless told otherwise. This gives a capable model the chance to emit a real + * batch, which is what reaches the parallel dispatch path. + */ + private static final String PARALLEL_SYSTEM_MESSAGE + = "You must call all the tools you need at once, in a single response, in parallel."; + + private static final String MULTI_TOOL_PROMPT + = "Add 3 and 4 using the add tool, and echo the word hello using the echo tool. Call both tools now, together."; + + @RegisterExtension + static McpEverythingService MCP_EVERYTHING = McpEverythingServiceFactory.createService(); + + @Override + protected RouteBuilder createRouteBuilder() { + String mcpConfig = "&mcpServer.everything.transportType=streamableHttp" + + "&mcpServer.everything.url=" + MCP_EVERYTHING.url() + + "&mcpProtocolVersions=" + MCP_PROTOCOL_VERSIONS; + + return new RouteBuilder() { + @Override + public void configure() { + from("direct:parallel") + .toF("openai:chat-completion?parallelToolExecution=true%s", mcpConfig); + + from("direct:parallel-timeout") + .toF("openai:chat-completion?parallelToolExecution=true¶llelToolTimeout=60000%s", mcpConfig); + + from("direct:sequential") + .toF("openai:chat-completion?%s", mcpConfig.substring(1)); + } + }; + } + + @Test + void parallelToolExecutionCompletesTheAgenticLoop() { + Exchange result = template.request("direct:parallel", this::multiToolRequest); + + assertThat(result.getException()).isNull(); + assertThat(result.getMessage().getBody(String.class)).contains("7").containsIgnoringCase("hello"); + assertThat(toolCalls(result)).contains("add", "echo"); + } + + @Test + void parallelToolExecutionHonorsABatchTimeout() { + // a generous timeout must not disturb tool calls that complete well within it + Exchange result = template.request("direct:parallel-timeout", this::multiToolRequest); + + assertThat(result.getException()).isNull(); + assertThat(result.getMessage().getBody(String.class)).contains("7"); + assertThat(toolCalls(result)).contains("add", "echo"); + } + + @Test + void parallelAndSequentialExecutionAgreeOnTheOutcome() { + Exchange sequential = template.request("direct:sequential", + e -> e.getIn().setBody("Use the add tool to add 15 and 27. What is the result?")); + Exchange parallel = template.request("direct:parallel", + e -> e.getIn().setBody("Use the add tool to add 15 and 27. What is the result?")); + + assertThat(sequential.getException()).isNull(); + assertThat(parallel.getException()).isNull(); + assertThat(sequential.getMessage().getBody(String.class)).contains("42"); + assertThat(parallel.getMessage().getBody(String.class)).contains("42"); + assertThat(toolCalls(parallel)).isEqualTo(toolCalls(sequential)); + } + + private void multiToolRequest(Exchange exchange) { + exchange.getIn().setHeader(OpenAIConstants.SYSTEM_MESSAGE, PARALLEL_SYSTEM_MESSAGE); + exchange.getIn().setBody(MULTI_TOOL_PROMPT); + } + + @SuppressWarnings("unchecked") + private static List<String> toolCalls(Exchange exchange) { + return exchange.getMessage().getHeader(OpenAIConstants.MCP_TOOL_CALLS, List.class); + } +} diff --git a/components/camel-ai/camel-openai/test_execution.md b/components/camel-ai/camel-openai/test_execution.md index c57033033d43..696bf1454899 100644 --- a/components/camel-ai/camel-openai/test_execution.md +++ b/components/camel-ai/camel-openai/test_execution.md @@ -32,6 +32,33 @@ Make sure the embedding model is pulled in Ollama before running the tests: ollama pull granite-embedding:30m ``` +### Parallel MCP tool execution tests + +`OpenAIMcpParallelToolExecutionIT` covers `parallelToolExecution=true` end to end. The parallel dispatch path is +only reached when the model emits more than one tool call in a single response, since a batch of one is executed +inline. That is the model's decision, not the component's, so the backend matters: + +* `qwen3.5:2b-mlx` reliably emits two-call batches for these prompts, and the MLX build is noticeably faster on + Apple Silicon. +* Smaller or older models frequently serialise the same prompt into one tool call per iteration, in which case the + tests still pass but only cover the inline path. + +```bash +ollama pull qwen3.5:2b-mlx +mvn verify -Dollama.endpoint=http://localhost:11434/ -Dollama.model=qwen3.5:2b-mlx -Dollama.instance.type=remote \ + -Dit.test=OpenAIMcpParallelToolExecutionIT +``` + +To confirm the parallel path was actually exercised, raise +`org.apache.camel.component.openai.McpToolCallExecutor` to `DEBUG` and look for +`Executing N tool call(s) in parallel` in `target/camel-openai-tests.log`. + +The deterministic proof that a batch is dispatched concurrently is in the unit tests +(`McpToolCallExecutorTest`, `OpenAIParallelToolExecutionTest`), which do not need Ollama at all. + +Note that agentic ITs asserting the model called a tool at all can flake with small models, which sometimes answer +directly instead of calling a tool. + ### Running with OpenAI To run tests against OpenAI API instead of Ollama: diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc index 0b12177f387c..35b8848e616b 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc @@ -429,6 +429,18 @@ Hosted tools: `builtinTools` (`web_search`, `file_search`, `code_interpreter`; ` on this operation. The full SDK response is stored in exchange property `CamelOpenAIResponsesResponse` when `storeFullResponse=true`. +==== Optional parallel MCP tool execution + +The new `parallelToolExecution` option (default `false`) executes the tool calls the model requests in a +single response concurrently rather than sequentially, and `parallelToolTimeout` (milliseconds, default `0` += disabled) bounds such a batch as a whole. Both apply to the agentic loop and to the `openai:tool-execution` +operation. Existing routes are unaffected unless they opt in. + +When `parallelToolExecution=true` is combined with `toolExecutionErrorStrategy=failExchange`, the sibling +tool calls that were already dispatched complete before the exchange fails, whereas sequential execution +abandons the remaining calls at the first failure. Routes whose tools have side effects and that rely on +a failure preventing later tool calls in the same batch should keep the sequential default. + === camel-management - Throughput MBean attribute uses EWMA smoothing The `Throughput` attribute on the `ManagedPerformanceCounter` JMX MBean now reports an EWMA
