SDK vs API Client

Anthropic offers two ways to use Claude. Understanding the difference helps you pick the right tool for your project.

query() — Communicating with the Agent

query() is the central function of the SDK. Send it a task in natural language and receive a stream of messages as the agent works.

Basic syntax

from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async for message in query(
    prompt="Analyze sales.csv and generate a summary report",
    options=ClaudeAgentOptions(
        system_prompt="You are an expert data analyst.",
        max_turns=10,              # max 10 action turns
        allowed_tools=["Read", "Bash"],
    )
):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                print(block.text)

Key parameters

Message types returned

Built-in Tools

The SDK ships with tools Claude can use immediately — no custom code needed. These are the agent's "hands".

Custom Tools & MCP

Beyond built-in tools, you can give the agent your own custom tools or connect it to external services via MCP.

Custom Tool in Python

Define a regular Python function, attach the @tool decorator, and register it with the agent. Claude will know when to call it.

from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeSDKClient

# Define your own tool
@tool("query_database", "Query the internal database", {"sql": str})
async def run_sql(args):
    result = your_db.execute(args["sql"])
    return {"content": [{"type": "text", "text": str(result)}]}

# Wrap as an internal MCP server (no separate process needed)
server = create_sdk_mcp_server(
    name="data-tools", version="1.0", tools=[run_sql]
)

# Pass it to the agent
client = ClaudeSDKClient(mcp_servers=[server])
await client.send("Analyze Q4 revenue from the database")

Connecting an External MCP Server

MCP (Model Context Protocol) is Anthropic's open tool-connection standard. You can connect Slack, GitHub, Notion, external databases with just a few lines of config.

options = ClaudeAgentOptions(
    mcp_servers=[
        {"name": "github",  "url": "https://mcp.github.com/sse"},
        {"name": "postgres", "url": "http://localhost:5432/mcp"},
    ],
    strict_mcp_config=True  # only use declared servers, ignore global config
)

Sessions & Context Management

Agents can work for hours continuously. Sessions help you manage memory, resume work after interruption, and avoid losing context when conversations grow long.

The problem: Context Window has limits

Resume a session after stopping

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

# Run 1: create session and save the ID
client = ClaudeSDKClient()
await client.send("Start refactoring the authentication module")
session_id = client.session_id  # save this

# Run 2: continue from exactly where it left off
client2 = ClaudeSDKClient(options=ClaudeAgentOptions(resume=session_id))
await client2.send("Continue — write unit tests for what you just refactored")

Eager session flushing for live UIs

Use session_store_flush="eager" if you need to display output in real-time to a terminal or dashboard — data is pushed immediately instead of waiting until the end of a turn.

Hooks — Controlling Agent Behavior

Hooks let you intercept between agent steps to inspect, block, or modify actions — like a security layer you control.

Pre-tool Hook (before tool use)

async def safety_check(context):
    # Block the agent from deleting files
    if context.tool_name == "Bash" and "rm -rf" in context.tool_input:
        return {"decision": "block", "reason": "File deletion not permitted"}
    # Auto-approve safe commands
    if context.tool_name in ["Read", "Glob"]:
        return {"decision": "approve"}

client = ClaudeSDKClient(pre_tool_use_hook=safety_check)

Post-tool Hook (after tool use)

Use to log actions, sanitize output before Claude sees it, or write an audit trail for compliance.

Streaming Output

The agent returns results as a stream — you receive each message the moment Claude finishes it, without waiting for the full task to complete.

Why streaming matters

For complex tasks (refactoring an entire codebase, analyzing a large dataset), the agent may run for several minutes. Streaming lets you display actual progress in a terminal or dashboard rather than staring at a blank screen.

async for message in query(prompt="Run the full test suite and fix failures"):
    match message:
        case AssistantMessage():
            print(f"🤖 Claude: {message.content[0].text}")
        case ToolUseMessage():
            print(f"🔧 Using tool: {message.name}")
        case ResultMessage(is_error=True):
            print(f"❌ Error: {message.error_text}")
        case ResultMessage():
            print(f"✅ Done after {message.num_turns} turns")

Structured Output

If you need JSON output (to process further in code), prompt Claude to return JSON and parse it:

result = await get_structured(
    prompt="Analyze the file. Return JSON: {errors: [], warnings: [], summary: ''}"
)
data = json.loads(result)

