How to Use the Claude Code SDK in Python

Post by BlueGrid.io Content Team
Reading time: 14 minutes

A practical, command-first walkthrough of the Claude Code SDK (now the Claude Agent SDK) for Python developers. Every capability below comes with a runnable snippet. Copy, run, adjust.

What you can do with the SDK

  • Send a prompt and get Claude’s answer in your Python process
  • Let Claude read, search, edit files, and run shell commands under your rules
  • Stream output token by token
  • Hold multi-turn conversations and resume old ones
  • Restrict or approve every tool call
  • Add your own Python functions as tools Claude can call
  • Read cost, token usage, and session metadata

1. Prerequisites and install

You need Python 3.10 or newer and Node.js 18 or newer. The Python SDK drives the Claude Code CLI under the hood, so the CLI must be installed too.

Check your versions:

python3 --version
node --version

Install the Claude Code CLI globally:

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

Verify the CLI:

claude --version

Create a project and a virtual environment (an isolated set of Python packages):

mkdir claude-sdk-demo
cd claude-sdk-demo
python3 -m venv .venv
source .venv/bin/activate

Install the SDK and an async runner:

pip install claude-agent-sdk anyio

Verify the import:

python3 -c "import claude_agent_sdk; print(claude_agent_sdk.__version__)"

2. Set your API key

Get a key from https://console.anthropic.com (API Keys, Create Key). Export it so the SDK can find it:

export ANTHROPIC_API_KEY="sk-ant-your-key-here"

To persist it across shells, add that line to ~/.bashrc then reload:

source ~/.bashrc

3. Hello world with query()

query() is the simplest entry point: send one prompt, iterate over the messages Claude sends back. Create hello.py:

import anyio
from claude_agent_sdk import query


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


anyio.run(main)

Run it:

python hello.py

You will see a stream of message objects ending in a ResultMessage. The next step makes that output readable.

4. Read the text out of messages

Claude’s replies arrive as AssistantMessage objects whose .content is a list of typed blocks. Pull out the TextBlock text. Create text.py:

import anyio
from claude_agent_sdk import query, AssistantMessage, TextBlock


async def main():
    async for message in query(prompt="Give me three uses for the color blue."):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)


anyio.run(main)

Run it:

python text.py

Other block types you may encounter: ThinkingBlock (extended reasoning), ToolUseBlock (Claude calling a tool), and ToolResultBlock (the tool’s output).

5. Get the result, cost, tokens, and session id

The final ResultMessage carries metadata. Create result.py:

import anyio
from claude_agent_sdk import query, ResultMessage


async def main():
    async for message in query(prompt="Summarize why unit tests matter in one sentence."):
        if isinstance(message, ResultMessage):
            print("Result text :", message.result)
            print("Session id  :", message.session_id)
            print("Turns       :", message.num_turns)
            print("Duration ms :", message.duration_ms)
            print("Cost USD    :", message.total_cost_usd)
            print("Usage       :", message.usage)


anyio.run(main)

Run it:

python result.py

6. Configure a run with ClaudeAgentOptions

ClaudeAgentOptions is the single dataclass that holds every setting. (In the older claude-code-sdk package this was ClaudeCodeOptions.) Create options.py:

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


async def main():
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-5",
        system_prompt="You are a terse assistant. Answer in under 20 words.",
        max_turns=1,
    )
    async for message in query(prompt="Explain a database index.", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)


anyio.run(main)

Run:

python options.py

To reuse Claude Code’s built-in system prompt instead of writing your own, pass a preset:

options = ClaudeAgentOptions(
    system_prompt={"type": "preset", "preset": "claude_code"},
)

Append to that preset rather than replacing it:

options = ClaudeAgentOptions(
    system_prompt={
        "type": "preset",
        "preset": "claude_code",
        "append": "Always prefer standard library solutions.",
    },
)

7. Point Claude at a working directory

Set cwd so file tools operate in a specific folder, and add_dirs to grant extra readable directories. Create workdir.py:

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


async def main():
    options = ClaudeAgentOptions(
        cwd="/home/ubuntu/example-pipeline",
        add_dirs=["/home/ubuntu/shared-configs"],
        allowed_tools=["Read", "Glob", "Grep"],
    )
    async for message in query(
        prompt="List the files here and tell me what this project does.",
        options=options,
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)


anyio.run(main)

Execute it:

python workdir.py

8. Control which tools Claude may use

Built-in tools include Read, Write, Edit, Bash, Grep, Glob, WebFetch, WebSearch, and TodoWrite. Allow only what you want.

Allowlist (read-only agent):

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Grep", "Glob"],
)

Denylist (everything except dangerous ones):

options = ClaudeAgentOptions(
    disallowed_tools=["Bash", "Write", "Edit"],
)

9. Set a permission mode

