Message Types

Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent
"""
Union of all message types.

When iterating over messages from query() or ClaudeSDKClient, you receive
Message objects which can be one of five types:

- UserMessage: User input messages
- AssistantMessage: Claude's responses with content
- SystemMessage: System notifications and metadata
- ResultMessage: Final result with cost and usage info
- StreamEvent: Partial message updates (when streaming enabled)

Use isinstance() to determine the specific message type and access
type-specific fields.
"""

@dataclass
class UserMessage:
    """
    User message.

    Represents a message sent by the user to Claude. This can be either
    your initial prompt or follow-up messages during conversation.

    Attributes:
        content: The message content (string or list of content blocks)
        parent_tool_use_id: Optional parent tool use ID for context
    """

    content: str | list[ContentBlock]
    """Message content.

    Can be:
    - A string for simple text messages
    - A list of ContentBlock objects for rich content (text, images, etc.)

    Example:
        "What is Python?"
        [TextBlock(text="Analyze this"), ...]
    """

    parent_tool_use_id: str | None = None
    """Optional parent tool use ID.

    When set, indicates this user message is in response to a tool use
    request from Claude. Used for tool result messages and maintaining
    conversation context.
    """

@dataclass
class AssistantMessage:
    """
    Assistant message with content blocks.

    Represents a complete response from Claude. The response is structured
    as a list of content blocks, which can include text, thinking, tool uses,
    and tool results.

    Attributes:
        content: List of content blocks making up the response
        model: The model used to generate this response
        parent_tool_use_id: Optional parent tool use ID for context
    """

    content: list[ContentBlock]
    """List of content blocks.

    Each block represents a different type of content in Claude's response:
    - TextBlock: Normal text output
    - ThinkingBlock: Extended thinking (when enabled)
    - ToolUseBlock: Request to execute a tool
    - ToolResultBlock: Result from a tool execution

    See content-blocks.md for details on each block type.
    """

    model: str
    """Model identifier.

    The Claude model that generated this response. Examples:
    - 'claude-sonnet-4-5-20250929'
    - 'claude-opus-4-1-20250805'
    - 'claude-opus-4-20250514'
    """

    parent_tool_use_id: str | None = None
    """Optional parent tool use ID.

    When set, indicates this assistant message is related to a specific
    tool execution context.
    """

@dataclass
class SystemMessage:
    """
    System message with metadata.

    System messages provide notifications, status updates, and metadata
    about the conversation flow. They're not part of the core user-assistant
    dialogue but provide important context.

    Attributes:
        subtype: Type of system message
        data: Message-specific data payload
    """

    subtype: str
    """System message subtype.

    Identifies the kind of system message. Common subtypes include:
    - 'status': Status updates
    - 'notification': System notifications
    - 'metadata': Additional metadata
    - Other internal message types
    """

    data: dict[str, Any]
    """Message data payload.

    Structure varies by subtype. Contains the actual information for
    this system message.

    Example:
        {"status": "processing", "step": 2, "total": 5}
    """

