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

Simple Query Interface

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

The query() function provides a straightforward async interface for one-shot, stateless interactions with Claude Code. It's ideal for simple questions, batch processing, automation scripts, and scenarios where all inputs are known upfront.

Capabilities

Query Function

Execute one-shot or unidirectional streaming interactions with Claude Code. Returns an async iterator of messages representing the conversation.

async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeAgentOptions | None = None,
    transport: Transport | None = None,
) -> AsyncIterator[Message]

Parameters:

prompt (str | AsyncIterable[dict[str, Any]]): The prompt to send. String for single-shot queries, AsyncIterable for streaming mode with multiple messages
options (ClaudeAgentOptions | None): Optional ClaudeAgentOptions configuration. Controls tool execution, working directory, model selection, permissions, and more. None uses default settings
transport (Transport | None): Optional custom Transport implementation. None uses bundled Claude Code CLI subprocess transport

Returns: AsyncIterator[Message] yielding messages in order:

UserMessage: User prompts and tool results
AssistantMessage: Claude's responses with text and tool uses
SystemMessage: System-level notifications
ResultMessage: Final result with metrics and cost (signals completion)
StreamEvent: Partial updates (only if include_partial_messages=True)

Raises:

CLINotFoundError: Claude Code CLI not found or not installed
CLIConnectionError: Unable to connect to Claude Code
ProcessError: CLI process failed during execution
CLIJSONDecodeError: Unable to decode JSON from CLI output

Usage:

Simple query:

import anyio
from claude_agent_sdk import query

async def main():
    async for message in query(prompt="What is 2 + 2?"):
        print(message)

anyio.run(main)

With options:

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are an expert Python developer",
        cwd="/home/user/project",
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="acceptEdits"
    )

    async for message in query(prompt="Create a Python web server", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)

Streaming mode (still unidirectional):

import anyio
from claude_agent_sdk import query
from typing import AsyncIterator

async def prompts() -> AsyncIterator[dict]:
    """Generate multiple prompts for streaming."""
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Hello"},
        "parent_tool_use_id": None,
        "session_id": "default"
    }
    yield {
        "type": "user",
        "message": {"role": "user", "content": "How are you?"},
        "parent_tool_use_id": None,
        "session_id": "default"
    }

async def main():
    async for message in query(prompt=prompts()):
        print(message)

anyio.run(main)

With custom transport:

import anyio
from claude_agent_sdk import query, Transport

class MyCustomTransport(Transport):
    """Custom transport implementation."""
    async def connect(self): ...
    async def write(self, data): ...
    async def read_messages(self): ...
    async def close(self): ...
    def is_ready(self): ...
    async def end_input(self): ...

async def main():
    transport = MyCustomTransport()
    async for message in query(prompt="Hello", transport=transport):
        print(message)

anyio.run(main)

Common Patterns

Pattern 1: Extract Text Response

Get only text from Claude's response:

import anyio
from claude_agent_sdk import query, AssistantMessage, TextBlock

async def get_text_response(prompt: str) -> str:
    """Extract text from Claude's response."""
    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 " ".join(texts)

async def main():
    response = await get_text_response("What is the capital of France?")
    print(response)

anyio.run(main)

Pattern 2: Batch Processing

Process multiple prompts in sequence:

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions

async def process_batch(prompts: list[str]):
    """Process multiple prompts in batch."""
    options = ClaudeAgentOptions(
        max_turns=1,  # Limit turns for efficiency
        allowed_tools=["Read"]
    )
    
    results = []
    for prompt in prompts:
        texts = []
        async for message in query(prompt=prompt, options=options):
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        texts.append(block.text)
        results.append(" ".join(texts))
    
    return results

async def main():
    prompts = [
        "Summarize file1.py",
        "Summarize file2.py",
        "Summarize file3.py"
    ]
    results = await process_batch(prompts)
    for i, result in enumerate(results):
        print(f"\nResult {i+1}:\n{result}")

anyio.run(main)

Pattern 3: Get Metrics

Extract cost and timing metrics:

import anyio
from claude_agent_sdk import query, ResultMessage

async def query_with_metrics(prompt: str):
    """Query and return result with metrics."""
    async for message in query(prompt=prompt):
        if isinstance(message, ResultMessage):
            return {
                "session_id": message.session_id,
                "duration_ms": message.duration_ms,
                "num_turns": message.num_turns,
                "cost_usd": message.total_cost_usd,
                "usage": message.usage,
                "is_error": message.is_error
            }
    return None

async def main():
    metrics = await query_with_metrics("Calculate 2 + 2")
    print(f"Duration: {metrics['duration_ms']}ms")
    print(f"Turns: {metrics['num_turns']}")
    print(f"Cost: ${metrics['cost_usd']:.4f}")
    print(f"Tokens: {metrics['usage']}")

anyio.run(main)

Pattern 4: Tool Usage Tracking

Track which tools Claude used:

