Claude Agent SDK in Python: First Agent to Workflows

The Claude Agent SDK (claude-agent-sdk) is Anthropic's official Python package for building autonomous AI agents. It wraps a bundled Claude Code CLI binary over stdio. Python code reaches file operations, terminal commands, and multi-step workflow chaining through that wrapper, and the SDK drives the tool loop itself.

Picking the Right Anthropic Python Package

Python developers building their first Claude agent face an immediate problem: two separate Anthropic packages exist, and most tutorials conflate them. The anthropic package is the official Python SDK for the Anthropic REST API, with synchronous and asynchronous clients powered by httpx. The claude-agent-sdk package wraps the Claude Code CLI subprocess. Agents built on it reach file operations, terminal commands, and web search on their own initiative.

Picking the right package is the first decision for any Python agent project. The Agent SDK handles the tool-call loop internally, so Claude autonomously reads files, runs terminal commands, and searches the web without application code managing each step. The Messages API takes a different approach, where developers write the tool execution loop in their application by following Anthropic's documented tool-use pattern. That approach offers flexibility at the cost of additional code.

The distinction matters because each package uses different classes, async patterns, and tool registration systems. Code using anthropic.Anthropic() and client.messages.create() does not use the Agent SDK at all. This guide covers the Agent SDK specifically, closing with Augment Cosmos, which addresses the operational problems that appear once agent work outgrows a single SDK session.

Installation and Setup on Python 3.10+

The Claude Agent SDK requires Python 3.10 or higher and supports 3.10, 3.11, 3.12, and 3.13. It ships as a single pip-installable package with the CLI binary bundled inside the official wheel.

pip install claude-agent-sdk

The installer skips any separate CLI download or PATH configuration. Wheel sizes vary by platform because the bundled CLI binary is compiled per architecture, so macOS Apple Silicon, macOS Intel, Linux x64, and Windows x64 each ship a different wheel.

API key configuration uses environment variables:

export ANTHROPIC_API_KEY=your-api-key

Alternative providers include AWS Bedrock (CLAUDE_CODE_USE_BEDROCK=1), Google Cloud Vertex AI (CLAUDE_CODE_USE_VERTEX=1), and Anthropic Foundry (CLAUDE_CODE_USE_FOUNDRY=1).

Migration note: The previous claude-code-sdk package is deprecated, and the class ClaudeCodeOptions was renamed to ClaudeAgentOptions. Any existing code using the old names requires updating.

Your First Claude Agent in Python

The Claude Agent SDK provides two interaction modes for Python developers. query() handles single exchanges, and ClaudeSDKClient maintains persistent multi-turn conversations. Both modes manage message streaming, tool execution, and the CLI subprocess automatically, so application code can focus on prompts and response handling.

Minimal Agent with query()

The query() function is the primary entry point and returns an AsyncIterator of response messages:

import asyncio
from claude_agent_sdk import query

async def main():
    async for message in query(prompt="What is 2 + 2?"):
        print(message)

asyncio.run(main())

The SDK depends on anyio, which runs on asyncio by default. Anthropic's Python reference uses asyncio.run(main()) throughout, while the package README quickstart uses anyio.run(main). Either entry point works, and switching between them is not a source of errors.

Multi-Turn Conversations with ClaudeSDKClient

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(model="claude-sonnet-5")
    async with ClaudeSDKClient(options=options) as client:
        await client.query("What files are in the current directory?")
        async for msg in client.receive_response():
            print(msg)
        await client.query("Now read the README.md file")
        async for msg in client.receive_response():
            print(msg)

asyncio.run(main())

ClaudeSDKClient maintains conversation history across query() calls, so the second message has full context of the first exchange. Each call to client.query() sends a message, and client.receive_response() returns an async iterator of response messages.

Configured Agent with Message Parsing

