CtrlK
BlogDocsLog inGet started
Tessl Logo

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

advanced.mddocs/reference/

Advanced Features

Advanced capabilities for building sophisticated AI applications including custom agents, plugins, session management, file checkpointing, structured outputs, beta features, and custom transports.

Capabilities

Custom Agents

Define custom agent configurations with specific tools, models, and prompts.

@dataclass
class AgentDefinition:
    description: str
    prompt: str
    tools: list[str] | None = None
    model: Literal["sonnet", "opus", "haiku", "inherit"] | None = None

Fields:

  • description: Human-readable agent description
  • prompt: System prompt for the agent
  • tools: List of tool names available to agent
  • model: AI model for agent ("sonnet", "opus", "haiku", or "inherit" from parent)

Usage:

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AgentDefinition

# Define custom agents
agents = {
    "code_reviewer": AgentDefinition(
        description="Code review specialist",
        prompt="You are an expert code reviewer. Focus on bugs, security, and best practices.",
        tools=["Read", "Grep", "Glob"],
        model="sonnet"
    ),
    "implementer": AgentDefinition(
        description="Implementation specialist",
        prompt="You implement code changes efficiently and correctly.",
        tools=["Read", "Write", "Edit", "Bash"],
        model="opus"
    ),
    "tester": AgentDefinition(
        description="Testing specialist",
        prompt="You write and run tests to verify functionality.",
        tools=["Read", "Write", "Bash"],
        model="haiku"
    )
}

options = ClaudeAgentOptions(agents=agents)

async with ClaudeSDKClient(options=options) as client:
    # Claude can now use @code_reviewer, @implementer, @tester
    await client.query("Review, implement, and test the authentication module")
    async for msg in client.receive_response():
        print(msg)

Session Management

Manage conversation sessions with resume, continue, and fork capabilities.

@dataclass
class ClaudeAgentOptions:
    continue_conversation: bool = False
    resume: str | None = None
    fork_session: bool = False

Fields:

  • continue_conversation: Continue existing session without forking
  • resume: Session ID to resume from
  • fork_session: Create new session ID when resuming (default: False)

Usage:

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    ResultMessage
)

# Initial conversation
async with ClaudeSDKClient() as client:
    await client.query("Start working on the project")

    async for msg in client.receive_response():
        if isinstance(msg, ResultMessage):
            session_id = msg.session_id
            print(f"Session ID: {session_id}")

# Resume session (continues in same session)
resume_options = ClaudeAgentOptions(
    resume=session_id,
    continue_conversation=True
)

async with ClaudeSDKClient(options=resume_options) as client:
    await client.query("Continue the previous task")
    async for msg in client.receive_response():
        print(msg)

# Fork session (creates new session from checkpoint)
fork_options = ClaudeAgentOptions(
    resume=session_id,
    fork_session=True
)

async with ClaudeSDKClient(options=fork_options) as client:
    await client.query("Try a different approach")
    async for msg in client.receive_response():
        if isinstance(msg, ResultMessage):
            new_session_id = msg.session_id
            print(f"Forked to: {new_session_id}")

File Checkpointing

Track and rewind file changes during conversations.

@dataclass
class ClaudeAgentOptions:
    enable_file_checkpointing: bool = False

class ClaudeSDKClient:
    async def rewind_files(self, user_message_id: str) -> dict[str, Any]

Configuration:

  • enable_file_checkpointing: Enable tracking of file changes (default: False)

Methods:

  • rewind_files(): Rewind tracked files to state at specific user message

Usage:

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    UserMessage
)

# Enable checkpointing
options = ClaudeAgentOptions(
    enable_file_checkpointing=True,
    permission_mode="acceptEdits"
)

message_ids = []

async with ClaudeSDKClient(options=options) as client:
    # Make changes
    await client.query("Refactor the authentication module")

    async for msg in client.receive_response():
        if isinstance(msg, UserMessage) and msg.uuid:
            message_ids.append(msg.uuid)

    # Make more changes
    await client.query("Add error handling")

    async for msg in client.receive_response():
        if isinstance(msg, UserMessage) and msg.uuid:
            message_ids.append(msg.uuid)

    # Rewind to first checkpoint
    if message_ids:
        result = await client.rewind_files(message_ids[0])
        print(f"Rewound files: {result}")

    # Continue from rewound state
    await client.query("Try a different implementation")
    async for msg in client.receive_response():
        print(msg)