import anyio
from claude_agent_sdk import query, AssistantMessage, ToolUseBlock, ClaudeAgentOptions

async def track_tool_usage(prompt: str):
    """Track tools used during query."""
    tools_used = []
    
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash", "Grep"]
    )
    
    async for message in query(prompt=prompt, options=options):
        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

async def main():
    tools = await track_tool_usage("Find all Python files and count lines")
    print("Tools used:")
    for tool in tools:
        print(f"  - {tool['name']}: {tool['input']}")

anyio.run(main)

Pattern 5: Error Handling

Robust error handling with retries:

import anyio
from claude_agent_sdk import (
    query,
    ClaudeSDKError,
    CLIConnectionError,
    ProcessError,
    CLINotFoundError
)

async def query_with_retry(prompt: str, max_retries: int = 3):
    """Query with automatic retry on transient errors."""
    for attempt in range(max_retries):
        try:
            messages = []
            async for message in query(prompt=prompt):
                messages.append(message)
            return messages  # Success
            
        except CLINotFoundError:
            # Don't retry - fatal error
            print("Error: Claude Code not installed")
            raise
            
        except (CLIConnectionError, ProcessError) as e:
            if attempt < max_retries - 1:
                print(f"Attempt {attempt + 1} failed: {e}")
                print(f"Retrying... ({max_retries - attempt - 1} attempts left)")
                await anyio.sleep(2 ** attempt)  # Exponential backoff
            else:
                print(f"All {max_retries} attempts failed")
                raise
                
        except ClaudeSDKError:
            # Don't retry other errors
            raise

async def main():
    try:
        messages = await query_with_retry("What is 2 + 2?")
        print(f"Success! Received {len(messages)} messages")
    except ClaudeSDKError as e:
        print(f"Final error: {e}")

anyio.run(main)

Pattern 6: Timeout Handling

Add timeout to query:

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions

async def query_with_timeout(prompt: str, timeout_seconds: int = 60):
    """Query with timeout."""
    options = ClaudeAgentOptions(
        max_turns=10  # Prevent infinite loops
    )
    
    with anyio.fail_after(timeout_seconds):
        messages = []
        async for message in query(prompt=prompt, options=options):
            messages.append(message)
        return messages

async def main():
    try:
        messages = await query_with_timeout("Complex task", timeout_seconds=30)
        print(f"Completed with {len(messages)} messages")
    except TimeoutError:
        print("Query timed out after 30 seconds")

anyio.run(main)

Pattern 7: File Analysis

Analyze files with Claude:

import anyio
from pathlib import Path
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def analyze_files(directory: Path):
    """Analyze files in directory."""
    options = ClaudeAgentOptions(
        cwd=str(directory),
        allowed_tools=["Read", "Grep", "Glob"],
        system_prompt="You are a code reviewer. Be concise."
    )
    
    prompt = "Analyze all Python files and identify potential issues"
    
    async for message in query(prompt=prompt, options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

async def main():
    await analyze_files(Path("/path/to/project"))

anyio.run(main)

Pattern 8: Code Generation

Generate code with specific requirements:

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def generate_code(requirements: str, output_file: str):
    """Generate code and save to file."""
    options = ClaudeAgentOptions(
        allowed_tools=["Write"],
        permission_mode="acceptEdits",
        system_prompt="You are an expert programmer. Write production-quality code."
    )
    
    prompt = f"Create {output_file} with the following requirements:\n{requirements}"
    
    async for message in query(prompt=prompt, options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

async def main():
    requirements = """
    - A FastAPI web server
    - Endpoint /health for health checks
    - Endpoint /api/users for user management
    - Include error handling
    - Add logging
    """
    await generate_code(requirements, "server.py")

anyio.run(main)

Pattern 9: Structured Output

Get structured results:

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def get_structured_output(prompt: str):
    """Get structured JSON output."""
    schema = {
        "type": "json_schema",
        "json_schema": {
            "name": "analysis_result",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "summary": {"type": "string"},
                    "issues": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "severity": {"type": "string", "enum": ["high", "medium", "low"]},
                                "description": {"type": "string"}
                            },
                            "required": ["severity", "description"],
                            "additionalProperties": False
                        }
                    }
                },
                "required": ["summary", "issues"],
                "additionalProperties": False
            }
        }
    }
    
    options = ClaudeAgentOptions(
        output_format=schema,
        allowed_tools=["Read", "Grep"]
    )
    
    async for message in query(prompt=prompt, options=options):
        if isinstance(message, ResultMessage) and message.structured_output:
            return message.structured_output
    
    return None

async def main():
    result = await get_structured_output("Analyze security in this codebase")
    if result:
        print(f"Summary: {result['summary']}")
        print(f"Issues found: {len(result['issues'])}")
        for issue in result['issues']:
            print(f"  [{issue['severity']}] {issue['description']}")

anyio.run(main)

Pattern 10: Cost Budget Control

