feat(bedrock): add guardrail_last_turn_only option

src/strands/models/bedrock.py

@@ -82,6 +82,8 @@ class BedrockConfig(TypedDict, total=False):
             guardrail_redact_input_message: If a Bedrock Input guardrail triggers, replace the input with this message.
             guardrail_redact_output: Flag to redact output if guardrail is triggered. Defaults to False.
             guardrail_redact_output_message: If a Bedrock Output guardrail triggers, replace output with this message.
+            guardrail_last_turn_only: Flag to send only the last turn to guardrails instead of full conversation.
+                Defaults to False.
             max_tokens: Maximum number of tokens to generate in the response
             model_id: The Bedrock model ID (e.g., "us.anthropic.claude-sonnet-4-20250514-v1:0")
             include_tool_result_status: Flag to include status field in tool results.
@@ -105,6 +107,7 @@ class BedrockConfig(TypedDict, total=False):
         guardrail_redact_input_message: Optional[str]
         guardrail_redact_output: Optional[bool]
         guardrail_redact_output_message: Optional[str]
+        guardrail_last_turn_only: Optional[bool]
         max_tokens: Optional[int]
         model_id: str
         include_tool_result_status: Optional[Literal["auto"] | bool]
@@ -206,9 +209,12 @@ def _format_request(
         Returns:
             A Bedrock converse stream request.
         """
+        messages_for_request = messages
+
         if not tool_specs:
             has_tool_content = any(
-                any("toolUse" in block or "toolResult" in block for block in msg.get("content", [])) for msg in messages
+                any("toolUse" in block or "toolResult" in block for block in msg.get("content", []))
+                for msg in messages_for_request
             )
             if has_tool_content:
                 tool_specs = [noop_tool.tool_spec]
@@ -224,7 +230,10 @@ def _format_request(
 
         return {
             "modelId": self.config["model_id"],
-            "messages": self._format_bedrock_messages(messages),
+            "messages": self._format_bedrock_messages(
+                messages_for_request,
+                guardrail_last_turn_only=bool(self.config.get("guardrail_last_turn_only", False)),
+            ),
             "system": system_blocks,
             **(
                 {
@@ -295,16 +304,20 @@ def _format_request(
             ),
         }
 
-    def _format_bedrock_messages(self, messages: Messages) -> list[dict[str, Any]]:
+    def _format_bedrock_messages(
+        self, messages: Messages, guardrail_last_turn_only: bool = False
+    ) -> list[dict[str, Any]]:
         """Format messages for Bedrock API compatibility.
 
         This function ensures messages conform to Bedrock's expected format by:
         - Filtering out SDK_UNKNOWN_MEMBER content blocks
         - Eagerly filtering content blocks to only include Bedrock-supported fields
         - Ensuring all message content blocks are properly formatted for the Bedrock API
+        - Optionally wrapping the last user message in guardrailConverseContent blocks
 
         Args:
             messages: List of messages to format
+            guardrail_last_turn_only: If True, wrap the last user message content in guardrailConverseContent blocks
 
         Returns:
             Messages formatted for Bedrock API compatibility
@@ -321,7 +334,17 @@ def _format_bedrock_messages(self, messages: Messages) -> list[dict[str, Any]]:
         filtered_unknown_members = False
         dropped_deepseek_reasoning_content = False
 
-        for message in messages:
+        # Find the index of the last user message if wrapping is enabled
+        last_user_idx = -1
+        if guardrail_last_turn_only and messages:
+            last_msg = messages[-1]
+            if last_msg["role"] == "user" and any(
+                "text" in block or "image" in block or "document" in block or "video" in block
+                for block in last_msg["content"]
+            ):
+                last_user_idx = len(messages) - 1
+
+        for idx, message in enumerate(messages):
             cleaned_content: list[dict[str, Any]] = []
 
             for content_block in message["content"]:
@@ -338,6 +361,11 @@ 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)
+
+                # Wrap text content in guardrailContent if this is the last user message
+                if guardrail_last_turn_only and idx == last_user_idx and "text" in formatted_content:
+                    formatted_content = {"guardContent": {"text": {"text": formatted_content["text"]}}}
+
                 cleaned_content.append(formatted_content)
 
             # Create new message with cleaned content (skip if empty)

tests/strands/models/test_bedrock.py

@@ -2196,3 +2196,82 @@ async def test_citations_content_preserves_tagged_union_structure(bedrock_client
         "(documentChar, documentPage, documentChunk, searchResultLocation, or web) "
         "with the location fields nested inside."
     )
