Build Production AI Agents with the Claude Agent SDK for Python

Anthropic built more than Claude — they packaged the exact agent harness that powers Claude Code and shipped it as a library. The Claude Agent SDK for Python gives you that same agent loop, tool execution, context management, and permission system as importable, composable code.

If you live in Python — data pipelines, backend services, automation scripts, ML tooling — this is the fastest way to add a genuinely autonomous agent to your stack. In this tutorial you will go from a blank virtual environment to an agent that reads files, runs commands, calls your own custom tools, and enforces safety rules you define.

What You Will Learn

By the end of this tutorial you will be able to:

• Set up a Python project with the Claude Agent SDK
• Run a one-shot autonomous agent with the query function
• Hold a multi-turn streaming conversation with ClaudeSDKClient
• Expose your own Python functions as tools via the @tool decorator
• Gate dangerous actions with a can_use_tool permission callback
• Observe and audit every tool call with lifecycle hooks
• Apply production patterns for cost, safety, and reliability

Prerequisites

Before starting, make sure you have:

• Python 3.10 or newer installed (python --version)
• Node.js 18+ installed — the SDK talks to the Claude Code CLI under the hood
• An Anthropic API key from the console, or an active Claude subscription
• Basic familiarity with Python async/await and asyncio

Step 1: Project Setup

Create an isolated environment and install the SDK. We will use the built-in venv, but uv or poetry work equally well.

mkdir claude-agent-python && cd claude-agent-python
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
 
pip install claude-agent-sdk

The SDK drives the Claude Code CLI, so install it too:

npm install -g @anthropic-ai/claude-code

Finally, export your key so the SDK can authenticate:

export ANTHROPIC_API_KEY="sk-ant-..."

Step 2: Your First Agent with query

The query function is the simplest entry point. It runs a fully autonomous agent loop and yields messages as they stream back. Create first_agent.py:

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions
 
async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are a concise senior Python engineer.",
        allowed_tools=["Read", "Glob", "Grep"],
        cwd=".",
    )
 
    async for message in query(
        prompt="List the Python files in this folder and summarize what each does.",
        options=options,
    ):
        print(message)
 
anyio.run(main)

Run it:

python first_agent.py

The agent decides on its own to call Glob to find files, Read to open them, and then writes a summary. You never wrote the orchestration — the SDK ran the loop. Note that we passed allowed_tools to restrict the agent to read-only operations; it cannot write or run shell commands here.

Understanding the message stream

query yields typed message objects. The most useful ones are:

• AssistantMessage — text and tool-use blocks the model produced
• UserMessage — tool results fed back into the loop
• ResultMessage — the final summary, including token usage and cost

Let's filter the stream to print only assistant text and the final cost:

from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, TextBlock, ResultMessage,
)
 
async def main():
    options = ClaudeAgentOptions(allowed_tools=["Read", "Glob"])
 
    async for message in query(prompt="Summarize README.md", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)
        elif isinstance(message, ResultMessage):
            print(f"\n--- done in {message.num_turns} turns, "
                  f"cost ${message.total_cost_usd:.4f} ---")

Step 3: Multi-Turn Conversations with ClaudeSDKClient

query is stateless — one prompt, one run. For interactive sessions where the agent remembers context across turns, use ClaudeSDKClient. It keeps the session alive so you can send follow-up prompts.

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    AssistantMessage, TextBlock,
)
 
async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful coding assistant.",
        allowed_tools=["Read", "Write", "Edit", "Bash"],
        permission_mode="acceptEdits",
    )
 
    async with ClaudeSDKClient(options=options) as client:
        # First turn
        await client.query("Create a file hello.py that prints today's date.")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
 
        # Follow-up turn — the agent remembers hello.py
        await client.query("Now add a function that returns the date as ISO 8601.")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
 
anyio.run(main)

The async with block manages the session lifecycle. client.query() sends a prompt; client.receive_response() streams messages until that turn completes. Because the client persists, the second prompt operates on the file created in the first.

Step 4: Custom Tools with the @tool Decorator

Built-in tools cover the filesystem and shell. To connect your agent to your world — a database, an internal API, a pricing engine — define custom tools. The SDK exposes them through an in-process MCP server, so there is no network hop.

Here is an agent that can look up orders and compute shipping, backed by plain Python functions:

import anyio
from claude_agent_sdk import (
    tool, create_sdk_mcp_server,
    ClaudeSDKClient, ClaudeAgentOptions,
    AssistantMessage, TextBlock,
)
 