Structured Outputs

Request structured JSON outputs that match a specific schema.

@dataclass
class ClaudeAgentOptions:
    output_format: dict[str, Any] | None = None

@dataclass
class ResultMessage:
    structured_output: Any = None

Configuration:

  • output_format: JSON Schema defining output structure (matches Messages API format)

Result:

  • structured_output: Parsed structured output from ResultMessage

Usage:

from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    ResultMessage
)

# Define output schema
schema = {
    "type": "json_schema",
    "json_schema": {
        "name": "code_review",
        "strict": True,
        "schema": {
            "type": "object",
            "properties": {
                "issues": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "severity": {
                                "type": "string",
                                "enum": ["critical", "warning", "info"]
                            },
                            "file": {"type": "string"},
                            "line": {"type": "integer"},
                            "description": {"type": "string"}
                        },
                        "required": ["severity", "file", "description"],
                        "additionalProperties": False
                    }
                },
                "summary": {"type": "string"}
            },
            "required": ["issues", "summary"],
            "additionalProperties": False
        }
    }
}

options = ClaudeAgentOptions(
    output_format=schema,
    allowed_tools=["Read", "Grep"]
)

async for message in query(
    prompt="Review the code and return structured results",
    options=options
):
    if isinstance(message, ResultMessage):
        # Access structured output
        output = message.structured_output
        print(f"Found {len(output['issues'])} issues")
        for issue in output["issues"]:
            print(f"{issue['severity']}: {issue['description']}")
        print(f"Summary: {output['summary']}")

Beta Features

Enable beta features and experimental capabilities.

SdkBeta = Literal["context-1m-2025-08-07"]

@dataclass
class ClaudeAgentOptions:
    betas: list[SdkBeta] = field(default_factory=list)

Available Betas:

  • "context-1m-2025-08-07": Extended context window support

Usage:

from claude_agent_sdk import ClaudeAgentOptions

# Enable beta features
options = ClaudeAgentOptions(
    betas=["context-1m-2025-08-07"]
)

Plugins

Load custom plugins to extend SDK functionality.

class SdkPluginConfig(TypedDict):
    type: Literal["local"]
    path: str

@dataclass
class ClaudeAgentOptions:
    plugins: list[SdkPluginConfig] = field(default_factory=list)

Configuration:

  • type: Plugin type (currently only "local" supported)
  • path: Path to plugin file

Usage:

from claude_agent_sdk import ClaudeAgentOptions

plugins = [
    {"type": "local", "path": "/path/to/custom_plugin.py"},
    {"type": "local", "path": "/path/to/another_plugin.py"}
]

options = ClaudeAgentOptions(plugins=plugins)

Setting Sources

Control which configuration sources to load.

SettingSource = Literal["user", "project", "local"]

@dataclass
class ClaudeAgentOptions:
    setting_sources: list[SettingSource] | None = None

Sources:

  • "user": User-level settings (e.g., ~/.config/claude)
  • "project": Project-level settings (e.g., .claude/ in project root)
  • "local": Local workspace settings

Usage:

from claude_agent_sdk import ClaudeAgentOptions

# Only load project settings, ignore user settings
options = ClaudeAgentOptions(
    setting_sources=["project"]
)

# Load specific combination
options = ClaudeAgentOptions(
    setting_sources=["user", "project"]  # Exclude local
)

Custom Transports

Implement custom transport for remote or alternative Claude Code connections.

class Transport(ABC):
    @abstractmethod
    async def connect(self) -> None: ...

    @abstractmethod
    async def write(self, data: str) -> None: ...

    @abstractmethod
    def read_messages(self) -> AsyncIterator[dict[str, Any]]: ...

    @abstractmethod
    async def close(self) -> None: ...

    @abstractmethod
    def is_ready(self) -> bool: ...

    @abstractmethod
    async def end_input(self) -> None: ...