+
+
+@pytest.mark.asyncio
+async def test_format_request_with_guardrail_last_turn_only(model):
+    """Test _format_request passes apply_last_turn flag correctly."""
+    model.update_config(guardrail_id="test-guardrail", guardrail_version="DRAFT", guardrail_last_turn_only=True)
+
+    messages = [
+        {"role": "user", "content": [{"text": "First message"}]},
+        {"role": "assistant", "content": [{"text": "First response"}]},
+        {"role": "user", "content": [{"text": "Latest message"}]},
+    ]
+
+    request = model._format_request(messages)
+
+    # All messages should be in the request
+    formatted_messages = request["messages"]
+    assert len(formatted_messages) == 3
+
+    # Last user message should be wrapped
+    assert "guardContent" in formatted_messages[2]["content"][0]
+    assert formatted_messages[2]["content"][0]["guardContent"]["text"]["text"] == "Latest message"
+
+    # First user message should NOT be wrapped
+    assert "text" in formatted_messages[0]["content"][0]
+    assert formatted_messages[0]["content"][0]["text"] == "First message"
+
+
+def test_format_bedrock_messages_multimodal_content(model):
+    """Test that only text blocks are wrapped, not images."""
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {"text": "Look at this image"},
+                {"image": {"format": "png", "source": {"bytes": b"fake_image_data"}}},
+            ],
+        }
+    ]
+
+    result = model._format_bedrock_messages(messages, guardrail_last_turn_only=True)
+
+    # Should have 2 content blocks
+    assert len(result[0]["content"]) == 2
+
+    # Text should be wrapped
+    assert "guardContent" in result[0]["content"][0]
+    assert result[0]["content"][0]["guardContent"]["text"]["text"] == "Look at this image"
+
+    # Image should NOT be wrapped
+    assert "image" in result[0]["content"][1]
+
+
+def test_format_bedrock_messages_wraps_last_user_text(model):
+    """Test that only the last user message text is wrapped in guardContent."""
+    messages = [
+        {"role": "user", "content": [{"text": "First message"}]},
+        {"role": "assistant", "content": [{"text": "First response"}]},
+        {"role": "user", "content": [{"text": "Latest message"}]},
+    ]
+
+    result = model._format_bedrock_messages(messages, guardrail_last_turn_only=True)
+
+    # All messages should be present
+    assert len(result) == 3
+
+    # First user message should NOT be wrapped
+    assert result[0]["role"] == "user"
+    assert "text" in result[0]["content"][0]
+    assert result[0]["content"][0]["text"] == "First message"
+
+    # Assistant message should be unchanged
+    assert result[1]["role"] == "assistant"
+    assert result[1]["content"][0]["text"] == "First response"
+
+    # Last user message should be wrapped in guardContent
+    assert result[2]["role"] == "user"
+    assert "guardContent" in result[2]["content"][0]
+    assert result[2]["content"][0]["guardContent"]["text"]["text"] == "Latest message"

tests_integ/test_bedrock_guardrails.py

@@ -289,6 +289,34 @@ def list_users() -> str:
     assert tool_result["content"][0]["text"] == INPUT_REDACT_MESSAGE
 
 
+def test_guardrail_last_turn_only(boto_session, bedrock_guardrail):
+    """Test that guardrail_last_turn_only only wraps the last user message in guardContent."""
+    bedrock_model = BedrockModel(
+        guardrail_id=bedrock_guardrail,
+        guardrail_version="DRAFT",
+        guardrail_last_turn_only=True,
+        boto_session=boto_session,
+    )
+
+    # Create agent with pre-existing conversation that contains blocked word
+    agent = Agent(
+        model=bedrock_model,
+        system_prompt="You are a helpful assistant.",
+        callback_handler=None,
+        messages=[
+            {"role": "user", "content": [{"text": "CACTUS"}]},
+            {"role": "assistant", "content": [{"text": "Hello!"}]},
+            {"role": "user", "content": [{"text": "How are you?"}]},
+        ],
+    )
+
+    # With guardrail_last_turn_only=True, the blocked word "CACTUS" in the conversation history
+    # should NOT trigger the guardrail because only the last user message ("How are you?")
+    # gets wrapped in guardContent
+    response = agent("Tell me about plants")
+    assert response.stop_reason != "guardrail_intervened"
+
+
 def test_guardrail_input_intervention_properly_redacts_in_session(boto_session, bedrock_guardrail, temp_dir):
     bedrock_model = BedrockModel(
         guardrail_id=bedrock_guardrail,