# A fake datastore for the demo
ORDERS = {
    "A100": {"customer": "Layla", "weight_kg": 2.5, "country": "TN"},
    "A101": {"customer": "Omar", "weight_kg": 8.0, "country": "SA"},
}
 
@tool("get_order", "Fetch an order by its ID", {"order_id": str})
async def get_order(args):
    order = ORDERS.get(args["order_id"])
    if not order:
        return {"content": [{"type": "text", "text": "Order not found"}]}
    return {"content": [{"type": "text", "text": str(order)}]}
 
@tool("shipping_cost", "Estimate shipping cost in USD",
      {"weight_kg": float, "country": str})
async def shipping_cost(args):
    base = 5.0 if args["country"] == "TN" else 12.0
    cost = base + args["weight_kg"] * 1.5
    return {"content": [{"type": "text", "text": f"${cost:.2f}"}]}
 
async def main():
    server = create_sdk_mcp_server(
        name="fulfillment",
        version="1.0.0",
        tools=[get_order, shipping_cost],
    )
 
    options = ClaudeAgentOptions(
        mcp_servers={"fulfillment": server},
        allowed_tools=[
            "mcp__fulfillment__get_order",
            "mcp__fulfillment__shipping_cost",
        ],
    )
 
    async with ClaudeSDKClient(options=options) as client:
        await client.query("What will it cost to ship order A101?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
 
anyio.run(main)

Three things to notice:

1. The @tool decorator takes a name, a description, and a schema mapping argument names to types. Claude reads the description to decide when to call it.
2. create_sdk_mcp_server bundles your tools into a server you register under a namespace.
3. Tools are addressed as mcp__<server>__<tool>. You must list them in allowed_tools or the agent cannot call them.

The agent will chain the tools on its own: fetch order A101, read the weight and country, then call shipping_cost with those values — no manual wiring required.

Step 5: Guarding Actions with a Permission Callback

Autonomy without guardrails is a liability. The can_use_tool callback runs before every tool call, letting you allow, deny, or rewrite it based on your own logic.

from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    PermissionResultAllow, PermissionResultDeny,
    ToolPermissionContext,
)
 
async def gatekeeper(tool_name: str, tool_input: dict,
                     context: ToolPermissionContext):
    # Block destructive shell commands outright
    if tool_name == "Bash":
        command = str(tool_input.get("command", ""))
        for danger in ("rm -rf", "sudo", "mkfs", ":(){"):
            if danger in command:
                return PermissionResultDeny(
                    message=f"Blocked dangerous command: {danger}",
                    interrupt=True,
                )
    # Everything else is allowed
    return PermissionResultAllow()
 
async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        can_use_tool=gatekeeper,
    )
    async with ClaudeSDKClient(options=options) as client:
        await client.query("Clean up temp files in this directory.")
        async for msg in client.receive_response():
            print(msg)

If the agent tries anything matching your blocklist, the callback returns PermissionResultDeny with interrupt=True, which stops the run immediately. Return PermissionResultAllow to proceed. This is your last line of defense — keep the logic here simple, explicit, and well-tested.

Step 6: Observability with Hooks

For anything running unattended, you need an audit trail. Hooks fire on lifecycle events so you can log, measure, or veto actions without touching your tool code. Register them with HookMatcher:

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher
 
async def audit_pre_tool(input_data, tool_use_id, context):
    print(f"[AUDIT] about to run {input_data['tool_name']} "
          f"(id={tool_use_id})")
    return {}  # empty dict = allow, continue normally
 
async def audit_post_tool(input_data, tool_use_id, context):
    print(f"[AUDIT] finished {input_data['tool_name']}")
    return {}
 
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Bash"],
    hooks={
        "PreToolUse": [HookMatcher(hooks=[audit_pre_tool])],
        "PostToolUse": [HookMatcher(hooks=[audit_post_tool])],
    },
)

The matcher argument on HookMatcher can target a specific tool (for example matcher="Bash") so a hook only fires for that tool. In production, swap the print calls for your logging framework and ship the records to your observability stack.

Step 7: A Complete Mini-Agent

Let's tie it together into a small but realistic agent: a repository assistant with a custom tool, a permission gate, and audit hooks.

import anyio
from claude_agent_sdk import (
    tool, create_sdk_mcp_server,
    ClaudeSDKClient, ClaudeAgentOptions,
    HookMatcher, PermissionResultAllow, PermissionResultDeny,
    AssistantMessage, TextBlock, ResultMessage,
)
 