permission_mode decides how tool calls are approved when running non-interactively.

  • "default" – standard permission checks
  • "acceptEdits" – auto-accept file edits
  • "plan" – Claude plans but does not execute
  • "bypassPermissions" – run everything without prompts (use with care)
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Edit"],
    permission_mode="acceptEdits",
)

10. Approve or deny each tool call in code

can_use_tool is a callback invoked before every tool runs. Return an allow or deny result. Create permit.py:

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
    PermissionResultAllow,
    PermissionResultDeny,
)

async def can_use_tool(tool_name, input_data, context):
    # Block any Bash command that deletes files.
    if tool_name == "Bash" and "rm " in input_data.get("command", ""):
        return PermissionResultDeny(message="Delete commands are blocked.")
    return PermissionResultAllow()

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Bash", "Read"],
        can_use_tool=can_use_tool,
    )
    async with ClaudeSDKClient(options=options) as client:
        await client.query("Show the current directory, then try to delete a file.")
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

anyio.run(main)

And then run:

python permit.py

The can_use_tool callback needs the streaming client (ClaudeSDKClient), which is why this example does not use query().

11. Guard tools with hooks

Hooks fire on lifecycle events such as PreToolUse and PostToolUse. They can inspect and block actions. Create hooks.py:

import anyio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    HookMatcher,
    AssistantMessage,
    TextBlock,
)

async def block_dangerous(input_data, tool_use_id, context):
    command = input_data.get("tool_input", {}).get("command", "")
    if "rm -rf" in command:
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": "rm -rf is not permitted.",
            }
        }
    return {}

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Bash"],
        hooks={
            "PreToolUse": [HookMatcher(matcher="Bash", hooks=[block_dangerous])],
        },
    )
    async for message in query(prompt="Run: echo hello", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)
python hooks.py

12. Hold a multi-turn conversation

Use ClaudeSDKClient to keep context across several prompts in one session. Create chat.py:

import anyio
from claude_agent_sdk import ClaudeSDKClient, AssistantMessage, TextBlock

async def print_reply(client):
    async for message in client.receive_response():
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

async def main():
    async with ClaudeSDKClient() as client:
        await client.query("My favorite language is Python. Remember that.")
        await print_reply(client)

        await client.query("What did I say my favorite language was?")
        await print_reply(client)

anyio.run(main)
python chat.py

receive_response() yields messages for the current answer up to and including its ResultMessage. Use receive_messages() if you want a continuous stream across turns.

13. Stream output token by token

Set include_partial_messages=True to get incremental text as Claude types. Create stream.py:

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(include_partial_messages=True)
    async for message in query(prompt="Write a four line poem about servers.", options=options):
        if type(message).__name__ == "StreamEvent":
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta["text"], end="", flush=True)
    print()

anyio.run(main)
python stream.py

14. Stream input (feed messages as a generator)

Instead of one string, pass an async generator of message dicts. Useful for piping data or building interactive loops. Create stream_in.py:

import anyio
from claude_agent_sdk import query, AssistantMessage, TextBlock

async def messages():
    yield {"type": "user", "message": {"role": "user", "content": "Count to three."}}
    yield {"type": "user", "message": {"role": "user", "content": "Now count to five."}}

async def main():
    async for message in query(prompt=messages()):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)
python stream_in.py

15. Resume and continue sessions

Save the session_id from a ResultMessage, then reuse it later. Create resume.py:

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage, AssistantMessage, TextBlock

async def first_run():
    session_id = None
    async for message in query(prompt="Pick a random animal and remember it."):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print("First:", block.text)
        if isinstance(message, ResultMessage):
            session_id = message.session_id
    return session_id

