tessl/pypi-claude-agent-sdk

Python SDK for programmatically interacting with Claude Code, enabling AI-powered automation workflows with support for bidirectional conversations, custom in-process tools, hooks, and fine-grained permission control

—

Quality

—

Does it follow best practices?

Run evals on this skill

Adds up to 20 points to the overall score

View guide

Securityby

—

The risk profile of this skill

This plugin was archived by the owner on Jun 3, 2026

Reason: Retiring all tiles created prior to the transition to plugin support

Overview

Eval results

Files

Messages and Content

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

Messages and content blocks provide structured communication between user and Claude. Messages are the top-level containers for conversation turns, while content blocks compose the actual content within messages.

Capabilities

Message Types

All messages are instances of one of five types, united by the Message type union.

Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent

UserMessage

Messages from the user to Claude.

@dataclass
class UserMessage:
    """Message from user to Claude."""

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

Fields:

content (str | list[ContentBlock]): The message content. Can be a simple string or a list of content blocks for complex messages with mixed content types.
parent_tool_use_id (str | None): Optional parent tool use ID for nested conversations. Used when this message is a response to a tool's sub-agent request.

Usage Example:

from claude_agent_sdk import UserMessage, TextBlock

# Simple string content
user_msg = UserMessage(content="What is 2 + 2?")

# Complex content with blocks
user_msg = UserMessage(
    content=[
        TextBlock(text="Analyze this code:")
    ],
    parent_tool_use_id=None
)

AssistantMessage

Messages from Claude to the user.

@dataclass
class AssistantMessage:
    """Message from Claude to user."""

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

Fields:

content (list[ContentBlock]): List of content blocks composing the message. Can include text, thinking blocks, and tool use/result blocks.
model (str): The AI model that generated this message (e.g., "claude-sonnet-4-5-20250929").
parent_tool_use_id (str | None): Optional parent tool use ID for nested conversations.

Usage Example:

from claude_agent_sdk import query, AssistantMessage, TextBlock, ThinkingBlock

async for message in query(prompt="Explain quantum computing"):
    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, ThinkingBlock):
                print(f"Thinking: {block.thinking}")

SystemMessage

System-level messages providing metadata and status information.

@dataclass
class SystemMessage:
    """System-level message."""

    subtype: str
    data: dict[str, Any]

Fields:

subtype (str): Type of system message (e.g., "info", "warning", "error").
data (dict[str, Any]): System message payload with arbitrary data.

Usage Example:

from claude_agent_sdk import query, SystemMessage

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

ResultMessage

Final result of a conversation turn, including cost and usage statistics.

@dataclass
class ResultMessage:
    """Final result of conversation turn."""

    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 (str): Result subtype indicating the nature of completion.
duration_ms (int): Total duration of the conversation turn in milliseconds.
duration_api_ms (int): API call duration in milliseconds (subset of total duration).
is_error (bool): Whether the result represents an error condition.
num_turns (int): Number of conversation turns in this interaction.
session_id (str): Session identifier for this conversation.
total_cost_usd (float | None): Total cost in USD for this turn, if available.
usage (dict[str, Any] | None): Token usage details including input_tokens, output_tokens, and cache information.
result (str | None): Optional result text summarizing the outcome.
structured_output (Any): Structured output when using output_format option, validated against the provided JSON schema.

Usage Example:

from claude_agent_sdk import query, ResultMessage

async for message in query(prompt="Calculate 42 * 17"):
    if isinstance(message, ResultMessage):
        print(f"Duration: {message.duration_ms}ms")
        print(f"Cost: ${message.total_cost_usd:.4f}")
        print(f"Turns: {message.num_turns}")
        print(f"Usage: {message.usage}")
        if message.is_error:
            print(f"Error: {message.result}")

Structured Output Example:

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

options = ClaudeAgentOptions(
    output_format={
        "type": "json_schema",
        "schema": {
            "type": "object",
            "properties": {
                "answer": {"type": "number"},
                "explanation": {"type": "string"}
            },
            "required": ["answer"]
        }
    }
)

async for message in query(prompt="What is 2 + 2?", options=options):
    if isinstance(message, ResultMessage):
        output = message.structured_output
        print(f"Answer: {output['answer']}")
        print(f"Explanation: {output.get('explanation', 'N/A')}")

StreamEvent

Raw stream events from the Anthropic API for advanced use cases.

@dataclass
class StreamEvent:
    """Raw stream event from Anthropic API."""

    uuid: str
    session_id: str
    event: dict[str, Any]
    parent_tool_use_id: str | None = None

Fields:

uuid (str): Unique identifier for this event.
session_id (str): Session identifier.
event (dict[str, Any]): Raw API event data from Anthropic's streaming API.
parent_tool_use_id (str | None): Optional parent tool use ID.

Usage Example:

from claude_agent_sdk import query, ClaudeAgentOptions, StreamEvent

options = ClaudeAgentOptions(include_partial_messages=True)

