Skip to content
Open
Show file tree
Hide file tree
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
74 changes: 74 additions & 0 deletions src/claude_agent_sdk/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -855,13 +855,21 @@ 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:
"""Text content block."""

text: str

def __repr__(self) -> str:
return f"TextBlock(text={_truncate(self.text)!r})"


@dataclass
class ThinkingBlock:
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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",
Expand Down Expand Up @@ -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:
Expand All @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand All @@ -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."""
Expand Down Expand Up @@ -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):
Expand All @@ -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):
Expand All @@ -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):
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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"]
Expand Down Expand Up @@ -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:
Expand All @@ -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
Expand Down
Loading