[2026 Guide] Claude Agent SDK in Production: A Complete Walkthrough

# Python
pip install claude-agent-sdk

# TypeScript / Node
npm install @anthropic-ai/agent-sdk

from claude_agent_sdk import Agent, tool
import httpx

@tool
def get_weather(city: str) -> str:
    """Return current weather for a city. Be specific (e.g. 'Lisbon, PT')."""
    r = httpx.get(
        "https://wttr.in/" + city,
        params={"format": "%C %t %h %w"},
        timeout=10.0,
    )
    r.raise_for_status()
    return r.text.strip()

agent = Agent(
    model="claude-sonnet-4-6",
    system="You are a concise travel assistant. Always answer in one sentence.",
    tools=[get_weather],
    max_steps=6,          # hard ceiling on loop iterations
    max_tokens_total=8000, # hard ceiling on tokens used per run
)

result = agent.run("Should I take a jacket to Lisbon today?")
print(result.text)
# → "Yes — Lisbon is 14°C with light rain, so a waterproof jacket is a good call."

from claude_agent_sdk import tool, ToolError
import hashlib, time

# 1. Idempotency — protect against retries
_seen_keys: set[str] = set()

@tool
def send_email(to: str, subject: str, body: str, idempotency_key: str) -> str:
    """Send an email. The idempotency_key MUST be unique per logical send."""
    key = f"{to}:{idempotency_key}"
    if key in _seen_keys:
        return f"already_sent:{idempotency_key}"
    _seen_keys.add(key)
    # ... actually call your mail provider
    return f"sent:{idempotency_key}"

# 2. Strict typing — return ToolError instead of raising
@tool
def lookup_user(user_id: str) -> dict:
    """Look up a user by ID. Returns {id, email, plan} or an error."""
    user = db.get_user(user_id)
    if user is None:
        # Returned to the model as a tool_result with is_error=True
        raise ToolError(f"no user with id={user_id}")
    return {"id": user.id, "email": user.email, "plan": user.plan}

# 3. Cost-bounded tools — never let a single call eat the whole budget
@tool
def search_docs(query: str, max_results: int = 5) -> list[dict]:
    """Search internal docs. Hard cap on result count to keep tokens bounded."""
    max_results = min(max_results, 10)  # never trust the model
    results = vector_store.search(query, k=max_results)
    return [{"title": r.title, "snippet": r.text[:300]} for r in results]

from claude_agent_sdk import Agent, ModelPicker

def pick_model(step: int, last_tool: str | None) -> str:
    """Per-step model selection. Cheap by default, expensive when it matters."""
    # The first synthesis step usually benefits from more reasoning
    if step == 0:
        return "claude-opus-4-7"
    # Long context summarization → Opus
    if last_tool == "fetch_full_document":
        return "claude-opus-4-7"
    # Everything else (tool dispatch, simple replies) → Sonnet
    return "claude-sonnet-4-6"

agent = Agent(
    model=ModelPicker(pick_model),
    system=PROMPT,
    tools=TOOLS,
    max_steps=8,
)

from claude_agent_sdk import Agent, McpServer

agent = Agent(
    model="claude-sonnet-4-6",
    system="You are an internal-tools assistant.",
    mcp_servers=[
        # stdio transport: spawn a local subprocess
        McpServer.stdio(
            name="filesystem",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "/workspace"],
        ),
        # HTTP transport: call a hosted MCP server
        McpServer.http(
            name="github",
            url="https://mcp.example.com/github",
            headers={"Authorization": f"Bearer {os.environ['GITHUB_MCP_TOKEN']}"},
        ),
        # SSE transport: server-sent events for streaming tools
        McpServer.sse(
            name="logs",
            url="https://logs.internal/mcp",
        ),
    ],
    tools=[],   # local tools optional — MCP brings its own
    max_steps=10,
)

result = agent.run(
    "Find every TODO in the repo, group them by file, and post a summary to GitHub issue #42."
)

from claude_agent_sdk import Agent, PostgresSessionStore, Summarizer

# Durable session storage — survives crashes, shared across workers
store = PostgresSessionStore(
    dsn=os.environ["DATABASE_URL"],
    table="agent_sessions",
)

# Summarize when the conversation crosses 60% of the context window
summarizer = Summarizer(
    trigger_at_token_pct=0.60,
    keep_last_n_turns=4,                # always keep the recent context verbatim
    summarizer_model="claude-sonnet-4-6", # cheap model for summarization itself
)

agent = Agent(
    model="claude-sonnet-4-6",
    system=PROMPT,
    tools=TOOLS,
    session_store=store,
    summarizer=summarizer,
    max_steps=20,
)

# Each session has a stable id — same id resumes the conversation
result = agent.run(
    "Pick up where we left off on issue #42.",
    session_id="user-7331",
)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent.runtime import agent
import logging, uuid

log = logging.getLogger("agent")
app = FastAPI(title="claude-agent", version="1.0")

class TurnRequest(BaseModel):
    message: str
    session_id: str | None = None
    user_id: str

@app.post("/v1/chat")
def chat(req: TurnRequest):
    session_id = req.session_id or str(uuid.uuid4())
    log.info("agent.run.start", extra={"session_id": session_id, "user_id": req.user_id})
    try:
        result = agent.run(
            req.message,
            session_id=session_id,
            metadata={"user_id": req.user_id},  # propagated to traces
        )
    except Exception as e:
        log.exception("agent.run.error", extra={"session_id": session_id})
        raise HTTPException(
            status_code=500,
            detail={"session_id": session_id, "error": str(e)},
        )
    log.info("agent.run.ok", extra={
        "session_id": session_id,
        "tokens_used": result.usage.total_tokens,
        "tool_calls": len(result.tool_calls),
        "steps": result.steps,
    })
    return {
        "session_id": session_id,
        "text": result.text,
        "usage": result.usage.model_dump(),
    }

@app.get("/healthz")
def health():
    return {"status": "ok"}

FROM python:3.12-slim

# Non-root — never run an agent as root
RUN useradd -m -u 10001 agent
WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY agent/ ./agent/
COPY api/ ./api/

USER agent
EXPOSE 8080

CMD ["gunicorn", "api.main:app", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--workers", "4", \
     "--bind", "0.0.0.0:8080", \
     "--timeout", "180", \
     "--graceful-timeout", "30"]

{
  "agent": "./agent/runtime.py:agent",
  "python_version": "3.12",
  "dependencies": ["."],
  "rapidclaw": {
    "session_store": "postgres",
    "min_replicas": 2,
    "max_replicas": 20,
    "scaling_metric": "in_flight_runs",
    "target_value": 8,
    "observability": {
      "otel": true,
      "log_level": "info",
      "include_tool_calls": true
    },
    "security": {
      "egress": "allowlist",
      "allowed_hosts": [
        "api.anthropic.com",
        "api.github.com",
        "wttr.in"
      ]
    }
  }
}

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from claude_agent_sdk import Agent, OtelTracer

trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint=os.environ["OTEL_ENDPOINT"]))
)

# The SDK ships a first-party OTel hook — every step, tool call, and
# model call becomes a span with token usage as attributes.
agent = Agent(
    model="claude-sonnet-4-6",
    system=PROMPT,
    tools=TOOLS,
    tracer=OtelTracer(service_name="travel-agent"),
)