tessl/pypi-claude-code-sdk

Python SDK for Claude Code providing simple query functions and advanced bidirectional interactive conversations with custom tool support

Overview

Eval results

Files

Message Types and Content

Name: tessl/pypi-claude-code-sdk
Author: tessl

Rich message system with structured content blocks supporting text, tool usage, tool results, thinking blocks, and streaming events. Includes complete type definitions for all message and content types used in Claude Code SDK communications.

Capabilities

Message Union Type

All possible message types that can be received from Claude Code.

Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent

User Messages

Messages representing user input and prompts.

@dataclass
class UserMessage:
    """User message."""

    content: str | list[ContentBlock]
    parent_tool_use_id: str | None = None

Fields:

content: User input as string or structured content blocks
parent_tool_use_id: Optional reference to parent tool use operation

Assistant Messages

Messages from Claude containing responses with rich content blocks.

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

    content: list[ContentBlock]
    model: str
    parent_tool_use_id: str | None = None

Fields:

content: List of content blocks (text, tool use, etc.)
model: Claude model identifier used for the response
parent_tool_use_id: Optional reference to parent tool use operation

System Messages

Internal system messages with metadata and notifications.

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

    subtype: str
    data: dict[str, Any]

Fields:

subtype: Type of system message (status, notification, etc.)
data: Structured data payload for the system message

Result Messages

Final result messages containing cost, usage, and completion information.

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

    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None

Fields:

subtype: Result type identifier
duration_ms: Total duration in milliseconds
duration_api_ms: API call duration in milliseconds
is_error: Whether the interaction resulted in an error
num_turns: Number of conversation turns
session_id: Session identifier
total_cost_usd: Total cost in USD (if available)
usage: Token usage and other metrics
result: Final result summary

Stream Events

Streaming updates during partial message processing.

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

    uuid: str
    session_id: str
    event: dict[str, Any]  # The raw Anthropic API stream event
    parent_tool_use_id: str | None = None

Fields:

uuid: Unique identifier for the stream event
session_id: Session identifier
event: Raw Anthropic API stream event data
parent_tool_use_id: Optional reference to parent tool use operation

Content Block Types

Content Block Union

All possible content block types within messages.

ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

Text Blocks

Plain text content blocks containing textual responses.

@dataclass
class TextBlock:
    """Text content block."""

    text: str

Fields:

text: The text content

Thinking Blocks

Claude's internal reasoning and thought process.

@dataclass
class ThinkingBlock:
    """Thinking content block."""

    thinking: str
    signature: str

Fields:

thinking: Claude's internal reasoning text
signature: Cryptographic signature for thinking content

Tool Use Blocks

Blocks representing Claude's intention to use a tool.

@dataclass
class ToolUseBlock:
    """Tool use content block."""

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

Fields:

id: Unique identifier for the tool use
name: Name of the tool to be used
input: Input parameters for the tool

Tool Result Blocks

Blocks containing the results of tool execution.

@dataclass
class ToolResultBlock:
    """Tool result content block."""

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

Fields:

tool_use_id: Reference to the tool use that generated this result
content: Tool execution result content
is_error: Whether the tool execution resulted in an error

Usage Examples

Processing Assistant Messages

from claude_code_sdk import query, AssistantMessage, TextBlock, ToolUseBlock, ToolResultBlock

async def main():
    async for message in query(prompt="Create a Python file and run it"):
        if isinstance(message, AssistantMessage):
            print(f"Response from {message.model}:")

            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Text: {block.text}")

                elif isinstance(block, ToolUseBlock):
                    print(f"Tool use: {block.name} (ID: {block.id})")
                    print(f"Input: {block.input}")

                elif isinstance(block, ToolResultBlock):
                    print(f"Tool result for {block.tool_use_id}:")
                    if block.is_error:
                        print(f"Error: {block.content}")
                    else:
                        print(f"Result: {block.content}")

                elif isinstance(block, ThinkingBlock):
                    print(f"Claude's thinking: {block.thinking[:100]}...")

Handling Different Message Types

from claude_code_sdk import (
    query, UserMessage, AssistantMessage, SystemMessage,
    ResultMessage, StreamEvent
)

