Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion strands-py/src/strands/models/anthropic.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,14 @@

T = TypeVar("T", bound=BaseModel)

_IMAGE_MEDIA_TYPES = {
"gif": "image/gif",
"jpeg": "image/jpeg",
"jpg": "image/jpeg",
"png": "image/png",
"webp": "image/webp",
}


class AnthropicModel(Model):
"""Anthropic model provider implementation."""
Expand Down Expand Up @@ -131,10 +139,13 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An
}

if "image" in content:
image_format = content["image"]["format"]
return {
"source": {
"data": base64.b64encode(content["image"]["source"]["bytes"]).decode("utf-8"),
"media_type": mimetypes.types_map.get(f".{content['image']['format']}", "application/octet-stream"),
"media_type": _IMAGE_MEDIA_TYPES.get(
image_format.lower(), mimetypes.types_map.get(f".{image_format}", "application/octet-stream")
),
"type": "base64",
},
"type": "image",
Expand Down
6 changes: 5 additions & 1 deletion strands-py/src/strands/models/bedrock.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
from .model import BaseModelConfig, CacheConfig, CacheToolsConfig, Model

logger = logging.getLogger(__name__)
_VIDEO_FORMAT_ALIASES = {"3gp": "three_gp"}

# See: `BedrockModel._get_default_model_with_warning` for why we need both
DEFAULT_BEDROCK_MODEL_ID = "global.anthropic.claude-sonnet-4-6"
Expand Down Expand Up @@ -702,7 +703,10 @@ 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}
result = {
"format": _VIDEO_FORMAT_ALIASES.get(video["format"], video["format"]),
"source": formatted_video_source,
}
return {"video": result}

# https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_CitationsContentBlock.html
Expand Down
2 changes: 1 addition & 1 deletion strands-py/src/strands/types/media.py
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,7 @@ class ImageContent(TypedDict):
source: ImageSource


VideoFormat = Literal["flv", "mkv", "mov", "mpeg", "mpg", "mp4", "three_gp", "webm", "wmv"]
VideoFormat = Literal["flv", "mkv", "mov", "mpeg", "mpg", "mp4", "3gp", "three_gp", "webm", "wmv"]
"""Supported video formats."""


Expand Down
22 changes: 22 additions & 0 deletions strands-py/tests/strands/models/test_anthropic.py
Original file line number Diff line number Diff line change
Expand Up @@ -254,6 +254,28 @@ def test_format_request_with_image(model, model_id, max_tokens):
assert tru_request == exp_request


def test_format_request_with_webp_image_uses_explicit_media_type(model, monkeypatch):
monkeypatch.delitem(strands.models.anthropic.mimetypes.types_map, ".webp", raising=False)

messages = [
{
"role": "user",
"content": [
{
"image": {
"format": "webp",
"source": {"bytes": b"webpimage"},
},
},
],
},
]

tru_request = model.format_request(messages)

assert tru_request["messages"][0]["content"][0]["source"]["media_type"] == "image/webp"


def test_format_request_with_reasoning(model, model_id, max_tokens):
messages = [
{
Expand Down
21 changes: 21 additions & 0 deletions strands-py/tests/strands/models/test_bedrock.py
Original file line number Diff line number Diff line change
Expand Up @@ -2254,6 +2254,27 @@ def test_format_request_video_s3_location(model, model_id):
assert video_source == {"s3Location": {"uri": "s3://my-bucket/video.mp4"}}


def test_format_request_video_3gp_maps_bedrock_enum(model, model_id):
messages = [
{
"role": "user",
"content": [
{
"video": {
"format": "3gp",
"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 = [
Expand Down