fix(bedrock): normalize 3gp video format

Author: he-yufeng

src/strands/models/bedrock.py

@@ -43,6 +43,12 @@
 _DEFAULT_BEDROCK_MODEL_ID = "{}.anthropic.claude-sonnet-4-6"
 DEFAULT_BEDROCK_REGION = "us-west-2"
 
+_BEDROCK_VIDEO_FORMAT_ALIASES = {
+    "3gp": "three_gp",
+    "3g2": "three_gp",
+    "3gpp": "three_gp",
+}
+
 BEDROCK_CONTEXT_WINDOW_OVERFLOW_MESSAGES = [
     "Input is too long for requested model",
     "input length and `max_tokens` exceed context limit",
@@ -702,7 +708,8 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An
                     return None
             elif "bytes" in source:
                 formatted_video_source = {"bytes": source["bytes"]}
-            result = {"format": video["format"], "source": formatted_video_source}
+            video_format = _BEDROCK_VIDEO_FORMAT_ALIASES.get(video["format"], video["format"])
+            result = {"format": video_format, "source": formatted_video_source}
             return {"video": result}
 
         # https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_CitationsContentBlock.html
@@ -852,13 +859,9 @@ async def count_tokens(
             logger.debug("model_id=<%s>, total_tokens=<%d> | native token count", self.config["model_id"], total_tokens)
             return total_tokens
         except Exception as e:
-            if (
-                isinstance(e, ClientError)
-                and e.response.get("Error", {}).get("Code") == "AccessDeniedException"
-            ):
+            if isinstance(e, ClientError) and e.response.get("Error", {}).get("Code") == "AccessDeniedException":
                 logger.warning(
-                    "model_id=<%s> | bedrock:CountTokens permission denied,"
-                    " falling back to heuristic estimation: %s",
+                    "model_id=<%s> | bedrock:CountTokens permission denied, falling back to heuristic estimation: %s",
                     model_id,
                     e,
                 )

tests/strands/models/test_bedrock.py

@@ -2254,6 +2254,28 @@ def test_format_request_video_s3_location(model, model_id):
     assert video_source == {"s3Location": {"uri": "s3://my-bucket/video.mp4"}}
 
 
+@pytest.mark.parametrize("video_format", ["3gp", "3g2", "3gpp"])
+def test_format_request_maps_3gp_video_formats(model, model_id, video_format):
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {
+                    "video": {
+                        "format": video_format,
+                        "source": {"bytes": b"video_data"},
+                    }
+                },
+            ],
+        }
+    ]
+
+    formatted_request = model._format_request(messages)
+
+    video_block = formatted_request["messages"][0]["content"][0]["video"]
+    assert video_block == {"format": "three_gp", "source": {"bytes": b"video_data"}}
+
+
 def test_format_request_filters_document_content_blocks(model, model_id):
     """Test that format_request filters extra fields from document content blocks."""
     messages = [
@@ -3505,8 +3527,7 @@ async def test_access_denied_logs_warning_with_full_error(
         self, model_with_client, bedrock_client, messages, caplog
     ):
         error_message = (
-            "User: arn:aws:sts::123456789012:assumed-role/role is not authorized"
-            " to perform: bedrock:CountTokens"
+            "User: arn:aws:sts::123456789012:assumed-role/role is not authorized to perform: bedrock:CountTokens"
         )
         bedrock_client.count_tokens.side_effect = ClientError(
             {"Error": {"Code": "AccessDeniedException", "Message": error_message}},