Use Case: AI Coding Agent

Apply the Claude Agent SDK to automate the full software development lifecycle — from code review to deployment.

async def auto_fix_bug(error_log: str, repo_path: str):
    async for msg in query(
        prompt=f"""
        Test suite failed with the following error:
        {error_log}

        Tasks:
        1. Read the stack trace and find the root cause
        2. Fix the code (only the broken part)
        3. Re-run: pytest tests/
        4. If passing: git add -A && git commit -m 'fix: ...'
        5. gh pr create --title 'Auto fix' --body 'Fixed by Claude Agent'
        """,
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Edit", "Bash", "Grep"],
            max_turns=20,
            system_prompt="You are a senior Python engineer. Fix bugs precisely, no unnecessary refactoring."
        )
    ):
        print(msg)

SDK features used

Use Case: Data Analytics Pipeline

Use the Agent SDK to build an automated large-scale data pipeline — from cleaning to visualization — without a data engineer running each step manually.

async def morning_pipeline(csv_path: str):
    async for msg in query(
        prompt=f"""
        New data file: {csv_path}

        Pipeline:
        1. Read and summarize (rows, columns, null %)
        2. Clean: handle nulls per domain rules, normalize date format
        3. Calculate KPIs: total revenue, by region, top 10 products
        4. Generate bar chart revenue by region, save to /reports/today.png
        5. Call send_slack #data-team with summary and chart path
        """,
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Write", "Bash"],
            max_turns=30,
            mcp_servers=[slack_server],
        )
    ):
        print(msg)

Use Case: Legal Document Review

Automate contract review — find risk clauses, flag unfavorable terms, compare against standard templates. Reduces junior lawyer time by 80%.

Use Case: Customer Support Triage

Automatically classify, prioritize, and resolve support tickets — the agent handles 70% of tier-1 tickets without human intervention, escalating complex cases with full context.

Use Case: Security Threat Detection

Agent continuously reads logs, correlates events, detects attack patterns, generates incident reports, and triggers automated response — an intelligent SIEM without rule-based limitations.

Use Case: Content Production Factory

Multi-channel marketing content pipeline — from a single brief, the agent produces a blog post, 10 social posts, an email newsletter, and SEO metadata, all consistent in voice and brand.

async def content_factory(brief: dict):
    # Orchestrator creates the blog first
    blog_content = await create_blog(brief)

    # 3 subagents run in parallel from the blog content
    social, email, seo = await asyncio.gather(
        create_social_posts(blog_content, platforms=["linkedin", "twitter", "threads"]),
        create_newsletter(blog_content),
        optimize_seo(blog_content, target_keyword=brief["keyword"]),
    )
    return {"blog": blog_content, "social": social, "email": email, "seo": seo}

Use Case: Employee Onboarding Agent

New employees get a personal AI agent — it reads the entire wiki, policy docs, and codebase, and answers any onboarding question 24/7 with full context about the company and their specific role.

Architecture Diagrams — How It All Connects

Understand how the Agent SDK, Claude model, tools, and your client interact with each other.

Agent Loop — Overall Diagram

Server Stdout → Client Flow

CLI / Stdout Demo — Real World

See how the agent operates in the terminal — from the invocation to each streamed message.

Method 1: Use the CLI directly (no code)

Method 2: Python SDK with stream output

Method 3: Subprocess — call the agent from any language

# Run agent non-interactively, output JSON to stdout
claude --print --output-format json-stream \
  -p "Check errors in src/ and list affected files" \
  | jq '.message'  # parse with jq

Live Console Animation

Real-time simulation of an agent running. Choose a scenario and click ▶ Run to watch each step appear like a real terminal.

LLM Pricing Calculator

Calculate the real cost of your Agent SDK pipeline. Adjust task volume, model mix, and token usage to estimate monthly cost.

Minor Features

Features rarely needed in basic projects but valuable in production.

Observability & Monitoring

• OpenTelemetry: Integrate trace/span to track agents in monitoring systems (Datadog, Jaeger).
• Cost Tracking: View per-request token cost via ResultMessage.usage.
• Todo Tracking: Agent auto-creates a todo list to stay on track in long tasks (via TodoWrite tool).

File & State Management

• File Checkpointing: Automatically snapshots files before Edit. Can rewind (undo) if agent breaks something.
• Session Store: Saves conversation transcript for resume after crash or step-by-step debugging.

