The Anthropic Agent SDK

The Claude Agent SDK is the same agent loop, tool execution, and context management that power Claude Code, as a Python/TypeScript library with built-in tools (Read, Edit, Bash, Glob, Grep, WebSearch, WebFetch). Unlike the lower-level Client SDK, it runs the tool loop for you.

pip install claude-agent-sdk   # requires Python 3.10+
export ANTHROPIC_API_KEY=your-api-key

The core primitive is query() — an async generator yielding each message (assistant turns, tool calls, results):

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    async for message in query(prompt="What files are here?",
            options=ClaudeAgentOptions(allowed_tools=["Bash", "Glob"])):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

allowed_tools pre-approves tools so the loop runs unattended; anything else triggers a permission decision (Step 5). For multi-turn sessions, use ClaudeSDKClient.

We are building a research workflow: a coordinator, search, synthesis, and fact-checker agent. The SDK's unit of specialization is the subagent, declared with AgentDefinition — each gets isolated context and scoped tools, the decomposition discipline from Lesson 4.

from claude_agent_sdk import AgentDefinition

AGENTS = {
    "searcher": AgentDefinition(
        description="Finds primary sources for a focused question.",
        prompt="Search the web and return JSON url/claim/quote objects. Do not synthesize.",
        tools=["WebSearch", "WebFetch"]),
    "synthesizer": AgentDefinition(
        description="Combines sources into a structured briefing.",
        prompt="Group claims by theme, cite each by url, add no facts absent from sources.",
        tools=["Read"]),
    "fact_checker": AgentDefinition(
        description="Independently verifies each claim against its source.",
        prompt="Fetch each claim's url; mark SUPPORTED/UNSUPPORTED/UNCERTAIN. Return the table.",
        tools=["WebFetch"]),
}

Each tools list is a strict subset of system capabilities — least-privilege at the agent boundary, and the topology is legible straight off the lists. Context is isolated per subagent: it sees only what is passed in and returns only what is relevant, so pass inputs in.

Subagents are invoked through the built-in Agent tool. Include Agent in allowed_tools and register specialists under agents so the coordinator delegates without approving every hop. This is the supervisor pattern from Lesson 2: one coordinator, many workers.

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

COORDINATOR = "Decompose; delegate to searcher/synthesizer/fact_checker; drop UNSUPPORTED claims; never do the work."

async def run_research(question: str):
    opts = ClaudeAgentOptions(
        system_prompt=COORDINATOR,
        allowed_tools=["Agent"],   # its only tool: delegate
        agents=AGENTS, permission_mode="default")
    async for message in query(prompt=question, options=opts):
        tag = getattr(message, "parent_tool_use_id", None)
        prefix = f"[subagent {tag}]" if tag else "[coordinator]"
        if hasattr(message, "result"):
            print(prefix, message.result)

asyncio.run(run_research("State of solid-state battery commercialization?"))

The coordinator's only move is delegation. Each message inside a subagent's run carries a parent_tool_use_id, attributing every line to the execution that produced it — the backbone of tracing (Step 6, Lesson 8). Branches are fault-isolated: a failed search fails one sub-question.

Two mechanisms share state. Sessions give continuity: capture the session ID from the init message, then resume or fork via resume=session_id. The second is the blackboard from Lesson 3: agents read and write a common store instead of passing everything through the prompt. Expose it as an in-process MCP tool via tool and create_sdk_mcp_server:

from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions

ARTIFACTS = {}  # in production: Redis, S3, or a database

@tool("save_artifact", "Persist a named JSON artifact", {"key": str, "value": str})
async def save(args):
    ARTIFACTS[args["key"]] = args["value"]
    return {"content": [{"type": "text", "text": f"saved {args['key']}"}]}

@tool("load_artifact", "Read an artifact by key", {"key": str})
async def load(args):  # MISSING sentinel lets readers degrade gracefully
    return {"content": [{"type": "text", "text": ARTIFACTS.get(args["key"], "MISSING")}]}

store = create_sdk_mcp_server("blackboard", "1.0.0", tools=[save, load])
options = ClaudeAgentOptions(mcp_servers={"blackboard": store},
    allowed_tools=["mcp__blackboard__save_artifact", "mcp__blackboard__load_artifact"])

The searcher writes sources:battery; the synthesizer reads it — by reference, not a blob in context.

Course 04 covered per-agent safety. Orchestration adds a tier the coordinator enforces on its workers. Permissions are coarse: allowed_tools pre-approves the safe set, permission_mode decides the rest. Hooks are fine-grained — callbacks at lifecycle points (PreToolUse, PostToolUse, Stop, and more) to validate, log, block, or transform behavior. A PreToolUse hook is your enforcement point: inspect the input and refuse before it runs.

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher

ALLOWED = {"docs.anthropic.com", "arxiv.org", "nature.com"}

async def restrict_fetch(input_data, tool_use_id, context):
    url = input_data.get("tool_input", {}).get("url", "")
    if url and not any(h in url for h in ALLOWED):   # a deny blocks the call
        return {"hookSpecificOutput": {"permissionDecision": "deny"}}
    return {}

opts = ClaudeAgentOptions(allowed_tools=["Agent", "WebFetch"], agents=AGENTS,
    hooks={"PreToolUse": [HookMatcher(matcher="WebFetch", hooks=[restrict_fetch])]})

Hooks run in-process and fire for subagent calls too, so one allowlist enforces a network policy across the fleet; a PostToolUse audit hook adds the trail.

A multi-agent run is interleaved — branches stream concurrently and a flat log is unreadable. Bucket each message by parent_tool_use_id (defaulting to "coordinator") for one timeline per subagent. Common failures and their signatures:

Log the artifact at each handoff boundary keyed by parent_tool_use_id — these become OpenTelemetry spans in Lesson 8.