The Anthropic Agent SDK lets you build networks of AI agents that collaborate, specialize, and execute complex tasks in parallel. This is where Claude goes from a tool to an autonomous engineering team.
Claude Code (CLI) is great for interactive development sessions. The Anthropic Agent SDK is for embedding agentic Claude behavior directly into your applications — as Python or TypeScript code that you control programmatically.
Claude Code CLI → you interact with Claude in your terminal. Great for development-time tasks.
Agent SDK → Python/TypeScript code that creates Claude agents programmatically as part of your production application. Great for automated workflows and multi-agent pipelines.
Every agentic Claude interaction follows the same fundamental observe-plan-act-evaluate cycle:
YOUR CODE sends task + tools
│
┌────▼────────────────┐
│ Claude API │
│ Decides: respond │
│ OR call a tool │
└────┬────────────────┘
│
┌────▼──────────────┐
│ text response? │──► DONE (task complete)
│ tool_use block? │──► Execute tool → get result
└───────────────────┘ │
└─► Send result back → Loop again
import anthropic client = anthropic.Anthropic() def run_agent(task: str, tools: list): messages = [{"role": "user", "content": task}] while True: r = client.messages.create(model="claude-sonnet-4-5", max_tokens=4096, tools=tools, messages=messages) messages.append({"role": "assistant", "content": r.content}) if r.stop_reason == "end_turn": return next(b.text for b in r.content if b.type == "text") tool_results = [] for b in r.content: if b.type == "tool_use": result = execute_tool(b.name, b.input) tool_results.append({"type": "tool_result", "tool_use_id": b.id, "content": str(result)}) messages.append({"role": "user", "content": tool_results})
A coordinator breaks down complex tasks and delegates to specialized subagents. Most common pattern for enterprise workflows.
User task
│
┌───▼──────────────────┐
│ COORDINATOR AGENT │ (Orchestrates, plans, merges)
└──┬────────┬─────────┬─┘
│ │ │
┌──▼──┐ ┌──▼──┐ ┌──▼──┐
│ Sub │ │ Sub │ │ Sub │ (Specialists run in PARALLEL)
│ A │ │ B │ │ C │
└──┬──┘ └──┬──┘ └──┬──┘
└────────┴─────────┘
│
┌───────▼──────────┐
│ SYNTHESIS AGENT │ (Combines all subagent outputs)
└──────────────────┘
Each agent hands off to the next. Great for ETL, content processing, and multi-stage validation workflows.
Ingest Agent → Validate Agent → Transform Agent → Load Agent → Verify Agent. Each stage's output becomes the next stage's input.
Same task runs on multiple independent agents simultaneously. Results are compared or voted on for higher reliability. Used for document review, code auditing, and confidence-critical decisions.
import asyncio async def research_company(company: str) -> dict: # Each subagent uses cheaper/faster Haiku model response = client.messages.create( model="claude-haiku-4-5", max_tokens=1500, messages=[{"role": "user", "content": f"Research {company}: products, strengths, weaknesses. JSON."}] ) return {"company": company, "data": response.content[0].text} async def coordinator(competitors: list) -> str: # Run ALL subagents in parallel — critical for performance results = await asyncio.gather(*[research_company(c) for c in competitors]) # Synthesis uses more capable Sonnet model final = client.messages.create( model="claude-sonnet-4-5", max_tokens=4000, messages=[{"role": "user", "content": f"Synthesize into competitive brief: {results}"}] ) return final.content[0].text
Use Claude Haiku for subagents doing focused, scoped tasks (cheaper, faster). Use Claude Sonnet for coordinator and synthesis (better reasoning). This asymmetric approach cuts cost by 60-80% vs. using Sonnet for everything.
Hooks intercept tool calls before or after execution. Essential for production safety, auditing, and data protection:
def before_tool(name: str, inputs: dict) -> dict: # 1. Audit log audit_log(name, inputs) # 2. Safety gate — block dangerous ops if name == "execute_sql" and "DROP" in inputs.get("query","").upper(): raise ValueError("DROP statements blocked") # 3. Inject context Claude shouldn't manage inputs["tenant_id"] = get_current_tenant() return inputs def after_tool(name: str, result: str) -> str: # Redact PII before Claude sees it if name == "get_customer": result = redact_fields(result, ["ssn", "dob", "credit_card"]) return result
Long-running agentic tasks need persistent sessions — so they can be paused, resumed, and audited across restarts:
import redis, json r = redis.Redis() def save(session_id, messages): r.setex(f"session:{session_id}", 86400, json.dumps(messages)) def resume(session_id): data = r.get(f"session:{session_id}") if not data: raise ValueError("Session expired") return json.loads(data) # Checkpoint after each major step save("research-2024-001", messages) # Next day: messages = resume("research-2024-001")
| Error Type | Handler | Action |
|---|---|---|
| Transient (timeout, rate limit) | Subagent auto-retry | Exponential backoff, max 3 attempts |
| Data error (bad inputs) | Coordinator | Validate and resubmit with corrected data |
| Authorization error | Human escalation | Pause task, alert team, await resolution |
| Catastrophic (data loss risk) | Immediate halt | Stop all agents, preserve state, alert on-call |
asyncio.gatherwrite_file(path, content) tool./output//etc/hosts — does the hook catch it?Resume Scanner → Culture Fit Evaluator → Skills Assessor → Decision Coordinator. 500 applications processed overnight; only top 10% go to human recruiters.
Parallel agents scan repos simultaneously — OWASP vulnerabilities, dependency issues, secrets in commits. Coordinator merges into prioritized report.
Monthly reports that took analysts 2 weeks now run as overnight agentic pipeline — data collection, validation, analysis, and narrative all automated.
Triage Agent classifies tickets → routes to specialist subagents (billing, technical, returns) → escalates to human only when confidence is low.