The ClaudeAgentOptions dataclass controls system prompts, model selection, and tool permissions, along with related configuration options:

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are a senior Python developer. Always follow PEP 8.",
        model="claude-opus-5",
        max_turns=5,
        allowed_tools=["Read", "Edit", "Glob"],
        permission_mode="acceptEdits",
    )
    async for message in query(
        prompt="Review my code and suggest improvements",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

asyncio.run(main())

Common misconception: allowed_tools is an auto-approval list, not a restriction list. Listed tools run without prompting, and unlisted tools remain available to Claude, falling through to the active permission mode and the can_use_tool callback. To remove a tool entirely, pass it to disallowed_tools, which strips the tool definition from the request so Claude never sees it. A scoped deny rule such as disallowed_tools=["Bash(rm *)"] keeps Bash available while blocking matching calls in every mode.

Error Handling

The SDK defines specific error types for common failure modes:

import asyncio
from claude_agent_sdk import query, CLINotFoundError, ProcessError

async def main():
    try:
        async for message in query(prompt="Hello"):
            print(message)
    except CLINotFoundError:
        print("Claude CLI not found. Verify installation with: pip install claude-agent-sdk")
    except ProcessError as e:
        print(f"CLI process failed with exit code: {e.exit_code}")

asyncio.run(main())

Adding Tools: Function Definitions and Response Handling

The Claude Agent SDK exposes the Claude Code built-in tools, including Read, Write, Edit, Bash, Glob, Grep, WebSearch, WebFetch, and Monitor. Developers extend this set with custom tools that run as in-process MCP servers, as shown in the Claude Agent SDK Python repository. Custom tool registration is the primary extension point for adapting agents to domain-specific workflows like database queries, internal APIs, or proprietary file formats.

Custom Tool Registration via MCP

Custom Python functions become in-process MCP servers with no subprocess overhead:

from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions

@tool("calculate", "Perform arithmetic calculation", {"expression": str})
async def calculate(args):
    try:
        result = eval(args["expression"], {"__builtins__": {}}, {})
        return {"content": [{"type": "text", "text": f"Result: {result}"}]}
    except Exception as e:
        return {"content": [{"type": "text", "text": f"Error: {str(e)}"}]}

server = create_sdk_mcp_server(
    name="math-tools", version="1.0.0", tools=[calculate]
)
options = ClaudeAgentOptions(
    mcp_servers={"math": server},
    allowed_tools=["mcp__math__calculate"]
)

Custom tools work with both query() and ClaudeSDKClient, in each case through the mcp_servers field on ClaudeAgentOptions. The Python dataclass uses snake_case field names, so mcp_servers and allowed_tools are correct here and the camelCase spellings belong to the TypeScript SDK. Tool permission names follow the pattern mcp__<server-name>__<tool-name>.

Messages API Tool Loop (anthropic Package)

For applications requiring direct control over tool execution, the anthropic package provides a manual tool-call loop where application code executes each tool and returns results. Claude returns stop_reason: "tool_use" when it wants to invoke a tool. Application code executes any requested tool, appends the result, and repeats while stop_reason == "tool_use", typically ending when the model returns stop_reason == "end_turn". One ordering constraint applies: tool_result blocks must come first in the content array, with any text after them, or the API returns a 400 error.

Async Patterns for Concurrent Agent Execution

The Claude Agent SDK is async-first, so production applications can run multiple agent conversations concurrently and stream responses without blocking. The patterns below use both the SDK's query() interface and the anthropic package's AsyncAnthropic client. Multi-agent workflows often combine the two, pairing the SDK for autonomous file and terminal operations with the Messages API for lightweight conversational agents that need no built-in tools.

Concurrent Multi-Agent Execution

asyncio.gather() runs independent agent conversations in parallel. While one request awaits the network response, the event loop processes others:

import asyncio
from anthropic import AsyncAnthropic

class AsyncAgent:
    def __init__(self, name: str, system_prompt: str):
        self.name = name
        self.system_prompt = system_prompt
        self.client = AsyncAnthropic()

    async def run(self, user_message: str) -> str:
        response = await self.client.messages.create(
            model="claude-sonnet-5", max_tokens=1024,
            system=self.system_prompt,
            messages=[{"role": "user", "content": user_message}]
        )
        return response.content[0].text

async def run_concurrent():
    agents = [
        AsyncAgent("Researcher", "You are a research specialist."),
        AsyncAgent("Summarizer", "You are a summarization specialist."),
    ]
    results = await asyncio.gather(
        agents[0].run("Research quantum computing trends"),
        agents[1].run("Summarize the history of AI"),
    )
    return results

This pattern works for independent tasks. Coordinating agents on a shared codebase is a harder problem, because parallel execution alone does nothing to stop two agents from editing overlapping files. Isolation has to come from the runtime rather than from the prompt.

Parallel Tool Execution

When Claude returns multiple tool_use blocks in a single response, application code should execute each tool call and return the corresponding tool_result blocks. Synchronous tools require wrapping with asyncio.to_thread() to prevent event loop blocking:

async def execute_tool(name: str, tool_input: dict):
    fn = TOOL_MAP[name]
    if asyncio.iscoroutinefunction(fn):
        return await fn(**tool_input)
    return await asyncio.to_thread(fn, **tool_input)

Known issue: GitHub issue #531 reports that the CLI does not trigger compaction for custom MCP tool results during a tool execution loop, and that parallel batches of any tool type overflow context without a compaction check. Built-in tools such as Read and Bash do compact in sequential mode. The issue was open at the time of writing, so long-running pipelines that lean on custom MCP tools should confirm its current status.

Multi-Step Workflows: Chaining Agents with State

Multi-step workflows in the Claude Agent SDK use explicit Python control flow in place of implicit model reasoning. Anthropic's research on building effective agents distinguishes workflows (predefined code paths) from agents (dynamic self-directed processes) and recommends workflows for reliability.

The workflow examples in this section use the anthropic Messages API for direct control over model calls and tool execution. The same patterns apply when wrapping Agent SDK query() calls, though the Messages API makes the control flow explicit for multi-step orchestration.

State Management Through Prompt Injection

A common inter-agent state mechanism serializes prior step outputs as JSON and injects them into each agent's user message:

from dataclasses import dataclass, field
import json

@dataclass
class WorkflowState:
    raw_input: str = ""
    step_outputs: dict = field(default_factory=dict)
    errors: list = field(default_factory=list)
    current_step: str = "init"
    completed_steps: list = field(default_factory=list)

    def record_step(self, step_name: str, output: str) -> None:
        self.step_outputs[step_name] = output
        self.completed_steps.append(step_name)

    def to_context_string(self) -> str:
        return json.dumps({
            "completed_steps": self.completed_steps,
            "outputs": self.step_outputs
        }, indent=2)

Each agent reads the full state and writes only to its own step slot. Same-session pipelines need no external database or shared memory store. The tradeoff appears when a pipeline outlives its process. The state object dies with the run, so nothing an agent learned on Monday survives to Tuesday, and a reviewer has to re-supply every correction by hand.

Sequential Pipeline with Asymmetric Model Selection

On a multi-agent search benchmark reported in Anthropic's Claude Opus 4.5 system card, Claude Opus 4.5 orchestrating Claude Haiku 4.5 subagents scored 87.0%, against 74.8% for Claude Opus 4.5 working alone. The same benchmark found the orchestrator choice mattered independently: Claude Opus 4.5 directing Claude Sonnet 4.5 subagents reached 85.4%, where Claude Sonnet 4.5 directing the same subagents reached 66.5%. Those numbers describe one search benchmark and one model generation, so teams should treat the pattern as the transferable part and re-benchmark whichever pair they deploy.

import anthropic

class ClaudeAgent:
    def __init__(self, name: str, system_prompt: str, model: str):
        self.name = name
        self.system_prompt = system_prompt
        self.model = model
        self.client = anthropic.Anthropic()

    def run(self, task: str, state: WorkflowState) -> str:
        context_message = (
            "## Workflow Context (read-only: prior step outputs)\n"
            f"{state.to_context_string()}\n\n"
            "## Your Task\n"
            f"{task}"
        )
        response = self.client.messages.create(
            model=self.model, max_tokens=4096,
            system=self.system_prompt,
            messages=[{"role": "user", "content": context_message}]
        )
        output = response.content[0].text
        state.record_step(self.name, output)
        return output

A three-agent content pipeline assigns models by task complexity:

Conditional Branching with Triage

A triage agent classifies incoming queries and routes to specialized branch agents. The classification step can run on a smaller model because routing decisions are simple compared to the work they dispatch:

class TriageAgent:
    def classify(self, query: str) -> str:
        response = self.client.messages.create(
            model="claude-haiku-4-5-20251001", max_tokens=512,
            system='Classify as: simple_answer | deep_research | escalate_human',
            messages=[{"role": "user", "content": query}]
        )
        return json.loads(response.content[0].text)["branch"]

Production Hardening: Guards, Retries, and Cost Control

Production agent deployments require safeguards against runaway execution, transient API failures, and uncontrolled costs. The patterns below apply to both claude-agent-sdk agents and anthropic Messages API tool loops.

Loop Guards and Token Limits

Production agents need a ceiling on iterations to prevent runaway tool cycles. A max_iterations guard of 20 catches infinite loops before they consume excessive tokens:

def run_with_guard(client, messages, tools, max_iterations=20):
    for iteration in range(max_iterations):
        response = client.messages.create(
            model="claude-sonnet-5", max_tokens=4096,
            tools=tools, messages=messages
        )
        if response.stop_reason == "end_turn":
            return response
        messages = append_tool_results(messages, response)
    raise RuntimeError(f"Exceeded {max_iterations} iterations")

Structured Retry Logic

API calls in production pipelines fail intermittently due to rate limits and transient server errors. Exponential backoff recovers from temporary failures without producing retry storms:

import time
from anthropic import RateLimitError, APIStatusError

def retry_with_backoff(fn, max_retries=3, base_delay=1.0):
    for attempt in range(max_retries):
        try:
            return fn()
        except RateLimitError:
            delay = base_delay * (2 ** attempt)
            time.sleep(delay)
        except APIStatusError as e:
            if e.status_code >= 500:
                delay = base_delay * (2 ** attempt)
                time.sleep(delay)
            else:
                raise
    raise RuntimeError(f"Failed after {max_retries} retries")

Non-retryable errors (400, 401, 403, 413) indicate request or configuration problems. Retrying these wastes time and tokens.

Structured Logging and Observability

The SDK supports observability through the Claude Code CLI, and the otel extra adds OpenTelemetry tracing. Logging the specific message types from the async iterator makes multi-step workflows debuggable in production, since each type signals a different phase of agent execution.

import asyncio
from claude_agent_sdk import query, AssistantMessage, ResultMessage, TextBlock

async def logged_query(prompt: str):
    async for message in query(prompt=prompt):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"[ASSISTANT] {block.text[:100]}...")
        elif isinstance(message, ResultMessage):
            print("[RESULT] Turn complete")

