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.
integration-patterns.mddocs/examples/
Integration Patterns
Patterns for integrating Claude Agent SDK with other tools and frameworks.
FastAPI Integration
Create an API endpoint with Claude:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
app = FastAPI()
class CodeReviewRequest(BaseModel):
code: str
language: str
class CodeReviewResponse(BaseModel):
issues: list[dict]
score: int
duration_ms: int
@app.post("/code-review", response_model=CodeReviewResponse)
async def review_code(request: CodeReviewRequest):
"""Review code and return analysis."""
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"},
"description": {"type": "string"}
},
"required": ["severity", "description"],
"additionalProperties": False
}
},
"score": {"type": "integer"}
},
"required": ["issues", "score"],
"additionalProperties": False
}
}
}
options = ClaudeAgentOptions(
output_format=schema,
max_budget_usd=0.10
)
prompt = f"Review this {request.language} code:\n\n{request.code}"
async for msg in query(prompt=prompt, options=options):
if isinstance(msg, ResultMessage):
if msg.is_error:
raise HTTPException(status_code=500, detail="Review failed")
return CodeReviewResponse(
issues=msg.structured_output["issues"],
score=msg.structured_output["score"],
duration_ms=msg.duration_ms
)
raise HTTPException(status_code=500, detail="No result received")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)Flask Integration
from flask import Flask, request, jsonify
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock
app = Flask(__name__)
@app.route('/ask', methods=['POST'])
def ask_claude():
"""Ask Claude a question."""
data = request.json
prompt = data.get('prompt')
if not prompt:
return jsonify({"error": "prompt required"}), 400
async def get_response():
options = ClaudeAgentOptions(max_turns=1)
texts = []
async for msg in query(prompt=prompt, options=options):
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
texts.append(block.text)
return " ".join(texts)
response = anyio.run(get_response)
return jsonify({"response": response})
if __name__ == '__main__':
app.run(debug=True)Django Integration
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
import anyio
import json
from claude_agent_sdk import query, ClaudeAgentOptions
@csrf_exempt
@require_http_methods(["POST"])
def analyze_view(request):
"""Analyze submitted code."""
try:
data = json.loads(request.body)
code = data.get('code')
async def analyze():
options = ClaudeAgentOptions(
allowed_tools=["Read"],
max_budget_usd=0.10
)
texts = []
async for msg in query(f"Analyze: {code}", options=options):
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
texts.append(block.text)
return " ".join(texts)
result = anyio.run(analyze)
return JsonResponse({"analysis": result})
except Exception as e:
return JsonResponse({"error": str(e)}, status=500)Celery Task Queue
from celery import Celery
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task(bind=True, max_retries=3)
def analyze_code_async(self, code: str):
"""Async code analysis task."""
try:
async def run_analysis():
options = ClaudeAgentOptions(
max_budget_usd=0.50,
max_turns=10
)
async for msg in query(f"Analyze: {code}", options=options):
if isinstance(msg, ResultMessage):
return {
"success": not msg.is_error,
"cost": msg.total_cost_usd,
"duration_ms": msg.duration_ms
}
result = anyio.run(run_analysis)
return result
except Exception as e:
# Retry with exponential backoff
raise self.retry(exc=e, countdown=2 ** self.request.retries)
# Usage
result = analyze_code_async.delay("def foo(): pass")SQLAlchemy Integration
Generate and validate models:
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions
Base = declarative_base()
async def generate_model_from_description(description: str):
"""Generate SQLAlchemy model from description."""
options = ClaudeAgentOptions(
output_format={
"type": "json_schema",
"json_schema": {
"name": "model_definition",
"schema": {
"type": "object",
"properties": {
"table_name": {"type": "string"},
"columns": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {"type": "string"},
"type": {"type": "string"},
"nullable": {"type": "boolean"}
}
}
}
}
}
}
}
)
prompt = f"Generate SQLAlchemy model definition for: {description}"
async for msg in query(prompt=prompt, options=options):
if isinstance(msg, ResultMessage) and msg.structured_output:
return msg.structured_output
# Usage
model_def = anyio.run(lambda: generate_model_from_description(
"User table with email, name, and age"
))Pytest Integration
Use Claude in tests:
import pytest
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions
@pytest.fixture
def claude_options():
return ClaudeAgentOptions(
max_budget_usd=0.05,
max_turns=1,
model="haiku"
)
@pytest.mark.asyncio
async def test_code_generation(claude_options):
"""Test that Claude can generate valid Python code."""
prompt = "Generate a function that calculates factorial"
code = []
async for msg in query(prompt=prompt, options=claude_options):
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
code.append(block.text)
code_text = " ".join(code)
# Verify generated code
assert "def" in code_text
assert "factorial" in code_text.lower()
# Try to execute it
exec(code_text)
@pytest.mark.asyncio
async def test_code_review(claude_options):
"""Test code review functionality."""
bad_code = "def foo():\n x = 1/0" # Division by zero
async for msg in query(f"Review: {bad_code}", options=claude_options):
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
# Should mention the division by zero
assert "division" in block.text.lower() or "zero" in block.text.lower()Docker Integration
Run in container with proper configuration:
FROM python:3.10
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt
# Copy application
COPY . .
# Set environment variables
ENV CLAUDE_SDK_CLI_PATH=/usr/local/bin/claude-code
ENV ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
CMD ["python", "app.py"]# app.py - Container-optimized configuration
import os
from claude_agent_sdk import ClaudeAgentOptions
options = ClaudeAgentOptions(
cli_path=os.getenv("CLAUDE_SDK_CLI_PATH"),
max_budget_usd=float(os.getenv("MAX_BUDGET", "1.0")),
permission_mode="bypassPermissions", # Non-interactive
setting_sources=["project"] # Reproducible
)GitHub Actions Integration
Use in CI/CD workflows:
# .github/workflows/claude-review.yml
name: Claude Code Review
on: [pull_request]
jobs:
review:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: '3.10'
- name: Install dependencies
run: |
pip install claude-agent-sdk
- name: Run Claude review
env:
ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
run: python .github/scripts/review.py# .github/scripts/review.py
import anyio
import sys
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
async def review_pr():
options = ClaudeAgentOptions(
allowed_tools=["Read", "Grep"],
max_budget_usd=0.50,
permission_mode="bypassPermissions"
)
async for msg in query("Review changed files for issues", options=options):
if isinstance(msg, ResultMessage):
if msg.is_error:
print("Review found critical issues")
sys.exit(1)
else:
print("Review passed")
sys.exit(0)
anyio.run(review_pr)Redis Cache Integration
Cache Claude responses:
import redis
import hashlib
import json
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def cache_key(prompt: str, options: dict) -> str:
"""Generate cache key from prompt and options."""
data = f"{prompt}:{json.dumps(options, sort_keys=True)}"
return hashlib.md5(data.encode()).hexdigest()
async def cached_query(prompt: str, options: ClaudeAgentOptions, ttl: int = 3600):
"""Query with Redis caching."""
# Generate cache key
key = cache_key(prompt, options.__dict__)
# Check cache
cached = redis_client.get(key)
if cached:
print("Cache hit!")
return json.loads(cached)
# Query Claude
print("Cache miss, querying Claude...")
result = []
async for msg in query(prompt=prompt, options=options):
result.append(str(msg))
# Store in cache
redis_client.setex(key, ttl, json.dumps(result))
return resultLogging Integration
Comprehensive logging:
import logging
import anyio
from claude_agent_sdk import (
ClaudeSDKClient,
ClaudeAgentOptions,
ClaudeSDKError
)
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('claude_sdk.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
async def logged_query(prompt: str):
"""Query with comprehensive logging."""
logger.info(f"Starting query: {prompt[:50]}...")
try:
options = ClaudeAgentOptions(
stderr=lambda msg: logger.debug(f"CLI stderr: {msg}"),
max_budget_usd=1.0
)
async with ClaudeSDKClient(options=options) as client:
logger.debug("Client connected")
await client.query(prompt)
logger.debug("Query sent")
async for msg in client.receive_response():
logger.debug(f"Received message: {type(msg).__name__}")
if isinstance(msg, ResultMessage):
logger.info(
f"Query completed - "
f"Duration: {msg.duration_ms}ms, "
f"Cost: ${msg.total_cost_usd:.4f}, "
f"Turns: {msg.num_turns}"
)
except ClaudeSDKError as e:
logger.error(f"Query failed: {e}", exc_info=True)
raise
anyio.run(lambda: logged_query("Analyze the codebase"))Monitoring and Metrics
Track usage metrics:
from dataclasses import dataclass
from datetime import datetime
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
@dataclass
class QueryMetrics:
prompt: str
timestamp: datetime
duration_ms: int
cost_usd: float
num_turns: int
success: bool
class MetricsCollector:
def __init__(self):
self.metrics = []
async def query_with_metrics(self, prompt: str, options: ClaudeAgentOptions):
"""Execute query and collect metrics."""
start_time = datetime.now()
async for msg in query(prompt=prompt, options=options):
if isinstance(msg, ResultMessage):
self.metrics.append(QueryMetrics(
prompt=prompt[:50],
timestamp=start_time,
duration_ms=msg.duration_ms,
cost_usd=msg.total_cost_usd or 0.0,
num_turns=msg.num_turns,
success=not msg.is_error
))
return msg
def report(self):
"""Generate metrics report."""
if not self.metrics:
return "No metrics collected"
total_cost = sum(m.cost_usd for m in self.metrics)
avg_duration = sum(m.duration_ms for m in self.metrics) / len(self.metrics)
success_rate = sum(1 for m in self.metrics if m.success) / len(self.metrics)
return f"""
Metrics Report
==============
Total Queries: {len(self.metrics)}
Total Cost: ${total_cost:.4f}
Avg Duration: {avg_duration:.0f}ms
Success Rate: {success_rate:.1%}
"""
# Usage
collector = MetricsCollector()
options = ClaudeAgentOptions(max_budget_usd=0.10)
anyio.run(lambda: collector.query_with_metrics("Task 1", options))
anyio.run(lambda: collector.query_with_metrics("Task 2", options))
print(collector.report())Next Steps
- Real-world scenarios: Real-World Scenarios
- Edge cases: Edge Cases
- API reference: Reference Documentation
Install with Tessl CLI
npx tessl i tessl/pypi-claude-agent-sdk