tessl/pypi-claude-agent-sdk

Python SDK for Claude Code enabling developers to build AI-powered applications and agents with support for custom tools, hooks, and bidirectional interactive conversations

Moderation error

Malicious code detected in tile.json: This tile describes a package named 'claude-agent-sdk' on PyPI, which appears to be typosquatting/impersonating official Anthropic SDK packages. The legitimate Anthropic Python SDK is 'anthropic', not 'claude-agent-sdk'. This naming pattern (claude-agent-sdk) is designed to deceive developers into installing a potentially malicious package by mimicking official Anthropic/Claude branding. This is a classic supply chain attack vector through package name confusion.

Overview

Eval results

Files

Message and Content Types

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

Strongly-typed message system with discriminated unions for all message types and content blocks. Provides type safety for message handling, pattern matching, and content extraction.

Capabilities

Message Types Union

Union type representing all possible message types.

Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent

UserMessage

User-originated message in the conversation.

@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

Fields:

content: Message content as string or list of content blocks
uuid: Message UUID (only present with replay-user-messages flag)
parent_tool_use_id: Parent tool use identifier for tool results
tool_use_result: Tool execution result if this is a tool response

Usage:

from claude_agent_sdk import UserMessage, TextBlock

# String content
user_msg = UserMessage(
    content="Hello Claude",
    uuid="msg-123"
)

# Block content
user_msg = UserMessage(
    content=[TextBlock(text="Hello Claude")],
    uuid="msg-123"
)

# Tool result message
user_msg = UserMessage(
    content="Tool completed successfully",
    parent_tool_use_id="tool-456",
    tool_use_result={"status": "success"}
)

AssistantMessage

Claude's response message containing content blocks.

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

Fields:

content: List of content blocks (TextBlock, ThinkingBlock, ToolUseBlock, ToolResultBlock)
model: Model identifier that generated this message
parent_tool_use_id: Parent tool use identifier if nested
error: Error type if message represents an error

Error Types:

AssistantMessageError = Literal[
    "authentication_failed",
    "billing_error",
    "rate_limit",
    "invalid_request",
    "server_error",
    "unknown"
]

"authentication_failed" - API key or authentication credentials are invalid
"billing_error" - Account has billing issues or insufficient credits
"rate_limit" - API rate limit has been exceeded
"invalid_request" - Request parameters are invalid or malformed
"server_error" - Anthropic API server encountered an error
"unknown" - An unclassified error occurred

Usage:

from claude_agent_sdk import AssistantMessage, TextBlock, ToolUseBlock

# Process assistant messages
async for message in query(prompt="Hello"):
    if isinstance(message, AssistantMessage):
        print(f"Model: {message.model}")

        for block in message.content:
            if isinstance(block, TextBlock):
                print(f"Text: {block.text}")
            elif isinstance(block, ToolUseBlock):
                print(f"Tool: {block.name}, Input: {block.input}")

        if message.error:
            print(f"Error: {message.error}")

SystemMessage

System-level message with metadata.

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

Fields:

subtype: Message subtype identifier
data: Message data dictionary

Usage:

from claude_agent_sdk import SystemMessage

async for message in query(prompt="Hello"):
    if isinstance(message, SystemMessage):
        print(f"System {message.subtype}: {message.data}")

ResultMessage

Final result message with cost and usage metrics.

@dataclass
class ResultMessage:
    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
    structured_output: Any = None

Fields:

subtype: Result subtype
duration_ms: Total duration in milliseconds
duration_api_ms: API call duration in milliseconds
is_error: Whether result represents an error
num_turns: Number of conversation turns
session_id: Session identifier
total_cost_usd: Total cost in USD (if available)
usage: Usage metrics dictionary with keys like input_tokens, output_tokens, total_tokens, and potentially cache_creation_input_tokens, cache_read_input_tokens for prompt caching
result: Result text (if any)
structured_output: Structured output if requested via output_format

Usage:

from claude_agent_sdk import ResultMessage

async for message in query(prompt="Calculate 2+2"):
    if isinstance(message, ResultMessage):
        print(f"Completed in {message.duration_ms}ms")
        print(f"Turns: {message.num_turns}")
        print(f"Session: {message.session_id}")

        if message.total_cost_usd:
            print(f"Cost: ${message.total_cost_usd:.4f}")

        if message.usage:
            print(f"Tokens: {message.usage.get('total_tokens', 0)}")

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

        if message.structured_output:
            print(f"Structured: {message.structured_output}")

StreamEvent

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

@dataclass
class StreamEvent:
    uuid: str
    session_id: str
    event: dict[str, Any]
    parent_tool_use_id: str | None = None

Fields:

uuid: Event UUID
session_id: Session identifier
event: Raw Anthropic API stream event
parent_tool_use_id: Parent tool use identifier if nested

Usage:

from claude_agent_sdk import query, ClaudeAgentOptions, StreamEvent

options = ClaudeAgentOptions(
    include_partial_messages=True
)

async for message in query(prompt="Tell me a story", options=options):
    if isinstance(message, StreamEvent):
        event_type = message.event.get("type")
        print(f"Stream event: {event_type}")

        if event_type == "content_block_delta":
            delta = message.event.get("delta", {})
            if delta.get("type") == "text_delta":
                print(delta.get("text", ""), end="", flush=True)

Content Block Types

ContentBlock Union

Union type representing all content block types.

ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

TextBlock

Text content block.

