Add __str__ methods to content blocks and message types

src/claude_agent_sdk/types.py

@@ -855,13 +855,28 @@ class SandboxSettings(TypedDict, total=False):
     enableWeakerNestedSandbox: bool
 
 
+# __str__ payload truncation budget — keeps tool-call / tool-result output
+# readable when the payload is a large file or code blob.
+_STR_TRUNCATE_LEN = 200
+
+
+def _truncate(obj: object, max_len: int = _STR_TRUNCATE_LEN) -> str:
+    s = str(obj)
+    if len(s) <= max_len:
+        return s
+    return s[: max_len - 3] + "..."
+
+
 # Content block types
 @dataclass
 class TextBlock:
     """Text content block."""
 
     text: str
 
+    def __str__(self) -> str:
+        return self.text
+
 
 @dataclass
 class ThinkingBlock:
@@ -870,6 +885,9 @@ class ThinkingBlock:
     thinking: str
     signature: str
 
+    def __str__(self) -> str:
+        return f"[Thinking] {self.thinking}"
+
 
 @dataclass
 class ToolUseBlock:
@@ -879,6 +897,9 @@ class ToolUseBlock:
     name: str
     input: dict[str, Any]
 
+    def __str__(self) -> str:
+        return f"[Tool use: {self.name}] {_truncate(self.input)}"
+
 
 @dataclass
 class ToolResultBlock:
@@ -888,6 +909,12 @@ class ToolResultBlock:
     content: str | list[dict[str, Any]] | None = None
     is_error: bool | None = None
 
+    def __str__(self) -> str:
+        label = "Tool error" if self.is_error else "Tool result"
+        if self.content is None:
+            return f"[{label}]"
+        return f"[{label}] {_truncate(self.content)}"
+
 
 ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock
 
@@ -912,6 +939,13 @@ class UserMessage:
     parent_tool_use_id: str | None = None
     tool_use_result: dict[str, Any] | None = None
 
+    def __str__(self) -> str:
+        if isinstance(self.content, str):
+            body = self.content
+        else:
+            body = " ".join(str(block) for block in self.content)
+        return f"User: {body}" if body else "User:"
+
 
 @dataclass
 class AssistantMessage:
@@ -927,6 +961,10 @@ class AssistantMessage:
     session_id: str | None = None
     uuid: str | None = None
 
+    def __str__(self) -> str:
+        body = " ".join(str(block) for block in self.content)
+        return f"Claude: {body}" if body else "Claude:"
+
 
 @dataclass
 class SystemMessage:
@@ -935,6 +973,9 @@ class SystemMessage:
     subtype: str
     data: dict[str, Any]
 
+    def __str__(self) -> str:
+        return f"[System: {self.subtype}]"
+
 
 class TaskUsage(TypedDict):
     """Usage statistics reported in task_progress and task_notification messages."""
@@ -964,6 +1005,9 @@ class TaskStartedMessage(SystemMessage):
     tool_use_id: str | None = None
     task_type: str | None = None
 
+    def __str__(self) -> str:
+        return f"[Task started] {self.description} (task_id={self.task_id})"
+
 
 @dataclass
 class TaskProgressMessage(SystemMessage):
@@ -982,6 +1026,12 @@ class TaskProgressMessage(SystemMessage):
     tool_use_id: str | None = None
     last_tool_name: str | None = None
 
+    def __str__(self) -> str:
+        return (
+            f"[Task progress] {self.description} "
+            f"(task_id={self.task_id}, tokens={self.usage['total_tokens']})"
+        )
+
 
 @dataclass
 class TaskNotificationMessage(SystemMessage):
@@ -1001,6 +1051,9 @@ class TaskNotificationMessage(SystemMessage):
     tool_use_id: str | None = None
     usage: TaskUsage | None = None
 
+    def __str__(self) -> str:
+        return f"[Task {self.status}] {self.summary} (task_id={self.task_id})"
+
 
 @dataclass
 class ResultMessage:
@@ -1022,6 +1075,15 @@ class ResultMessage:
     errors: list[str] | None = None
     uuid: str | None = None
 
+    def __str__(self) -> str:
+        cost = (
+            f", cost=${self.total_cost_usd:.4f}"
+            if self.total_cost_usd is not None
+            else ""
+        )
+        status = "error" if self.is_error else self.subtype
+        return f"[Result: {status}] duration={self.duration_ms}ms{cost}"
+
 
 @dataclass
 class StreamEvent:
@@ -1032,6 +1094,10 @@ class StreamEvent:
     event: dict[str, Any]  # The raw Anthropic API stream event
     parent_tool_use_id: str | None = None
 
+    def __str__(self) -> str:
+        event_type = self.event.get("type", "unknown")
+        return f"[StreamEvent: {event_type}]"
+
 
 # Rate limit types — see https://docs.claude.com/en/docs/claude-code/rate-limits
 RateLimitStatus = Literal["allowed", "allowed_warning", "rejected"]
@@ -1079,6 +1145,14 @@ class RateLimitEvent:
     uuid: str
     session_id: str
 
+    def __str__(self) -> str:
+        info = self.rate_limit_info
+        window = info.rate_limit_type or "n/a"
+        util = (
+            f" @ {info.utilization * 100:.0f}%" if info.utilization is not None else ""
+        )
+        return f"[RateLimit: {info.status}] {window}{util}"
+
 
 Message = (
     UserMessage

tests/test_types.py

@@ -14,6 +14,13 @@
 from claude_agent_sdk.types import (
     PostToolUseHookSpecificOutput,
     PreToolUseHookSpecificOutput,
+    RateLimitEvent,
+    RateLimitInfo,
+    StreamEvent,
+    SystemMessage,
+    TaskNotificationMessage,
+    TaskProgressMessage,
+    TaskStartedMessage,
     TextBlock,
     ThinkingBlock,
     ToolResultBlock,
@@ -619,3 +626,186 @@ def test_new_fields_omitted_when_none(self):
         assert "background" not in payload
         assert "effort" not in payload
         assert "permissionMode" not in payload
+
+
+class TestMessageStr:
+    """Test __str__ methods produce human-readable output for print()."""
+
+    def test_text_block_str(self):
+        assert str(TextBlock(text="hello")) == "hello"
+
+    def test_thinking_block_str(self):
+        assert str(ThinkingBlock(thinking="hmm", signature="sig")) == "[Thinking] hmm"
+
+    def test_tool_use_block_str(self):
+        block = ToolUseBlock(id="t1", name="Read", input={"path": "/x"})
+        assert str(block) == "[Tool use: Read] {'path': '/x'}"
+
+    def test_tool_use_block_str_truncates_large_input(self):
+        block = ToolUseBlock(id="t1", name="Write", input={"content": "x" * 5000})
+        out = str(block)
+        assert out.startswith("[Tool use: Write] ")
+        assert out.endswith("...")
+        assert len(out) < 300
+
+    def test_tool_result_block_str(self):
+        block = ToolResultBlock(tool_use_id="t1", content="ok")
+        assert str(block) == "[Tool result] ok"
+
+    def test_tool_result_block_str_error(self):
+        block = ToolResultBlock(tool_use_id="t1", content="boom", is_error=True)
+        assert str(block) == "[Tool error] boom"
+
+    def test_tool_result_block_str_none_content(self):
+        block = ToolResultBlock(tool_use_id="t1")
+        assert str(block) == "[Tool result]"
+
+    def test_tool_result_block_str_truncates_large_content(self):
+        block = ToolResultBlock(tool_use_id="t1", content="y" * 5000)
+        out = str(block)
+        assert out.startswith("[Tool result] ")
+        assert out.endswith("...")
+        assert len(out) < 300
+
+    def test_user_message_str_from_string(self):
+        assert str(UserMessage(content="hi")) == "User: hi"
+
+    def test_user_message_str_from_blocks(self):
+        msg = UserMessage(content=[TextBlock(text="a"), TextBlock(text="b")])
+        assert str(msg) == "User: a b"
+
+    def test_user_message_str_empty(self):
+        assert str(UserMessage(content="")) == "User:"
+        assert str(UserMessage(content=[])) == "User:"
+
+    def test_assistant_message_str(self):
+        msg = AssistantMessage(
+            content=[
+                TextBlock(text="sure,"),
+                ToolUseBlock(id="t1", name="Read", input={"path": "/x"}),
+            ],
+            model="claude-opus-4-1-20250805",
+        )
+        assert str(msg) == "Claude: sure, [Tool use: Read] {'path': '/x'}"
+
+    def test_assistant_message_str_preserves_block_order(self):
+        """Mixed-block AssistantMessage renders blocks in their list order."""
+        msg = AssistantMessage(
+            content=[
+                ThinkingBlock(thinking="hmm", signature="sig"),
+                TextBlock(text="answer"),
+                ToolUseBlock(id="t1", name="Bash", input={"cmd": "ls"}),
+            ],
+            model="claude-opus-4-1-20250805",
+        )
+        assert str(msg) == (
+            "Claude: [Thinking] hmm answer [Tool use: Bash] {'cmd': 'ls'}"
+        )
+
+    def test_assistant_message_str_empty(self):
+        msg = AssistantMessage(content=[], model="claude-opus-4-1-20250805")
+        assert str(msg) == "Claude:"
+
+    def test_system_message_str(self):
+        assert str(SystemMessage(subtype="init", data={})) == "[System: init]"
+
+    def test_task_started_message_str(self):
+        msg = TaskStartedMessage(
+            subtype="task_started",
+            data={},
+            task_id="tsk_1",
+            description="Refactor tests",
+            uuid="u",
+            session_id="s",
+        )
+        assert str(msg) == "[Task started] Refactor tests (task_id=tsk_1)"
+
+    def test_task_progress_message_str(self):
+        msg = TaskProgressMessage(
+            subtype="task_progress",
+            data={},
+            task_id="tsk_1",
+            description="Refactor tests",
+            usage={"total_tokens": 1200, "tool_uses": 3, "duration_ms": 5000},
+            uuid="u",
+            session_id="s",
+        )
+        assert str(msg) == (
+            "[Task progress] Refactor tests (task_id=tsk_1, tokens=1200)"
+        )
+
+    def test_task_notification_message_str(self):
+        msg = TaskNotificationMessage(
+            subtype="task_notification",
+            data={},
+            task_id="tsk_1",
+            status="completed",
+            output_file="/tmp/out",
+            summary="All green",
+            uuid="u",
+            session_id="s",
+        )
+        assert str(msg) == "[Task completed] All green (task_id=tsk_1)"
+
+    def test_stream_event_str(self):
+        event = StreamEvent(
+            uuid="u", session_id="s", event={"type": "content_block_delta"}
+        )
+        assert str(event) == "[StreamEvent: content_block_delta]"
+
+    def test_stream_event_str_unknown_type(self):
+        event = StreamEvent(uuid="u", session_id="s", event={})
+        assert str(event) == "[StreamEvent: unknown]"
+
+    def test_rate_limit_event_str(self):
+        info = RateLimitInfo(
+            status="allowed_warning",
+            rate_limit_type="five_hour",
+            utilization=0.85,
+        )
+        event = RateLimitEvent(rate_limit_info=info, uuid="u", session_id="s")
+        assert str(event) == "[RateLimit: allowed_warning] five_hour @ 85%"
+
+    def test_rate_limit_event_str_without_utilization(self):
+        info = RateLimitInfo(status="rejected", rate_limit_type="overage")
+        event = RateLimitEvent(rate_limit_info=info, uuid="u", session_id="s")
+        assert str(event) == "[RateLimit: rejected] overage"
+
+    def test_result_message_str(self):
+        msg = ResultMessage(
+            subtype="success",
+            duration_ms=123,
+            duration_api_ms=100,
+            is_error=False,
+            num_turns=1,
+            session_id="s1",
+            total_cost_usd=0.0015,
+        )
+        assert str(msg) == "[Result: success] duration=123ms, cost=$0.0015"
+
+    def test_result_message_str_without_cost(self):
+        msg = ResultMessage(
+            subtype="success",
+            duration_ms=50,
+            duration_api_ms=40,
+            is_error=False,
+            num_turns=1,
+            session_id="s1",
+        )
+        assert str(msg) == "[Result: success] duration=50ms"
+
+    def test_result_message_str_error(self):
+        msg = ResultMessage(
+            subtype="success",
+            duration_ms=10,
+            duration_api_ms=5,
+            is_error=True,
+            num_turns=1,
+            session_id="s1",
+        )
+        assert str(msg) == "[Result: error] duration=10ms"
+
+    def test_dataclass_repr_preserved(self):
+        """Default dataclass __repr__ is still available for debugging."""
+        block = TextBlock(text="hello")
+        assert repr(block) == "TextBlock(text='hello')"