@dataclass
class ResultMessage:
    """
    Result message with cost and usage information.

    Marks the completion of a conversation turn and provides detailed
    metrics including API usage, costs, duration, and session information.

    This message type signals that Claude has finished processing and
    you can proceed with follow-up actions or end the conversation.

    Attributes:
        subtype: Result message subtype
        duration_ms: Total duration in milliseconds
        duration_api_ms: API call duration in milliseconds
        is_error: Whether an error occurred
        num_turns: Number of conversation turns
        session_id: Session identifier
        total_cost_usd: Total cost in USD (if available)
        usage: Detailed usage information (if available)
        result: Optional result string
    """

    subtype: str
    """Result subtype.

    Identifies the kind of result. Common subtypes:
    - 'success': Normal completion
    - 'error': Error occurred
    - 'interrupted': User interrupted
    - 'max_turns': Hit max turns limit
    """

    duration_ms: int
    """Total duration in milliseconds.

    Time from start of request to completion, including all tool
    executions and API calls.
    """

    duration_api_ms: int
    """API call duration in milliseconds.

    Time spent in Anthropic API calls only, excluding tool execution
    time and other overhead.
    """

    is_error: bool
    """Whether an error occurred.

    True if the conversation ended due to an error, False for normal
    completion.
    """

    num_turns: int
    """Number of conversation turns.

    Count of back-and-forth exchanges in this conversation session.
    One turn = one user message + one assistant response.
    """

    session_id: str
    """Session identifier.

    Unique ID for this conversation session. Use with the `resume`
    option in ClaudeAgentOptions to continue the conversation later.
    """

    total_cost_usd: float | None = None
    """Total cost in USD.

    The total cost of API calls for this conversation turn. May be None
    if cost information is not available or tracking is disabled.

    Example: 0.0025 for $0.0025 (0.25 cents)
    """

    usage: dict[str, Any] | None = None
    """Detailed usage information.

    Token usage and other metrics. Structure varies but typically includes:
    - input_tokens: Number of input tokens
    - output_tokens: Number of output tokens
    - cache_read_tokens: Tokens read from cache
    - cache_creation_tokens: Tokens written to cache

    Example:
        {
            "input_tokens": 1250,
            "output_tokens": 850,
            "cache_read_tokens": 500,
            "cache_creation_tokens": 0
        }

    May be None if usage tracking is disabled.
    """

    result: str | None = None
    """Optional result string.

    Additional result information or error message. Content depends
    on the subtype and completion status.
    """

@dataclass
class StreamEvent:
    """
    Stream event for partial message updates during streaming.

    When include_partial_messages is enabled in ClaudeAgentOptions, the SDK
    emits StreamEvent messages containing partial content as it's generated
    in real-time. This enables live updates in UIs.

    Attributes:
        uuid: Unique event identifier
        session_id: Session identifier
        event: Raw Anthropic API stream event
        parent_tool_use_id: Optional parent tool use ID
    """

    uuid: str
    """Unique event identifier.

    Each stream event has a unique UUID for tracking and deduplication.
    """

    session_id: str
    """Session identifier.

    The session this stream event belongs to. Same as ResultMessage.session_id.
    """

    event: dict[str, Any]
    """Raw Anthropic API stream event.

    The unprocessed stream event from the Anthropic API. Structure follows
    the Anthropic streaming format with types like:
    - content_block_start
    - content_block_delta
    - content_block_stop
    - message_start
    - message_delta
    - message_stop

    Example:
        {
            "type": "content_block_delta",
            "index": 0,
            "delta": {"type": "text_delta", "text": "Hello"}
        }

    See Anthropic's streaming documentation for full event schema.
    """

    parent_tool_use_id: str | None = None
    """Optional parent tool use ID.

    When set, indicates this stream event is related to a specific
    tool execution context.
    """

from claude_agent_sdk import (
    query, UserMessage, AssistantMessage, SystemMessage,
    ResultMessage, TextBlock
)

async for msg in query(prompt="What is Python?"):
    if isinstance(msg, UserMessage):
        print(f"User: {msg.content}")

    elif isinstance(msg, AssistantMessage):
        for block in msg.content:
            if isinstance(block, TextBlock):
                print(f"Claude: {block.text}")

    elif isinstance(msg, SystemMessage):
        print(f"System [{msg.subtype}]: {msg.data}")

    elif isinstance(msg, ResultMessage):
        print(f"Session: {msg.session_id}")
        print(f"Cost: ${msg.total_cost_usd:.4f}")
        print(f"Duration: {msg.duration_ms}ms")

from claude_agent_sdk import query, AssistantMessage, TextBlock

async for msg in query(prompt="Explain decorators"):
    if isinstance(msg, AssistantMessage):
        # Extract all text blocks
        texts = [
            block.text
            for block in msg.content
            if isinstance(block, TextBlock)
        ]

        full_response = "\n".join(texts)
        print(full_response)