Methods:

  • connect() -> None: Establish connection to Claude Code. Called automatically when starting a session
  • write(data: str) -> None: Write raw JSON data string to the transport channel
  • read_messages() -> AsyncIterator[dict[str, Any]]: Async iterator that yields parsed JSON message dictionaries from Claude
  • close() -> None: Close the connection and perform cleanup operations
  • is_ready() -> bool: Returns True if the transport is connected and ready for communication
  • end_input() -> None: Signal end of input stream (equivalent to closing stdin in subprocess transport)

Warning: This is an internal API that may change in future releases. Use for custom transport implementations only.

Usage:

from claude_agent_sdk import Transport, ClaudeSDKClient
from collections.abc import AsyncIterator
from typing import Any
import anyio

class RemoteTransport(Transport):
    """Custom transport for remote Claude Code connection."""

    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port
        self.stream = None
        self._ready = False

    async def connect(self) -> None:
        # Connect to remote Claude Code server
        self.stream = await anyio.connect_tcp(self.host, self.port)
        self._ready = True

    async def write(self, data: str) -> None:
        if not self.stream:
            raise RuntimeError("Not connected")
        await self.stream.send_all(data.encode())

    async def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        if not self.stream:
            raise RuntimeError("Not connected")

        import json
        buffer = ""
        async for chunk in self.stream:
            buffer += chunk.decode()
            while "\n" in buffer:
                line, buffer = buffer.split("\n", 1)
                if line:
                    yield json.loads(line)

    async def close(self) -> None:
        if self.stream:
            await self.stream.aclose()
            self._ready = False

    def is_ready(self) -> bool:
        return self._ready

    async def end_input(self) -> None:
        # Send end-of-input signal
        if self.stream:
            await self.stream.send_eof()

# Use custom transport
transport = RemoteTransport(host="claude-server.local", port=9000)
async with ClaudeSDKClient(transport=transport) as client:
    await client.query("Hello via remote transport")
    async for msg in client.receive_response():
        print(msg)

Extended Thinking

Control extended thinking with max tokens.

@dataclass
class ClaudeAgentOptions:
    max_thinking_tokens: int | None = None

Configuration:

  • max_thinking_tokens: Maximum tokens for thinking blocks

Usage:

from claude_agent_sdk import ClaudeAgentOptions

# Allow extended thinking
options = ClaudeAgentOptions(
    max_thinking_tokens=10000,
    model="opus"
)

Additional CLI Arguments

Pass arbitrary CLI flags for advanced customization.

@dataclass
class ClaudeAgentOptions:
    extra_args: dict[str, str | None] = field(default_factory=dict)

Configuration:

  • extra_args: Dictionary of CLI flag names to values (None for boolean flags)

Usage:

from claude_agent_sdk import ClaudeAgentOptions

# Pass custom CLI arguments
options = ClaudeAgentOptions(
    extra_args={
        "--custom-flag": "value",
        "--enable-feature": None,  # Boolean flag
        "--timeout": "30000"
    }
)

Complete Advanced Examples

Example 1: Multi-Agent Workflow

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AgentDefinition
)

async def main():
    # Define specialized agents
    agents = {
        "architect": AgentDefinition(
            description="System architect for design decisions",
            prompt="You design system architecture and data models.",
            tools=["Read", "Grep", "Glob"],
            model="opus"
        ),
        "developer": AgentDefinition(
            description="Full-stack developer",
            prompt="You implement features following the architecture.",
            tools=["Read", "Write", "Edit", "Bash"],
            model="sonnet"
        ),
        "qa": AgentDefinition(
            description="Quality assurance engineer",
            prompt="You write tests and verify quality.",
            tools=["Read", "Write", "Bash"],
            model="haiku"
        )
    }

    options = ClaudeAgentOptions(
        agents=agents,
        permission_mode="acceptEdits"
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query(
            "Build a user authentication system. "
            "Use @architect for design, @developer for implementation, "
            "and @qa for testing."
        )
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

Example 2: Session Branching

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    ResultMessage
)