Advanced Customization

• Slash Commands: Define shortcuts like /reset, /status to control the agent mid-session.
• Skills: Package sets of instructions/tools into a reusable "skill" across multiple projects.
• Plugins: Extension point for platform-level customization.
• Subagents: Main agent spawns child agents that run in parallel (parallelism).
• Tool Search: When you have hundreds of tools, the agent auto-finds the right ones instead of loading all of them.
• Effort Level: Adjust effort from low to xhigh (for Opus models).
• Strict MCP Config: Lock the exact list of MCP servers the agent can use, preventing unintended connections.

Multi-cloud Support

• Amazon Bedrock: Use CLAUDE_CODE_USE_BEDROCK=1 to route through AWS instead of Anthropic directly.
• Google Vertex AI: Use CLAUDE_CODE_USE_VERTEX=1 for Google Cloud.
• Microsoft Azure AI Foundry: Use CLAUDE_CODE_USE_FOUNDRY=1 for Azure.

Deploy & Billing

What you need to know when taking Agent SDK to production.

Billing Notice (from 15/6/2026)

Production best practices

Optimal Architecture — Multi-Skill, Multi-Document

When a project has dozens of Skills and thousands of documents, you can't load everything into context at once. This architecture describes how to intelligently layer things so the agent works accurately with minimal token cost.

4-Layer Architecture Map

Core Architecture Principles

Scan · Embed · Store — How Agents Read Large Data

When you have thousands of documents and skills, you need a data preparation pipeline before the agent runs. Here's the 3-step process.

1. Scan — Crawl and Chunk

2. Embed — Convert to Vectors

import voyageai
from qdrant_client import QdrantClient

vo = voyageai.Client()
qdrant = QdrantClient(url="http://localhost:6333")

# 1. Chunk documents (500 tokens/chunk, overlap 50)
chunks = chunk_documents(docs, size=500, overlap=50)

# 2. Embed with Voyage AI (Anthropic recommended)
doc_embeddings = vo.embed(
    [c.text for c in chunks],
    model="voyage-4", input_type="document"
).embeddings

# 3. Store in Qdrant
qdrant.upload_points(
    collection_name="project_docs",
    points=[{"id": i, "vector": emb,
             "payload": {"text": chunks[i].text, "source": chunks[i].source}}
            for i, emb in enumerate(doc_embeddings)]
)

3. Memory Store Layers

Tool Search for Skills — 90% context savings

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("all-MiniLM-L6-v2")  # local, free

# Offline: embed all skill descriptions once
skill_texts = [f"{s.name}: {s.description}" for s in all_skills]
skill_vectors = model.encode(skill_texts)  # [N_skills, 384]

def search_skills(query: str, top_k: int = 5, min_score: float = 0.70):
    q_vec = model.encode([query])[0]
    scores = np.dot(skill_vectors, q_vec)  # cosine similarity
    top_idx = np.argsort(scores)[::-1][:top_k]
    return [all_skills[i] for i in top_idx if scores[i] >= min_score]

Agent Scoring & Predefined Rubric

How do you know if the agent answered well? A rubric is a pre-defined set of rules — the agent "takes the test" against the rubric, it doesn't grade itself.

What is a Rubric?

Scoring Pipeline Architecture

Rubric YAML — Defining the Scale

# Rubric for data analysis task
task_type: data_analysis
version: "2.1"

dimensions:
  relevance:
    weight: 0.35
    description: "Does the result actually answer the question?"
    scoring:
      1.0: "Answers directly, correct metric asked for"
      0.7: "Answered but missing some dimensions"
      0.4: "Partially off-topic"

  accuracy:
    weight: 0.30
    validators:
      - type: cross_check_sql
      - type: range_sanity  # revenue cannot be negative

  completeness:
    weight: 0.20
    required_fields: [summary, kpis, anomalies, recommendation]

  cost_efficiency:
    weight: 0.15
    max_tokens_per_task: 50000
    penalty_over_budget: 0.5

thresholds:
  pass: 0.75
  warn: 0.60
  retry: 0.45
  escalate: 0.30

retry_strategy:
  max_retries: 2
  on_fail_dimension: relevance  # re-prompt focused on failing dimension

Scoring Methods — Implementation Tabs

