claude-py: Python SDK for Claude API with Streaming Support

# Install with uv (recommended)
uv pip install -e .

# Or with pip
pip install -e .

import asyncio
from claude import ClaudeAgentClient

async def main():
    async with ClaudeAgentClient() as client:
        # Stream response in real-time
        async for chunk in client.send_message_stream("Tell me a joke"):
            if chunk.text_delta:
                print(chunk.text_delta, end='', flush=True)
        print()

asyncio.run(main())

Why did the programmer quit his job?

Because he didn't get arrays! 😄

import asyncio
from claude import ClaudeAgentClient, AgentOptions

async def main():
    options = AgentOptions(stream=False)

    async with ClaudeAgentClient(options=options) as client:
        # Get complete response at once
        response = await client.send_message("What is 2+2?")
        print(response.content[0]['text'])

asyncio.run(main())

import asyncio
from claude import ChatClient

async def main():
    async with ChatClient() as c:
        # One-shot (non-streaming)
        r = await c.chat("Explain monads in one sentence")
        print(r.content[0]["text"])

        # Streaming
        async for chunk in c.stream("Write a haiku about recursion"):
            if chunk.text_delta:
                print(chunk.text_delta, end="", flush=True)
        print()

        # Streaming, return text only
        text = await c.collect("Say hello")
        print(text)

        # Per-call overrides
        r = await c.chat(
            "Who are you?",
            system="You are a pirate.",
            model="claude-haiku-4-5-20251001",
            max_tokens=256,
        )
        print(r.content[0]["text"])

        # Multi-turn via explicit message list
        r = await c.chat([
            {"role": "user", "content": "My name is Alice"},
            {"role": "assistant", "content": "Hi Alice!"},
            {"role": "user", "content": "What is my name?"},
        ])
        print(r.content[0]["text"])

asyncio.run(main())

from claude import query

# One-shot query
result = await query("Explain quantum computing in one sentence")
print(result.content[0]['text'])

async with ClaudeAgentClient() as client:
    # First message
    async for chunk in client.send_message_stream("What's the capital of France?"):
        if chunk.text_delta:
            print(chunk.text_delta, end='')
    print()

    # Second message (continues conversation)
    async for chunk in client.send_message_stream("What's the population?"):
        if chunk.text_delta:
            print(chunk.text_delta, end='')
    print()

def on_text(text: str):
    """Process text as it arrives."""
    # Update UI, log, analyze, etc.
    print(text, end='', flush=True)

async for chunk in client.send_message_stream("Write a haiku", on_text=on_text):
    # Text already printed by callback
    # Could also process chunks here
    pass

from claude import ToolResult

async with ClaudeAgentClient() as client:
    # Assistant requests tool use
    async for chunk in client.send_message_stream("Read the file data.txt"):
        if chunk.text_delta:
            print(chunk.text_delta, end='')
        # Detect tool uses in chunks...

    # Execute tool (simplified)
    with open('data.txt', 'r') as f:
        content = f.read()

    # Send tool result back
    tool_result = ToolResult(
        type="tool_result",
        tool_use_id="toolu_xxx",
        content=content
    )

    async for chunk in client.send_message_stream("", tool_results=[tool_result]):
        if chunk.text_delta:
            print(chunk.text_delta, end='')

from claude import AgentOptions

options = AgentOptions(
    # Model selection
    model="claude-sonnet-4-5-20250929",  # sonnet, opus, haiku

    # Token limit
    max_tokens=32000,  # Claude Code default

    # Streaming (default: True)
    stream=True,

    # Tool filtering
    allowed_tools=["Read", "Write", "Bash"],
    disallowed_tools=["WebSearch"],

    # Custom system prompt
    system_prompt="You are a helpful assistant.",
    append_system_prompt="Additional instructions...",

    # Session management
    session_id="custom-uuid",  # Or None for auto-generated

    # Working directory
    cwd="/path/to/workspace",
)

client = ClaudeAgentClient(options=options)

# Sessions are client-side - just maintain the client instance
async with ClaudeAgentClient() as client:
    # Message 1
    await client.send_message("Hello")

    # Message 2 (continues conversation)
    await client.send_message("How are you?")

    # All messages stored in client.messages
    print(f"Conversation has {len(client.messages)} messages")