from claude_agent_sdk import query, Message

# Collect all messages
messages: list[Message] = []

async for msg in query(prompt="Hello"):
    messages.append(msg)

print(f"Received {len(messages)} messages")

from claude_agent_sdk import (
    query, AssistantMessage, ToolUseBlock, ToolResultBlock,
    ClaudeAgentOptions
)

options = ClaudeAgentOptions(
    allowed_tools=["Bash"]
)

async for msg in query(prompt="List files", options=options):
    if isinstance(msg, AssistantMessage):
        for block in msg.content:
            if isinstance(block, ToolUseBlock):
                print(f"Tool: {block.name}")
                print(f"Input: {block.input}")

            elif isinstance(block, ToolResultBlock):
                print(f"Result: {block.content}")
                if block.is_error:
                    print("(Error occurred)")

from claude_agent_sdk import query, ResultMessage

result_msg = None

async for msg in query(prompt="What is 2 + 2?"):
    if isinstance(msg, ResultMessage):
        result_msg = msg

if result_msg:
    print(f"Session ID: {result_msg.session_id}")
    print(f"Turns: {result_msg.num_turns}")
    print(f"Duration: {result_msg.duration_ms}ms")
    print(f"API time: {result_msg.duration_api_ms}ms")

    if result_msg.total_cost_usd:
        print(f"Cost: ${result_msg.total_cost_usd:.6f}")

    if result_msg.usage:
        print(f"Input tokens: {result_msg.usage.get('input_tokens')}")
        print(f"Output tokens: {result_msg.usage.get('output_tokens')}")

    if result_msg.is_error:
        print(f"Error: {result_msg.result}")

from claude_agent_sdk import query, StreamEvent, ClaudeAgentOptions

options = ClaudeAgentOptions(
    include_partial_messages=True
)

async for msg in query(prompt="Write a long story", options=options):
    if isinstance(msg, StreamEvent):
        # Handle streaming updates
        event = msg.event
        event_type = event.get("type")

        if event_type == "content_block_delta":
            delta = event.get("delta", {})
            if delta.get("type") == "text_delta":
                # Print partial text as it arrives
                print(delta.get("text", ""), end="", flush=True)

        elif event_type == "message_stop":
            print("\n[Stream complete]")

from claude_agent_sdk import query, AssistantMessage, ResultMessage

# Only process assistant messages
assistant_messages = []

async for msg in query(prompt="Explain Python"):
    if isinstance(msg, AssistantMessage):
        assistant_messages.append(msg)

print(f"Claude sent {len(assistant_messages)} responses")

# Get just the result
async for msg in query(prompt="What is 2 + 2?"):
    if isinstance(msg, ResultMessage):
        final_result = msg
        break

from claude_agent_sdk import (
    query, UserMessage, AssistantMessage, TextBlock, ClaudeAgentOptions
)

chat_history = []

# First message
async for msg in query(prompt="What is Python?"):
    if isinstance(msg, UserMessage):
        chat_history.append({"role": "user", "content": msg.content})

    elif isinstance(msg, AssistantMessage):
        texts = [b.text for b in msg.content if isinstance(b, TextBlock)]
        chat_history.append({"role": "assistant", "content": " ".join(texts)})

# Continue conversation
options = ClaudeAgentOptions(continue_conversation=True)

async for msg in query(prompt="Can you show an example?", options=options):
    if isinstance(msg, UserMessage):
        chat_history.append({"role": "user", "content": msg.content})

    elif isinstance(msg, AssistantMessage):
        texts = [b.text for b in msg.content if isinstance(b, TextBlock)]
        chat_history.append({"role": "assistant", "content": " ".join(texts)})

# Display history
for entry in chat_history:
    print(f"{entry['role']}: {entry['content']}")

from claude_agent_sdk import query, ResultMessage, ToolResultBlock, AssistantMessage

has_error = False