async def second_run(session_id):
    options = ClaudeAgentOptions(resume=session_id)
    async for message in query(prompt="Which animal did you pick?", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print("Second:", block.text)

async def main():
    session_id = await first_run()
    if session_id:
        await second_run(session_id)


anyio.run(main)
python resume.py

To resume the most recent session without tracking an id, use continue_conversation=True:

options = ClaudeAgentOptions(continue_conversation=True)

16. Add your own Python tools

Expose Python functions to Claude with the @tool decorator and an in-process MCP server. MCP is the protocol the SDK uses to register tools. Create custom_tool.py:

import anyio
from claude_agent_sdk import (
    query,
    tool,
    create_sdk_mcp_server,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
)

@tool("add", "Add two numbers", {"a": float, "b": float})
async def add(args):
    total = args["a"] + args["b"]
    return {"content": [{"type": "text", "text": f"The sum is {total}"}]}

@tool("uppercase", "Uppercase a string", {"text": str})
async def uppercase(args):
    return {"content": [{"type": "text", "text": args["text"].upper()}]}

async def main():
    server = create_sdk_mcp_server(name="utils", version="1.0.0", tools=[add, uppercase])
    options = ClaudeAgentOptions(
        mcp_servers={"utils": server},
        allowed_tools=["mcp__utils__add", "mcp__utils__uppercase"],
    )
    async for message in query(prompt="Add 21 and 21, then uppercase 'done'.", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)
python custom_tool.py

Tool names follow the pattern mcp__{server_name}__{tool_name}, which is why the allowlist uses mcp__utils__add.

17. Connect an external MCP server

Point the SDK at a separate MCP server process (here, a filesystem server run through npx). Create external_mcp.py:

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

async def main():
    options = ClaudeAgentOptions(
        mcp_servers={
            "filesystem": {
                "type": "stdio",
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", "/home/ubuntu/example-pipeline"],
            }
        },
        allowed_tools=["mcp__filesystem__read_file", "mcp__filesystem__list_directory"],
    )
    async for message in query(prompt="List the files in the pipeline folder.", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)
python external_mcp.py

18. Load project settings and CLAUDE.md

By default the Agent SDK does not read filesystem settings or CLAUDE.md. Opt in with setting_sources. Create settings.py:

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

async def main():
    options = ClaudeAgentOptions(
        cwd="/home/ubuntu/example-pipeline",
        setting_sources=["project"],
        system_prompt={"type": "preset", "preset": "claude_code"},
    )
    async for message in query(prompt="Follow the project conventions and describe them.", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)
python settings.py

Valid sources are "user", "project", and "local".

19. Interrupt a running task

Stop Claude mid-task with the streaming client. Create interrupt.py:

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(allowed_tools=["Bash"], permission_mode="bypassPermissions")
    async with ClaudeSDKClient(options=options) as client:
        await client.query("Count slowly from 1 to 100 using sleep between numbers.")
        await anyio.sleep(3)
        await client.interrupt()
        print("Interrupted the task.")

anyio.run(main)
python interrupt.py

20. Handle errors

Catch the SDK’s typed exceptions. Create errors.py:

import anyio
from claude_agent_sdk import (
    query,
    CLINotFoundError,
    ProcessError,
    CLIJSONDecodeError,
    ClaudeSDKError,
)

async def main():
    try:
        async for message in query(prompt="Hello"):
            print(message)
    except CLINotFoundError:
        print("Install the CLI: npm install -g @anthropic-ai/claude-code")
    except ProcessError as e:
        print(f"Process failed with exit code {e.exit_code}")
    except CLIJSONDecodeError:
        print("Could not parse the CLI response.")
    except ClaudeSDKError as e:
        print(f"General SDK error: {e}")

anyio.run(main)
python errors.py

21. Use Bedrock or Vertex instead of the Anthropic API

The SDK reads provider settings from environment variables. For Amazon Bedrock:

export CLAUDE_CODE_USE_BEDROCK=1
export AWS_REGION="us-east-1"

For Google Vertex AI:

export CLAUDE_CODE_USE_VERTEX=1
export CLOUD_ML_REGION="us-east5"

You can also pass environment variables per run:

options = ClaudeAgentOptions(env={"CLAUDE_CODE_USE_BEDROCK": "1", "AWS_REGION": "us-east-1"})

22. A complete read-only code assistant

This ties the pieces together: a sandboxed assistant that can only read and search a directory, streams its answer, and reports cost. Create assistant.py:

import anyio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
    ResultMessage,
)

async def main():
    options = ClaudeAgentOptions(
        cwd="/home/ubuntu/example-pipeline",
        allowed_tools=["Read", "Grep", "Glob"],
        permission_mode="default",
        system_prompt="You are a code reviewer. Be specific and cite file names.",
        include_partial_messages=True,
        max_turns=8,
    )

    prompt = "Review this project and list the three biggest risks you can find."

    async for message in query(prompt=prompt, options=options):
        if type(message).__name__ == "StreamEvent":
            event = message.event
            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta":
                    print(delta["text"], end="", flush=True)
        elif isinstance(message, ResultMessage):
            print("\n---")
            print(f"Cost USD: {message.total_cost_usd}  Turns: {message.num_turns}")

anyio.run(main)
python assistant.py

Error codes

Status codeError type
400BadRequestError
401AuthenticationError
403PermissionDeniedError
404NotFoundError
409ConflictError
422UnprocessableEntityError
429RateLimitError
>=500InternalServerError
N/AAPIConnectionError

For some additional information, visit the official page: https://platform.claude.com/docs/en/cli-sdks-libraries/sdks/python

Enjoy scripting!

BlueGrid.io Content Team

Three people pose together against a plain white background. The woman on the left is smiling with her hand on her hip, while the two men beside her stand closely, one in a hoodie and the other in a plaid shirt.

BlueGrid.io Content Team

BlueGrid.io Team is an editorial collective of engineers, practitioners, and contributors sharing insights across technology, operations, company culture, and the people behind the systems. Content is created through interviews, hands-on experience, internal collaboration, and editorial review, reflecting both how systems are built and how teams work together in real-world environments.

More from BlueGrid.io Content Team
Share this post

Share this link via

Or copy link