async def main():
    # Initial session
    print("Starting initial session...")
    async with ClaudeSDKClient() as client:
        await client.query("Design a caching strategy")

        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage):
                base_session = msg.session_id
                print(f"Base session: {base_session}")

    # Branch 1: Try Redis
    print("\nBranch 1: Redis approach...")
    redis_options = ClaudeAgentOptions(
        resume=base_session,
        fork_session=True
    )

    async with ClaudeSDKClient(options=redis_options) as client:
        await client.query("Implement using Redis")
        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage):
                redis_session = msg.session_id
                print(f"Redis session: {redis_session}")

    # Branch 2: Try Memcached
    print("\nBranch 2: Memcached approach...")
    memcached_options = ClaudeAgentOptions(
        resume=base_session,
        fork_session=True
    )

    async with ClaudeSDKClient(options=memcached_options) as client:
        await client.query("Implement using Memcached")
        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage):
                memcached_session = msg.session_id
                print(f"Memcached session: {memcached_session}")

anyio.run(main)

Example 3: File Checkpointing for Safe Refactoring

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    UserMessage,
    ResultMessage
)

async def main():
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits"
    )

    checkpoints = []

    async with ClaudeSDKClient(options=options) as client:
        # Checkpoint 1: Initial refactoring
        await client.query("Refactor the database layer")
        async for msg in client.receive_response():
            if isinstance(msg, UserMessage) and msg.uuid:
                checkpoints.append(("db_refactor", msg.uuid))

        # Checkpoint 2: Add caching
        await client.query("Add caching to the refactored code")
        async for msg in client.receive_response():
            if isinstance(msg, UserMessage) and msg.uuid:
                checkpoints.append(("add_cache", msg.uuid))

        # Checkpoint 3: Optimize queries
        await client.query("Optimize the database queries")
        async for msg in client.receive_response():
            if isinstance(msg, UserMessage) and msg.uuid:
                checkpoints.append(("optimize", msg.uuid))

        # Test the result
        await client.query("Run the tests")
        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage):
                if msg.is_error:
                    print("Tests failed! Rewinding to last checkpoint...")

                    # Rewind to checkpoint before optimization
                    rewind_name, rewind_id = checkpoints[-2]
                    await client.rewind_files(rewind_id)
                    print(f"Rewound to: {rewind_name}")

                    # Try different approach
                    await client.query("Try a different optimization strategy")
                    async for retry_msg in client.receive_response():
                        print(retry_msg)

anyio.run(main)

Example 4: Structured Code Review

import anyio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    ResultMessage
)

async def main():
    # Define strict schema for code review output
    review_schema = {
        "type": "json_schema",
        "json_schema": {
            "name": "code_review",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "files_reviewed": {"type": "integer"},
                    "issues": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "severity": {
                                    "type": "string",
                                    "enum": ["critical", "major", "minor", "info"]
                                },
                                "category": {
                                    "type": "string",
                                    "enum": ["security", "performance", "maintainability", "style"]
                                },
                                "file": {"type": "string"},
                                "line": {"type": "integer"},
                                "description": {"type": "string"},
                                "suggestion": {"type": "string"}
                            },
                            "required": ["severity", "category", "file", "description"],
                            "additionalProperties": False
                        }
                    },
                    "score": {
                        "type": "integer",
                        "minimum": 0,
                        "maximum": 100
                    },
                    "summary": {"type": "string"}
                },
                "required": ["files_reviewed", "issues", "score", "summary"],
                "additionalProperties": False
            }
        }
    }

    options = ClaudeAgentOptions(
        output_format=review_schema,
        allowed_tools=["Read", "Grep", "Glob"]
    )

    async for message in query(
        prompt="Review all Python files in the src/ directory",
        options=options
    ):
        if isinstance(message, ResultMessage) and message.structured_output:
            review = message.structured_output

            print(f"\nCode Review Results")
            print(f"Files Reviewed: {review['files_reviewed']}")
            print(f"Overall Score: {review['score']}/100")
            print(f"\nIssues Found: {len(review['issues'])}")

            for issue in review["issues"]:
                print(f"\n[{issue['severity'].upper()}] {issue['category']}")
                print(f"File: {issue['file']}:{issue.get('line', 'N/A')}")
                print(f"Issue: {issue['description']}")
                if "suggestion" in issue:
                    print(f"Suggestion: {issue['suggestion']}")

            print(f"\nSummary: {review['summary']}")

anyio.run(main)

Example 5: Extended Context with Beta Features

import anyio
from claude_agent_sdk import ClaudeAgentOptions, query