@dataclass
class TextBlock:
    text: str

Usage:

from claude_agent_sdk import AssistantMessage, TextBlock

async for message in query(prompt="Hello"):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                print(f"Claude says: {block.text}")

ThinkingBlock

Extended thinking content block (requires extended thinking model feature).

@dataclass
class ThinkingBlock:
    thinking: str
    signature: str

Fields:

thinking: Thinking content
signature: Thinking signature

Usage:

from claude_agent_sdk import AssistantMessage, ThinkingBlock

async for message in query(prompt="Solve this problem"):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, ThinkingBlock):
                print(f"Claude's thinking: {block.thinking}")
                print(f"Signature: {block.signature}")

ToolUseBlock

Tool invocation block indicating Claude wants to use a tool.

@dataclass
class ToolUseBlock:
    id: str
    name: str
    input: dict[str, Any]

Fields:

id: Unique tool use identifier
name: Tool name
input: Tool input parameters

Usage:

from claude_agent_sdk import AssistantMessage, ToolUseBlock

async for message in query(
    prompt="Read the file config.json",
    options=ClaudeAgentOptions(allowed_tools=["Read"])
):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, ToolUseBlock):
                print(f"Tool: {block.name}")
                print(f"ID: {block.id}")
                print(f"Input: {block.input}")

                if block.name == "Read":
                    file_path = block.input.get("file_path")
                    print(f"Reading: {file_path}")

ToolResultBlock

Tool execution result block.

@dataclass
class ToolResultBlock:
    tool_use_id: str
    content: str | list[dict[str, Any]] | None = None
    is_error: bool | None = None

Fields:

tool_use_id: Tool use identifier this result corresponds to
content: Result content as string, list of content dicts, or None
is_error: Whether this result represents an error

Usage:

from claude_agent_sdk import UserMessage, ToolResultBlock

async for message in query(prompt="Read config.json"):
    if isinstance(message, UserMessage):
        if isinstance(message.content, list):
            for block in message.content:
                if isinstance(block, ToolResultBlock):
                    print(f"Tool result for: {block.tool_use_id}")
                    print(f"Content: {block.content}")
                    if block.is_error:
                        print("Tool execution failed")

Message Handling Patterns

Pattern Matching

Use isinstance checks for type-safe message handling:

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

async for message in query(prompt="Hello"):
    match message:
        case AssistantMessage():
            for block in message.content:
                match block:
                    case TextBlock(text=text):
                        print(f"Claude: {text}")
                    case ToolUseBlock(name=name, input=input):
                        print(f"Using tool {name}: {input}")

        case UserMessage(content=content):
            print(f"User: {content}")

        case SystemMessage(subtype=subtype, data=data):
            print(f"System {subtype}: {data}")

        case ResultMessage() as result:
            print(f"Completed: {result.duration_ms}ms, Cost: ${result.total_cost_usd}")

        case StreamEvent():
            print(f"Stream event: {message.event}")

Extracting Text

Extract all text from assistant messages:

from claude_agent_sdk import AssistantMessage, TextBlock

async def get_assistant_text(prompt: str) -> list[str]:
    texts = []
    async for message in query(prompt=prompt):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    texts.append(block.text)
    return texts

texts = await get_assistant_text("What is 2+2?")
print(" ".join(texts))

Tracking Tool Usage

Track all tools used in a conversation:

from claude_agent_sdk import AssistantMessage, ToolUseBlock

async def track_tools(prompt: str) -> list[dict]:
    tools_used = []
    async for message in query(prompt=prompt):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    tools_used.append({
                        "id": block.id,
                        "name": block.name,
                        "input": block.input
                    })
    return tools_used

tools = await track_tools("Read all Python files and summarize them")
for tool in tools:
    print(f"Used {tool['name']}: {tool['input']}")

Complete Example

import anyio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    AssistantMessage,
    UserMessage,
    ResultMessage,
    TextBlock,
    ToolUseBlock,
    ToolResultBlock
)

async def process_conversation(prompt: str):
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write"],
        permission_mode="acceptEdits"
    )

    async for message in query(prompt=prompt, options=options):
        if isinstance(message, AssistantMessage):
            print(f"\n[Assistant - {message.model}]")

            if message.error:
                print(f"ERROR: {message.error}")
                continue

            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

                elif isinstance(block, ToolUseBlock):
                    print(f"\n→ Using tool: {block.name}")
                    print(f"  ID: {block.id}")
                    print(f"  Input: {block.input}")

        elif isinstance(message, UserMessage):
            if isinstance(message.content, str):
                print(f"\n[User] {message.content}")
            elif isinstance(message.content, list):
                for block in message.content:
                    if isinstance(block, ToolResultBlock):
                        status = "ERROR" if block.is_error else "SUCCESS"
                        print(f"\n[Tool Result - {status}]")
                        print(f"  For: {block.tool_use_id}")
                        if block.content:
                            print(f"  Content: {block.content}")

        elif isinstance(message, ResultMessage):
            print(f"\n[Result]")
            print(f"  Duration: {message.duration_ms}ms")
            print(f"  Turns: {message.num_turns}")
            print(f"  Session: {message.session_id}")
            if message.total_cost_usd:
                print(f"  Cost: ${message.total_cost_usd:.4f}")
            if message.is_error:
                print(f"  Error: {message.result}")

anyio.run(lambda: process_conversation("Analyze the codebase"))

Install with Tessl CLI