You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2025/10/29 11:47:15 UTC
(camel) branch CAMEL-22622 created (now 40c06750fde3)
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a change to branch CAMEL-22622
in repository https://gitbox.apache.org/repos/asf/camel.git
at 40c06750fde3 CAMEL-22622 - Camel-AWS-Bedrock: Support Converse API
This branch includes the following new commits:
new 40c06750fde3 CAMEL-22622 - Camel-AWS-Bedrock: Support Converse API
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 01/01: CAMEL-22622 - Camel-AWS-Bedrock: Support Converse API
Posted by ac...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch CAMEL-22622
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 40c06750fde3ef558ad3634ced5ae290a84429b8
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Wed Oct 29 12:46:35 2025 +0100
CAMEL-22622 - Camel-AWS-Bedrock: Support Converse API
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
.../camel/catalog/components/aws-bedrock.json | 14 +-
.../aws2/bedrock/runtime/aws-bedrock.json | 14 +-
.../aws2/bedrock/runtime/BedrockConstants.java | 17 ++
.../aws2/bedrock/runtime/BedrockOperations.java | 6 +-
.../aws2/bedrock/runtime/BedrockProducer.java | 239 +++++++++++++++++++++
.../runtime/stream/ConverseStreamHandler.java | 172 +++++++++++++++
.../runtime/integration/BedrockProducerIT.java | 71 ++++++
.../integration/BedrockProducerStreamingIT.java | 4 +-
.../runtime/stream/ConverseStreamHandlerTest.java | 79 +++++++
.../dsl/BedrockEndpointBuilderFactory.java | 100 +++++++++
10 files changed, 706 insertions(+), 10 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-bedrock.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-bedrock.json
index 4b75e1b80f4f..c392de4f2189 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-bedrock.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-bedrock.json
@@ -28,7 +28,7 @@
"includeStreamingMetadata": { "index": 1, "kind": "property", "displayName": "Include Streaming Metadata", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "Whether to include streaming metadata in the response hea [...]
"lazyStartProducer": { "index": 2, "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "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 producer may otherwise fail [...]
"modelId": { "index": 3, "kind": "property", "displayName": "Model Id", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "java.lang.String", "enum": [ "amazon.titan-text-express-v1", "amazon.titan-text-lite-v1", "amazon.titan-image-generator-v1", "amazon.titan-embed-text-v1", "amazon.titan-embed-image-v1", "amazon.titan-text-premier-v1:0", "amazon.titan-embed-text-v2:0", "amazon.titan-image-generator-v2:0", "amazon.nova-canvas-v1:0", "amazon.nova-lite-v [...]
- "operation": { "index": 4, "kind": "property", "displayName": "Operation", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "org.apache.camel.component.aws2.bedrock.runtime.BedrockOperations", "enum": [ "invokeTextModel", "invokeImageModel", "invokeEmbeddingsModel", "invokeTextModelStreaming", "invokeImageModelStreaming", "invokeEmbeddingsModelStreaming" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationCl [...]
+ "operation": { "index": 4, "kind": "property", "displayName": "Operation", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "org.apache.camel.component.aws2.bedrock.runtime.BedrockOperations", "enum": [ "invokeTextModel", "invokeImageModel", "invokeEmbeddingsModel", "invokeTextModelStreaming", "invokeImageModelStreaming", "invokeEmbeddingsModelStreaming", "converse", "converseStream" ], "deprecated": false, "deprecationNote": "", "autowired": false, "se [...]
"overrideEndpoint": { "index": 5, "kind": "property", "displayName": "Override Endpoint", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "Set the need for overriding the endpoint. This option needs to be used in [...]
"pojoRequest": { "index": 6, "kind": "property", "displayName": "Pojo Request", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "If we want to use a POJO request as body or not" },
"profileCredentialsName": { "index": 7, "kind": "property", "displayName": "Profile Credentials Name", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "If using a profile credentials provider, this para [...]
@@ -58,13 +58,21 @@
"CamelAwsBedrockStreamOutputMode": { "index": 3, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The streaming output mode (complete or chunks)", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#STREAM_OUTPUT_MODE" },
"CamelAwsBedrockCompletionReason": { "index": 4, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The completion reason for streaming response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#STREAMING_COMPLETION_REASON" },
"CamelAwsBedrockTokenCount": { "index": 5, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The number of tokens generated in streaming response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#STREAMING_TOKEN_COUNT" },
- "CamelAwsBedrockChunkCount": { "index": 6, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The number of chunks received in streaming response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#STREAMING_CHUNK_COUNT" }
+ "CamelAwsBedrockChunkCount": { "index": 6, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The number of chunks received in streaming response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#STREAMING_CHUNK_COUNT" },
+ "CamelAwsBedrockConverseMessages": { "index": 7, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "List<Message>", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The conversation messages for Converse API", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_MESSAGES" },
+ "CamelAwsBedrockConverseSystem": { "index": 8, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "List<SystemContentBlock>", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The system prompts for Converse API", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_SYSTEM" },
+ "CamelAwsBedrockConverseInferenceConfig": { "index": 9, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "InferenceConfiguration", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The inference configuration for Converse API", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_INFERENCE_CONFIG" },
+ "CamelAwsBedrockConverseToolConfig": { "index": 10, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "ToolConfiguration", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The tool configuration for Converse API", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_TOOL_CONFIG" },
+ "CamelAwsBedrockConverseAdditionalFields": { "index": 11, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "software.amazon.awssdk.core.document.Document", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The additional model request fields for Converse API", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_ADDITIONAL_MODEL_REQUEST_FIELDS" },
+ "CamelAwsBedrockConverseStopReason": { "index": 12, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The stop reason from Converse API response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_STOP_REASON" },
+ "CamelAwsBedrockConverseUsage": { "index": 13, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "TokenUsage", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The usage metrics from Converse API response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_USAGE" },
+ "CamelAwsBedrockConverseOutputMessage": { "index": 14, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Message", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The output message from Converse API response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_OUTPUT_MESSAGE" }
},
"properties": {
"label": { "index": 0, "kind": "path", "displayName": "Label", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "Logical name" },
"includeStreamingMetadata": { "index": 1, "kind": "parameter", "displayName": "Include Streaming Metadata", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "Whether to include streaming metadata in the response he [...]
"modelId": { "index": 2, "kind": "parameter", "displayName": "Model Id", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "java.lang.String", "enum": [ "amazon.titan-text-express-v1", "amazon.titan-text-lite-v1", "amazon.titan-image-generator-v1", "amazon.titan-embed-text-v1", "amazon.titan-embed-image-v1", "amazon.titan-text-premier-v1:0", "amazon.titan-embed-text-v2:0", "amazon.titan-image-generator-v2:0", "amazon.nova-canvas-v1:0", "amazon.nova-lite- [...]
- "operation": { "index": 3, "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "org.apache.camel.component.aws2.bedrock.runtime.BedrockOperations", "enum": [ "invokeTextModel", "invokeImageModel", "invokeEmbeddingsModel", "invokeTextModelStreaming", "invokeImageModelStreaming", "invokeEmbeddingsModelStreaming" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationC [...]
+ "operation": { "index": 3, "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "org.apache.camel.component.aws2.bedrock.runtime.BedrockOperations", "enum": [ "invokeTextModel", "invokeImageModel", "invokeEmbeddingsModel", "invokeTextModelStreaming", "invokeImageModelStreaming", "invokeEmbeddingsModelStreaming", "converse", "converseStream" ], "deprecated": false, "deprecationNote": "", "autowired": false, "s [...]
"overrideEndpoint": { "index": 4, "kind": "parameter", "displayName": "Override Endpoint", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "Set the need for overriding the endpoint. This option needs to be used i [...]
"pojoRequest": { "index": 5, "kind": "parameter", "displayName": "Pojo Request", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "If we want to use a POJO request as body or not" },
"profileCredentialsName": { "index": 6, "kind": "parameter", "displayName": "Profile Credentials Name", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "If using a profile credentials provider, this par [...]
diff --git a/components/camel-aws/camel-aws-bedrock/src/generated/resources/META-INF/org/apache/camel/component/aws2/bedrock/runtime/aws-bedrock.json b/components/camel-aws/camel-aws-bedrock/src/generated/resources/META-INF/org/apache/camel/component/aws2/bedrock/runtime/aws-bedrock.json
index 4b75e1b80f4f..c392de4f2189 100644
--- a/components/camel-aws/camel-aws-bedrock/src/generated/resources/META-INF/org/apache/camel/component/aws2/bedrock/runtime/aws-bedrock.json
+++ b/components/camel-aws/camel-aws-bedrock/src/generated/resources/META-INF/org/apache/camel/component/aws2/bedrock/runtime/aws-bedrock.json
@@ -28,7 +28,7 @@
"includeStreamingMetadata": { "index": 1, "kind": "property", "displayName": "Include Streaming Metadata", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "Whether to include streaming metadata in the response hea [...]
"lazyStartProducer": { "index": 2, "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "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 producer may otherwise fail [...]
"modelId": { "index": 3, "kind": "property", "displayName": "Model Id", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "java.lang.String", "enum": [ "amazon.titan-text-express-v1", "amazon.titan-text-lite-v1", "amazon.titan-image-generator-v1", "amazon.titan-embed-text-v1", "amazon.titan-embed-image-v1", "amazon.titan-text-premier-v1:0", "amazon.titan-embed-text-v2:0", "amazon.titan-image-generator-v2:0", "amazon.nova-canvas-v1:0", "amazon.nova-lite-v [...]
- "operation": { "index": 4, "kind": "property", "displayName": "Operation", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "org.apache.camel.component.aws2.bedrock.runtime.BedrockOperations", "enum": [ "invokeTextModel", "invokeImageModel", "invokeEmbeddingsModel", "invokeTextModelStreaming", "invokeImageModelStreaming", "invokeEmbeddingsModelStreaming" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationCl [...]
+ "operation": { "index": 4, "kind": "property", "displayName": "Operation", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "org.apache.camel.component.aws2.bedrock.runtime.BedrockOperations", "enum": [ "invokeTextModel", "invokeImageModel", "invokeEmbeddingsModel", "invokeTextModelStreaming", "invokeImageModelStreaming", "invokeEmbeddingsModelStreaming", "converse", "converseStream" ], "deprecated": false, "deprecationNote": "", "autowired": false, "se [...]
"overrideEndpoint": { "index": 5, "kind": "property", "displayName": "Override Endpoint", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "Set the need for overriding the endpoint. This option needs to be used in [...]
"pojoRequest": { "index": 6, "kind": "property", "displayName": "Pojo Request", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "If we want to use a POJO request as body or not" },
"profileCredentialsName": { "index": 7, "kind": "property", "displayName": "Profile Credentials Name", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "If using a profile credentials provider, this para [...]
@@ -58,13 +58,21 @@
"CamelAwsBedrockStreamOutputMode": { "index": 3, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The streaming output mode (complete or chunks)", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#STREAM_OUTPUT_MODE" },
"CamelAwsBedrockCompletionReason": { "index": 4, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The completion reason for streaming response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#STREAMING_COMPLETION_REASON" },
"CamelAwsBedrockTokenCount": { "index": 5, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The number of tokens generated in streaming response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#STREAMING_TOKEN_COUNT" },
- "CamelAwsBedrockChunkCount": { "index": 6, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The number of chunks received in streaming response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#STREAMING_CHUNK_COUNT" }
+ "CamelAwsBedrockChunkCount": { "index": 6, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The number of chunks received in streaming response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#STREAMING_CHUNK_COUNT" },
+ "CamelAwsBedrockConverseMessages": { "index": 7, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "List<Message>", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The conversation messages for Converse API", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_MESSAGES" },
+ "CamelAwsBedrockConverseSystem": { "index": 8, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "List<SystemContentBlock>", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The system prompts for Converse API", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_SYSTEM" },
+ "CamelAwsBedrockConverseInferenceConfig": { "index": 9, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "InferenceConfiguration", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The inference configuration for Converse API", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_INFERENCE_CONFIG" },
+ "CamelAwsBedrockConverseToolConfig": { "index": 10, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "ToolConfiguration", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The tool configuration for Converse API", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_TOOL_CONFIG" },
+ "CamelAwsBedrockConverseAdditionalFields": { "index": 11, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "software.amazon.awssdk.core.document.Document", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The additional model request fields for Converse API", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_ADDITIONAL_MODEL_REQUEST_FIELDS" },
+ "CamelAwsBedrockConverseStopReason": { "index": 12, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The stop reason from Converse API response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_STOP_REASON" },
+ "CamelAwsBedrockConverseUsage": { "index": 13, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "TokenUsage", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The usage metrics from Converse API response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_USAGE" },
+ "CamelAwsBedrockConverseOutputMessage": { "index": 14, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Message", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The output message from Converse API response", "constantName": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_OUTPUT_MESSAGE" }
},
"properties": {
"label": { "index": 0, "kind": "path", "displayName": "Label", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "Logical name" },
"includeStreamingMetadata": { "index": 1, "kind": "parameter", "displayName": "Include Streaming Metadata", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "Whether to include streaming metadata in the response he [...]
"modelId": { "index": 2, "kind": "parameter", "displayName": "Model Id", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "java.lang.String", "enum": [ "amazon.titan-text-express-v1", "amazon.titan-text-lite-v1", "amazon.titan-image-generator-v1", "amazon.titan-embed-text-v1", "amazon.titan-embed-image-v1", "amazon.titan-text-premier-v1:0", "amazon.titan-embed-text-v2:0", "amazon.titan-image-generator-v2:0", "amazon.nova-canvas-v1:0", "amazon.nova-lite- [...]
- "operation": { "index": 3, "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "org.apache.camel.component.aws2.bedrock.runtime.BedrockOperations", "enum": [ "invokeTextModel", "invokeImageModel", "invokeEmbeddingsModel", "invokeTextModelStreaming", "invokeImageModelStreaming", "invokeEmbeddingsModelStreaming" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationC [...]
+ "operation": { "index": 3, "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "", "required": true, "type": "enum", "javaType": "org.apache.camel.component.aws2.bedrock.runtime.BedrockOperations", "enum": [ "invokeTextModel", "invokeImageModel", "invokeEmbeddingsModel", "invokeTextModelStreaming", "invokeImageModelStreaming", "invokeEmbeddingsModelStreaming", "converse", "converseStream" ], "deprecated": false, "deprecationNote": "", "autowired": false, "s [...]
"overrideEndpoint": { "index": 4, "kind": "parameter", "displayName": "Override Endpoint", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "Set the need for overriding the endpoint. This option needs to be used i [...]
"pojoRequest": { "index": 5, "kind": "parameter", "displayName": "Pojo Request", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "If we want to use a POJO request as body or not" },
"profileCredentialsName": { "index": 6, "kind": "parameter", "displayName": "Profile Credentials Name", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "false", "configurationClass": "org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration", "configurationField": "configuration", "description": "If using a profile credentials provider, this par [...]
diff --git a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockConstants.java b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockConstants.java
index f0df52b81a4b..09929fe76515 100644
--- a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockConstants.java
+++ b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockConstants.java
@@ -36,4 +36,21 @@ public interface BedrockConstants {
String STREAMING_TOKEN_COUNT = "CamelAwsBedrockTokenCount";
@Metadata(description = "The number of chunks received in streaming response", javaType = "Integer")
String STREAMING_CHUNK_COUNT = "CamelAwsBedrockChunkCount";
+ @Metadata(description = "The conversation messages for Converse API", javaType = "List<Message>")
+ String CONVERSE_MESSAGES = "CamelAwsBedrockConverseMessages";
+ @Metadata(description = "The system prompts for Converse API", javaType = "List<SystemContentBlock>")
+ String CONVERSE_SYSTEM = "CamelAwsBedrockConverseSystem";
+ @Metadata(description = "The inference configuration for Converse API", javaType = "InferenceConfiguration")
+ String CONVERSE_INFERENCE_CONFIG = "CamelAwsBedrockConverseInferenceConfig";
+ @Metadata(description = "The tool configuration for Converse API", javaType = "ToolConfiguration")
+ String CONVERSE_TOOL_CONFIG = "CamelAwsBedrockConverseToolConfig";
+ @Metadata(description = "The additional model request fields for Converse API",
+ javaType = "software.amazon.awssdk.core.document.Document")
+ String CONVERSE_ADDITIONAL_MODEL_REQUEST_FIELDS = "CamelAwsBedrockConverseAdditionalFields";
+ @Metadata(description = "The stop reason from Converse API response", javaType = "String")
+ String CONVERSE_STOP_REASON = "CamelAwsBedrockConverseStopReason";
+ @Metadata(description = "The usage metrics from Converse API response", javaType = "TokenUsage")
+ String CONVERSE_USAGE = "CamelAwsBedrockConverseUsage";
+ @Metadata(description = "The output message from Converse API response", javaType = "Message")
+ String CONVERSE_OUTPUT_MESSAGE = "CamelAwsBedrockConverseOutputMessage";
}
diff --git a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockOperations.java b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockOperations.java
index b541f6e4c206..079c82fd0497 100644
--- a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockOperations.java
+++ b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockOperations.java
@@ -28,5 +28,9 @@ public enum BedrockOperations {
invokeImageModelStreaming,
- invokeEmbeddingsModelStreaming
+ invokeEmbeddingsModelStreaming,
+
+ converse,
+
+ converseStream
}
diff --git a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducer.java b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducer.java
index 7ee309ea8983..001ed6bc0c4e 100644
--- a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducer.java
+++ b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducer.java
@@ -35,9 +35,16 @@ import org.slf4j.LoggerFactory;
import software.amazon.awssdk.awscore.exception.AwsServiceException;
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.services.bedrockruntime.BedrockRuntimeClient;
+import software.amazon.awssdk.services.bedrockruntime.model.ContentBlock;
+import software.amazon.awssdk.services.bedrockruntime.model.ConverseRequest;
+import software.amazon.awssdk.services.bedrockruntime.model.ConverseResponse;
+import software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamRequest;
+import software.amazon.awssdk.services.bedrockruntime.model.InferenceConfiguration;
import software.amazon.awssdk.services.bedrockruntime.model.InvokeModelRequest;
import software.amazon.awssdk.services.bedrockruntime.model.InvokeModelResponse;
import software.amazon.awssdk.services.bedrockruntime.model.InvokeModelWithResponseStreamRequest;
+import software.amazon.awssdk.services.bedrockruntime.model.SystemContentBlock;
+import software.amazon.awssdk.services.bedrockruntime.model.ToolConfiguration;
/**
* A Producer which sends messages to the Amazon Bedrock Service <a href="http://aws.amazon.com/bedrock/">AWS
@@ -73,6 +80,12 @@ public class BedrockProducer extends DefaultProducer {
case invokeEmbeddingsModelStreaming:
invokeEmbeddingsModelStreaming(getEndpoint().getBedrockRuntimeClient(), exchange);
break;
+ case converse:
+ converse(getEndpoint().getBedrockRuntimeClient(), exchange);
+ break;
+ case converseStream:
+ converseStream(exchange);
+ break;
default:
throw new IllegalArgumentException("Unsupported operation");
}
@@ -506,6 +519,232 @@ public class BedrockProducer extends DefaultProducer {
message.setHeader(BedrockConstants.STREAMING_CHUNK_COUNT, metadata.getChunkCount());
}
+ private void converse(BedrockRuntimeClient bedrockRuntimeClient, Exchange exchange) throws InvalidPayloadException {
+ ConverseRequest request;
+
+ if (getConfiguration().isPojoRequest()) {
+ Object payload = exchange.getMessage().getMandatoryBody();
+ if (payload instanceof ConverseRequest) {
+ request = (ConverseRequest) payload;
+ } else {
+ throw new IllegalArgumentException(
+ "Converse operation requires ConverseRequest in POJO mode");
+ }
+ } else {
+ // Build request from headers and body
+ ConverseRequest.Builder builder = ConverseRequest.builder();
+
+ // Set model ID
+ builder.modelId(getConfiguration().getModelId());
+
+ // Get messages from header or body
+ @SuppressWarnings("unchecked")
+ List<software.amazon.awssdk.services.bedrockruntime.model.Message> messages
+ = exchange.getMessage().getHeader(BedrockConstants.CONVERSE_MESSAGES, List.class);
+ if (messages != null) {
+ builder.messages(messages);
+ } else {
+ throw new IllegalArgumentException(
+ "Converse operation requires messages in header " + BedrockConstants.CONVERSE_MESSAGES);
+ }
+
+ // Optional: System prompts
+ @SuppressWarnings("unchecked")
+ List<SystemContentBlock> system
+ = exchange.getMessage().getHeader(BedrockConstants.CONVERSE_SYSTEM, List.class);
+ if (system != null) {
+ builder.system(system);
+ }
+
+ // Optional: Inference configuration
+ InferenceConfiguration inferenceConfig
+ = exchange.getMessage().getHeader(BedrockConstants.CONVERSE_INFERENCE_CONFIG, InferenceConfiguration.class);
+ if (inferenceConfig != null) {
+ builder.inferenceConfig(inferenceConfig);
+ }
+
+ // Optional: Tool configuration
+ ToolConfiguration toolConfig
+ = exchange.getMessage().getHeader(BedrockConstants.CONVERSE_TOOL_CONFIG, ToolConfiguration.class);
+ if (toolConfig != null) {
+ builder.toolConfig(toolConfig);
+ }
+
+ // Optional: Additional model request fields
+ software.amazon.awssdk.core.document.Document additionalFields = exchange.getMessage()
+ .getHeader(BedrockConstants.CONVERSE_ADDITIONAL_MODEL_REQUEST_FIELDS,
+ software.amazon.awssdk.core.document.Document.class);
+ if (additionalFields != null) {
+ builder.additionalModelRequestFields(additionalFields);
+ }
+
+ request = builder.build();
+ }
+
+ try {
+ ConverseResponse response = bedrockRuntimeClient.converse(request);
+
+ org.apache.camel.Message message = getMessageForResponse(exchange);
+
+ // Set the output message content as body
+ if (response.output() != null && response.output().message() != null) {
+ software.amazon.awssdk.services.bedrockruntime.model.Message outputMessage = response.output().message();
+ message.setHeader(BedrockConstants.CONVERSE_OUTPUT_MESSAGE, outputMessage);
+
+ // Extract text content from the message
+ StringBuilder textContent = new StringBuilder();
+ for (ContentBlock content : outputMessage.content()) {
+ if (content.text() != null) {
+ textContent.append(content.text());
+ }
+ }
+ message.setBody(textContent.toString());
+ }
+
+ // Set metadata headers
+ if (response.stopReason() != null) {
+ message.setHeader(BedrockConstants.CONVERSE_STOP_REASON, response.stopReason().toString());
+ }
+ if (response.usage() != null) {
+ message.setHeader(BedrockConstants.CONVERSE_USAGE, response.usage());
+ }
+
+ } catch (AwsServiceException ase) {
+ LOG.trace("Converse command returned the error code {}", ase.awsErrorDetails().errorCode());
+ throw ase;
+ }
+ }
+
+ private void converseStream(Exchange exchange) throws InvalidPayloadException {
+ ConverseStreamRequest request;
+
+ if (getConfiguration().isPojoRequest()) {
+ Object payload = exchange.getMessage().getMandatoryBody();
+ if (payload instanceof ConverseStreamRequest) {
+ request = (ConverseStreamRequest) payload;
+ } else {
+ throw new IllegalArgumentException(
+ "ConverseStream operation requires ConverseStreamRequest in POJO mode");
+ }
+ } else {
+ // Build request from headers and body
+ ConverseStreamRequest.Builder builder = ConverseStreamRequest.builder();
+
+ // Set model ID
+ builder.modelId(getConfiguration().getModelId());
+
+ // Get messages from header or body
+ @SuppressWarnings("unchecked")
+ List<software.amazon.awssdk.services.bedrockruntime.model.Message> messages
+ = exchange.getMessage().getHeader(BedrockConstants.CONVERSE_MESSAGES, List.class);
+ if (messages != null) {
+ builder.messages(messages);
+ } else {
+ throw new IllegalArgumentException(
+ "ConverseStream operation requires messages in header " + BedrockConstants.CONVERSE_MESSAGES);
+ }
+
+ // Optional: System prompts
+ @SuppressWarnings("unchecked")
+ List<SystemContentBlock> system
+ = exchange.getMessage().getHeader(BedrockConstants.CONVERSE_SYSTEM, List.class);
+ if (system != null) {
+ builder.system(system);
+ }
+
+ // Optional: Inference configuration
+ InferenceConfiguration inferenceConfig
+ = exchange.getMessage().getHeader(BedrockConstants.CONVERSE_INFERENCE_CONFIG, InferenceConfiguration.class);
+ if (inferenceConfig != null) {
+ builder.inferenceConfig(inferenceConfig);
+ }
+
+ // Optional: Tool configuration
+ ToolConfiguration toolConfig
+ = exchange.getMessage().getHeader(BedrockConstants.CONVERSE_TOOL_CONFIG, ToolConfiguration.class);
+ if (toolConfig != null) {
+ builder.toolConfig(toolConfig);
+ }
+
+ // Optional: Additional model request fields
+ software.amazon.awssdk.core.document.Document additionalFields = exchange.getMessage()
+ .getHeader(BedrockConstants.CONVERSE_ADDITIONAL_MODEL_REQUEST_FIELDS,
+ software.amazon.awssdk.core.document.Document.class);
+ if (additionalFields != null) {
+ builder.additionalModelRequestFields(additionalFields);
+ }
+
+ request = builder.build();
+ }
+
+ processConverseStreamingRequest(request, exchange);
+ }
+
+ private void processConverseStreamingRequest(ConverseStreamRequest request, Exchange exchange) {
+ try {
+ String streamOutputMode = getConfiguration().getStreamOutputMode();
+ if (streamOutputMode == null) {
+ streamOutputMode = "complete";
+ }
+
+ // Check if mode is overridden in headers
+ if (ObjectHelper.isNotEmpty(exchange.getMessage().getHeader(BedrockConstants.STREAM_OUTPUT_MODE))) {
+ streamOutputMode = exchange.getIn().getHeader(BedrockConstants.STREAM_OUTPUT_MODE, String.class);
+ }
+
+ org.apache.camel.Message message = getMessageForResponse(exchange);
+ org.apache.camel.component.aws2.bedrock.runtime.stream.ConverseStreamHandler.StreamMetadata metadata
+ = new org.apache.camel.component.aws2.bedrock.runtime.stream.ConverseStreamHandler.StreamMetadata();
+
+ if ("chunks".equals(streamOutputMode)) {
+ // Chunks mode - emit each chunk as separate message
+ List<String> allChunks = new ArrayList<>();
+ getEndpoint().getBedrockRuntimeAsyncClient().converseStream(
+ request,
+ org.apache.camel.component.aws2.bedrock.runtime.stream.ConverseStreamHandler.createChunksHandler(
+ metadata,
+ allChunks,
+ null))
+ .join();
+
+ message.setBody(allChunks);
+ if (getConfiguration().isIncludeStreamingMetadata()) {
+ setConverseStreamingMetadata(message, metadata);
+ }
+ } else {
+ // Complete mode - accumulate all chunks and return complete response
+ StringBuilder fullText = new StringBuilder();
+ getEndpoint().getBedrockRuntimeAsyncClient().converseStream(
+ request,
+ org.apache.camel.component.aws2.bedrock.runtime.stream.ConverseStreamHandler.createCompleteHandler(
+ metadata,
+ fullText))
+ .join();
+
+ message.setBody(fullText.toString());
+ if (getConfiguration().isIncludeStreamingMetadata()) {
+ setConverseStreamingMetadata(message, metadata);
+ }
+ }
+
+ } catch (AwsServiceException ase) {
+ LOG.trace("Converse Stream command returned the error code {}", ase.awsErrorDetails().errorCode());
+ throw ase;
+ }
+ }
+
+ private void setConverseStreamingMetadata(
+ org.apache.camel.Message message,
+ org.apache.camel.component.aws2.bedrock.runtime.stream.ConverseStreamHandler.StreamMetadata metadata) {
+ if (metadata.getStopReason() != null) {
+ message.setHeader(BedrockConstants.CONVERSE_STOP_REASON, metadata.getStopReason());
+ }
+ if (metadata.getUsage() != null) {
+ message.setHeader(BedrockConstants.CONVERSE_USAGE, metadata.getUsage());
+ }
+ message.setHeader(BedrockConstants.STREAMING_CHUNK_COUNT, metadata.getChunkCount());
+ }
+
public static Message getMessageForResponse(final Exchange exchange) {
return exchange.getMessage();
}
diff --git a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandler.java b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandler.java
new file mode 100644
index 000000000000..1b242d5481a5
--- /dev/null
+++ b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandler.java
@@ -0,0 +1,172 @@
+/*
+ * 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.aws2.bedrock.runtime.stream;
+
+import java.util.List;
+import java.util.function.Consumer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamResponseHandler;
+import software.amazon.awssdk.services.bedrockruntime.model.TokenUsage;
+
+/**
+ * Utility class for handling Converse API streaming responses from Bedrock models
+ */
+public final class ConverseStreamHandler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ConverseStreamHandler.class);
+
+ private ConverseStreamHandler() {
+ // Utility class
+ }
+
+ /**
+ * Create a response handler for complete mode (accumulates all chunks)
+ *
+ * @param metadata the metadata object to populate
+ * @param fullText the string builder to accumulate text
+ * @return the response handler
+ */
+ public static ConverseStreamResponseHandler createCompleteHandler(
+ StreamMetadata metadata,
+ StringBuilder fullText) {
+
+ int[] chunkCount = { 0 };
+
+ return ConverseStreamResponseHandler.builder()
+ .subscriber(ConverseStreamResponseHandler.Visitor.builder()
+ .onContentBlockDelta(delta -> {
+ if (delta.delta() != null && delta.delta().text() != null) {
+ fullText.append(delta.delta().text());
+ }
+ chunkCount[0]++;
+ })
+ .onMetadata(metadataEvent -> {
+ if (metadataEvent.usage() != null) {
+ metadata.setUsage(metadataEvent.usage());
+ }
+ })
+ .onMessageStop(stop -> {
+ if (stop.stopReason() != null) {
+ metadata.setStopReason(stop.stopReason().toString());
+ }
+ })
+ .build())
+ .onComplete(() -> {
+ metadata.setChunkCount(chunkCount[0]);
+ metadata.setFullText(fullText.toString());
+ })
+ .build();
+ }
+
+ /**
+ * Create a response handler for chunks mode (emits each chunk)
+ *
+ * @param metadata the metadata object to populate
+ * @param chunks the list to collect chunks
+ * @param chunkConsumer consumer that receives each chunk
+ * @return the response handler
+ */
+ public static ConverseStreamResponseHandler createChunksHandler(
+ StreamMetadata metadata,
+ List<String> chunks,
+ Consumer<String> chunkConsumer) {
+
+ int[] chunkCount = { 0 };
+
+ return ConverseStreamResponseHandler.builder()
+ .subscriber(ConverseStreamResponseHandler.Visitor.builder()
+ .onContentBlockDelta(delta -> {
+ if (delta.delta() != null && delta.delta().text() != null) {
+ String text = delta.delta().text();
+ chunks.add(text);
+ if (chunkConsumer != null) {
+ chunkConsumer.accept(text);
+ }
+ }
+ chunkCount[0]++;
+ })
+ .onMetadata(metadataEvent -> {
+ if (metadataEvent.usage() != null) {
+ metadata.setUsage(metadataEvent.usage());
+ }
+ })
+ .onMessageStop(stop -> {
+ if (stop.stopReason() != null) {
+ metadata.setStopReason(stop.stopReason().toString());
+ }
+ })
+ .build())
+ .onComplete(() -> {
+ metadata.setChunkCount(chunkCount[0]);
+ metadata.setChunks(chunks);
+ })
+ .build();
+ }
+
+ /**
+ * Metadata extracted from Converse streaming response
+ */
+ public static class StreamMetadata {
+ private String fullText;
+ private List<String> chunks;
+ private String stopReason;
+ private TokenUsage usage;
+ private int chunkCount;
+
+ public String getFullText() {
+ return fullText;
+ }
+
+ public void setFullText(String fullText) {
+ this.fullText = fullText;
+ }
+
+ public List<String> getChunks() {
+ return chunks;
+ }
+
+ public void setChunks(List<String> chunks) {
+ this.chunks = chunks;
+ }
+
+ public String getStopReason() {
+ return stopReason;
+ }
+
+ public void setStopReason(String stopReason) {
+ this.stopReason = stopReason;
+ }
+
+ public TokenUsage getUsage() {
+ return usage;
+ }
+
+ public void setUsage(TokenUsage usage) {
+ this.usage = usage;
+ }
+
+ public int getChunkCount() {
+ return chunkCount;
+ }
+
+ public void setChunkCount(int chunkCount) {
+ this.chunkCount = chunkCount;
+ }
+ }
+}
diff --git a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/integration/BedrockProducerIT.java b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/integration/BedrockProducerIT.java
index ce51d3f7ef1d..6455c673a2b8 100644
--- a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/integration/BedrockProducerIT.java
+++ b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/integration/BedrockProducerIT.java
@@ -26,6 +26,7 @@ import org.apache.camel.component.aws2.bedrock.BedrockModels;
import org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.condition.EnabledIfSystemProperties;
@@ -45,6 +46,11 @@ class BedrockProducerIT extends CamelTestSupport {
@EndpointInject("mock:result")
private MockEndpoint result;
+ @BeforeEach
+ public void resetMocks() {
+ result.reset();
+ }
+
@Test
public void testInvokeTitanExpressModel() throws InterruptedException {
@@ -867,6 +873,59 @@ class BedrockProducerIT extends CamelTestSupport {
MockEndpoint.assertIsSatisfied(context);
}
+ @Test
+ public void testConverseWithClaudeModel() throws InterruptedException {
+ result.expectedMessageCount(1);
+ final Exchange result = template.send("direct:converse_claude", exchange -> {
+ // Create a message using the Converse API
+ java.util.List<software.amazon.awssdk.services.bedrockruntime.model.Message> messages = new java.util.ArrayList<>();
+ messages.add(software.amazon.awssdk.services.bedrockruntime.model.Message.builder()
+ .role(software.amazon.awssdk.services.bedrockruntime.model.ConversationRole.USER)
+ .content(software.amazon.awssdk.services.bedrockruntime.model.ContentBlock
+ .fromText("What is the capital of France?"))
+ .build());
+
+ exchange.getMessage().setHeader(BedrockConstants.CONVERSE_MESSAGES, messages);
+
+ // Optional: Add inference configuration
+ software.amazon.awssdk.services.bedrockruntime.model.InferenceConfiguration inferenceConfig
+ = software.amazon.awssdk.services.bedrockruntime.model.InferenceConfiguration.builder()
+ .maxTokens(100)
+ .temperature(0.7f)
+ .build();
+ exchange.getMessage().setHeader(BedrockConstants.CONVERSE_INFERENCE_CONFIG, inferenceConfig);
+ });
+
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ @Test
+ public void testConverseStreamWithClaudeModel() throws InterruptedException {
+ result.expectedMessageCount(1);
+ final Exchange result = template.send("direct:converse_stream_claude", exchange -> {
+ // Create a message using the Converse API
+ java.util.List<software.amazon.awssdk.services.bedrockruntime.model.Message> messages = new java.util.ArrayList<>();
+ messages.add(software.amazon.awssdk.services.bedrockruntime.model.Message.builder()
+ .role(software.amazon.awssdk.services.bedrockruntime.model.ConversationRole.USER)
+ .content(software.amazon.awssdk.services.bedrockruntime.model.ContentBlock
+ .fromText("Tell me a short joke about Java programming"))
+ .build());
+
+ exchange.getMessage().setHeader(BedrockConstants.CONVERSE_MESSAGES, messages);
+ exchange.getMessage().setHeader(BedrockConstants.STREAM_OUTPUT_MODE, "complete");
+
+ // Optional: Add inference configuration
+ software.amazon.awssdk.services.bedrockruntime.model.InferenceConfiguration inferenceConfig
+ = software.amazon.awssdk.services.bedrockruntime.model.InferenceConfiguration.builder()
+ .maxTokens(200)
+ .temperature(0.9f)
+ .build();
+ exchange.getMessage().setHeader(BedrockConstants.CONVERSE_INFERENCE_CONFIG, inferenceConfig);
+ });
+
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -1057,6 +1116,18 @@ class BedrockProducerIT extends CamelTestSupport {
+ BedrockModels.MISTRAL_SMALL_2402.model)
.log("Completions: ${body}")
.to(result);
+
+ from("direct:converse_claude")
+ .to("aws-bedrock:label?accessKey=RAW({{aws.manual.access.key}})&secretKey=RAW({{aws.manual.secret.key}})®ion=us-east-1&operation=converse&modelId="
+ + BedrockModels.ANTROPHIC_CLAUDE_V3.model)
+ .log("Converse response: ${body}")
+ .to(result);
+
+ from("direct:converse_stream_claude")
+ .to("aws-bedrock:label?accessKey=RAW({{aws.manual.access.key}})&secretKey=RAW({{aws.manual.secret.key}})®ion=us-east-1&operation=converseStream&modelId="
+ + BedrockModels.ANTROPHIC_CLAUDE_V3.model)
+ .log("Converse stream response: ${body}")
+ .to(result);
}
};
}
diff --git a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/integration/BedrockProducerStreamingIT.java b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/integration/BedrockProducerStreamingIT.java
index 6efe1bc9a7db..62e189dbe5b8 100644
--- a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/integration/BedrockProducerStreamingIT.java
+++ b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/integration/BedrockProducerStreamingIT.java
@@ -61,10 +61,8 @@ class BedrockProducerStreamingIT extends CamelTestSupport {
private MockEndpoint result;
@BeforeEach
- public void resetMocks() throws InterruptedException {
+ public void resetMocks() {
result.reset();
- // Add small delay to avoid AWS rate limiting between tests
- Thread.sleep(1000);
}
@Test
diff --git a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandlerTest.java b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandlerTest.java
new file mode 100644
index 000000000000..a07aa1a728ee
--- /dev/null
+++ b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandlerTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.aws2.bedrock.runtime.stream;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamResponseHandler;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+class ConverseStreamHandlerTest {
+
+ @Test
+ void testCreateCompleteHandler() {
+ ConverseStreamHandler.StreamMetadata metadata = new ConverseStreamHandler.StreamMetadata();
+ StringBuilder fullText = new StringBuilder();
+
+ ConverseStreamResponseHandler handler = ConverseStreamHandler.createCompleteHandler(metadata, fullText);
+
+ assertNotNull(handler, "Handler should not be null");
+ assertNotNull(metadata, "Metadata should not be null");
+ assertNotNull(fullText, "Full text builder should not be null");
+ }
+
+ @Test
+ void testCreateChunksHandler() {
+ ConverseStreamHandler.StreamMetadata metadata = new ConverseStreamHandler.StreamMetadata();
+ List<String> chunks = new ArrayList<>();
+
+ ConverseStreamResponseHandler handler
+ = ConverseStreamHandler.createChunksHandler(metadata, chunks, null);
+
+ assertNotNull(handler, "Handler should not be null");
+ assertNotNull(metadata, "Metadata should not be null");
+ assertNotNull(chunks, "Chunks list should not be null");
+ }
+
+ @Test
+ void testStreamMetadata() {
+ ConverseStreamHandler.StreamMetadata metadata = new ConverseStreamHandler.StreamMetadata();
+
+ // Test setting and getting fullText
+ metadata.setFullText("Test response");
+ assertEquals("Test response", metadata.getFullText());
+
+ // Test setting and getting chunks
+ List<String> chunks = List.of("chunk1", "chunk2");
+ metadata.setChunks(chunks);
+ assertEquals(chunks, metadata.getChunks());
+
+ // Test setting and getting stopReason
+ metadata.setStopReason("end_turn");
+ assertEquals("end_turn", metadata.getStopReason());
+
+ // Test setting and getting chunkCount
+ metadata.setChunkCount(5);
+ assertEquals(5, metadata.getChunkCount());
+
+ // Test usage is null initially
+ assertEquals(null, metadata.getUsage());
+ }
+}
diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BedrockEndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BedrockEndpointBuilderFactory.java
index aca85d17480b..8cdb93811d2f 100644
--- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BedrockEndpointBuilderFactory.java
+++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BedrockEndpointBuilderFactory.java
@@ -772,6 +772,106 @@ public interface BedrockEndpointBuilderFactory {
public String awsBedrockChunkCount() {
return "CamelAwsBedrockChunkCount";
}
+ /**
+ * The conversation messages for Converse API.
+ *
+ * The option is a: {@code List<Message>} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code AwsBedrockConverseMessages}.
+ */
+ public String awsBedrockConverseMessages() {
+ return "CamelAwsBedrockConverseMessages";
+ }
+ /**
+ * The system prompts for Converse API.
+ *
+ * The option is a: {@code List<SystemContentBlock>} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code AwsBedrockConverseSystem}.
+ */
+ public String awsBedrockConverseSystem() {
+ return "CamelAwsBedrockConverseSystem";
+ }
+ /**
+ * The inference configuration for Converse API.
+ *
+ * The option is a: {@code InferenceConfiguration} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code
+ * AwsBedrockConverseInferenceConfig}.
+ */
+ public String awsBedrockConverseInferenceConfig() {
+ return "CamelAwsBedrockConverseInferenceConfig";
+ }
+ /**
+ * The tool configuration for Converse API.
+ *
+ * The option is a: {@code ToolConfiguration} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code AwsBedrockConverseToolConfig}.
+ */
+ public String awsBedrockConverseToolConfig() {
+ return "CamelAwsBedrockConverseToolConfig";
+ }
+ /**
+ * The additional model request fields for Converse API.
+ *
+ * The option is a: {@code
+ * software.amazon.awssdk.core.document.Document} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code
+ * AwsBedrockConverseAdditionalFields}.
+ */
+ public String awsBedrockConverseAdditionalFields() {
+ return "CamelAwsBedrockConverseAdditionalFields";
+ }
+ /**
+ * The stop reason from Converse API response.
+ *
+ * The option is a: {@code String} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code AwsBedrockConverseStopReason}.
+ */
+ public String awsBedrockConverseStopReason() {
+ return "CamelAwsBedrockConverseStopReason";
+ }
+ /**
+ * The usage metrics from Converse API response.
+ *
+ * The option is a: {@code TokenUsage} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code AwsBedrockConverseUsage}.
+ */
+ public String awsBedrockConverseUsage() {
+ return "CamelAwsBedrockConverseUsage";
+ }
+ /**
+ * The output message from Converse API response.
+ *
+ * The option is a: {@code Message} type.
+ *
+ * Group: producer
+ *
+ * @return the name of the header {@code
+ * AwsBedrockConverseOutputMessage}.
+ */
+ public String awsBedrockConverseOutputMessage() {
+ return "CamelAwsBedrockConverseOutputMessage";
+ }
}
static BedrockEndpointBuilder endpointBuilder(String componentName, String path) {
class BedrockEndpointBuilderImpl extends AbstractEndpointBuilder implements BedrockEndpointBuilder, AdvancedBedrockEndpointBuilder {