Pairing this with structured JSON logging of trace IDs and token counts correlates agent behavior with cost and performance metrics.

Cost Optimization Through Model Selection

Production pipelines benefit from asymmetric model assignment, following the same orchestrator-plus-worker pattern measured in the Opus 4.5 system card. Routing simple decisions to smaller models while reserving larger models for complex reasoning reduces pipeline cost. The actual saving depends on the traffic mix, so teams should measure it before projecting it.

Security and Deployment

Production deployments should avoid bypassPermissions mode, which approves every call that reaches the permission step regardless of what allowed_tools contains. Automated pipelines that need file write access should use acceptEdits, and disallowed_tools should name any tool the agent must never invoke. API keys belong in a secret manager, never in committed .env files or container images.

Wrapping query() in a FastAPI endpoint is a common pattern for HTTP-based invocation. The bundled CLI ships inside the wheel, so Docker images need only pip install claude-agent-sdk, at the cost of a larger image that can affect serverless cold starts. Each query() call can spawn multiple internal tool calls. Application-level semaphores should cap simultaneous sessions against the account's rate limits, and long-running tasks should return a job ID for polling rather than holding an HTTP connection open through the full agent loop.

Running Python Agents at Team Scale with Cosmos

Everything above lives inside one process. The agent starts when Python calls it, its context dies with the run, and any isolation between parallel agents has to be built by hand. Those constraints are fine for a script and expensive for a team.