@tool("count_todos", "Count TODO comments in a source file",
      {"path": str})
async def count_todos(args):
    try:
        with open(args["path"], encoding="utf-8") as fh:
            n = sum(1 for line in fh if "TODO" in line)
        return {"content": [{"type": "text", "text": f"{n} TODOs"}]}
    except FileNotFoundError:
        return {"content": [{"type": "text", "text": "File not found"}]}
 
async def gatekeeper(tool_name, tool_input, context):
    if tool_name == "Bash" and "rm " in str(tool_input.get("command", "")):
        return PermissionResultDeny(message="No deletions allowed")
    return PermissionResultAllow()
 
async def audit(input_data, tool_use_id, context):
    print(f"[hook] {input_data['tool_name']}")
    return {}
 
async def main():
    server = create_sdk_mcp_server(name="repo", tools=[count_todos])
    options = ClaudeAgentOptions(
        system_prompt="You are a meticulous code auditor.",
        mcp_servers={"repo": server},
        allowed_tools=["Read", "Glob", "Grep", "mcp__repo__count_todos"],
        can_use_tool=gatekeeper,
        hooks={"PreToolUse": [HookMatcher(hooks=[audit])]},
        max_turns=10,
    )
 
    async with ClaudeSDKClient(options=options) as client:
        await client.query(
            "Find the Python file with the most TODO comments and report it."
        )
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
            elif isinstance(msg, ResultMessage):
                print(f"\nUsage: {msg.usage}")
 
anyio.run(main)

This agent globs the tree, reads candidate files, calls your count_todos tool, and reports the winner — all while every tool call is audited and deletions are blocked. Note max_turns=10, which caps the loop so a confused agent cannot run forever (and run up cost).

Testing Your Implementation

Verify each piece works:

1. Read-only agent — run Step 2 in a folder with a couple of .py files. You should see a per-file summary and no writes.
2. Custom tools — run Step 4 and confirm the agent calls both get_order and shipping_cost for order A101.
3. Permission gate — ask the Step 5 agent to run rm -rf /tmp/x and confirm it is blocked with your denial message.
4. Hooks — confirm [AUDIT] lines print before every tool call in Step 6.

Troubleshooting

• CLINotFoundError or "command not found" — the Claude Code CLI is missing. Run npm install -g @anthropic-ai/claude-code.
• Authentication errors — confirm ANTHROPIC_API_KEY is exported in the same shell running the script.
• The agent refuses to call a custom tool — check it is listed in allowed_tools with the exact mcp__server__tool name, and that your description clearly states when to use it.
• Runaway loops or high cost — set max_turns and inspect ResultMessage.total_cost_usd to catch expensive prompts early.
• RuntimeError about the event loop — always drive the SDK with anyio.run(main) or asyncio.run(main()); never call async SDK functions synchronously.

Production Patterns

• Least privilege by default. Start with an empty allowed_tools list and add capabilities one at a time as the task requires them.
• Cap turns and budget. Use max_turns and log total_cost_usd from every ResultMessage to a metrics store.
• Keep tools pure and fast. Custom tools should be small, deterministic, and quick. Push slow I/O behind a queue rather than blocking the agent loop.
• Log everything with hooks. For cron jobs and background workers, hook-based audit logs are non-negotiable — they are how you debug a run that happened at 3 a.m.
• Pin the model. Set model in ClaudeAgentOptions explicitly so a provider default change never silently alters behavior or cost.

Next Steps

• Compare with the TypeScript version in our Claude Agent SDK for TypeScript tutorial
• Learn to build tools your agents can consume in Build an MCP Server with TypeScript
• Explore multi-agent orchestration patterns for larger workloads
• Wrap this agent in a FastAPI endpoint or a Telegram bot to make it interactive

Conclusion

The Claude Agent SDK for Python turns the agent harness behind Claude Code into a library you control. You have run a one-shot agent with query, held a stateful conversation with ClaudeSDKClient, exposed your own logic through the @tool decorator, and layered in permission callbacks and hooks for safety and observability.

The pattern scales cleanly: describe the goal, whitelist the tools, guard the dangerous ones, and let the SDK run the loop. Start with a read-only agent on a real task in your codebase, then grant capabilities deliberately as you build trust. That is how you ship autonomous agents that are both useful and safe.