async def main():
    # Enable extended context window
    options = ClaudeAgentOptions(
        betas=["context-1m-2025-08-07"],
        model="opus",
        max_thinking_tokens=50000
    )

    async for message in query(
        prompt="Analyze all files in this large codebase and provide insights",
        options=options
    ):
        print(message)

anyio.run(main)

Example 6: Remote Transport with Load Balancing

import anyio
from claude_agent_sdk import Transport, ClaudeSDKClient
from collections.abc import AsyncIterator
from typing import Any
import random

class LoadBalancedTransport(Transport):
    """Transport that load balances across multiple Claude servers."""

    def __init__(self, servers: list[tuple[str, int]]):
        self.servers = servers
        self.current_transport = None

    async def connect(self) -> None:
        # Pick random server
        host, port = random.choice(self.servers)
        print(f"Connecting to {host}:{port}")

        # Create transport for chosen server
        # (In real implementation, would create actual connection)
        self.current_transport = RemoteTransport(host, port)
        await self.current_transport.connect()

    async def write(self, data: str) -> None:
        if not self.current_transport:
            raise RuntimeError("Not connected")
        await self.current_transport.write(data)

    async def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        if not self.current_transport:
            raise RuntimeError("Not connected")
        async for msg in self.current_transport.read_messages():
            yield msg

    async def close(self) -> None:
        if self.current_transport:
            await self.current_transport.close()

    def is_ready(self) -> bool:
        return self.current_transport and self.current_transport.is_ready()

    async def end_input(self) -> None:
        if self.current_transport:
            await self.current_transport.end_input()

async def main():
    # Configure server pool
    servers = [
        ("claude1.example.com", 9000),
        ("claude2.example.com", 9000),
        ("claude3.example.com", 9000),
    ]

    transport = LoadBalancedTransport(servers)

    async with ClaudeSDKClient(transport=transport) as client:
        await client.query("Process this request")
        async for msg in client.receive_response():
            print(msg)

# Note: RemoteTransport class from previous example would be needed
# anyio.run(main)

Example 7: Plugin System Integration

import anyio
from claude_agent_sdk import ClaudeAgentOptions, query

async def main():
    # Load custom plugins
    plugins = [
        {"type": "local", "path": "/plugins/custom_logger.py"},
        {"type": "local", "path": "/plugins/metrics_collector.py"},
        {"type": "local", "path": "/plugins/security_validator.py"}
    ]

    options = ClaudeAgentOptions(
        plugins=plugins,
        permission_mode="acceptEdits"
    )

    async for message in query(
        prompt="Work on the project with plugin enhancements",
        options=options
    ):
        print(message)

anyio.run(main)

Example 8: Settings Source Control

import anyio
from claude_agent_sdk import ClaudeAgentOptions, query

async def main():
    # CI/CD: Only use project settings, ignore user settings
    ci_options = ClaudeAgentOptions(
        setting_sources=["project"],  # Reproducible builds
        permission_mode="bypassPermissions"
    )

    async for message in query(
        prompt="Run automated tests",
        options=ci_options
    ):
        print(message)

anyio.run(main)

Example 9: Complex Structured Output with Nested Schema

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def main():
    # Complex schema for test results
    test_schema = {
        "type": "json_schema",
        "json_schema": {
            "name": "test_results",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "summary": {
                        "type": "object",
                        "properties": {
                            "total": {"type": "integer"},
                            "passed": {"type": "integer"},
                            "failed": {"type": "integer"},
                            "skipped": {"type": "integer"}
                        },
                        "required": ["total", "passed", "failed", "skipped"],
                        "additionalProperties": False
                    },
                    "suites": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "name": {"type": "string"},
                                "tests": {
                                    "type": "array",
                                    "items": {
                                        "type": "object",
                                        "properties": {
                                            "name": {"type": "string"},
                                            "status": {
                                                "type": "string",
                                                "enum": ["passed", "failed", "skipped"]
                                            },
                                            "duration_ms": {"type": "integer"},
                                            "error": {"type": "string"}
                                        },
                                        "required": ["name", "status", "duration_ms"],
                                        "additionalProperties": False
                                    }
                                }
                            },
                            "required": ["name", "tests"],
                            "additionalProperties": False
                        }
                    }
                },
                "required": ["summary", "suites"],
                "additionalProperties": False
            }
        }
    }

    options = ClaudeAgentOptions(
        output_format=test_schema,
        allowed_tools=["Bash", "Read"]
    )

    async for message in query(
        prompt="Run the test suite and return structured results",
        options=options
    ):
        if isinstance(message, ResultMessage) and message.structured_output:
            results = message.structured_output

            summary = results["summary"]
            print(f"\nTest Results Summary")
            print(f"Total: {summary['total']}")
            print(f"Passed: {summary['passed']}")
            print(f"Failed: {summary['failed']}")
            print(f"Skipped: {summary['skipped']}")

            for suite in results["suites"]:
                print(f"\n{suite['name']}:")
                for test in suite["tests"]:
                    status_icon = "✓" if test["status"] == "passed" else "✗"
                    print(f"  {status_icon} {test['name']} ({test['duration_ms']}ms)")
                    if test.get("error"):
                        print(f"    Error: {test['error']}")