async for message in query(prompt="Generate code", options=options):
    if isinstance(message, StreamEvent):
        # Process raw streaming events
        event_type = message.event.get("type")
        if event_type == "content_block_delta":
            # Handle partial content updates
            delta = message.event.get("delta", {})
            print(f"Partial: {delta}")

Content Block Types

Content blocks are the building blocks of message content, united by the ContentBlock type union.

ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

TextBlock

Plain text content.

@dataclass
class TextBlock:
    """Plain text content."""

    text: str

Fields:

text (str): The text content.

Usage Example:

from claude_agent_sdk import query, AssistantMessage, TextBlock

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

ThinkingBlock

Extended thinking content showing Claude's reasoning process.

@dataclass
class ThinkingBlock:
    """Extended thinking content."""

    thinking: str
    signature: str

Fields:

thinking (str): The thinking/reasoning content.
signature (str): Cryptographic signature verifying the thinking content.

Usage Example:

from claude_agent_sdk import query, AssistantMessage, ThinkingBlock

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

ToolUseBlock

Tool invocation request from Claude.

@dataclass
class ToolUseBlock:
    """Tool invocation request."""

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

Fields:

id (str): Unique identifier for this tool use request.
name (str): Name of the tool being invoked (e.g., "Read", "Write", "Bash").
input (dict[str, Any]): Tool input parameters as a dictionary.

Usage Example:

from claude_agent_sdk import query, AssistantMessage, ToolUseBlock

async for message in query(prompt="Read file.txt"):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, ToolUseBlock):
                print(f"Tool: {block.name}")
                print(f"Input: {block.input}")
                print(f"ID: {block.id}")

ToolResultBlock

Tool execution result returned to Claude.

@dataclass
class ToolResultBlock:
    """Tool execution result."""

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

Fields:

tool_use_id (str): ID of the corresponding ToolUseBlock that triggered this result.
content (str | list[dict[str, Any]] | None): Result content. Can be a string, a list of content items, or None.
is_error (bool | None): Whether the tool execution resulted in an error.

Usage Example:

from claude_agent_sdk import query, AssistantMessage, ToolResultBlock

async for message in query(prompt="Read config.json"):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, ToolResultBlock):
                if block.is_error:
                    print(f"Tool error: {block.content}")
                else:
                    print(f"Tool result: {block.content}")

Message Flow Patterns

Simple Query-Response

from claude_agent_sdk import query, AssistantMessage, TextBlock, ResultMessage

async def simple_flow():
    async for message in query(prompt="What is Python?"):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Response: {block.text}")
        elif isinstance(message, ResultMessage):
            print(f"Complete in {message.duration_ms}ms")

Tool Usage Flow

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

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

    async for message in query(prompt="Create hello.py", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude says: {block.text}")
                elif isinstance(block, ToolUseBlock):
                    print(f"Using tool: {block.name}")
                    print(f"With input: {block.input}")
                elif isinstance(block, ToolResultBlock):
                    print(f"Tool result: {block.content}")

Extended Thinking Flow

from claude_agent_sdk import query, AssistantMessage, TextBlock, ThinkingBlock

async def thinking_flow():
    async for message in query(prompt="Design a database schema"):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ThinkingBlock):
                    print(f"[THINKING]: {block.thinking}")
                elif isinstance(block, TextBlock):
                    print(f"[RESPONSE]: {block.text}")

Interactive Multi-Turn Flow

from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def interactive_flow():
    async with ClaudeSDKClient() as client:
        # Turn 1
        await client.query("What is Python?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Turn 1: {block.text}")

        # Turn 2 - follow-up in same session
        await client.query("What are its main features?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Turn 2: {block.text}")

Type Guards and Pattern Matching

Python's pattern matching (3.10+) works well with message and content types:

from claude_agent_sdk import query

async for message in query(prompt="Hello"):
    match message:
        case AssistantMessage(content=blocks):
            for block in blocks:
                match block:
                    case TextBlock(text=t):
                        print(f"Text: {t}")
                    case ThinkingBlock(thinking=t):
                        print(f"Thinking: {t}")
                    case ToolUseBlock(name=n, input=i):
                        print(f"Tool {n}: {i}")
        case ResultMessage(total_cost_usd=cost, num_turns=turns):
            print(f"Done: {turns} turns, ${cost}")
        case SystemMessage(subtype=st, data=d):
            print(f"System [{st}]: {d}")

Traditional type guards also work:

from claude_agent_sdk import (
    query, AssistantMessage, TextBlock, ThinkingBlock,
    ToolUseBlock, ToolResultBlock, ResultMessage
)

async for message in query(prompt="Hello"):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                handle_text(block.text)
            elif isinstance(block, ThinkingBlock):
                handle_thinking(block.thinking)
            elif isinstance(block, ToolUseBlock):
                handle_tool_use(block.name, block.input)
            elif isinstance(block, ToolResultBlock):
                handle_tool_result(block.content, block.is_error)
    elif isinstance(message, ResultMessage):
        handle_result(message)

tessl/pypi-claude-agent-sdk