diff --git a/src/claude_agent_sdk/types.py b/src/claude_agent_sdk/types.py index a82a8b9b..13faf29e 100644 --- a/src/claude_agent_sdk/types.py +++ b/src/claude_agent_sdk/types.py @@ -855,6 +855,18 @@ 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: @@ -862,6 +874,9 @@ class TextBlock: 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 diff --git a/tests/test_types.py b/tests/test_types.py index fbd07509..f298b6d9 100644 --- a/tests/test_types.py +++ b/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')"