async for msg in query(prompt="Delete a non-existent file"):
    if isinstance(msg, AssistantMessage):
        for block in msg.content:
            if isinstance(block, ToolResultBlock) and block.is_error:
                print(f"Tool error: {block.content}")
                has_error = True

    elif isinstance(msg, ResultMessage):
        if msg.is_error:
            print(f"Result error: {msg.result}")
            has_error = True

if has_error:
    print("Errors occurred during execution")

from claude_agent_sdk import query, ResultMessage, ClaudeAgentOptions

# Get session ID from result
session_id = None

async for msg in query(prompt="Remember this: my favorite color is blue"):
    if isinstance(msg, ResultMessage):
        session_id = msg.session_id
        print(f"Session ID: {session_id}")

# Resume the session later
if session_id:
    options = ClaudeAgentOptions(resume=session_id)

    async for msg in query(prompt="What is my favorite color?", options=options):
        if isinstance(msg, ResultMessage):
            print(f"Resumed session: {msg.session_id}")

from claude_agent_sdk import query, ResultMessage

async for msg in query(prompt="Explain machine learning"):
    if isinstance(msg, ResultMessage):
        if msg.usage:
            input_tokens = msg.usage.get('input_tokens', 0)
            output_tokens = msg.usage.get('output_tokens', 0)
            cache_read = msg.usage.get('cache_read_tokens', 0)
            total_tokens = input_tokens + output_tokens

            print(f"Total tokens: {total_tokens}")
            print(f"  Input: {input_tokens}")
            print(f"  Output: {output_tokens}")
            print(f"  Cache read: {cache_read}")

            # Approximate cost calculation (example rates)
            input_cost = (input_tokens / 1_000_000) * 3.00
            output_cost = (output_tokens / 1_000_000) * 15.00
            estimated_cost = input_cost + output_cost

            print(f"Estimated cost: ${estimated_cost:.6f}")

from claude_agent_sdk import (
    query, AssistantMessage, UserMessage, ToolUseBlock,
    ClaudeAgentOptions
)

options = ClaudeAgentOptions(allowed_tools=["Bash"])

async for msg in query(prompt="Run a command", options=options):
    if isinstance(msg, AssistantMessage):
        for block in msg.content:
            if isinstance(block, ToolUseBlock):
                print(f"Tool use ID: {block.id}")

    elif isinstance(msg, UserMessage):
        if msg.parent_tool_use_id:
            print(f"This message relates to tool use: {msg.parent_tool_use_id}")

from claude_agent_sdk import (
    query, UserMessage, AssistantMessage, SystemMessage, ResultMessage
)

counts = {
    "user": 0,
    "assistant": 0,
    "system": 0,
    "result": 0
}

async for msg in query(prompt="Analyze this project"):
    if isinstance(msg, UserMessage):
        counts["user"] += 1
    elif isinstance(msg, AssistantMessage):
        counts["assistant"] += 1
    elif isinstance(msg, SystemMessage):
        counts["system"] += 1
    elif isinstance(msg, ResultMessage):
        counts["result"] += 1

print("Message statistics:")
for msg_type, count in counts.items():
    print(f"  {msg_type}: {count}")

from claude_agent_sdk import query, StreamEvent, ClaudeAgentOptions

options = ClaudeAgentOptions(include_partial_messages=True)

async for msg in query(prompt="Generate code", options=options):
    if isinstance(msg, StreamEvent):
        event = msg.event
        event_type = event.get("type")

        if event_type == "message_start":
            print("Message started")

        elif event_type == "content_block_start":
            print(f"Content block {event.get('index')} started")

        elif event_type == "content_block_delta":
            delta = event.get("delta", {})
            print(f"Delta: {delta}")

        elif event_type == "content_block_stop":
            print(f"Content block {event.get('index')} stopped")

        elif event_type == "message_delta":
            print(f"Message delta: {event.get('delta')}")

        elif event_type == "message_stop":
            print("Message complete")