# Count tokens for current conversation
token_count = await client.count_tokens()
print(f"Current conversation uses {token_count} tokens")

# Set your API key
export ANTHROPIC_API_KEY="sk-ant-..."

# Run examples
python .agents/tests/streaming_basic.py
python .agents/tests/streaming_chat.py
python .agents/tests/debug_streaming_chat.py

export ANTHROPIC_API_KEY="sk-ant-api03-YOUR_KEY_HERE"

export ANTHROPIC_API_KEY="sk-ant-oat01-YOUR_TOKEN_HERE"

User Request
    ↓
Client.send_message_stream()
    ↓
POST /v1/messages?beta=true
{
  "stream": true,
  "messages": [...],
  ...
}
    ↓
Server-Sent Events (SSE) Stream
    ↓
event: message_start
event: content_block_start
event: content_block_delta  ← Text chunks arrive
event: content_block_delta  ← More text
event: content_block_stop
event: message_stop
    ↓
StreamParser accumulates chunks
    ↓
Yields StreamChunk objects
    ↓
User processes in real-time

User Request
    ↓
Client.send_message()
    ↓
POST /v1/messages?beta=true
{
  "stream": false,
  "messages": [...],
  ...
}
    ↓
Complete JSON Response
    ↓
AssistantMessage returned

class ClaudeAgentClient:
    async def send_message_stream(
        prompt: str,
        tool_results: Optional[List[ToolResult]] = None,
        on_text: Optional[Callable[[str], None]] = None,
    ) -> AsyncIterator[StreamChunk]:
        """Stream response chunks."""

    async def send_message(
        prompt: str,
        tool_results: Optional[List[ToolResult]] = None,
    ) -> AssistantMessage:
        """Get complete response."""

    async def count_tokens() -> int:
        """Count tokens in conversation."""

    async def close():
        """Close HTTP client."""

class StreamChunk:
    event_type: str              # "message_start", "content_block_delta", etc.
    data: Dict[str, Any]         # Raw event data
    text_delta: Optional[str]    # Text for this chunk (if text event)
    content_block: Optional[Dict]  # Content block data (if applicable)

# Install with http2 and sse support
uv pip install "httpx[http2]>=0.27.0" "httpx-sse>=0.4.0"

options = AgentOptions(stream=True)

print(chunk.text_delta, end='', flush=True)  # flush=True is important!

claude-py/
├── src/claude/              # Core SDK implementation
│   ├── client.py           # Main client with streaming support
│   ├── streaming.py        # SSE parser and StreamParser
│   ├── types.py            # Type definitions
│   ├── tools.json          # All 17 tools from Claude Code
│   └── system_prompt.json  # 13KB system prompt from Claude Code
├── .agents/                # Self-generated documentation & tests
│   ├── *.md                # All internal documentation
│   ├── INDEX.md            # Directory navigation guide
│   ├── verify_installation.py  # Verify SDK setup
│   ├── reinstall.sh        # Reinstall with latest fixes
│   ├── data/               # Analysis assets
│   │   ├── *.mitm          # mitmproxy traffic captures
│   │   ├── real_system_prompt.json
│   │   └── real_tools.json
│   ├── research/           # Research notes & analysis
│   └── tests/              # All examples, tests & debug tools
│       ├── streaming_*.py  # Streaming examples
│       ├── debug_*.py      # Debug utilities
│       ├── test_*.py       # Test scripts
│       └── README.md       # Test documentation
├── README.md               # This file (user documentation)
└── pyproject.toml          # Package configuration

# Old (v0.1) - non-streaming only
response = await client.send_message("Hello")
print(response.content[0]['text'])

# New (v0.2) - streaming by default
async for chunk in client.send_message_stream("Hello"):
    if chunk.text_delta:
        print(chunk.text_delta, end='', flush=True)

# Or keep non-streaming behavior
options = AgentOptions(stream=False)
client = ClaudeAgentClient(options=options)
response = await client.send_message("Hello")  # Same as v0.1