fix: convert ContentBlock to proper Union type

src/strands/agent/agent.py

@@ -58,7 +58,16 @@
 from ..tools.watcher import ToolWatcher
 from ..types._events import AgentResultEvent, EventLoopStopEvent, InitEventLoopEvent, ModelStreamChunkEvent, TypedEvent
 from ..types.agent import AgentInput
-from ..types.content import ContentBlock, Message, Messages, SystemContentBlock
+from ..types.content import (
+    CONTENT_BLOCK_KEYS,
+    ContentBlock,
+    ContentBlockText,
+    Message,
+    Messages,
+    SystemContentBlock,
+    is_tool_result_block,
+    is_tool_use_block,
+)
 from ..types.exceptions import ContextWindowOverflowException
 from ..types.traces import AttributeValue
 from .agent_result import AgentResult
@@ -720,7 +729,9 @@ async def _convert_prompt_to_messages(self, prompt: AgentInput) -> Messages:
                     "Agents latest message is toolUse, appending a toolResult message to have valid conversation."
                 )
                 tool_use_ids = [
-                    content["toolUse"]["toolUseId"] for content in self.messages[-1]["content"] if "toolUse" in content
+                    content["toolUse"]["toolUseId"]
+                    for content in self.messages[-1]["content"]
+                    if is_tool_use_block(content)
                 ]
                 await self._append_messages(
                     {
@@ -743,7 +754,7 @@ async def _convert_prompt_to_messages(self, prompt: AgentInput) -> Messages:
                         messages = cast(Messages, prompt)
 
                     # Check if all items are content blocks
-                    elif all(any(key in ContentBlock.__annotations__.keys() for key in item) for item in prompt):
+                    elif all(any(key in CONTENT_BLOCK_KEYS for key in item) for item in prompt):
                         # Treat as List[ContentBlock] input - convert to user message
                         # This allows invalid structures to be passed through to the model
                         messages = [{"role": "user", "content": cast(list[ContentBlock], prompt)}]
@@ -838,14 +849,14 @@ def _redact_user_content(self, content: list[ContentBlock], redact_message: str)
             - otherwise, the entire content of the message is replaced
                 with a single text block with the redact message.
         """
-        redacted_content = []
+        redacted_content: list[ContentBlock] = []
         for block in content:
-            if "toolResult" in block:
+            if is_tool_result_block(block):
                 block["toolResult"]["content"] = [{"text": redact_message}]
                 redacted_content.append(block)
 
         if not redacted_content:
             # Text content is added only if no toolResult blocks were found
-            redacted_content = [{"text": redact_message}]
+            redacted_content = [ContentBlockText(text=redact_message)]
 
         return redacted_content

src/strands/agent/agent_result.py

@@ -10,7 +10,7 @@
 
 from ..interrupt import Interrupt
 from ..telemetry.metrics import EventLoopMetrics
-from ..types.content import Message
+from ..types.content import Message, is_text_block
 from ..types.streaming import StopReason
 
 
@@ -48,8 +48,8 @@ def __str__(self) -> str:
 
         result = ""
         for item in content_array:
-            if isinstance(item, dict) and "text" in item:
-                result += item.get("text", "") + "\n"
+            if is_text_block(item):
+                result += item["text"] + "\n"
 
         if not result and self.structured_output:
             result = self.structured_output.model_dump_json()

src/strands/agent/conversation_manager/sliding_window_conversation_manager.py

@@ -1,13 +1,13 @@
 """Sliding window conversation history management."""
 
 import logging
-from typing import TYPE_CHECKING, Any, Optional
+from typing import TYPE_CHECKING, Any, Optional, cast
 
 if TYPE_CHECKING:
     from ...agent.agent import Agent
 
 from ...hooks import BeforeModelCallEvent, HookRegistry
-from ...types.content import Messages
+from ...types.content import ContentBlockToolResult, Messages, is_tool_result_block
 from ...types.exceptions import ContextWindowOverflowException
 from .conversation_manager import ConversationManager
 
@@ -216,21 +216,23 @@ def _truncate_tool_results(self, messages: Messages, msg_idx: int) -> bool:
         changes_made = False
         tool_result_too_large_message = "The tool result was too large!"
         for i, content in enumerate(message.get("content", [])):
-            if isinstance(content, dict) and "toolResult" in content:
+            if is_tool_result_block(content):
                 tool_result_content_text = next(
                     (item["text"] for item in content["toolResult"]["content"] if "text" in item),
                     "",
                 )
+                # Cast to ensure type narrowing for indexed access
+                content_block = cast(ContentBlockToolResult, message["content"][i])
                 # make the overwriting logic togglable
                 if (
-                    message["content"][i]["toolResult"]["status"] == "error"
+                    content_block["toolResult"]["status"] == "error"
                     and tool_result_content_text == tool_result_too_large_message
                 ):
                     logger.info("ToolResult has already been updated, skipping overwrite")
                     return False
                 # Update status to error with informative message
-                message["content"][i]["toolResult"]["status"] = "error"
-                message["content"][i]["toolResult"]["content"] = [{"text": tool_result_too_large_message}]
+                content_block["toolResult"]["status"] = "error"
+                content_block["toolResult"]["content"] = [{"text": tool_result_too_large_message}]
                 changes_made = True
 
         return changes_made

src/strands/event_loop/_recover_message_on_max_tokens_reached.py

@@ -7,8 +7,7 @@
 
 import logging
 
-from ..types.content import ContentBlock, Message
-from ..types.tools import ToolUse
+from ..types.content import ContentBlock, Message, is_tool_use_block
 
 logger = logging.getLogger(__name__)
 
@@ -52,11 +51,12 @@ def recover_message_on_max_tokens_reached(message: Message) -> Message:
 
     valid_content: list[ContentBlock] = []
     for content in message["content"] or []:
-        tool_use: ToolUse | None = content.get("toolUse")
-        if not tool_use:
+        if not is_tool_use_block(content):
             valid_content.append(content)
             continue
 
+        tool_use = content["toolUse"]
+
         # Replace all tool uses with error messages when max_tokens is reached
         display_name = tool_use.get("name") or "<unknown>"
         logger.warning("tool_name=<%s> | replacing with error message due to max_tokens truncation.", display_name)

src/strands/event_loop/streaming.py

@@ -22,7 +22,15 @@
     TypedEvent,
 )
 from ..types.citations import CitationsContentBlock
-from ..types.content import ContentBlock, Message, Messages, SystemContentBlock
+from ..types.content import (
+    ContentBlock,
+    ContentBlockReasoningContent,
+    Message,
+    Messages,
+    SystemContentBlock,
+    is_text_block,
+    is_tool_use_block,
+)
 from ..types.streaming import (
     ContentBlockDeltaEvent,
     ContentBlockStart,
@@ -69,7 +77,7 @@ def _normalize_messages(messages: Messages) -> Messages:
             # Ensure the tool-uses always have valid names before sending
             # https://github.com/strands-agents/sdk-python/issues/1069
             for item in content:
-                if "toolUse" in item:
+                if is_tool_use_block(item):
                     has_tool_use = True
                     tool_use: ToolUse = item["toolUse"]
 
@@ -82,13 +90,13 @@ def _normalize_messages(messages: Messages) -> Messages:
             if has_tool_use:
                 # Remove blank 'text' items for assistant messages
                 before_len = len(content)
-                content[:] = [item for item in content if "text" not in item or item["text"].strip()]
+                content[:] = [item for item in content if not is_text_block(item) or item["text"].strip()]
                 if not removed_blank_message_content_text and before_len != len(content):
                     removed_blank_message_content_text = True
             else:
                 # Replace blank 'text' with '[blank text]' for assistant messages
                 for item in content:
-                    if "text" in item and not item["text"].strip():
+                    if is_text_block(item) and not item["text"].strip():
                         replaced_blank_message_content_text = True
                         item["text"] = "[blank text]"
 
@@ -136,13 +144,13 @@ def remove_blank_messages_content_text(messages: Messages) -> Messages:
             if has_tool_use:
                 # Remove blank 'text' items for assistant messages
                 before_len = len(content)
-                content[:] = [item for item in content if "text" not in item or item["text"].strip()]
+                content[:] = [item for item in content if not is_text_block(item) or item["text"].strip()]
                 if not removed_blank_message_content_text and before_len != len(content):
                     removed_blank_message_content_text = True
             else:
                 # Replace blank 'text' with '[blank text]' for assistant messages
                 for item in content:
-                    if "text" in item and not item["text"].strip():
+                    if is_text_block(item) and not item["text"].strip():
                         replaced_blank_message_content_text = True
                         item["text"] = "[blank text]"
 
@@ -298,7 +306,7 @@ def handle_content_block_stop(state: dict[str, Any]) -> dict[str, Any]:
         state["text"] = ""
 
     elif reasoning_text:
-        content_block: ContentBlock = {
+        content_block: ContentBlockReasoningContent = {
             "reasoningContent": {
                 "reasoningText": {
                     "text": state["reasoningText"],

src/strands/models/anthropic.py

@@ -15,7 +15,16 @@
 
 from ..event_loop.streaming import process_stream
 from ..tools.structured_output.structured_output_utils import convert_pydantic_to_tool_spec
-from ..types.content import ContentBlock, Messages
+from ..types.content import (
+    ContentBlock,
+    Messages,
+    is_document_block,
+    is_image_block,
+    is_reasoning_content_block,
+    is_text_block,
+    is_tool_result_block,
+    is_tool_use_block,
+)
 from ..types.exceptions import ContextWindowOverflowException, ModelThrottledException
 from ..types.streaming import StreamEvent
 from ..types.tools import ToolChoice, ToolChoiceToolDict, ToolSpec
@@ -108,7 +117,7 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An
         Raises:
             TypeError: If the content block type cannot be converted to an Anthropic-compatible format.
         """
-        if "document" in content:
+        if is_document_block(content):
             mime_type = mimetypes.types_map.get(f".{content['document']['format']}", "application/octet-stream")
             return {
                 "source": {
@@ -124,7 +133,7 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An
                 "type": "document",
             }
 
-        if "image" in content:
+        if is_image_block(content):
             return {
                 "source": {
                     "data": base64.b64encode(content["image"]["source"]["bytes"]).decode("utf-8"),
@@ -134,25 +143,25 @@ def _format_request_message_content(self, content: ContentBlock) -> dict[str, An
                 "type": "image",
             }
 
-        if "reasoningContent" in content:
+        if is_reasoning_content_block(content):
             return {
                 "signature": content["reasoningContent"]["reasoningText"]["signature"],
                 "thinking": content["reasoningContent"]["reasoningText"]["text"],
                 "type": "thinking",
             }
 
-        if "text" in content:
+        if is_text_block(content):
             return {"text": content["text"], "type": "text"}
 
-        if "toolUse" in content:
+        if is_tool_use_block(content):
             return {
                 "id": content["toolUse"]["toolUseId"],
                 "input": content["toolUse"]["input"],
                 "name": content["toolUse"]["name"],
                 "type": "tool_use",
             }
 
-        if "toolResult" in content:
+        if is_tool_result_block(content):
             return {
                 "content": [
                     self._format_request_message_content(