async def main():
    async for message in query(prompt="What is the capital of France?"):
        if isinstance(message, UserMessage):
            print(f"User: {message.content}")

        elif isinstance(message, AssistantMessage):
            print(f"Assistant ({message.model}): Processing {len(message.content)} content blocks")

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

        elif isinstance(message, ResultMessage):
            print(f"Result: {message.num_turns} turns, {message.duration_ms}ms")
            if message.total_cost_usd:
                print(f"Cost: ${message.total_cost_usd:.4f}")

        elif isinstance(message, StreamEvent):
            print(f"Stream event: {message.uuid}")

Collecting Complete Responses

from claude_code_sdk import ClaudeSDKClient, AssistantMessage, ResultMessage, TextBlock

async def main():
    async with ClaudeSDKClient() as client:
        await client.query("Explain quantum computing")

        complete_response = []
        async for msg in client.receive_response():
            complete_response.append(msg)

            if isinstance(msg, ResultMessage):
                # Response is complete
                break

        # Process collected messages
        for msg in complete_response:
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

Working with Tool Interactions

from claude_code_sdk import (
    query, ClaudeCodeOptions, AssistantMessage,
    ToolUseBlock, ToolResultBlock
)

async def main():
    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="acceptEdits"
    )

    tool_uses = []
    tool_results = []

    async for message in query(
        prompt="Create a hello.py file and run it",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    tool_uses.append(block)
                    print(f"Claude wants to use {block.name}:")
                    print(f"  ID: {block.id}")
                    print(f"  Input: {block.input}")

                elif isinstance(block, ToolResultBlock):
                    tool_results.append(block)
                    print(f"Tool result for {block.tool_use_id}:")
                    if block.is_error:
                        print(f"  Error: {block.content}")
                    else:
                        print(f"  Success: {block.content}")

    print(f"Total tool uses: {len(tool_uses)}")
    print(f"Total tool results: {len(tool_results)}")

Stream Event Processing

from claude_code_sdk import ClaudeCodeOptions, query, StreamEvent

async def main():
    options = ClaudeCodeOptions(
        include_partial_messages=True  # Enable streaming events
    )

    async for message in query(
        prompt="Write a long story about space exploration",
        options=options
    ):
        if isinstance(message, StreamEvent):
            event_type = message.event.get("type", "unknown")
            print(f"Stream event ({event_type}): {message.uuid}")

            # Handle different stream event types
            if event_type == "content_block_start":
                print("  Starting content block")
            elif event_type == "content_block_delta":
                delta = message.event.get("delta", {})
                if "text" in delta:
                    print(f"  Text delta: {delta['text']}")
            elif event_type == "content_block_stop":
                print("  Finished content block")

Error Handling in Messages

from claude_code_sdk import query, ResultMessage, ToolResultBlock

async def main():
    async for message in query(prompt="Run an invalid command"):
        if isinstance(message, ResultMessage):
            if message.is_error:
                print(f"Conversation ended with error: {message.result}")
            else:
                print(f"Conversation completed successfully")

        # Check for tool errors in content blocks
        if hasattr(message, 'content'):
            for block in getattr(message, 'content', []):
                if isinstance(block, ToolResultBlock) and block.is_error:
                    print(f"Tool error in {block.tool_use_id}: {block.content}")

Message Filtering and Analysis

from claude_code_sdk import query, AssistantMessage, TextBlock
from collections import Counter

async def main():
    message_types = Counter()
    content_types = Counter()
    total_text_length = 0

    async for message in query(prompt="Analyze this codebase and suggest improvements"):
        message_types[type(message).__name__] += 1

        if isinstance(message, AssistantMessage):
            for block in message.content:
                content_types[type(block).__name__] += 1

                if isinstance(block, TextBlock):
                    total_text_length += len(block.text)

    print("Message type distribution:")
    for msg_type, count in message_types.items():
        print(f"  {msg_type}: {count}")

    print("Content block distribution:")
    for content_type, count in content_types.items():
        print(f"  {content_type}: {count}")

    print(f"Total text content length: {total_text_length} characters")