Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 63 additions & 0 deletions src/claude_agent_sdk/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -768,6 +768,11 @@ class TextBlock:

text: str

def __repr__(self) -> str:
preview = self.text[:60].replace("\n", "\\n")
suffix = "..." if len(self.text) > 60 else ""
return f"TextBlock(text={preview!r}{suffix})"


@dataclass
class ThinkingBlock:
Expand All @@ -776,6 +781,11 @@ class ThinkingBlock:
thinking: str
signature: str

def __repr__(self) -> str:
preview = self.thinking[:60].replace("\n", "\\n")
suffix = "..." if len(self.thinking) > 60 else ""
return f"ThinkingBlock(thinking={preview!r}{suffix})"


@dataclass
class ToolUseBlock:
Expand All @@ -785,6 +795,10 @@ class ToolUseBlock:
name: str
input: dict[str, Any]

def __repr__(self) -> str:
keys = list(self.input.keys())
return f"ToolUseBlock(name={self.name!r}, input_keys={keys})"


@dataclass
class ToolResultBlock:
Expand All @@ -794,6 +808,18 @@ class ToolResultBlock:
content: str | list[dict[str, Any]] | None = None
is_error: bool | None = None

def __repr__(self) -> str:
if isinstance(self.content, str):
preview = self.content[:60].replace("\n", "\\n")
suffix = "..." if len(self.content) > 60 else ""
content_repr = f"{preview!r}{suffix}"
elif isinstance(self.content, list):
content_repr = f"[{len(self.content)} block(s)]"
else:
content_repr = "None"
error_suffix = ", is_error=True" if self.is_error else ""
return f"ToolResultBlock(tool_use_id={self.tool_use_id!r}, content={content_repr}{error_suffix})"


ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

Expand All @@ -818,6 +844,15 @@ class UserMessage:
parent_tool_use_id: str | None = None
tool_use_result: dict[str, Any] | None = None

def __repr__(self) -> str:
if isinstance(self.content, str):
preview = self.content[:80].replace("\n", "\\n")
suffix = "..." if len(self.content) > 80 else ""
content_repr = f"{preview!r}{suffix}"
else:
content_repr = f"[{len(self.content)} block(s)]"
return f"UserMessage(content={content_repr})"


@dataclass
class AssistantMessage:
Expand All @@ -833,6 +868,26 @@ class AssistantMessage:
session_id: str | None = None
uuid: str | None = None

def __repr__(self) -> str:
text_preview = next(
(b.text[:60].replace("\n", "\\n") for b in self.content if isinstance(b, TextBlock)),
None,
)
if text_preview is not None:
has_more_text = any(
isinstance(b, TextBlock) and len(b.text) > 60 for b in self.content
)
content_repr = f"'{text_preview}{'...' if has_more_text else ''}'"
else:
content_repr = f"[{len(self.content)} block(s)]"
extras = []
if self.stop_reason:
extras.append(f"stop_reason={self.stop_reason!r}")
if self.error:
extras.append(f"error={self.error!r}")
extra_str = (", " + ", ".join(extras)) if extras else ""
return f"AssistantMessage(model={self.model!r}, content={content_repr}{extra_str})"


@dataclass
class SystemMessage:
Expand All @@ -841,6 +896,9 @@ class SystemMessage:
subtype: str
data: dict[str, Any]

def __repr__(self) -> str:
return f"SystemMessage(subtype={self.subtype!r})"


class TaskUsage(TypedDict):
"""Usage statistics reported in task_progress and task_notification messages."""
Expand Down Expand Up @@ -928,6 +986,11 @@ class ResultMessage:
errors: list[str] | None = None
uuid: str | None = None

def __repr__(self) -> str:
status = "error" if self.is_error else "ok"
cost = f", cost=${self.total_cost_usd:.4f}" if self.total_cost_usd is not None else ""
return f"ResultMessage(status={status!r}, turns={self.num_turns}{cost})"


@dataclass
class StreamEvent:
Expand Down