Augment Cosmos is Augment Code's agent orchestration platform, generally available and included on paid plans. It runs agents in the cloud against a team's repositories, and it exists to solve the operational half of the problem that the SDK leaves to application code.

How Cosmos Structures Agent Work

Cosmos organizes work into core building blocks rather than function calls. Environments define where an agent runs and what it can touch, across laptops, Dev VMs, and cloud sandboxes. Experts define how an agent behaves, including its prompt, the tools and MCP servers it reaches through Capabilities, and the events it subscribes to. Sessions capture each run as an auditable, replayable record that stays private to one engineer or gets promoted into shared organizational capability.

Triggers replace the calling convention entirely. Instead of a Python function invoking an agent, Experts wake on events from GitHub, Slack, Linear, PagerDuty, webhooks, and cron. A shared file system with tenant-level and user-level memory carries patterns, conventions, and corrections between runs, which is the layer the prompt-injected WorkflowState object above cannot provide once the process exits.

Underneath sits Augment's Context Engine, which maps a codebase across repositories, services, and commit history, understanding relationships between hundreds of thousands of files. It retrieves the slice a task touches instead of replaying broad file searches: on Terminal Bench 2.0 with Opus 4.7, Augment spent 33% less than Claude Code while solving tasks at effectively the same rate.

Where Each Tool Fits

The SDK and Cosmos operate at different layers, and the split becomes clearest when mapped against the runtime concerns each one owns.

MCP is the practical bridge between the two. A tool written for an SDK agent already speaks the protocol that Cosmos Experts use to reach external tools, so domain-specific capabilities built during SDK prototyping carry forward as MCP servers rather than being rewritten. Teams comparing this layer against other options can start with a survey of agentic OS platforms before committing.

Start with One Agent, Then Decide What It Needs to Outlive

The Claude Agent SDK is the right starting point for Python agent work. Most teams should build the first agent exactly as described here, using query() for single exchanges, ClaudeSDKClient for multi-turn sessions, the @tool decorator for custom tools, and explicit control flow for anything multi-step.

The question worth answering early is what has to survive the process exiting. If the answer is nothing, application code is enough. Corrections that need to compound, runs that need an audit trail, and agents that wake on a pull request rather than a function call are all runtime concerns. Building that runtime a second time inside application code is the expensive path.

Frequently Asked Questions

Related

• 6 Best Spec-Driven Development Tools for AI Coding in 2026
• 5 Best Agentic Development Environments for Enterprise Teams in 2026
• 9 Open-Source Agent Orchestrators for AI Coding (2026)
• 9 Best AI Coding Agent Desktop Apps in 2026 (Ranked by Real-World Performance)
• 6 Best Devin Alternatives for AI Agent Orchestration in 2026