From 73e3640def17eea0be8a91b71f36fc1650e6c9f5 Mon Sep 17 00:00:00 2001 From: Junhyuk Lee Date: Wed, 22 Apr 2026 23:45:34 -0500 Subject: [PATCH 1/3] feat(types): add __repr__ to content block and message types Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- src/claude_agent_sdk/types.py | 74 +++++++++++++++++++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/src/claude_agent_sdk/types.py b/src/claude_agent_sdk/types.py index a5d7185b..c1e3ec3b 100644 --- a/src/claude_agent_sdk/types.py +++ b/src/claude_agent_sdk/types.py @@ -855,6 +855,11 @@ class SandboxSettings(TypedDict, total=False): enableWeakerNestedSandbox: bool +def _truncate(s: str, max_len: int = 100) -> str: + """Truncate a string to max_len chars, appending '...' if truncated.""" + return s if len(s) <= max_len else s[:max_len] + "..." + + # Content block types @dataclass class TextBlock: @@ -862,6 +867,9 @@ class TextBlock: text: str + def __repr__(self) -> str: + return f"TextBlock(text={_truncate(self.text)!r})" + @dataclass class ThinkingBlock: @@ -870,6 +878,9 @@ class ThinkingBlock: thinking: str signature: str + def __repr__(self) -> str: + return f"ThinkingBlock(thinking={_truncate(self.thinking)!r})" + @dataclass class ToolUseBlock: @@ -879,6 +890,10 @@ class ToolUseBlock: name: str input: dict[str, Any] + def __repr__(self) -> str: + input_repr = _truncate(repr(self.input)) + return f"ToolUseBlock(id={self.id!r}, name={self.name!r}, input={input_repr})" + @dataclass class ToolResultBlock: @@ -888,6 +903,10 @@ class ToolResultBlock: content: str | list[dict[str, Any]] | None = None is_error: bool | None = None + def __repr__(self) -> str: + content_repr = _truncate(repr(self.content)) + return f"ToolResultBlock(tool_use_id={self.tool_use_id!r}, is_error={self.is_error!r}, content={content_repr})" + ServerToolName = Literal[ "advisor", @@ -915,6 +934,10 @@ class ServerToolUseBlock: name: ServerToolName input: dict[str, Any] + def __repr__(self) -> str: + input_repr = _truncate(repr(self.input)) + return f"ServerToolUseBlock(id={self.id!r}, name={self.name!r}, input={input_repr})" + @dataclass class ServerToolResultBlock: @@ -928,6 +951,10 @@ class ServerToolResultBlock: tool_use_id: str content: dict[str, Any] + def __repr__(self) -> str: + content_repr = _truncate(repr(self.content)) + return f"ServerToolResultBlock(tool_use_id={self.tool_use_id!r}, content={content_repr})" + ContentBlock = ( TextBlock @@ -959,6 +986,14 @@ class UserMessage: parent_tool_use_id: str | None = None tool_use_result: dict[str, Any] | None = None + def __repr__(self) -> str: + content_summary = ( + f"[{len(self.content)} items]" + if isinstance(self.content, list) + else _truncate(repr(self.content)) + ) + return f"UserMessage(content={content_summary}, uuid={self.uuid!r})" + @dataclass class AssistantMessage: @@ -974,6 +1009,12 @@ class AssistantMessage: session_id: str | None = None uuid: str | None = None + def __repr__(self) -> str: + return ( + f"AssistantMessage(model={self.model!r}, stop_reason={self.stop_reason!r}," + f" content=[{len(self.content)} items])" + ) + @dataclass class SystemMessage: @@ -982,6 +1023,9 @@ class SystemMessage: subtype: str data: dict[str, Any] + def __repr__(self) -> str: + return f"SystemMessage(subtype={self.subtype!r}, data={_truncate(repr(self.data))})" + class TaskUsage(TypedDict): """Usage statistics reported in task_progress and task_notification messages.""" @@ -1011,6 +1055,9 @@ class TaskStartedMessage(SystemMessage): tool_use_id: str | None = None task_type: str | None = None + def __repr__(self) -> str: + return f"TaskStartedMessage(subtype={self.subtype!r}, session_id={self.session_id!r})" + @dataclass class TaskProgressMessage(SystemMessage): @@ -1029,6 +1076,9 @@ class TaskProgressMessage(SystemMessage): tool_use_id: str | None = None last_tool_name: str | None = None + def __repr__(self) -> str: + return f"TaskProgressMessage(subtype={self.subtype!r}, description={_truncate(self.description)})" + @dataclass class TaskNotificationMessage(SystemMessage): @@ -1048,6 +1098,11 @@ class TaskNotificationMessage(SystemMessage): tool_use_id: str | None = None usage: TaskUsage | None = None + def __repr__(self) -> str: + return ( + f"TaskNotificationMessage(subtype={self.subtype!r}, status={self.status!r})" + ) + @dataclass class MirrorErrorMessage(SystemMessage): @@ -1065,6 +1120,9 @@ class MirrorErrorMessage(SystemMessage): key: "SessionKey | None" = None error: str = "" + def __repr__(self) -> str: + return f"MirrorErrorMessage(subtype={self.subtype!r}, error={_truncate(self.error)})" + @dataclass class ResultMessage: @@ -1086,6 +1144,12 @@ class ResultMessage: errors: list[str] | None = None uuid: str | None = None + def __repr__(self) -> str: + return ( + f"ResultMessage(subtype={self.subtype!r}, is_error={self.is_error!r}," + f" duration_ms={self.duration_ms!r}, session_id={self.session_id!r})" + ) + @dataclass class StreamEvent: @@ -1096,6 +1160,10 @@ class StreamEvent: event: dict[str, Any] # The raw Anthropic API stream event parent_tool_use_id: str | None = None + def __repr__(self) -> str: + event_type = self.event.get("type") + return f"StreamEvent(event_type={event_type!r}, session_id={self.session_id!r})" + # Rate limit types — see https://docs.claude.com/en/docs/claude-code/rate-limits RateLimitStatus = Literal["allowed", "allowed_warning", "rejected"] @@ -1129,6 +1197,9 @@ class RateLimitInfo: overage_disabled_reason: str | None = None raw: dict[str, Any] = field(default_factory=dict) + def __repr__(self) -> str: + return f"RateLimitInfo(status={self.status!r}, raw={_truncate(repr(self.raw))})" + @dataclass class RateLimitEvent: @@ -1143,6 +1214,9 @@ class RateLimitEvent: uuid: str session_id: str + def __repr__(self) -> str: + return f"RateLimitEvent(rate_limit_info={self.rate_limit_info!r})" + Message = ( UserMessage From 39042ff5bdba99f4381850cebda3a9726971da4e Mon Sep 17 00:00:00 2001 From: Junhyuk Lee Date: Wed, 22 Apr 2026 23:45:41 -0500 Subject: [PATCH 2/3] test(types): add repr tests for all message types Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- tests/test_types.py | 222 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 222 insertions(+) diff --git a/tests/test_types.py b/tests/test_types.py index fbd07509..dbb59cd9 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -12,8 +12,19 @@ SubagentStartHookSpecificOutput, ) from claude_agent_sdk.types import ( + MirrorErrorMessage, PostToolUseHookSpecificOutput, PreToolUseHookSpecificOutput, + RateLimitEvent, + RateLimitInfo, + ServerToolResultBlock, + ServerToolUseBlock, + StreamEvent, + SystemMessage, + TaskNotificationMessage, + TaskProgressMessage, + TaskStartedMessage, + TaskUsage, TextBlock, ThinkingBlock, ToolResultBlock, @@ -619,3 +630,214 @@ def test_new_fields_omitted_when_none(self): assert "background" not in payload assert "effort" not in payload assert "permissionMode" not in payload + + +class TestRepr: + """Tests for __repr__ methods on all 17 repr-carrying types.""" + + def test_text_block_repr(self) -> None: + """TextBlock repr includes class name and truncated text field.""" + obj = TextBlock(text="Hello, world!") + r = repr(obj) + assert r.startswith("TextBlock(") + assert "Hello, world!" in r + + def test_thinking_block_repr(self) -> None: + """ThinkingBlock repr includes class name and truncated thinking field.""" + obj = ThinkingBlock( + thinking="I am reasoning step by step.", signature="sig-abc" + ) + r = repr(obj) + assert r.startswith("ThinkingBlock(") + assert "I am reasoning step by step." in r + + def test_tool_use_block_repr(self) -> None: + """ToolUseBlock repr includes id, name, and input.""" + obj = ToolUseBlock( + id="toolu_abc123", name="Read", input={"file_path": "/src/main.py"} + ) + r = repr(obj) + assert r.startswith("ToolUseBlock(") + assert "toolu_abc123" in r + assert "Read" in r + assert "input=" in r + + def test_tool_result_block_repr(self) -> None: + """ToolResultBlock repr includes tool_use_id and is_error.""" + obj = ToolResultBlock( + tool_use_id="toolu_abc123", content="file contents", is_error=False + ) + r = repr(obj) + assert r.startswith("ToolResultBlock(") + assert "toolu_abc123" in r + assert "False" in r + + def test_server_tool_use_block_repr(self) -> None: + """ServerToolUseBlock repr includes id, name, and input.""" + obj = ServerToolUseBlock( + id="srv-001", name="web_search", input={"query": "python repr"} + ) + r = repr(obj) + assert r.startswith("ServerToolUseBlock(") + assert "srv-001" in r + assert "web_search" in r + + def test_server_tool_result_block_repr(self) -> None: + """ServerToolResultBlock repr includes tool_use_id and content.""" + obj = ServerToolResultBlock( + tool_use_id="srv-001", content={"type": "text", "text": "results here"} + ) + r = repr(obj) + assert r.startswith("ServerToolResultBlock(") + assert "srv-001" in r + assert "content=" in r + + def test_user_message_repr_str_branch(self) -> None: + """UserMessage repr (str content) includes truncated content and uuid.""" + obj = UserMessage(content="Hello, Claude!", uuid="uuid-001") + r = repr(obj) + assert r.startswith("UserMessage(") + assert "Hello, Claude!" in r + assert "uuid-001" in r + + def test_user_message_repr_list_branch(self) -> None: + """UserMessage repr (list content) shows item count instead of raw content.""" + blocks = [TextBlock(text="hi"), TextBlock(text="there")] + obj = UserMessage(content=blocks, uuid="uuid-002") + r = repr(obj) + assert r.startswith("UserMessage(") + assert "[2 items]" in r + assert "uuid-002" in r + + def test_assistant_message_repr(self) -> None: + """AssistantMessage repr includes model, stop_reason, and content count.""" + obj = AssistantMessage( + content=[TextBlock(text="Here is my answer.")], + model="claude-opus-4-5", + stop_reason="end_turn", + ) + r = repr(obj) + assert r.startswith("AssistantMessage(") + assert "claude-opus-4-5" in r + assert "end_turn" in r + assert "[1 items]" in r + + def test_system_message_repr(self) -> None: + """SystemMessage repr includes subtype and truncated data.""" + obj = SystemMessage( + subtype="init", data={"session": "sess-1", "model": "claude"} + ) + r = repr(obj) + assert r.startswith("SystemMessage(") + assert "init" in r + assert "data=" in r + + def test_result_message_repr(self) -> None: + """ResultMessage repr includes subtype, is_error, duration_ms, session_id.""" + obj = ResultMessage( + subtype="success", + duration_ms=1500, + duration_api_ms=1200, + is_error=False, + num_turns=3, + session_id="sess-xyz", + ) + r = repr(obj) + assert r.startswith("ResultMessage(") + assert "success" in r + assert "False" in r + assert "1500" in r + assert "sess-xyz" in r + + def test_task_started_message_repr(self) -> None: + """TaskStartedMessage repr includes subtype and session_id.""" + obj = TaskStartedMessage( + subtype="task_started", + data={}, + task_id="task-001", + description="Run linter", + uuid="uuid-ts", + session_id="sess-ts", + ) + r = repr(obj) + assert r.startswith("TaskStartedMessage(") + assert "task_started" in r + assert "sess-ts" in r + + def test_task_progress_message_repr(self) -> None: + """TaskProgressMessage repr includes subtype and description.""" + usage: TaskUsage = {"total_tokens": 500, "tool_uses": 2, "duration_ms": 800} + obj = TaskProgressMessage( + subtype="task_progress", + data={}, + task_id="task-001", + description="Running tests now", + usage=usage, + uuid="uuid-tp", + session_id="sess-tp", + ) + r = repr(obj) + assert r.startswith("TaskProgressMessage(") + assert "task_progress" in r + assert "Running tests now" in r + + def test_task_notification_message_repr(self) -> None: + """TaskNotificationMessage repr includes subtype and status.""" + obj = TaskNotificationMessage( + subtype="task_notification", + data={}, + task_id="task-001", + status="completed", + output_file="/tmp/output.txt", + summary="All done", + uuid="uuid-tn", + session_id="sess-tn", + ) + r = repr(obj) + assert r.startswith("TaskNotificationMessage(") + assert "task_notification" in r + assert "completed" in r + + def test_mirror_error_message_repr(self) -> None: + """MirrorErrorMessage repr includes subtype and error.""" + obj = MirrorErrorMessage( + subtype="mirror_error", + data={}, + error="DB write timed out", + ) + r = repr(obj) + assert r.startswith("MirrorErrorMessage(") + assert "mirror_error" in r + assert "DB write timed out" in r + + def test_stream_event_repr(self) -> None: + """StreamEvent repr includes event_type and session_id.""" + obj = StreamEvent( + uuid="uuid-se", + session_id="sess-se", + event={"type": "content_block_delta", "index": 0}, + ) + r = repr(obj) + assert r.startswith("StreamEvent(") + assert "content_block_delta" in r + assert "sess-se" in r + + def test_rate_limit_info_repr(self) -> None: + """RateLimitInfo repr includes status and raw.""" + obj = RateLimitInfo(status="allowed_warning", raw={"utilization": 0.85}) + r = repr(obj) + assert r.startswith("RateLimitInfo(") + assert "allowed_warning" in r + assert "raw=" in r + + def test_rate_limit_event_repr(self) -> None: + """RateLimitEvent repr embeds the RateLimitInfo repr.""" + info = RateLimitInfo(status="rejected", raw={}) + obj = RateLimitEvent( + rate_limit_info=info, uuid="uuid-rle", session_id="sess-rle" + ) + r = repr(obj) + assert r.startswith("RateLimitEvent(") + assert "RateLimitInfo(" in r + assert "rejected" in r + From ac2c3e4971f6ab2f46e9d17ea4275d8b95101a35 Mon Sep 17 00:00:00 2001 From: Junhyuk Lee Date: Thu, 23 Apr 2026 16:06:28 -0500 Subject: [PATCH 3/3] test(types): add edge case tests for truncation, None, empty list, and unicode --- tests/test_types.py | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/tests/test_types.py b/tests/test_types.py index dbb59cd9..d4d47090 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -841,3 +841,37 @@ def test_rate_limit_event_repr(self) -> None: assert "RateLimitInfo(" in r assert "rejected" in r + def test_tool_use_block_repr_truncation(self) -> None: + """ToolUseBlock repr truncates large input dicts with '...'.""" + obj = ToolUseBlock(id="x", name="fn", input={"k": "a" * 200}) + r = repr(obj) + assert "ToolUseBlock" in repr(obj) + assert "..." in r + + def test_tool_result_block_repr_truncation(self) -> None: + """ToolResultBlock repr truncates large content with '...'.""" + obj = ToolResultBlock(tool_use_id="x", content=["a" * 200], is_error=False) + r = repr(obj) + assert "ToolResultBlock" in repr(obj) + assert "..." in r + + def test_repr_none_fields(self) -> None: + """ToolResultBlock repr renders None fields as 'None'.""" + obj = ToolResultBlock(tool_use_id="x", content=None, is_error=None) + r = repr(obj) + assert "ToolResultBlock" in r + assert "None" in r + + def test_repr_empty_list(self) -> None: + """UserMessage repr with empty list content shows [0 items].""" + obj = UserMessage(content=[]) + r = repr(obj) + assert "UserMessage" in r + assert "[0 items]" in r + + def test_repr_unicode(self) -> None: + """TextBlock repr preserves unicode content correctly.""" + obj = TextBlock(text="こんにちは世界") + r = repr(obj) + assert "TextBlock" in r + assert "こんにちは世界" in r