diff --git a/src/strands/models/bedrock.py b/src/strands/models/bedrock.py index a3cea7cfe..b053b70fb 100644 --- a/src/strands/models/bedrock.py +++ b/src/strands/models/bedrock.py @@ -17,6 +17,8 @@ from pydantic import BaseModel from typing_extensions import TypedDict, Unpack, override +from strands.types.media import S3Location, SourceLocation + from .._exception_notes import add_exception_note from ..event_loop import streaming from ..tools import convert_pydantic_to_tool_spec @@ -407,6 +409,8 @@ def _format_bedrock_messages(self, messages: Messages) -> list[dict[str, Any]]: # Format content blocks for Bedrock API compatibility formatted_content = self._format_request_message_content(content_block) + if formatted_content is None: + continue # Wrap text or image content in guardrailContent if this is the last user message if ( @@ -459,7 +463,19 @@ def _should_include_tool_result_status(self) -> bool: else: # "auto" return any(model in self.config["model_id"] for model in _MODELS_INCLUDE_STATUS) - def _format_request_message_content(self, content: ContentBlock) -> dict[str, Any]: + def _handle_location(self, location: SourceLocation) -> dict[str, Any] | None: + """Convert location content block to Bedrock format if its an S3Location.""" + if location["type"] == "s3": + s3_location = cast(S3Location, location) + formatted_document_s3: dict[str, Any] = {"uri": s3_location["uri"]} + if "bucketOwner" in s3_location: + formatted_document_s3["bucketOwner"] = s3_location["bucketOwner"] + return {"s3Location": formatted_document_s3} + else: + logger.warning("Non s3 location sources are not supported by Bedrock, skipping content block") + return None + + def _format_request_message_content(self, content: ContentBlock) -> dict[str, Any] | None: """Format a Bedrock content block. Bedrock strictly validates content blocks and throws exceptions for unknown fields. @@ -489,9 +505,17 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An if "format" in document: result["format"] = document["format"] - # Handle source + # Handle source - supports bytes or location if "source" in document: - result["source"] = {"bytes": document["source"]["bytes"]} + source = document["source"] + formatted_document_source: dict[str, Any] | None + if "location" in source: + formatted_document_source = self._handle_location(source["location"]) + if formatted_document_source is None: + return None + elif "bytes" in source: + formatted_document_source = {"bytes": source["bytes"]} + result["source"] = formatted_document_source # Handle optional fields if "citations" in document and document["citations"] is not None: @@ -512,10 +536,14 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An if "image" in content: image = content["image"] source = image["source"] - formatted_source = {} - if "bytes" in source: - formatted_source = {"bytes": source["bytes"]} - result = {"format": image["format"], "source": formatted_source} + formatted_image_source: dict[str, Any] | None + if "location" in source: + formatted_image_source = self._handle_location(source["location"]) + if formatted_image_source is None: + return None + elif "bytes" in source: + formatted_image_source = {"bytes": source["bytes"]} + result = {"format": image["format"], "source": formatted_image_source} return {"image": result} # https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ReasoningContentBlock.html @@ -550,9 +578,12 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An # Handle json field since not in ContentBlock but valid in ToolResultContent formatted_content.append({"json": tool_result_content["json"]}) else: - formatted_content.append( - self._format_request_message_content(cast(ContentBlock, tool_result_content)) + formatted_message_content = self._format_request_message_content( + cast(ContentBlock, tool_result_content) ) + if formatted_message_content is None: + continue + formatted_content.append(formatted_message_content) result = { "content": formatted_content, @@ -577,10 +608,14 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An if "video" in content: video = content["video"] source = video["source"] - formatted_source = {} - if "bytes" in source: - formatted_source = {"bytes": source["bytes"]} - result = {"format": video["format"], "source": formatted_source} + formatted_video_source: dict[str, Any] | None + if "location" in source: + formatted_video_source = self._handle_location(source["location"]) + if formatted_video_source is None: + return None + elif "bytes" in source: + formatted_video_source = {"bytes": source["bytes"]} + result = {"format": video["format"], "source": formatted_video_source} return {"video": result} # https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_CitationsContentBlock.html diff --git a/src/strands/types/media.py b/src/strands/types/media.py index 462d8af34..b1240dffb 100644 --- a/src/strands/types/media.py +++ b/src/strands/types/media.py @@ -5,9 +5,9 @@ - Bedrock docs: https://docs.aws.amazon.com/bedrock/latest/APIReference/API_Types_Amazon_Bedrock_Runtime.html """ -from typing import Literal +from typing import Literal, TypeAlias -from typing_extensions import TypedDict +from typing_extensions import Required, TypedDict from .citations import CitationsConfig @@ -15,14 +15,50 @@ """Supported document formats.""" -class DocumentSource(TypedDict): +class Location(TypedDict, total=False): + """A location for a document. + + This type is a generic location for a document. Its usage is determined by the underlying model provider. + """ + + type: Required[str] + + +class S3Location(Location, total=False): + """A storage location in an Amazon S3 bucket. + + Used by Bedrock to reference media files stored in S3 instead of passing raw bytes. + + - Docs: https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_S3Location.html + + Attributes: + type: s3 + uri: An object URI starting with `s3://`. Required. + bucketOwner: If the bucket belongs to another AWS account, specify that account's ID. Optional. + """ + + # mypy doesn't like overriding this field since its a subclass, but since its just a literal string, this is fine. + + type: Literal["s3"] # type: ignore[misc] + uri: Required[str] + bucketOwner: str + + +SourceLocation: TypeAlias = Location | S3Location + + +class DocumentSource(TypedDict, total=False): """Contains the content of a document. + Only one of `bytes` or `s3Location` should be specified. + Attributes: bytes: The binary content of the document. + location: Location of the document. """ bytes: bytes + location: SourceLocation class DocumentContent(TypedDict, total=False): @@ -45,14 +81,18 @@ class DocumentContent(TypedDict, total=False): """Supported image formats.""" -class ImageSource(TypedDict): +class ImageSource(TypedDict, total=False): """Contains the content of an image. + Only one of `bytes` or `s3Location` should be specified. + Attributes: bytes: The binary content of the image. + location: Location of the image. """ bytes: bytes + location: SourceLocation class ImageContent(TypedDict): @@ -71,14 +111,18 @@ class ImageContent(TypedDict): """Supported video formats.""" -class VideoSource(TypedDict): +class VideoSource(TypedDict, total=False): """Contains the content of a video. + Only one of `bytes` or `s3Location` should be specified. + Attributes: bytes: The binary content of the video. + location: Location of the video. """ bytes: bytes + location: SourceLocation class VideoContent(TypedDict): diff --git a/tests/strands/agent/hooks/test_events.py b/tests/strands/agent/hooks/test_events.py index 762b77452..de551d137 100644 --- a/tests/strands/agent/hooks/test_events.py +++ b/tests/strands/agent/hooks/test_events.py @@ -206,8 +206,6 @@ def test_invocation_state_is_available_in_model_call_events(agent): assert after_event.invocation_state["request_id"] == "req-456" - - def test_before_invocation_event_messages_default_none(agent): """Test that BeforeInvocationEvent.messages defaults to None for backward compatibility.""" event = BeforeInvocationEvent(agent=agent) diff --git a/tests/strands/models/test_bedrock.py b/tests/strands/models/test_bedrock.py index e92018f35..761434258 100644 --- a/tests/strands/models/test_bedrock.py +++ b/tests/strands/models/test_bedrock.py @@ -1,3 +1,5 @@ +import copy +import logging import os import sys import traceback @@ -1519,7 +1521,6 @@ async def test_add_note_on_validation_exception_throughput(bedrock_client, model @pytest.mark.asyncio async def test_stream_logging(bedrock_client, model, messages, caplog, alist): """Test that stream method logs debug messages at the expected stages.""" - import logging # Set the logger to debug level to capture debug messages caplog.set_level(logging.DEBUG, logger="strands.models.bedrock") @@ -1787,8 +1788,8 @@ def test_format_request_filters_image_content_blocks(model, model_id): assert "metadata" not in image_block -def test_format_request_filters_nested_image_s3_fields(model, model_id): - """Test that s3Location is filtered out and only bytes source is preserved.""" +def test_format_request_image_s3_location_only(model, model_id): + """Test that image with only s3Location is properly formatted.""" messages = [ { "role": "user", @@ -1797,8 +1798,7 @@ def test_format_request_filters_nested_image_s3_fields(model, model_id): "image": { "format": "png", "source": { - "bytes": b"image_data", - "s3Location": {"bucket": "my-bucket", "key": "image.png", "extraField": "filtered"}, + "location": {"type": "s3", "uri": "s3://my-bucket/image.png"}, }, } } @@ -1809,8 +1809,146 @@ def test_format_request_filters_nested_image_s3_fields(model, model_id): formatted_request = model._format_request(messages) image_source = formatted_request["messages"][0]["content"][0]["image"]["source"] + assert image_source == {"s3Location": {"uri": "s3://my-bucket/image.png"}} + + +def test_format_request_image_bytes_only(model, model_id): + """Test that image with only bytes source is properly formatted.""" + messages = [ + { + "role": "user", + "content": [ + { + "image": { + "format": "png", + "source": {"bytes": b"image_data"}, + } + } + ], + } + ] + + formatted_request = model._format_request(messages) + image_source = formatted_request["messages"][0]["content"][0]["image"]["source"] + assert image_source == {"bytes": b"image_data"} - assert "s3Location" not in image_source + + +def test_format_request_document_s3_location(model, model_id): + """Test that document with s3Location is properly formatted.""" + messages = [ + { + "role": "user", + "content": [ + { + "document": { + "name": "report.pdf", + "format": "pdf", + "source": { + "location": {"type": "s3", "uri": "s3://my-bucket/report.pdf"}, + }, + } + }, + { + "document": { + "name": "report.pdf", + "format": "pdf", + "source": { + "location": { + "type": "s3", + "uri": "s3://my-bucket/report.pdf", + "bucketOwner": "123456789012", + }, + }, + } + }, + ], + } + ] + + formatted_request = model._format_request(messages) + document = formatted_request["messages"][0]["content"][0]["document"] + document_with_bucket_owner = formatted_request["messages"][0]["content"][1]["document"] + + assert document["source"] == {"s3Location": {"uri": "s3://my-bucket/report.pdf"}} + + assert document_with_bucket_owner["source"] == { + "s3Location": {"uri": "s3://my-bucket/report.pdf", "bucketOwner": "123456789012"} + } + + +def test_format_request_unsupported_location(model, caplog): + """Test that document with s3Location is properly formatted.""" + + caplog.set_level(logging.WARNING, logger="strands.models.bedrock") + + messages = [ + { + "role": "user", + "content": [ + {"text": "Hello!"}, + { + "document": { + "name": "report.pdf", + "format": "pdf", + "source": { + "location": { + "type": "other", + }, + }, + } + }, + { + "video": { + "format": "mp4", + "source": { + "location": { + "type": "other", + }, + }, + } + }, + { + "image": { + "format": "png", + "source": { + "location": { + "type": "other", + }, + }, + } + }, + ], + } + ] + + formatted_request = model._format_request(messages) + assert len(formatted_request["messages"][0]["content"]) == 1 + 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.""" + messages = [ + { + "role": "user", + "content": [ + { + "video": { + "format": "mp4", + "source": { + "location": {"type": "s3", "uri": "s3://my-bucket/video.mp4"}, + }, + } + }, + ], + } + ] + + formatted_request = 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_filters_document_content_blocks(model, model_id): @@ -2310,7 +2448,6 @@ def test_inject_cache_point_skipped_for_non_claude(bedrock_client): def test_format_bedrock_messages_does_not_mutate_original(bedrock_client): """Test that _format_bedrock_messages does not mutate original messages.""" - import copy model = BedrockModel( model_id="us.anthropic.claude-sonnet-4-20250514-v1:0", cache_config=CacheConfig(strategy="auto") diff --git a/tests/strands/tools/mcp/test_mcp_client.py b/tests/strands/tools/mcp/test_mcp_client.py index f784da414..a2ef369ea 100644 --- a/tests/strands/tools/mcp/test_mcp_client.py +++ b/tests/strands/tools/mcp/test_mcp_client.py @@ -632,7 +632,7 @@ def test_call_tool_sync_embedded_nested_base64_textual_mime(mock_transport, mock def test_call_tool_sync_embedded_image_blob(mock_transport, mock_session): """EmbeddedResource.resource (blob with image MIME) should map to image content.""" # Read yellow.png file - with open("tests_integ/yellow.png", "rb") as image_file: + with open("tests_integ/resources/yellow.png", "rb") as image_file: png_data = image_file.read() payload = base64.b64encode(png_data).decode() diff --git a/tests/strands/types/test_media.py b/tests/strands/types/test_media.py new file mode 100644 index 000000000..2fa8c3621 --- /dev/null +++ b/tests/strands/types/test_media.py @@ -0,0 +1,99 @@ +"""Tests for media type definitions.""" + +from strands.types.media import ( + DocumentSource, + ImageSource, + S3Location, + VideoSource, +) + + +class TestS3Location: + """Tests for S3Location TypedDict.""" + + def test_s3_location_with_uri_only(self): + """Test S3Location with only uri field.""" + s3_loc: S3Location = {"uri": "s3://my-bucket/path/to/file.pdf"} + + assert s3_loc["uri"] == "s3://my-bucket/path/to/file.pdf" + assert "bucketOwner" not in s3_loc + + def test_s3_location_with_bucket_owner(self): + """Test S3Location with both uri and bucketOwner fields.""" + s3_loc: S3Location = { + "uri": "s3://my-bucket/path/to/file.pdf", + "bucketOwner": "123456789012", + } + + assert s3_loc["uri"] == "s3://my-bucket/path/to/file.pdf" + assert s3_loc["bucketOwner"] == "123456789012" + + +class TestDocumentSource: + """Tests for DocumentSource TypedDict.""" + + def test_document_source_with_bytes(self): + """Test DocumentSource with bytes content.""" + doc_source: DocumentSource = {"bytes": b"document content"} + + assert doc_source["bytes"] == b"document content" + assert "s3Location" not in doc_source + + def test_document_source_with_s3_location(self): + """Test DocumentSource with s3Location.""" + doc_source: DocumentSource = { + "s3Location": { + "uri": "s3://my-bucket/docs/report.pdf", + "bucketOwner": "123456789012", + } + } + + assert "bytes" not in doc_source + assert doc_source["s3Location"]["uri"] == "s3://my-bucket/docs/report.pdf" + assert doc_source["s3Location"]["bucketOwner"] == "123456789012" + + +class TestImageSource: + """Tests for ImageSource TypedDict.""" + + def test_image_source_with_bytes(self): + """Test ImageSource with bytes content.""" + img_source: ImageSource = {"bytes": b"image content"} + + assert img_source["bytes"] == b"image content" + assert "s3Location" not in img_source + + def test_image_source_with_s3_location(self): + """Test ImageSource with s3Location.""" + img_source: ImageSource = { + "s3Location": { + "uri": "s3://my-bucket/images/photo.png", + } + } + + assert "bytes" not in img_source + assert img_source["s3Location"]["uri"] == "s3://my-bucket/images/photo.png" + + +class TestVideoSource: + """Tests for VideoSource TypedDict.""" + + def test_video_source_with_bytes(self): + """Test VideoSource with bytes content.""" + vid_source: VideoSource = {"bytes": b"video content"} + + assert vid_source["bytes"] == b"video content" + assert "s3Location" not in vid_source + + def test_video_source_with_s3_location(self): + """Test VideoSource with s3Location.""" + vid_source: VideoSource = { + "s3Location": { + "uri": "s3://my-bucket/videos/clip.mp4", + "bucketOwner": "987654321098", + } + } + + assert "bytes" not in vid_source + assert vid_source["s3Location"]["uri"] == "s3://my-bucket/videos/clip.mp4" + assert vid_source["s3Location"]["bucketOwner"] == "987654321098" diff --git a/tests_integ/conftest.py b/tests_integ/conftest.py index 9de00089b..dbe25d685 100644 --- a/tests_integ/conftest.py +++ b/tests_integ/conftest.py @@ -133,14 +133,21 @@ def pytest_sessionstart(session): @pytest.fixture def yellow_img(pytestconfig): - path = pytestconfig.rootdir / "tests_integ/yellow.png" + path = pytestconfig.rootdir / "tests_integ/resources/yellow.png" with open(path, "rb") as fp: return fp.read() @pytest.fixture def letter_pdf(pytestconfig): - path = pytestconfig.rootdir / "tests_integ/letter.pdf" + path = pytestconfig.rootdir / "tests_integ/resources/letter.pdf" + with open(path, "rb") as fp: + return fp.read() + + +@pytest.fixture +def blue_video(pytestconfig): + path = pytestconfig.rootdir / "tests_integ/resources/blue.mp4" with open(path, "rb") as fp: return fp.read() diff --git a/tests_integ/mcp/echo_server.py b/tests_integ/mcp/echo_server.py index 8fa1fb2b2..363c588ee 100644 --- a/tests_integ/mcp/echo_server.py +++ b/tests_integ/mcp/echo_server.py @@ -90,7 +90,7 @@ def get_weather(location: Literal["New York", "London", "Tokyo"] = "New York"): ] elif location.lower() == "tokyo": # Read yellow.png file for weather icon - with open("tests_integ/yellow.png", "rb") as image_file: + with open("tests_integ/resources/yellow.png", "rb") as image_file: png_data = image_file.read() return [ EmbeddedResource( diff --git a/tests_integ/mcp/test_mcp_client.py b/tests_integ/mcp/test_mcp_client.py index 298272df5..4e192c935 100644 --- a/tests_integ/mcp/test_mcp_client.py +++ b/tests_integ/mcp/test_mcp_client.py @@ -43,7 +43,7 @@ def calculator(x: int, y: int) -> int: @mcp.tool(description="Generates a custom image") def generate_custom_image() -> MCPImageContent: try: - with open("tests_integ/yellow.png", "rb") as image_file: + with open("tests_integ/resources/yellow.png", "rb") as image_file: encoded_image = base64.b64encode(image_file.read()) return MCPImageContent(type="image", data=encoded_image, mimeType="image/png") except Exception as e: diff --git a/tests_integ/resources/blue.mp4 b/tests_integ/resources/blue.mp4 new file mode 100644 index 000000000..5989bb4b0 Binary files /dev/null and b/tests_integ/resources/blue.mp4 differ diff --git a/tests_integ/letter.pdf b/tests_integ/resources/letter.pdf similarity index 100% rename from tests_integ/letter.pdf rename to tests_integ/resources/letter.pdf diff --git a/tests_integ/yellow.png b/tests_integ/resources/yellow.png similarity index 100% rename from tests_integ/yellow.png rename to tests_integ/resources/yellow.png diff --git a/tests_integ/test_a2a_executor.py b/tests_integ/test_a2a_executor.py index ddca0bfa6..43a6026bf 100644 --- a/tests_integ/test_a2a_executor.py +++ b/tests_integ/test_a2a_executor.py @@ -17,7 +17,7 @@ async def test_a2a_executor_with_real_image(): """Test A2A server processes a real image file correctly via HTTP.""" # Read the test image file - test_image_path = os.path.join(os.path.dirname(__file__), "yellow.png") + test_image_path = os.path.join(os.path.dirname(__file__), "resources/yellow.png") with open(test_image_path, "rb") as f: original_image_bytes = f.read() @@ -80,7 +80,7 @@ async def test_a2a_executor_with_real_image(): def test_a2a_executor_image_roundtrip(): """Test that image data survives the A2A base64 encoding/decoding roundtrip.""" # Read the test image - test_image_path = os.path.join(os.path.dirname(__file__), "yellow.png") + test_image_path = os.path.join(os.path.dirname(__file__), "resources/yellow.png") with open(test_image_path, "rb") as f: original_bytes = f.read() diff --git a/tests_integ/test_bedrock_s3_location.py b/tests_integ/test_bedrock_s3_location.py new file mode 100644 index 000000000..9b28e88be --- /dev/null +++ b/tests_integ/test_bedrock_s3_location.py @@ -0,0 +1,177 @@ +"""Integration tests for S3 location support in media content types.""" + +import time + +import boto3 +import pytest + +from strands import Agent +from strands.models.bedrock import BedrockModel + + +@pytest.fixture +def boto_session(): + """Create a boto3 session for testing.""" + return boto3.Session(region_name="us-west-2") + + +@pytest.fixture +def account_id(boto_session): + """Get the current AWS account ID.""" + sts_client = boto_session.client("sts") + return sts_client.get_caller_identity()["Account"] + + +@pytest.fixture +def s3_client(boto_session): + """Create an S3 client.""" + return boto_session.client("s3") + + +@pytest.fixture +def test_bucket(s3_client, account_id): + """Create a test S3 bucket for the tests. + + Creates a bucket with account-specific name and cleans it up after tests. + """ + bucket_name = f"strands-integ-tests-resources-{account_id}" + + # Create the bucket if it doesn't exist + try: + s3_client.head_bucket(Bucket=bucket_name) + print(f"Bucket {bucket_name} already exists") + except s3_client.exceptions.ClientError: + try: + s3_client.create_bucket( + Bucket=bucket_name, + CreateBucketConfiguration={"LocationConstraint": "us-west-2"}, + ) + print(f"Created test bucket: {bucket_name}") + # Wait for bucket to be available + time.sleep(2) + except s3_client.exceptions.BucketAlreadyOwnedByYou: + print(f"Bucket {bucket_name} already exists") + + yield bucket_name + + # Note: We don't delete the bucket to allow reuse across test runs + # Objects will be overwritten on subsequent runs + + +@pytest.fixture +def s3_document(s3_client, test_bucket, letter_pdf): + """Upload a test document to S3 and return its URI.""" + document_key = "test-documents/letter.pdf" + + # Upload the document using existing letter_pdf fixture + s3_client.put_object( + Bucket=test_bucket, + Key=document_key, + Body=letter_pdf, + ContentType="application/pdf", + ) + print(f"Uploaded test document to s3://{test_bucket}/{document_key}") + + return f"s3://{test_bucket}/{document_key}" + + +@pytest.fixture +def s3_image(s3_client, test_bucket, yellow_img): + """Upload a test image to S3 and return its URI.""" + image_key = "test-images/yellow.png" + + # Upload the image using existing yellow_img fixture + s3_client.put_object( + Bucket=test_bucket, + Key=image_key, + Body=yellow_img, + ContentType="image/png", + ) + print(f"Uploaded test image to s3://{test_bucket}/{image_key}") + + return f"s3://{test_bucket}/{image_key}" + + +@pytest.fixture +def s3_video(s3_client, test_bucket, blue_video): + """Upload a test video to S3 and return its URI.""" + video_key = "test-videos/blue.mp4" + + # Upload the video using existing blue_video fixture + s3_client.put_object( + Bucket=test_bucket, + Key=video_key, + Body=blue_video, + ContentType="video/mp4", + ) + print(f"Uploaded test video to s3://{test_bucket}/{video_key}") + + return f"s3://{test_bucket}/{video_key}" + + +def test_document_s3_location(s3_document, account_id): + """Test that Bedrock correctly formats a document with S3 location.""" + messages = [ + { + "role": "user", + "content": [ + {"text": "Please tell me about this document?"}, + { + "document": { + "format": "pdf", + "name": "letter", + "source": {"location": {"type": "s3", "uri": s3_document, "bucketOwner": account_id}}, + }, + }, + ], + }, + ] + + agent = Agent(model=BedrockModel(model_id="us.amazon.nova-2-lite-v1:0", region_name="us-west-2")) + result = agent(messages) + + # The actual recognition capabilities of these models is not great, so just asserting that the call actually worked. + assert len(str(result)) > 0 + + +def test_image_s3_location(s3_image): + """Test that Bedrock correctly formats an image with S3 location.""" + messages = [ + { + "role": "user", + "content": [ + {"text": "Please tell me about this image?"}, + { + "image": { + "format": "png", + "source": {"location": {"type": "s3", "uri": s3_image}}, + }, + }, + ], + }, + ] + + agent = Agent(model=BedrockModel(model_id="us.amazon.nova-2-lite-v1:0", region_name="us-west-2")) + result = agent(messages) + + # The actual recognition capabilities of these models is not great, so just asserting that the call actually worked. + assert len(str(result)) > 0 + + +def test_video_s3_location(s3_video): + """Test that Bedrock correctly formats a video with S3 location.""" + messages = [ + { + "role": "user", + "content": [ + {"text": "Describe the colors is in this video?"}, + {"video": {"format": "mp4", "source": {"location": {"type": "s3", "uri": s3_video}}}}, + ], + }, + ] + + agent = Agent(model=BedrockModel(model_id="us.amazon.nova-pro-v1:0", region_name="us-west-2")) + result = agent(messages) + + # The actual recognition capabilities of these models is not great, so just asserting that the call actually worked. + assert len(str(result)) > 0