Limit costs for queries:

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def query_with_budget(prompt: str, max_budget_usd: float = 0.10):
    """Query with cost budget limit."""
    options = ClaudeAgentOptions(
        max_budget_usd=max_budget_usd,
        model="haiku"  # Use cheaper model
    )
    
    async for message in query(prompt=prompt, options=options):
        if isinstance(message, ResultMessage):
            if message.is_error:
                print(f"Query stopped: {message.result}")
            if message.total_cost_usd:
                print(f"Cost: ${message.total_cost_usd:.4f}")
            return message

async def main():
    result = await query_with_budget(
        "Analyze this large codebase",
        max_budget_usd=0.50
    )

anyio.run(main)

When to Use query()

Use query() for:

Simple one-off questions ("What is X?", "How do I do Y?")
Batch processing of independent prompts
Code generation or analysis tasks
Automated scripts and CI/CD pipelines
When you know all inputs upfront
Fire-and-forget automation
Stateless operations

Use ClaudeSDKClient instead for:

Interactive conversations with follow-ups
Chat applications or REPL-like interfaces
When you need to send messages based on Claude's responses
When you need interrupt capabilities
Long-running sessions with state
Real-time applications with user input
Dynamic permission or model changes mid-conversation

Comparison with ClaudeSDKClient

Feature	query()	ClaudeSDKClient
Conversation style	Unidirectional	Bidirectional
State	Stateless	Stateful
Follow-up messages	No	Yes
Interrupt support	No	Yes
Dynamic config changes	No	Yes (set_permission_mode, set_model)
Use case	One-shot, automation	Interactive, chat
Complexity	Simple	More complex

Troubleshooting

Issue: Query hangs indefinitely

Cause: Long-running operation or infinite loop
Solution: Add max_turns limit in options or use timeout with anyio.fail_after()

Issue: "Tool X not allowed"

Cause: Tool not in allowed_tools list
Solution: Add tool name to allowed_tools. For MCP tools use format "mcp__<server>__<tool>"

Issue: High costs

Cause: Complex operations or expensive model
Solution: Set max_budget_usd limit, use cheaper model ("haiku"), or reduce max_turns

Issue: Permission prompts in automation

Cause: permission_mode="default" requires interactive prompts
Solution: Use permission_mode="acceptEdits" or implement can_use_tool callback

Issue: Empty or incomplete responses

Cause: Process terminated early, budget exceeded, or max_turns reached
Solution: Check ResultMessage.is_error, increase limits, or review stderr

Performance Tips

Choose appropriate model:

Use "haiku" for simple tasks (fastest, cheapest)
Use "sonnet" for balanced performance (default)
Use "opus" for complex reasoning (slowest, most expensive)

Optimize tool usage:

Limit allowed_tools to only what's needed
Use permission_mode="acceptEdits" to avoid prompts
Consider max_turns to prevent runaway costs

Batch efficiently:

Process independent queries in parallel with asyncio.gather()
Reuse ClaudeAgentOptions across queries
Consider using single query with multiple questions for related tasks

Monitor costs:

Always check ResultMessage.total_cost_usd
Set max_budget_usd for cost control
Track usage metrics in ResultMessage.usage

tessl/pypi-claude-agent-sdk

query.mddocs/reference/

Simple Query Interface

Capabilities

Query Function

Common Patterns

Pattern 1: Extract Text Response

Pattern 2: Batch Processing

Pattern 3: Get Metrics

Pattern 4: Tool Usage Tracking

Pattern 5: Error Handling

Pattern 6: Timeout Handling

Pattern 7: File Analysis

Pattern 8: Code Generation

Pattern 9: Structured Output

Pattern 10: Cost Budget Control

When to Use query()

Comparison with ClaudeSDKClient

Troubleshooting

Issue: Query hangs indefinitely

Issue: "Tool X not allowed"

Issue: High costs

Issue: Permission prompts in automation

Issue: Empty or incomplete responses

Performance Tips

See Also

tessl/pypi-claude-agent-sdk

query.md.css-3qkkll{font-size:var(--chakra-font-sizes-sm);font-weight:var(--chakra-font-weights-normal);color:var(--chakra-colors-gray-300);}docs/reference/

Simple Query Interface

Capabilities

Query Function

Common Patterns

Pattern 1: Extract Text Response

Pattern 2: Batch Processing

Pattern 3: Get Metrics

Pattern 4: Tool Usage Tracking

Pattern 5: Error Handling

Pattern 6: Timeout Handling

Pattern 7: File Analysis

Pattern 8: Code Generation

Pattern 9: Structured Output

Pattern 10: Cost Budget Control

When to Use query()

Comparison with ClaudeSDKClient

Troubleshooting

Issue: Query hangs indefinitely

Issue: "Tool X not allowed"

Issue: High costs

Issue: Permission prompts in automation

Issue: Empty or incomplete responses

Performance Tips

See Also

query.mddocs/reference/