Message Types

Message types for Claude Agent SDK conversations

Overview

The Claude Agent SDK uses strongly-typed message classes to represent conversation turns, system events, and streaming updates.

Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent

UserMessage

Represents a message from the user.

@dataclass
class UserMessage:
    content: str | list[ContentBlock]
    uuid: str | None = None
    parent_tool_use_id: str | None = None
    tool_use_result: dict[str, Any] | None = None

content · str | list[ContentBlock] · required

Message content as a string or list of content blocks.

uuid · str | None

Unique message identifier.

parent_tool_use_id · str | None

ID of the parent tool use if this is a tool result message.

tool_use_result · dict[str, Any] | None

Tool result data if this message contains tool output.

Example

message = UserMessage(
    content="What files are in the current directory?",
    uuid="msg-123"
)

await client.send_message(message)

AssistantMessage

Represents a response from Claude.

@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str
    parent_tool_use_id: str | None = None
    error: AssistantMessageError | None = None

content · list[ContentBlock] · required

List of content blocks (text, thinking, tool use, etc.).

See Content Blocks for details.

model · str · required

Model that generated the response (e.g., "claude-sonnet-4-20250514").

parent_tool_use_id · str | None

ID of the parent tool use if this is from a sub-agent.

error · AssistantMessageError | None

Error type if the message failed.

• "authentication_failed"
• "billing_error"
• "rate_limit"
• "invalid_request"
• "server_error"
• "unknown"

Example

async for message in client:
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                print(block.text)
            elif isinstance(block, ToolUseBlock):
                print(f"Using tool: {block.name}")

SystemMessage

Represents system-level events and metadata.

@dataclass
class SystemMessage:
    subtype: str
    data: dict[str, Any]

subtype · str · required

System message type (e.g., "task_started", "task_progress", "task_notification").

data · dict[str, Any] · required

Message-specific data payload.

Task Messages

The SDK provides specialized subclasses for task-related system messages:

Show TaskStartedMessage

@dataclass
class TaskStartedMessage(SystemMessage):
    task_id: str
    description: str
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    task_type: str | None = None

Emitted when a task (sub-agent) starts.

Show TaskProgressMessage

@dataclass
class TaskProgressMessage(SystemMessage):
    task_id: str
    description: str
    usage: TaskUsage
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    last_tool_name: str | None = None

Emitted during task execution with usage statistics.

TaskUsage contains:

• total_tokens: int
• tool_uses: int
• duration_ms: int

Show TaskNotificationMessage

@dataclass
class TaskNotificationMessage(SystemMessage):
    task_id: str
    status: TaskNotificationStatus  # "completed" | "failed" | "stopped"
    output_file: str
    summary: str
    uuid: str
    session_id: str
    tool_use_id: str | None = None
    usage: TaskUsage | None = None

Emitted when a task completes, fails, or is stopped.

Example

async for message in client:
    if isinstance(message, TaskStartedMessage):
        print(f"Task started: {message.description}")
    elif isinstance(message, TaskProgressMessage):
        print(f"Progress: {message.usage['total_tokens']} tokens used")
    elif isinstance(message, TaskNotificationMessage):
        print(f"Task {message.status}: {message.summary}")

ResultMessage

Final result message with cost and usage information.

@dataclass
class ResultMessage:
    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    stop_reason: str | None = None
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None
    structured_output: Any = None

subtype · str · required

Result type (typically "result").

duration_ms · int · required

Total session duration in milliseconds.

duration_api_ms · int · required

API call duration in milliseconds.

is_error · bool · required

Whether the session ended with an error.

num_turns · int · required

Number of conversation turns.

session_id · str · required

Session identifier.

stop_reason · str | None

Reason for stopping (e.g., "max_turns", "end_turn").

total_cost_usd · float | None

Total cost in USD.

usage · dict[str, Any] | None

Token usage statistics.

result · str | None

Final result text.

structured_output · Any

Structured output if output_format was specified in options.

Example

result = await query("Analyze this codebase", options)

if isinstance(result, ResultMessage):
    print(f"Completed in {result.duration_ms}ms")
    print(f"Used {result.num_turns} turns")
    print(f"Cost: ${result.total_cost_usd:.4f}")
    print(f"Result: {result.result}")

StreamEvent

Partial message update during streaming (requires include_partial_messages=True).

@dataclass
class StreamEvent:
    uuid: str
    session_id: str
    event: dict[str, Any]  # Raw Anthropic API stream event
    parent_tool_use_id: str | None = None

uuid · str · required

Message UUID being updated.

session_id · str · required

Session identifier.

event · dict[str, Any] · required

Raw Anthropic API stream event data.

parent_tool_use_id · str | None

Parent tool use ID if from a sub-agent.

Example

options = ClaudeAgentOptions(include_partial_messages=True)
client = ClaudeSDKClient(options)

async for message in client:
    if isinstance(message, StreamEvent):
        # Handle partial updates
        if message.event.get("type") == "content_block_delta":
            delta = message.event.get("delta", {})
            if "text" in delta:
                print(delta["text"], end="", flush=True)

Session History Types

Types for reading historical session data:

SDKSessionInfo

@dataclass
class SDKSessionInfo:
    session_id: str
    summary: str
    last_modified: int
    file_size: int
    custom_title: str | None = None
    first_prompt: str | None = None
    git_branch: str | None = None
    cwd: str | None = None

Returned by list_sessions() with session metadata.

SessionMessage

@dataclass
class SessionMessage:
    type: Literal["user", "assistant"]
    uuid: str
    session_id: str
    message: Any  # Raw Anthropic API message dict
    parent_tool_use_id: None = None

Returned by get_session_messages() for reading conversation history.