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
106 changes: 92 additions & 14 deletions src/claude_agent_sdk/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -731,38 +731,67 @@ class SandboxSettings(TypedDict, total=False):


# Content block types
@dataclass


def _truncate(text: str, max_length: int = 80) -> str:
"""Truncate text for repr display."""
if len(text) <= max_length:
return text
return text[: max_length - 3] + "..."


@dataclass(repr=False)
class TextBlock:
"""Text content block."""

text: str

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

@dataclass

@dataclass(repr=False)
class ThinkingBlock:
"""Thinking content block."""

thinking: str
signature: str

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

@dataclass

@dataclass(repr=False)
class ToolUseBlock:
"""Tool use content block."""

id: str
name: str
input: dict[str, Any]

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

@dataclass

@dataclass(repr=False)
class ToolResultBlock:
"""Tool result content block."""

tool_use_id: str
content: str | list[dict[str, Any]] | None = None
is_error: bool | None = None

def __repr__(self) -> str:
parts = [f"tool_use_id={self.tool_use_id!r}"]
if self.content is not None:
if isinstance(self.content, str):
parts.append(f"content={_truncate(self.content)!r}")
else:
parts.append(f"content={self.content!r}")
if self.is_error:
parts.append("is_error=True")
return f"ToolResultBlock({', '.join(parts)})"


ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

Expand All @@ -778,7 +807,7 @@ class ToolResultBlock:
]


@dataclass
@dataclass(repr=False)
class UserMessage:
"""User message."""

Expand All @@ -787,8 +816,13 @@ 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):
return f"UserMessage(content={_truncate(self.content)!r})"
return f"UserMessage(content={self.content!r})"

@dataclass

@dataclass(repr=False)
class AssistantMessage:
"""Assistant message with content blocks."""

Expand All @@ -798,14 +832,23 @@ class AssistantMessage:
error: AssistantMessageError | None = None
usage: dict[str, Any] | None = None

def __repr__(self) -> str:
parts = [f"model={self.model!r}", f"content={self.content!r}"]
if self.error is not None:
parts.append(f"error={self.error!r}")
return f"AssistantMessage({', '.join(parts)})"

@dataclass

@dataclass(repr=False)
class SystemMessage:
"""System message with metadata."""

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 All @@ -819,7 +862,7 @@ class TaskUsage(TypedDict):
TaskNotificationStatus = Literal["completed", "failed", "stopped"]


@dataclass
@dataclass(repr=False)
class TaskStartedMessage(SystemMessage):
"""System message emitted when a task starts.

Expand All @@ -835,8 +878,11 @@ class TaskStartedMessage(SystemMessage):
tool_use_id: str | None = None
task_type: str | None = None

def __repr__(self) -> str:
return f"TaskStartedMessage(task_id={self.task_id!r}, description={_truncate(self.description)!r})"

@dataclass

@dataclass(repr=False)
class TaskProgressMessage(SystemMessage):
"""System message emitted while a task is in progress.

Expand All @@ -853,8 +899,11 @@ class TaskProgressMessage(SystemMessage):
tool_use_id: str | None = None
last_tool_name: str | None = None

def __repr__(self) -> str:
return f"TaskProgressMessage(task_id={self.task_id!r}, description={_truncate(self.description)!r})"

@dataclass

@dataclass(repr=False)
class TaskNotificationMessage(SystemMessage):
"""System message emitted when a task completes, fails, or is stopped.

Expand All @@ -872,8 +921,13 @@ class TaskNotificationMessage(SystemMessage):
tool_use_id: str | None = None
usage: TaskUsage | None = None

def __repr__(self) -> str:
return (
f"TaskNotificationMessage(task_id={self.task_id!r}, status={self.status!r})"
)

@dataclass

@dataclass(repr=False)
class ResultMessage:
"""Result message with cost and usage information."""

Expand All @@ -889,8 +943,18 @@ class ResultMessage:
result: str | None = None
structured_output: Any = None

def __repr__(self) -> str:
parts = [f"num_turns={self.num_turns}"]
if self.is_error:
parts.append("is_error=True")
if self.total_cost_usd is not None:
parts.append(f"total_cost_usd={self.total_cost_usd}")
if self.stop_reason is not None:
parts.append(f"stop_reason={self.stop_reason!r}")
return f"ResultMessage({', '.join(parts)})"

@dataclass

@dataclass(repr=False)
class StreamEvent:
"""Stream event for partial message updates during streaming."""

Expand All @@ -899,6 +963,9 @@ class StreamEvent:
event: dict[str, Any] # The raw Anthropic API stream event
parent_tool_use_id: str | None = None

def __repr__(self) -> str:
return f"StreamEvent(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 All @@ -907,7 +974,7 @@ class StreamEvent:
]


@dataclass
@dataclass(repr=False)
class RateLimitInfo:
"""Rate limit status emitted by the CLI when rate limit state changes.

Expand All @@ -932,8 +999,16 @@ class RateLimitInfo:
overage_disabled_reason: str | None = None
raw: dict[str, Any] = field(default_factory=dict)

def __repr__(self) -> str:
parts = [f"status={self.status!r}"]
if self.utilization is not None:
parts.append(f"utilization={self.utilization}")
if self.rate_limit_type is not None:
parts.append(f"rate_limit_type={self.rate_limit_type!r}")
return f"RateLimitInfo({', '.join(parts)})"

@dataclass

@dataclass(repr=False)
class RateLimitEvent:
"""Rate limit event emitted when rate limit info changes.

Expand All @@ -946,6 +1021,9 @@ class RateLimitEvent:
uuid: str
session_id: str

def __repr__(self) -> str:
return f"RateLimitEvent(status={self.rate_limit_info.status!r})"


Message = (
UserMessage
Expand Down
Loading