From a7b8700b48cc6a2722c6f61fd9c422b5e6617a8e Mon Sep 17 00:00:00 2001 From: Zelys Date: Sun, 5 Apr 2026 16:29:29 -0500 Subject: [PATCH] fix(bedrock): skip S3 location sources for models that do not support them S3 location sources for documents, images, and videos are only supported by Amazon Nova models. Other model families (Anthropic Claude, Meta Llama, Mistral, Cohere) reject them with a ValidationException. This adds a `_MODELS_SUPPORT_S3_LOCATION` capability constant, a `_supports_s3_location` property, and updates `_handle_location` to warn and skip the content block when the current model does not support S3 locations. Fixes #1744 --- src/strands/models/bedrock.py | 33 +++++++++++++- tests/strands/models/test_bedrock.py | 67 ++++++++++++++++++++++++---- 2 files changed, 90 insertions(+), 10 deletions(-) diff --git a/src/strands/models/bedrock.py b/src/strands/models/bedrock.py index 5b7a2f34e..a1d832a63 100644 --- a/src/strands/models/bedrock.py +++ b/src/strands/models/bedrock.py @@ -52,6 +52,13 @@ "anthropic.claude", ] +# Models that support S3 location sources for documents, images, and videos in the Bedrock Converse API. +# Other model families (Anthropic Claude, Meta Llama, Mistral, Cohere, etc.) only accept inline bytes +# and reject S3 location sources with a ValidationException. +_MODELS_SUPPORT_S3_LOCATION = [ + "amazon.nova", +] + T = TypeVar("T", bound=BaseModel) DEFAULT_READ_TIMEOUT = 120 @@ -492,9 +499,33 @@ def _should_include_tool_result_status(self) -> bool: else: # "auto" return any(model in self.config["model_id"] for model in _MODELS_INCLUDE_STATUS) + @property + def _supports_s3_location(self) -> bool: + """Whether this model supports S3 location sources in the Bedrock Converse API. + + Only Amazon Nova models support S3 location sources for documents, images, and videos. + Other families (Anthropic Claude, Meta Llama, Mistral, Cohere, etc.) only accept inline + bytes and reject S3 location sources with a ValidationException. + + Returns: + True if the model supports S3 location sources, False otherwise. + """ + model_id = self.config.get("model_id", "").lower() + return any(prefix in model_id for prefix in _MODELS_SUPPORT_S3_LOCATION) + def _handle_location(self, location: SourceLocation) -> dict[str, Any] | None: - """Convert location content block to Bedrock format if its an S3Location.""" + """Convert location content block to Bedrock format if its an S3Location. + + Returns None if the location type is unsupported or if the current model does not + support S3 location sources. + """ if location["type"] == "s3": + if not self._supports_s3_location: + logger.warning( + "model_id=<%s> | S3 location sources are not supported by this model; skipping content block", + self.config.get("model_id"), + ) + return None s3_location = cast(S3Location, location) formatted_document_s3: dict[str, Any] = {"uri": s3_location["uri"]} if "bucketOwner" in s3_location: diff --git a/tests/strands/models/test_bedrock.py b/tests/strands/models/test_bedrock.py index 9c565d4f4..6706840af 100644 --- a/tests/strands/models/test_bedrock.py +++ b/tests/strands/models/test_bedrock.py @@ -1779,8 +1779,10 @@ def test_format_request_filters_image_content_blocks(model, model_id): assert "metadata" not in image_block -def test_format_request_image_s3_location_only(model, model_id): - """Test that image with only s3Location is properly formatted.""" +def test_format_request_image_s3_location_only(bedrock_client): + """Test that image with only s3Location is properly formatted for Nova models.""" + nova_model_id = "amazon.nova-pro-v1:0" + nova_model = BedrockModel(model_id=nova_model_id) messages = [ { "role": "user", @@ -1797,7 +1799,7 @@ def test_format_request_image_s3_location_only(model, model_id): } ] - formatted_request = model._format_request(messages) + formatted_request = nova_model._format_request(messages) image_source = formatted_request["messages"][0]["content"][0]["image"]["source"] assert image_source == {"s3Location": {"uri": "s3://my-bucket/image.png"}} @@ -1825,8 +1827,10 @@ def test_format_request_image_bytes_only(model, model_id): assert image_source == {"bytes": b"image_data"} -def test_format_request_document_s3_location(model, model_id): - """Test that document with s3Location is properly formatted.""" +def test_format_request_document_s3_location(bedrock_client): + """Test that document with s3Location is properly formatted for Nova models.""" + nova_model_id = "amazon.nova-pro-v1:0" + nova_model = BedrockModel(model_id=nova_model_id) messages = [ { "role": "user", @@ -1857,7 +1861,7 @@ def test_format_request_document_s3_location(model, model_id): } ] - formatted_request = model._format_request(messages) + formatted_request = nova_model._format_request(messages) document = formatted_request["messages"][0]["content"][0]["document"] document_with_bucket_owner = formatted_request["messages"][0]["content"][1]["document"] @@ -1918,8 +1922,10 @@ def test_format_request_unsupported_location(model, caplog): assert "Non s3 location sources are not supported by Bedrock | skipping content block" in caplog.text -def test_format_request_video_s3_location(model, model_id): - """Test that video with s3Location is properly formatted.""" +def test_format_request_video_s3_location(bedrock_client): + """Test that video with s3Location is properly formatted for Nova models.""" + nova_model_id = "amazon.nova-pro-v1:0" + nova_model = BedrockModel(model_id=nova_model_id) messages = [ { "role": "user", @@ -1936,12 +1942,55 @@ def test_format_request_video_s3_location(model, model_id): } ] - formatted_request = model._format_request(messages) + formatted_request = nova_model._format_request(messages) video_source = formatted_request["messages"][0]["content"][0]["video"]["source"] assert video_source == {"s3Location": {"uri": "s3://my-bucket/video.mp4"}} +def test_format_request_s3_location_skipped_for_unsupported_models(bedrock_client, model_id, caplog): + """S3 location sources are skipped with a warning for models that do not support them.""" + caplog.set_level(logging.WARNING, logger="strands.models.bedrock") + + # model_id fixture is "m1" (not an Amazon Nova model) + non_nova_model = BedrockModel(model_id=model_id) + messages = [ + { + "role": "user", + "content": [ + {"text": "analyze this"}, + { + "document": { + "name": "report.pdf", + "format": "pdf", + "source": {"location": {"type": "s3", "uri": "s3://my-bucket/report.pdf"}}, + } + }, + { + "image": { + "format": "png", + "source": {"location": {"type": "s3", "uri": "s3://my-bucket/image.png"}}, + } + }, + { + "video": { + "format": "mp4", + "source": {"location": {"type": "s3", "uri": "s3://my-bucket/video.mp4"}}, + } + }, + ], + } + ] + + formatted_request = non_nova_model._format_request(messages) + content = formatted_request["messages"][0]["content"] + + # Only the text block should remain; all three S3 location blocks are dropped + assert len(content) == 1 + assert content[0] == {"text": "analyze this"} + assert "S3 location sources are not supported by this model" in caplog.text + + def test_format_request_filters_document_content_blocks(model, model_id): """Test that format_request filters extra fields from document content blocks.""" messages = [