anyio.run(main)

Example 10: Full-Featured Application

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AgentDefinition,
    ResultMessage,
    UserMessage
)

async def main():
    # Define agents
    agents = {
        "planner": AgentDefinition(
            description="Project planner",
            prompt="Break down projects into tasks",
            tools=["Read", "Grep"],
            model="sonnet"
        ),
        "executor": AgentDefinition(
            description="Task executor",
            prompt="Execute tasks efficiently",
            tools=["Read", "Write", "Edit", "Bash"],
            model="opus"
        )
    }

    # Configure sandbox
    sandbox = {
        "enabled": True,
        "autoAllowBashIfSandboxed": True,
        "excludedCommands": ["docker", "git"]
    }

    # Load plugins
    plugins = [
        {"type": "local", "path": "/plugins/task_tracker.py"}
    ]

    # Structured output schema
    progress_schema = {
        "type": "json_schema",
        "json_schema": {
            "name": "progress_report",
            "schema": {
                "type": "object",
                "properties": {
                    "completed": {"type": "array", "items": {"type": "string"}},
                    "in_progress": {"type": "array", "items": {"type": "string"}},
                    "pending": {"type": "array", "items": {"type": "string"}}
                },
                "required": ["completed", "in_progress", "pending"]
            }
        }
    }

    # Full configuration
    options = ClaudeAgentOptions(
        agents=agents,
        sandbox=sandbox,
        plugins=plugins,
        output_format=progress_schema,
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",
        betas=["context-1m-2025-08-07"],
        max_thinking_tokens=10000,
        setting_sources=["project", "local"]
    )

    checkpoints = []

    async with ClaudeSDKClient(options=options) as client:
        # Initial planning
        await client.query(
            "Use @planner to create a plan, then @executor to implement"
        )

        async for msg in client.receive_response():
            if isinstance(msg, UserMessage) and msg.uuid:
                checkpoints.append(msg.uuid)

            if isinstance(msg, ResultMessage):
                if msg.structured_output:
                    print("\nProgress Report:")
                    print(f"Completed: {msg.structured_output['completed']}")
                    print(f"In Progress: {msg.structured_output['in_progress']}")
                    print(f"Pending: {msg.structured_output['pending']}")

        # Continue with next phase
        await client.query("Continue with the next tasks")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

Best Practices

Custom Agents: Use specialized agents for complex workflows requiring different expertise levels.

Session Management: Use fork_session for experimental branches, continue_conversation for linear progression.

File Checkpointing: Enable for refactoring tasks where rollback might be needed.

Structured Outputs: Define strict schemas for programmatic consumption of results.

Beta Features: Enable carefully and be prepared for API changes.

Custom Transports: Only implement for specialized needs like remote Claude Code servers.

Plugins: Use for cross-cutting concerns like logging, metrics, and validation.

Settings Sources: Control configuration hierarchy for reproducible builds in CI/CD.

Install with Tessl CLI

npx tessl i tessl/pypi-claude-agent-sdk

docs

reference

advanced.md

client.md

configuration.md

errors.md

hooks.md

mcp.md

messages.md

permissions.md

query.md

sandbox.md

index.md

tile.json