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

Interactive Sessions Guide

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

Build chat applications and multi-turn conversations with ClaudeSDKClient.

When to Use Interactive Sessions

Use ClaudeSDKClient when you need:

Multi-turn conversations with context
Chat applications or REPL interfaces
Ability to react to Claude's responses
Dynamic permission or model changes mid-conversation
Interrupt capability for long-running operations

Use query() for:

Simple one-off questions
Batch processing
Automation scripts

Basic Interactive Session

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="acceptEdits"
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # First query
        await client.query("What files are in this directory?")
        async for msg in client.receive_response():
            print(msg)
        
        # Follow-up with context
        await client.query("Read the README file")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

Chat REPL

Build an interactive REPL:

import anyio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def chat_repl():
    async with ClaudeSDKClient() as client:
        print("Chat started. Type 'exit' to quit.\n")
        
        while True:
            user_input = input("You: ").strip()
            if user_input.lower() == 'exit':
                break
            
            await client.query(user_input)
            
            print("Claude: ", end="")
            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(block.text, end=" ")
            print("\n")

anyio.run(chat_repl)

Dynamic Permission Control

Change permissions during conversation:

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Edit"],
        permission_mode="default"  # Start with prompts
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # Phase 1: Review (default permissions)
        await client.query("Review this code for issues")
        async for msg in client.receive_response():
            print(msg)
        
        # Phase 2: Implementation (auto-approve edits)
        await client.set_permission_mode("acceptEdits")
        await client.query("Now fix the issues")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

Model Switching

Switch models mid-conversation:

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def main():
    async with ClaudeSDKClient() as client:
        # Simple question with default model
        await client.query("What's 2+2?")
        async for msg in client.receive_response():
            print(msg)
        
        # Complex reasoning with more capable model
        await client.set_model("opus")
        await client.query("Explain quantum entanglement")
        async for msg in client.receive_response():
            print(msg)
        
        # Back to faster model
        await client.set_model("haiku")

anyio.run(main)

Interrupting Long Operations

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Grep", "Glob"]
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # Start long-running analysis
        await client.query("Analyze all files in this large repository")
        
        try:
            # Set timeout
            with anyio.fail_after(30):  # 30 seconds
                async for msg in client.receive_response():
                    print(msg)
        except TimeoutError:
            print("Operation timed out, interrupting...")
            await client.interrupt()

anyio.run(main)

File Checkpointing

Save and rewind file states:

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

async def main():
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,  # Enable tracking
        permission_mode="acceptEdits"
    )
    
    checkpoints = []
    
    async with ClaudeSDKClient(options=options) as client:
        # Checkpoint 1: Initial changes
        await client.query("Refactor the database layer")
        async for msg in client.receive_response():
            if isinstance(msg, UserMessage) and msg.uuid:
                checkpoints.append(("refactor", msg.uuid))
        
        # Checkpoint 2: Add features
        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(("caching", msg.uuid))
        
        # Test
        await client.query("Run the tests")
        error_occurred = False
        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage) and msg.is_error:
                error_occurred = True
        
        # Rewind on error
        if error_occurred and checkpoints:
            print("Tests failed! Rewinding to last checkpoint...")
            name, checkpoint_id = checkpoints[-2]  # Go back one checkpoint
            await client.rewind_files(checkpoint_id)
            print(f"Rewound to: {name}")

anyio.run(main)

Parallel Sessions

Run multiple independent conversations:

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def process_session(client, session_id: str, task: str):
    await client.query(task, session_id=session_id)
    async for msg in client.receive_response():
        print(f"[{session_id}] {msg}")

async def main():
    async with ClaudeSDKClient() as client:
        async with anyio.create_task_group() as tg:
            tg.start_soon(process_session, client, "docs", "Document the API")
            tg.start_soon(process_session, client, "tests", "Write unit tests")
            tg.start_soon(process_session, client, "refactor", "Refactor utils")

anyio.run(main)

Progressive Workflow Pattern

Implement phased workflows:

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def progressive_workflow():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Edit", "Bash"],
        permission_mode="default"
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # Phase 1: Analysis
        print("Phase 1: Analysis")
        await client.query("Analyze the codebase for issues")
        async for msg in client.receive_response():
            pass  # Process messages
        
        # Phase 2: Planning
        print("\nPhase 2: Planning")
        await client.set_permission_mode("plan")
        await client.query("Create a detailed fix plan")
        async for msg in client.receive_response():
            pass
        
        # [User reviews plan here]
        
        # Phase 3: Implementation
        print("\nPhase 3: Implementation")
        await client.set_permission_mode("acceptEdits")
        await client.query("Implement the fixes")
        async for msg in client.receive_response():
            pass

anyio.run(progressive_workflow)

Best Practices

Always Use Context Manager

# Good
async with ClaudeSDKClient(options=options) as client:
    # ... use client ...

# Avoid (manual cleanup)
client = ClaudeSDKClient(options=options)
await client.connect()
# ... use client ...
await client.disconnect()

Always Receive Responses

# Good
await client.query("Do something")
async for msg in client.receive_response():
    print(msg)

# Bad (query won't complete)
await client.query("Do something")
# Missing receive_response() call!

Handle Errors Gracefully

from claude_agent_sdk import ClaudeSDKError

try:
    async with ClaudeSDKClient() as client:
        await client.query("...")
        async for msg in client.receive_response():
            print(msg)
except ClaudeSDKError as e:
    print(f"Error: {e}")
    # Handle or log error

Common Patterns

Extract Only Text

from claude_agent_sdk import AssistantMessage, TextBlock

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

Track Tool Usage

from claude_agent_sdk import AssistantMessage, ToolUseBlock

async def track_tools(client, prompt: str) -> list[str]:
    await client.query(prompt)
    tools = []
    async for msg in client.receive_response():
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, ToolUseBlock):
                    tools.append(block.name)
    return tools

Monitor Costs

from claude_agent_sdk import ResultMessage

async def query_with_cost(client, prompt: str) -> dict:
    await client.query(prompt)
    async for msg in client.receive_response():
        if isinstance(msg, ResultMessage):
            return {
                "cost": msg.total_cost_usd,
                "duration_ms": msg.duration_ms,
                "turns": msg.num_turns
            }
    return {}