fix(bedrock): skip S3 location sources for models that do not support them

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:

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