Each rubric dimension is scored using a different method. No single method is "best" — what matters is matching the dimension's nature.

import anthropic

async def score_relevance(question: str, agent_output: str, criteria: dict) -> float:
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=100,
        messages=[{
            "role": "user",
            "content": f"""
Original question: {question}

Agent output:
{agent_output[:2000]}

Score Relevance (0.0 - 1.0) on this scale:
1.0 = {criteria['scoring'][1.0]}
0.7 = {criteria['scoring'][0.7]}
0.4 = {criteria['scoring'][0.4]}
0.0 = Completely irrelevant

Return only a single decimal number, no explanation.
"""
        }]
    )
    try:
        score = float(response.content[0].text.strip())
        return max(0.0, min(1.0, score))
    except:
        return 0.5  # fallback on parse error

def validate_range_sanity(output: dict) -> float:
    violations = 0
    checks = 0
    for key in ["revenue", "gmv", "arr"]:
        if key in output.get("kpis", {}):
            checks += 1
            if output["kpis"][key] < 0:
                violations += 1
    if checks == 0: return 1.0
    return 1.0 - (violations / checks)

def validate_cross_check_sql(output: dict, source_db) -> float:
    try:
        sql_result = source_db.query(
            "SELECT SUM(revenue) FROM orders WHERE date >= %s",
            [output["period_start"]]
        )
        expected = sql_result[0][0]
        reported = output["kpis"]["revenue"]
        diff_pct = abs(reported - expected) / expected
        if   diff_pct < 0.001: return 1.0
        elif diff_pct < 0.01:  return 0.85
        elif diff_pct < 0.05:  return 0.6
        else:                   return 0.2
    except:
        return 0.5

def score_completeness(output: dict, required_fields: list) -> float:
    found = 0
    for field in required_fields:
        parts = field.split(".")
        val = output
        try:
            for p in parts: val = val[p]
            has_value = val is not None and val != "" and val != []
            if has_value: found += 1
        except KeyError:
            pass
    return found / len(required_fields)

def score_cost_efficiency(result_message, max_tokens: int, penalty: float = 0.5) -> float:
    usage = result_message.usage
    total = usage.input_tokens + usage.output_tokens
    if total <= max_tokens:
        usage_pct = total / max_tokens
        return 1.0 - (usage_pct * 0.15)
    else:
        overrun_pct = (total - max_tokens) / max_tokens
        base_score = 1.0 * penalty
        return max(0.1, base_score - overrun_pct * 0.2)

async def evaluate_with_rubric(question, agent_output, result_msg, rubric_path) -> RubricResult:
    rubric = yaml.safe_load(open(rubric_path))
    dims = rubric["dimensions"]
    scores = {}
    scores["relevance"]      = await score_relevance(question, str(agent_output), dims["relevance"])
    scores["accuracy"]       = score_accuracy(agent_output, dims["accuracy"]["validators"])
    scores["completeness"]   = score_completeness(agent_output, dims["completeness"]["required_fields"])
    scores["cost_efficiency"] = score_cost_efficiency(result_msg, dims["cost_efficiency"]["max_tokens_per_task"])
    final = sum(scores[d] * dims[d]["weight"] for d in scores)
    t = rubric["thresholds"]
    verdict = ("PASS" if final >= t["pass"] else
               "WARN" if final >= t["warn"] else
               "RETRY" if final >= t["retry"] else "ESCALATE")
    failed = [d for d, s in scores.items() if s < 0.6]
    return RubricResult(scores, final, verdict, failed)

Interactive Scoring Calculator

Enter scores for each dimension to see how the weighted sum and verdict work in real time.

Radar Chart — Score by Dimension

Console Demo — Full End-to-End Run

Watch the entire pipeline run in a terminal: from receiving the prompt, scanning documents, selecting skills, spawning subagents, to scoring the result.

Step 1 — Setup & Index Data

Step 2 — Run Pipeline with Real Prompt

Cost Optimization — Strategy Comparison

Fact-Check Report — Claim-by-Claim Validation

All claims in this document were independently searched and verified from official sources (Anthropic docs, PyPI, GitHub, pricing pages). Checked: 14/05/2026.

1. SDK Identity & Package Names

2. Core API & Parameters

3. Built-in Tools

4. Custom Tools & MCP

5. Pricing & Model Names

6. Architecture Claims

Summary