Chapter 09 · Deep Dive: Anthropic SDK

Part of Part III · Deep Dives

The Anthropic SDK is the sharpest tool in the shed. Tool Runner. Managed Agents. Computer Use. This chapter is the manual that should have shipped with it.

Here is the same agent built three ways. First, raw HTTP requests:

import requests
import json

response = requests.post(
    "https://api.anthropic.com/v1/messages",
    headers={
        "x-api-key": os.environ["ANTHROPIC_API_KEY"],
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
    },
    json={
        "model": "claude-sonnet-4-20250514",
        "max_tokens": 1024,
        "system": "You are a helpful assistant.",
        "messages": [{"role": "user", "content": "What is 15% of 87?"}],
    },
)

data = response.json()
if response.status_code != 200:
    raise Exception(f"API error: {data}")
text = data["content"][0]["text"]
print(text)

Twenty lines. Manual header management. Manual JSON parsing. Manual error checking. No streaming. No retries. No type hints. Every feature you add is another layer of string formatting and status-code checking. This is the basement. It works, but you do not want to live here.

Now the same thing with a heavy framework:

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate

llm = ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "What is 15% of 87?"),
])
chain = prompt | llm
result = chain.invoke({})
print(result.content)

Eight lines, but you have imported three modules, instantiated an abstraction you do not control, and composed a chain whose internals you cannot see without reading framework source code. When it works, it is magic. When it breaks, you are debugging someone else's design decisions.

Now the Anthropic SDK:

import anthropic

client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system="You are a helpful assistant.",
    messages=[{"role": "user", "content": "What is 15% of 87?"}],
)
print(response.content[0].text)

Seven lines. No manual headers. No JSON parsing. The response is a typed object with attributes, not a dictionary you index by string. Streaming is a parameter, not a separate code path. Tool use is built in, not bolted on. This is the sweet spot: thin enough to understand, powerful enough to be productive.

This chapter is a complete walkthrough of the Anthropic SDK -- the Messages API, tool use, Tool Runner, Managed Agents, Computer Use, and prompt caching. By the end, you will be able to build Claude-powered agents without a framework getting in your way.

The Anchor

The Anthropic SDK is purpose-built for agents. It is not a general-purpose HTTP client that happens to work with Claude. Every feature in the SDK -- from the Tool Runner loop to the Computer Use action space to the prompt-caching primitives -- is designed for the observe-think-act cycle you learned in Chapter 4.

Three capabilities set it apart from every other LLM SDK:

Tool Runner handles the agent loop for you. You provide tools and a system prompt. It calls Claude, detects tool-use requests, executes your tools, feeds results back, and repeats until Claude produces a final text response. The loop you built by hand in Chapter 4 is now a single method call.

Managed Agents give you hosted, persistent agents. State lives on Anthropic's servers. You send messages to an agent by ID, and it maintains conversation history, tool state, and context across API calls. Your agent survives disconnections, server restarts, and deployment boundaries.

Computer Use lets Claude see screenshots and control a mouse and keyboard. It can navigate websites, fill out forms, and extract information from visual interfaces -- the same way a human would, but programmatically.

This chapter covers all three, with running code. You will build each one, understand when to use it, and know its limitations.

Section 1: SDK Fundamentals

Installation and Authentication

pip install anthropic

That is it. The SDK has no required dependencies beyond httpx and pydantic. No LangChain. No framework. Just the client.

Authentication uses the ANTHROPIC_API_KEY environment variable by default:

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

You can also pass the key explicitly:

client = anthropic.Anthropic(api_key="sk-ant-...")

Or use a third-party provider that serves the Anthropic API (Amazon Bedrock, Google Vertex AI):

# Amazon Bedrock
client = anthropic.AnthropicBedrock(
    aws_access_key="...",
    aws_secret_key="...",
    aws_region="us-east-1",
)

# Google Vertex AI
client = anthropic.AnthropicVertex(
    project_id="my-project",
    region="us-east5",
)

The client is thread-safe. Create one instance and reuse it across your application. Do not create a new client per request -- the underlying HTTP connection pool is expensive to recreate.

The Messages API

The core method is client.messages.create(). Every interaction with Claude flows through it:

import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system="You are a helpful assistant. Answer concisely.",
    messages=[
        {"role": "user", "content": "Explain recursion in one sentence."}
    ],
)

The parameters:

• model: The model ID. Use claude-sonnet-4-20250514 for the best balance of capability and cost. Use claude-opus-4-20250514 for the hardest reasoning tasks. Use claude-haiku-3-5-20241022 for speed and cost when the task is simple.
• max_tokens: The maximum number of tokens Claude can generate in its response. This is a hard cap. Claude will stop mid-sentence if it hits this limit. Set it high enough for your use case -- 4096 is a safe default for most agent tasks.
• system: The system prompt. This is not a message in the conversation. It is a separate parameter that sets Claude's behavior, personality, and constraints. The system prompt is the most powerful lever you have for controlling agent behavior.
• messages: The conversation history. A list of message objects, each with a role ("user" or "assistant") and content. The content can be a string or a list of content blocks (for multimodal input and tool results).

The response is a Message object with typed attributes:

print(response.id)           # "msg_01ABC123..."
print(response.model)        # "claude-sonnet-4-20250514"
print(response.stop_reason)  # "end_turn"
print(response.usage)        # Usage(input_tokens=15, output_tokens=12)

# Content is a list of blocks
for block in response.content:
    print(block.type)  # "text"
    print(block.text)  # "Recursion is a function that calls itself..."

The stop_reason tells you why Claude stopped generating. The values you will encounter:

stop_reason	Meaning
end_turn	Claude finished its response naturally
max_tokens	Claude hit the max_tokens limit mid-response
tool_use	Claude wants to call a tool (Section 2)
stop_sequence	Claude encountered a custom stop sequence you defined

Multi-Turn Conversations

The Messages API is stateless. Claude does not remember previous API calls. You maintain the conversation history yourself by appending each response to the messages list:

messages = []

# Turn 1
messages.append({"role": "user", "content": "What is the capital of France?"})
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=messages,
)
messages.append({"role": "assistant", "content": response.content[0].text})

# Turn 2
messages.append({"role": "user", "content": "What is its population?"})
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=messages,
)

This is the same pattern you used in Chapter 4. The SDK does not hide it from you. You control the message list. You decide what to include and what to drop. This is a feature, not a limitation -- it means you can trim context, inject tool results, and manage memory exactly how you want.

Streaming

Add stream=True to receive tokens as they are generated:

with client.messages.stream(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system="You are a helpful assistant.",
    messages=[{"role": "user", "content": "Write a haiku about Python."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

# After the stream completes, you can access the final message
final_message = stream.get_final_message()
print(f"\n\nTokens used: {final_message.usage.output_tokens}")

The stream.text_stream iterator yields text deltas as they arrive. Use it for real-time display. The stream.get_final_message() method returns the complete Message object after the stream finishes -- including token usage, stop reason, and the full content blocks.

Streaming is not optional for agents. Users will not wait 30 seconds for a complete response. Stream tokens as they arrive so the user can read along and interrupt if the agent goes off track. Every agent you build from here forward should stream.

A Complete Chat Function

Here is a production-quality chat function with error handling, retries, and streaming:

import anthropic
from anthropic import APIError, APITimeoutError, RateLimitError, APIStatusError
import time

def chat(
    system: str,
    prompt: str,
    model: str = "claude-sonnet-4-20250514",
    max_tokens: int = 4096,
    max_retries: int = 3,
) -> str:
    """
    Send a prompt to Claude and return the response text.

    Handles retries for transient errors, rate limits, and timeouts.
    Streams the response to stdout in real time.
    """
    client = anthropic.Anthropic()

    for attempt in range(max_retries):
        try:
            full_text = []
            with client.messages.stream(
                model=model,
                max_tokens=max_tokens,
                system=system,
                messages=[{"role": "user", "content": prompt}],
            ) as stream:
                for text in stream.text_stream:
                    print(text, end="", flush=True)
                    full_text.append(text)

            print()  # newline after stream
            return "".join(full_text)

        except RateLimitError:
            wait = 2 ** attempt
            print(f"\n[Rate limited. Retrying in {wait}s...]")
            time.sleep(wait)

        except APITimeoutError:
            wait = 2 ** attempt
            print(f"\n[Timeout. Retrying in {wait}s...]")
            time.sleep(wait)

        except APIStatusError as e:
            if e.status_code >= 500:
                wait = 2 ** attempt
                print(f"\n[Server error {e.status_code}. Retrying in {wait}s...]")
                time.sleep(wait)
            else:
                raise  # 4xx errors are not retryable

        except APIError as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt
            print(f"\n[API error: {e}. Retrying in {wait}s...]")
            time.sleep(wait)

    raise RuntimeError(f"Failed after {max_retries} attempts")

This function handles the four failure modes you will encounter in production:

• Rate limits (429): Exponential backoff. Claude's rate limits vary by tier. Free tier is 5 RPM. Build tier is 50 RPM. Scale tier is 200+ RPM.
• Timeouts: Network issues, server load. Retry with backoff.
• Server errors (5xx): Transient infrastructure issues. Retry.
• Client errors (4xx): Your fault. Bad API key, invalid parameters, account issues. Do not retry -- fix the problem.

Client Configuration

The Anthropic client constructor accepts configuration that applies to every request:

client = anthropic.Anthropic(
    api_key="sk-ant-...",
    max_retries=2,                # SDK-level retry (default: 2)
    timeout=60.0,                 # Request timeout in seconds (default: 600)
    default_headers={"X-Custom": "value"},
)

The SDK-level max_retries handles connection errors and 429/5xx responses automatically. You do not need to wrap every call in a retry loop -- the SDK does it for you. The timeout parameter is per-request, not cumulative. A 60-second timeout means each individual API call can take up to 60 seconds.

For agent workloads, increase the timeout. Tool execution can take seconds. Claude's reasoning can take seconds. A 60-second timeout is reasonable for simple chat. For agents with multiple tool calls, set it to 120 or 180 seconds.

Section 2: Tool Use with the Anthropic SDK

Tools are how your agent acts on the world. The Anthropic SDK's tool-use implementation is the cleanest in the industry. No separate function-calling API. No special message types. Tools are defined alongside messages in the same create() call, and Claude signals tool use through the same response object.

Defining Tools

A tool is a JSON Schema object with a name, description, and input schema:

tools = [
    {
        "name": "get_weather",
        "description": "Get the current weather for a city. Returns temperature, conditions, and humidity.",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "The city name, e.g. 'San Francisco, CA'"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit. Defaults to fahrenheit."
                },
            },
            "required": ["city"],
        },
    },
    {
        "name": "calculate",
        "description": "Evaluate a mathematical expression. Supports +, -, *, /, **, sqrt, sin, cos, log.",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "The mathematical expression to evaluate, e.g. 'sqrt(144) + 5 * 3'"
                },
            },
            "required": ["expression"],
        },
    },
]

Three rules for tool definitions:

The description is the most important field. Claude uses it to decide whether to call the tool. Write it for a colleague who has never seen your codebase. Include what the tool does, what it returns, and when to use it versus other tools.

The input schema is the contract. Claude will generate arguments that conform to this schema. If you mark a field as required, Claude will always include it. If you provide an enum, Claude will only use those values. The schema is enforced server-side -- Claude cannot produce arguments that violate it.

Names must be unique and descriptive. get_weather is good. tool_1 is not. Claude uses the name to understand the tool's purpose. A good name reduces the chance Claude calls the wrong tool.

How Claude Signals a Tool Call

When Claude decides to use a tool, the response changes:

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system="You are a helpful assistant with access to tools.",
    messages=[{"role": "user", "content": "What is the weather in Tokyo?"}],
    tools=tools,
)

print(response.stop_reason)  # "tool_use"

for block in response.content:
    if block.type == "tool_use":
        print(f"Tool: {block.name}")
        print(f"ID: {block.id}")
        print(f"Input: {block.input}")
        # Tool: get_weather
        # ID: toolu_01ABC123...
        # Input: {"city": "Tokyo, Japan"}

The stop_reason is "tool_use" instead of "end_turn". The content blocks include tool_use blocks alongside optional text blocks. Claude can explain what it is doing in text before issuing the tool call -- this is useful for debugging and user transparency.

Executing Tools and Returning Results

You execute the tool and return the result as a tool_result content block:

def execute_tool(name: str, input: dict) -> str:
    """Execute a tool and return the result as a string."""
    if name == "get_weather":
        city = input["city"]
        unit = input.get("unit", "fahrenheit")
        # In production, call a real weather API
        return f"Weather in {city}: 72°{unit[0].upper()}, partly cloudy, 65% humidity"

    elif name == "calculate":
        import math
        expression = input["expression"]
        # WARNING: eval() is dangerous in production. Use a sandboxed evaluator.
        allowed_names = {"sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
                         "log": math.log, "pi": math.pi, "e": math.e}
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return str(result)

    else:
        return f"Error: Unknown tool '{name}'"

Then append the result to the conversation:

# After receiving a tool_use response from Claude
for block in response.content:
    if block.type == "tool_use":
        result = execute_tool(block.name, block.input)

        # Append the assistant's tool-use request to the conversation
        messages.append({
            "role": "assistant",
            "content": [{"type": "tool_use", "id": block.id,
                         "name": block.name, "input": block.input}]
        })

        # Append the tool result as a user message
        messages.append({
            "role": "user",
            "content": [{"type": "tool_result", "tool_use_id": block.id,
                         "content": result}]
        })

The tool_result block must reference the tool_use_id from the corresponding tool_use block. This is how Claude knows which tool call produced which result. If you have multiple tool calls in a single response, each result must match its call by ID.

The Complete Tool-Use Loop

Here is the full agent loop with tool use:

import anthropic

client = anthropic.Anthropic()

def run_agent_with_tools(system_prompt: str, task: str, tools: list, max_turns: int = 10):
    """Run an agent loop with tool use."""
    messages = [{"role": "user", "content": task}]

    for turn in range(max_turns):
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system=system_prompt,
            messages=messages,
            tools=tools,
        )

        # If Claude is done, return the text
        if response.stop_reason == "end_turn":
            text_blocks = [b.text for b in response.content if b.type == "text"]
            return "\n".join(text_blocks)

        # If Claude wants to use tools, execute them
        if response.stop_reason == "tool_use":
            # Build the assistant message with all content blocks
            assistant_content = []
            tool_results = []

            for block in response.content:
                if block.type == "text":
                    assistant_content.append({"type": "text", "text": block.text})
                elif block.type == "tool_use":
                    assistant_content.append({
                        "type": "tool_use",
                        "id": block.id,
                        "name": block.name,
                        "input": block.input,
                    })
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })

            messages.append({"role": "assistant", "content": assistant_content})
            messages.append({"role": "user", "content": tool_results})
            continue

        # If Claude hit max_tokens, warn and continue
        if response.stop_reason == "max_tokens":
            messages.append({"role": "assistant",
                             "content": [{"type": "text", "text": response.content[0].text}]})
            messages.append({"role": "user",
                             "content": "You were cut off. Continue from where you stopped."})
            continue

    return "Agent did not finish within turn limit."

This loop handles all three stop reasons: end_turn (Claude is done), tool_use (Claude wants to call tools), and max_tokens (Claude was cut off mid-response). The max_tokens case is important -- if you set max_tokens too low, Claude might stop mid-tool-call or mid-sentence. The loop detects this and prompts Claude to continue.

Parallel Tool Calls

Claude can request multiple tools in a single response. When it does, the response contains multiple tool_use blocks:

# User: "Compare the weather in Tokyo, London, and New York"
# Claude's response might contain three tool_use blocks for get_weather

for block in response.content:
    if block.type == "tool_use":
        print(f"Calling {block.name} with {block.input}")
        # Execute all three in parallel

The loop above already handles this -- it iterates over all tool_use blocks and collects all results into a single user message. Claude receives all results at once and synthesizes them.

Execute independent tool calls in parallel. If Claude requests weather for three cities, do not call them sequentially. Use asyncio.gather() or a thread pool. The user is waiting. Every second you save is a second the user does not spend staring at a spinner.

Tool Choice Control

The tool_choice parameter controls whether and how Claude uses tools:

# Auto: Claude decides whether to use tools (default)
response = client.messages.create(..., tools=tools, tool_choice={"type": "auto"})

# Any: Claude must use at least one tool
response = client.messages.create(..., tools=tools, tool_choice={"type": "any"})

# Tool: Claude must use a specific tool
response = client.messages.create(
    ..., tools=tools,
    tool_choice={"type": "tool", "name": "get_weather"}
)

# None: Claude cannot use tools (useful for disabling tools mid-conversation)
response = client.messages.create(..., tools=tools, tool_choice={"type": "none"})

Use "any" when you know the task requires a tool call and you want to prevent Claude from attempting to answer from its training data. Use "tool" with a specific name when you are building a pipeline where each step has a predetermined tool. Use "none" when you want a text-only response even though tools are available -- for example, when the user asks a clarification question.

Section 3: Tool Runner -- The Agent Loop, Handled for You

The tool-use loop in Section 2 is 40 lines. It works. But it is boilerplate. Every agent you build will contain some variation of that loop. The Anthropic SDK provides a built-in version: Tool Runner.

What Tool Runner Is

Tool Runner is client.beta.messages.tool_runner. It handles the observe-think-act loop automatically. You provide a system prompt, tools, and tool implementations. It calls Claude, detects tool-use requests, executes your tools, feeds results back, and repeats until Claude produces a final text response.

Here is the same agent from Section 2, now with Tool Runner:

import anthropic

client = anthropic.Anthropic()

def get_weather(city: str, unit: str = "fahrenheit") -> str:
    return f"Weather in {city}: 72°{unit[0].upper()}, partly cloudy, 65% humidity"

def calculate(expression: str) -> str:
    import math
    allowed = {"sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
               "log": math.log, "pi": math.pi, "e": math.e}
    return str(eval(expression, {"__builtins__": {}}, allowed))

tools = [
    {
        "name": "get_weather",
        "description": "Get current weather for a city.",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "City name"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
            },
            "required": ["city"],
        },
    },
    {
        "name": "calculate",
        "description": "Evaluate a mathematical expression.",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "Math expression"},
            },
            "required": ["expression"],
        },
    },
]

tool_map = {"get_weather": get_weather, "calculate": calculate}

# The agent, in 10 lines
runner = client.beta.messages.tool_runner(
    model="claude-sonnet-4-20250514",
    system="You are a helpful assistant. Use tools when needed.",
    tools=tools,
    tool_map=tool_map,
    max_tokens=4096,
)

result = runner.run("What is the weather in Tokyo? Also compute sqrt(144) + 5 * 3.")
print(result)

Ten lines. The tool_map dictionary maps tool names to Python callables. Tool Runner calls them automatically when Claude requests a tool. The run() method returns the final text response after all tool calls are resolved.

How Tool Runner Works Internally

Tool Runner is not magic. It is the same loop you wrote in Section 2, packaged by Anthropic. Here is what happens when you call runner.run():

1. Send the user message to Claude with the system prompt and tool definitions.
2. If Claude returns stop_reason: "end_turn", return the text.
3. If Claude returns stop_reason: "tool_use", look up each requested tool in tool_map, call it with the arguments Claude provided, and feed the results back as tool_result blocks.
4. Go to step 1.

The loop continues until Claude produces a text response or hits a configurable max_turns limit (default: 10). Tool Runner also handles parallel tool calls -- if Claude requests three tools, it executes all three before sending results back.

Streaming with Tool Runner

Tool Runner supports streaming. Use run_streamed() instead of run():

with runner.run_streamed("What is the weather in Tokyo?") as stream:
    for event in stream:
        if event.type == "text":
            print(event.text, end="", flush=True)
        elif event.type == "tool_use":
            print(f"\n[Calling {event.tool_name}...]")
        elif event.type == "tool_result":
            print(f"[{event.tool_name} returned: {event.content[:80]}...]")

The stream yields events of three types: text (Claude is generating text), tool_use (Claude requested a tool), and tool_result (a tool returned a result). This gives you visibility into the agent's actions without building the loop yourself.

When to Use Tool Runner vs Building Your Own Loop

Use Tool Runner when:

• Your agent follows a standard pattern: user task, tool calls, final answer.
• You want to move fast and do not need custom logic between turns.
• Your tools are simple functions with no side effects that need tracking.
• You are building a prototype or an internal tool.

Build your own loop when:

• You need custom state management between turns (updating a UI, logging to a database, modifying the system prompt mid-conversation).
• You want custom reasoning strategies -- reflection steps, self-critique, plan revision.
• You are orchestrating multiple agents and need to intercept tool calls for routing.
• You need fine-grained control over which messages are included in the context window (trimming, summarization, context compression).
• You are building a production system where every API call needs custom observability, cost tracking, or approval gates.

Tool Runner is a convenience, not a constraint. If you find yourself fighting it, drop down to the raw loop. The SDK makes both paths equally clean. You are not locked into one or the other.

Tool Runner Limitations

Tool Runner is beta software. Its API may change. It does not support all edge cases:

• No mid-loop intervention. You cannot inspect Claude's reasoning between tool calls and decide to redirect. The loop runs until completion or max turns.
• No tool call approval. In production, you often want a human to approve sensitive tool calls (sending email, making purchases, deleting data). Tool Runner executes everything automatically.
• No custom stop conditions. Tool Runner stops when Claude produces text or hits max turns. You cannot add custom termination logic like "stop if the user's question has been answered with confidence > 0.9."
• Limited error recovery. If a tool raises an exception, Tool Runner feeds the error message back to Claude. It does not retry with different arguments or fall back to an alternative tool.

For production agents, start with Tool Runner. When you hit a limitation, extract the loop into your own code. The transition is straightforward because the underlying API is the same client.messages.create() call.

Section 4: Managed Agents (Beta)

Tool Runner runs in your process. When your process dies, the agent dies. Managed Agents solve this by running on Anthropic's infrastructure.

What Managed Agents Are

A Managed Agent is a persistent agent hosted by Anthropic. You create it once with a configuration (model, system prompt, tools). Then you send messages to it by agent ID. Anthropic maintains the conversation state, tool execution context, and message history server-side.

import anthropic

client = anthropic.Anthropic()

# Create a managed agent
agent = client.beta.agents.create(
    name="research-assistant",
    model="claude-sonnet-4-20250514",
    system="You are a research assistant. You have access to web search and a calculator. "
           "Be thorough. Cite your sources.",
    tools=[
        {
            "name": "web_search",
            "description": "Search the web for information.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                },
                "required": ["query"],
            },
        },
        {
            "name": "calculate",
            "description": "Evaluate a math expression.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "Math expression"},
                },
                "required": ["expression"],
            },
        },
    ],
    tool_map={
        "web_search": lambda query: f"Search results for '{query}': ...",
        "calculate": lambda expression: str(eval(expression)),
    },
)

print(f"Agent ID: {agent.id}")  # "agent_01ABC123..."

The agent now exists on Anthropic's servers. You can send messages to it from any process, any server, any deployment:

# Send a message to the agent
response = client.beta.agents.messages.create(
    agent_id=agent.id,
    messages=[{"role": "user", "content": "Research the latest developments in fusion energy."}],
)

# The response includes the full conversation, including any tool calls
for block in response.content:
    if block.type == "text":
        print(block.text)

The agent persists between calls. Send a follow-up message and it remembers the previous conversation:

# The agent remembers the fusion energy research from the previous call
response = client.beta.agents.messages.create(
    agent_id=agent.id,
    messages=[{"role": "user", "content": "Now summarize that in three bullet points."}],
)

How Managed Agents Differ from Tool Runner

Dimension	Tool Runner	Managed Agents
State location	Your process memory	Anthropic's servers
Persistence	Lost on process exit	Survives across API calls
Tool execution	Your code, your process	Your code, but called by Anthropic's infrastructure
Conversation history	You manage it	Anthropic manages it
Concurrency	Single user per runner	Multiple users can share an agent
API surface	tool_runner.run()	agents.create() + agents.messages.create()

Use Cases

Customer support agents. Create one agent per support tier. Users send messages. The agent maintains context across days. If a user returns a week later, the agent remembers the previous conversation.

Long-running research agents. Start a research task. The agent works for hours, making tool calls, reading results, synthesizing findings. You check in periodically to see progress. The agent survives your laptop going to sleep.

Shared team agents. Create an agent for your engineering team. Everyone sends it questions about the codebase. The agent builds up context about your team's preferences, common issues, and internal knowledge.

Limitations

Managed Agents are beta. The API may change. The tool execution model -- where your tools run on your infrastructure but are called by Anthropic's servers -- requires careful network configuration. Your tool endpoints must be reachable from Anthropic's infrastructure.

There is also vendor lock-in. A Managed Agent lives on Anthropic's platform. You cannot export it, migrate it to another provider, or run it locally. If you need provider independence, build your own agent loop with the Messages API.

Managed Agents are the right abstraction for the right problem. If you need persistent, always-on agents that survive process boundaries, they are the best option available today. If you need full control over state, execution, and deployment, build your own.

Section 5: Computer Use

Computer Use is the most ambitious feature in the Anthropic SDK. Claude can see screenshots and control a mouse and keyboard. It can navigate websites, fill out forms, click buttons, and extract information from visual interfaces -- the same way a human would.

What Computer Use Is

The computer use tool is a special tool type: "computer_20241022". When you include it in your tool definitions, Claude can request actions like moving the mouse, clicking, typing, and taking screenshots. You execute these actions on a real or virtual display and feed the results back.

The available actions:

Action	Description
key	Press a key or key combination (e.g., "Enter", "Ctrl+C")
type	Type a string of text
mouse_move	Move the mouse to (x, y) coordinates
left_click	Click at the current mouse position
left_click_drag	Click and drag from one position to another
right_click	Right-click at the current mouse position
middle_click	Middle-click at the current mouse position
double_click	Double-click at the current mouse position
screenshot	Take a screenshot and return it
cursor_position	Get the current cursor (x, y) coordinates

The Computer Use Loop

The loop is: screenshot, Claude decides an action, you execute it, repeat:

import anthropic
import base64
import pyautogui  # for controlling mouse/keyboard
from PIL import Image
import io

client = anthropic.Anthropic()

def take_screenshot() -> str:
    """Take a screenshot and return as base64-encoded PNG."""
    screenshot = pyautogui.screenshot()
    buffer = io.BytesIO()
    screenshot.save(buffer, format="PNG")
    return base64.b64encode(buffer.getvalue()).decode()

def execute_action(action: dict):
    """Execute a computer use action."""
    action_type = action["type"]
    if action_type == "left_click":
        pyautogui.click()
    elif action_type == "right_click":
        pyautogui.rightClick()
    elif action_type == "double_click":
        pyautogui.doubleClick()
    elif action_type == "mouse_move":
        x, y = action["coordinate"]
        pyautogui.moveTo(x, y)
    elif action_type == "type":
        pyautogui.write(action["text"])
    elif action_type == "key":
        pyautogui.hotkey(action["text"])
    elif action_type == "screenshot":
        pass  # Handled in the loop
    else:
        print(f"Unknown action: {action_type}")

def run_computer_use(task: str, max_steps: int = 20):
    """Run a computer-use agent to accomplish a task."""
    messages = [{"role": "user", "content": task}]

    computer_tool = {
        "name": "computer",
        "type": "computer_20241022",
        "display_width_px": 1920,
        "display_height_px": 1080,
        "display_number": 0,
    }

    for step in range(max_steps):
        # Take a screenshot and add it to the conversation
        screenshot_b64 = take_screenshot()
        if step == 0:
            # First turn: include the screenshot with the task
            messages[0]["content"] = [
                {"type": "text", "text": task},
                {"type": "image", "source": {
                    "type": "base64",
                    "media_type": "image/png",
                    "data": screenshot_b64,
                }},
            ]
        else:
            # Subsequent turns: screenshot is a tool result
            pass  # Handled below

        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system="You are controlling a computer. Use the computer tool to accomplish the task. "
                   "Think step by step. After each action, a new screenshot will be provided.",
            messages=messages,
            tools=[computer_tool],
        )

        if response.stop_reason == "end_turn":
            text = "".join(b.text for b in response.content if b.type == "text")
            return text

        if response.stop_reason == "tool_use":
            assistant_content = []
            tool_results = []

            for block in response.content:
                if block.type == "text":
                    assistant_content.append({"type": "text", "text": block.text})
                    print(block.text)
                elif block.type == "tool_use":
                    print(f"\n[Action: {block.input}]")
                    assistant_content.append({
                        "type": "tool_use",
                        "id": block.id,
                        "name": block.name,
                        "input": block.input,
                    })

                    # Execute the action
                    execute_action(block.input)

                    # Take a new screenshot after the action
                    import time
                    time.sleep(0.5)  # Wait for UI to update
                    new_screenshot = take_screenshot()

                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": [
                            {"type": "image", "source": {
                                "type": "base64",
                                "media_type": "image/png",
                                "data": new_screenshot,
                            }},
                        ],
                    })

            messages.append({"role": "assistant", "content": assistant_content})
            messages.append({"role": "user", "content": tool_results})

    return "Agent did not finish within step limit."

This is a simplified version. A production computer-use agent needs coordinate scaling (Claude works in a normalized coordinate space), error handling for failed actions, and safety checks.

A Real Example: Web Scraping with Computer Use

Here is a practical example. You need to extract the top story headlines from a news website that uses heavy JavaScript rendering -- traditional HTTP scraping will not work. Computer Use can:

task = """
Go to news.ycombinator.com and extract the titles of the top 5 stories on the front page.
Return them as a numbered list.
"""

result = run_computer_use(task, max_steps=15)
print(result)

Claude will: open the browser, navigate to the URL, wait for the page to load, take a screenshot, read the headlines from the screenshot, and return them as text. It does not need a CSS selector. It does not need to parse HTML. It sees what a human sees and acts accordingly.

Safety Considerations

Computer Use is powerful and dangerous. Three rules:

1. Never give Claude access to your real desktop. Run it in a virtual machine, a Docker container, or a dedicated machine with no sensitive data. One wrong click can delete files, send emails, or post to social media.

2. Validate actions before executing. In production, add an approval step for destructive actions:

DESTRUCTIVE_ACTIONS = {"key": ["Delete", "Backspace"], "left_click": True}

def is_safe(action: dict) -> bool:
    if action["type"] in DESTRUCTIVE_ACTIONS:
        print(f"WARNING: Destructive action requested: {action}")
        return input("Approve? (y/n): ").lower() == "y"
    return True

3. Set a step limit. Computer Use agents can loop indefinitely, clicking around without making progress. Always set max_steps and a timeout.

Use Cases

Web scraping resistant to anti-bot measures. Traditional scrapers fail on JavaScript-heavy sites, CAPTCHAs, and bot detection. Computer Use renders the page in a real browser and reads it visually. It is slower and more expensive than HTTP scraping, but it works on sites that are otherwise impenetrable.

Automating legacy UIs. That internal tool from 2008 with no API? Computer Use can fill out its forms, click its buttons, and extract its data. It is a bridge between old software and new automation.

Testing web applications. Instead of writing Selenium scripts, describe the test in natural language: "Go to the checkout page, add an item to the cart, verify the total includes tax." Claude navigates the UI and reports what it sees.

Section 6: Prompt Caching

Agents make many API calls. Each call sends the system prompt and tool definitions. For a complex agent, that can be thousands of tokens per call, repeated dozens of times per task. Prompt caching eliminates this redundancy.

What Prompt Caching Is

Prompt caching stores frequently-used prompt content on Anthropic's servers. Mark portions of your prompt as cacheable. Claude caches them and reuses the cached computation on subsequent calls. You pay less for cached tokens and get lower latency.

How It Works

Add cache_control: {"type": "ephemeral"} to any content block you want to cache:

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "You are a helpful assistant with access to tools...",
            "cache_control": {"type": "ephemeral"},
        }
    ],
    messages=[{"role": "user", "content": "What is the weather in Tokyo?"}],
    tools=[
        {
            "name": "get_weather",
            "cache_control": {"type": "ephemeral"},
            # ... rest of tool definition
        }
    ],
)

The cache_control marker tells Claude: "Cache everything up to this point." The cache is read sequentially from the beginning of the prompt. The first cache_control marker caches everything before it. Subsequent markers create additional cache breakpoints.

Cache Breakpoints

Caching is sequential. Put static content first, dynamic content last:

system = [
    # Static: cache this
    {"type": "text", "text": "You are an agent. You have these tools...",
     "cache_control": {"type": "ephemeral"}},
    # Dynamic: do not cache this (changes per task)
    {"type": "text", "text": f"Current date: {datetime.now().isoformat()}"},
]

The cache is read from the start. If you put dynamic content before a cache_control marker, the cache invalidates every time the dynamic content changes. Structure your prompts so that:

1. System prompt (static) comes first, with a cache_control marker at the end.
2. Tool definitions (static) come next, with a cache_control marker at the end.
3. Conversation history (dynamic) comes last, with no cache markers.

Implementing Prompt Caching in the Agent Loop

Here is the tool-use loop from Section 2, updated with prompt caching:

import anthropic
from datetime import datetime

client = anthropic.Anthropic()

def run_agent_with_caching(system_prompt: str, task: str, tools: list, max_turns: int = 10):
    """Run an agent loop with prompt caching enabled."""
    messages = [{"role": "user", "content": task}]

    # Build the system prompt with cache markers
    system_content = [
        {
            "type": "text",
            "text": system_prompt,
            "cache_control": {"type": "ephemeral"},
        },
        {
            "type": "text",
            "text": f"Current date: {datetime.now().isoformat()}",
        },
    ]

    # Add cache markers to tool definitions
    cached_tools = []
    for tool in tools:
        cached_tool = tool.copy()
        cached_tool["cache_control"] = {"type": "ephemeral"}
        cached_tools.append(cached_tool)

    for turn in range(max_turns):
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system=system_content,
            messages=messages,
            tools=cached_tools,
        )

        # Check cache usage
        if hasattr(response, 'usage') and hasattr(response.usage, 'cache_read_input_tokens'):
            cache_hit = response.usage.cache_read_input_tokens or 0
            total_input = response.usage.input_tokens
            if cache_hit > 0:
                print(f"[Cache hit: {cache_hit}/{total_input} input tokens from cache]")

        if response.stop_reason == "end_turn":
            text_blocks = [b.text for b in response.content if b.type == "text"]
            return "\n".join(text_blocks)

        if response.stop_reason == "tool_use":
            assistant_content = []
            tool_results = []

            for block in response.content:
                if block.type == "text":
                    assistant_content.append({"type": "text", "text": block.text})
                elif block.type == "tool_use":
                    assistant_content.append({
                        "type": "tool_use", "id": block.id,
                        "name": block.name, "input": block.input,
                    })
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })

            messages.append({"role": "assistant", "content": assistant_content})
            messages.append({"role": "user", "content": tool_results})
            continue

    return "Agent did not finish within turn limit."

On the first call, Claude processes the full system prompt and tool definitions. On subsequent calls, it reads them from cache. The cache_read_input_tokens field in the usage object tells you how many tokens were served from cache.

Cost Savings

Cached tokens cost 90% less than uncached tokens. For an agent with a 2,000-token system prompt and 1,000 tokens of tool definitions making 20 API calls per task:

• Without caching: 20 calls x 3,000 input tokens = 60,000 tokens at full price.
• With caching: 1 call x 3,000 tokens at full price + 19 calls x 3,000 tokens at 10% price = 3,000 + 5,700 = 8,700 effective tokens.

That is an 85% reduction in input token cost. The savings compound with more calls and larger system prompts.

When Caching Helps

• Long system prompts. Agents with detailed instructions, few-shot examples, and role definitions benefit the most.
• Many tool definitions. Each tool definition is tens or hundreds of tokens. With 10+ tools, caching is significant.
• Repeated context. If your agent makes many calls with the same prefix (system prompt + tools), every call after the first benefits.

When Caching Does Not Help

• Short prompts. If your system prompt is 50 tokens, the overhead of managing cache breakpoints exceeds the savings.
• Highly dynamic content. If your system prompt changes on every call (e.g., it includes the current conversation summary), the cache never hits.
• Single-call workloads. Caching only helps on the second call and beyond. If you make one call per task, there is nothing to cache.

Cache the system prompt and tool definitions. Always. There is no downside. If the cache hits, you save money and latency. If it does not, you lose nothing. It is the closest thing to a free lunch in the Anthropic API.

Section 7: Putting It All Together

You now know every major feature of the Anthropic SDK. Let us assemble them into a single agent that uses Tool Runner, prompt caching, streaming, and proper error handling.

The Complete Agent

import anthropic
from anthropic import APIError, RateLimitError, APITimeoutError
import time
import math
from datetime import datetime

# ---------------------------------------------------------------------------
# Tool implementations
# ---------------------------------------------------------------------------

def web_search(query: str) -> str:
    """Search the web. In production, call a real search API."""
    # Simulated search results
    results = {
        "tokyo weather": "Tokyo: 28C, partly cloudy, humidity 70%",
        "python 3.13 release": "Python 3.13 released October 2024 with new JIT compiler",
        "anthropic sdk": "Anthropic SDK v0.39.0, supports Tool Runner, Managed Agents, Computer Use",
    }
    for key, value in results.items():
        if key in query.lower():
            return value
    return f"No results found for '{query}'. Try different search terms."

def calculate(expression: str) -> str:
    """Safely evaluate a mathematical expression."""
    allowed = {
        "sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
        "log": math.log, "pi": math.pi, "e": math.e,
        "abs": abs, "round": round, "pow": pow,
    }
    try:
        result = eval(expression, {"__builtins__": {}}, allowed)
        return str(result)
    except Exception as e:
        return f"Error evaluating '{expression}': {e}"

def get_current_time() -> str:
    """Return the current date and time in ISO format."""
    return datetime.now().isoformat()

# ---------------------------------------------------------------------------
# Tool definitions with cache markers
# ---------------------------------------------------------------------------

TOOLS = [
    {
        "name": "web_search",
        "description": "Search the web for current information. Use this for facts, news, "
                       "and anything that might have changed since your training cutoff.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "The search query"},
            },
            "required": ["query"],
        },
        "cache_control": {"type": "ephemeral"},
    },
    {
        "name": "calculate",
        "description": "Evaluate a mathematical expression. Supports sqrt, sin, cos, log, "
                       "abs, round, pow, pi, e, and standard arithmetic operators.",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "Math expression to evaluate"},
            },
            "required": ["expression"],
        },
        "cache_control": {"type": "ephemeral"},
    },
    {
        "name": "get_current_time",
        "description": "Get the current date and time.",
        "input_schema": {
            "type": "object",
            "properties": {},
            "required": [],
        },
        "cache_control": {"type": "ephemeral"},
    },
]

TOOL_MAP = {
    "web_search": web_search,
    "calculate": calculate,
    "get_current_time": get_current_time,
}

# ---------------------------------------------------------------------------
# System prompt with cache marker
# ---------------------------------------------------------------------------

SYSTEM_PROMPT = [
    {
        "type": "text",
        "text": (
            "You are a capable research assistant. You have access to web search, "
            "a calculator, and the current time. Follow these rules:\n\n"
            "1. For factual questions, search the web before answering.\n"
            "2. For calculations, use the calculator tool. Do not do math in your head.\n"
            "3. Cite your sources when using web search results.\n"
            "4. If you are unsure, say so. Do not fabricate information.\n"
            "5. Be concise. The user wants answers, not essays."
        ),
        "cache_control": {"type": "ephemeral"},
    },
]

# ---------------------------------------------------------------------------
# The agent
# ---------------------------------------------------------------------------

def run_agent(task: str, max_retries: int = 3) -> str:
    """Run the complete agent with Tool Runner, caching, and streaming."""
    client = anthropic.Anthropic(max_retries=2, timeout=120.0)

    for attempt in range(max_retries):
        try:
            runner = client.beta.messages.tool_runner(
                model="claude-sonnet-4-20250514",
                system=SYSTEM_PROMPT,
                tools=TOOLS,
                tool_map=TOOL_MAP,
                max_tokens=4096,
            )

            print(f"\n{'='*60}")
            print(f"TASK: {task}")
            print(f"{'='*60}\n")

            with runner.run_streamed(task) as stream:
                full_text = []
                for event in stream:
                    if event.type == "text":
                        print(event.text, end="", flush=True)
                        full_text.append(event.text)
                    elif event.type == "tool_use":
                        print(f"\n  [{event.tool_name}({event.tool_input})]")
                    elif event.type == "tool_result":
                        preview = event.content[:100].replace("\n", " ")
                        print(f"  -> {preview}...")

                print("\n")
                return "".join(full_text)

        except RateLimitError:
            wait = 2 ** attempt
            print(f"\n[Rate limited. Waiting {wait}s...]")
            time.sleep(wait)

        except APITimeoutError:
            wait = 2 ** attempt
            print(f"\n[Timeout. Waiting {wait}s...]")
            time.sleep(wait)

        except APIError as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt
            print(f"\n[API error: {e}. Waiting {wait}s...]")
            time.sleep(wait)

    raise RuntimeError(f"Agent failed after {max_retries} attempts")

# ---------------------------------------------------------------------------
# Run it
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    result = run_agent(
        "What is the square root of the number of days between January 1, 2024 "
        "and today? Also, what is the weather in Tokyo right now?"
    )
    print(f"\nFINAL RESULT:\n{result}")

One hundred lines. Tool Runner for the agent loop. Prompt caching for the system prompt and tool definitions. Streaming for real-time output. Error handling with retries. This is a production-quality agent skeleton.

The Trace

Here is what happens when you run it:

============================================================
TASK: What is the square root of the number of days between
January 1, 2024 and today? Also, what is the weather in Tokyo
right now?
============================================================

I'll break this into two parts. First, let me get the current
date and the Tokyo weather.
  [get_current_time()]
  -> 2026-07-27T14:32:15.123456...
  [web_search(query='Tokyo weather')]
  -> Tokyo: 28C, partly cloudy, humidity 70%...

Now I know today is July 27, 2026. Let me calculate the number
of days from January 1, 2024 to today, then take the square root.

From Jan 1, 2024 to Jan 1, 2025: 365 days (2024 is a leap year).
From Jan 1, 2025 to Jan 1, 2026: 365 days.
From Jan 1, 2026 to July 27, 2026: 208 days.
Total: 365 + 365 + 208 = 938 days.

  [calculate(expression='sqrt(938)')]
  -> 30.62678566222711...

Here are the results:

1. **Square root of days since Jan 1, 2024:** sqrt(938) = **30.63**
   (938 days from January 1, 2024 to July 27, 2026)

2. **Tokyo weather:** 28C, partly cloudy, humidity 70%

FINAL RESULT:
[the text above]

The agent decomposed the task, called three tools in parallel where possible, did the intermediate reasoning, and synthesized a clear answer. Tool Runner handled the loop. Caching saved tokens on the system prompt and tool definitions. Streaming showed the user what was happening in real time.

The Turn

You now have deep knowledge of the Anthropic SDK. You can build Claude-powered agents with tool use, computer use, and managed agents. You understand when to use Tool Runner and when to build your own loop. You know how to cache prompts, handle errors, and stream responses.

The SDK is a tool, not a crutch. It gives you clean primitives -- the Messages API, Tool Runner, Managed Agents, Computer Use -- and gets out of your way. You are never locked into an abstraction you do not understand. When Tool Runner is the right choice, you use it. When you need more control, you drop down to the raw loop. The transition is a few lines of code, not a rewrite.

This is the difference between a framework and an SDK. A framework owns the control flow and calls your code. An SDK provides building blocks and lets you decide how to assemble them. The Anthropic SDK is the latter. You are the architect. The SDK is your material.

Close

You have mastered the Anthropic SDK. You can build agents that search the web, do math, control computers, and maintain state across sessions -- all with clean, framework-free Python.

But the agent ecosystem is bigger than one provider. What about LangChain -- the framework that powers thousands of production agents? What about LangGraph -- the state machine engine for complex agent workflows? What about the patterns that let you swap Claude for GPT-4 without rewriting your entire codebase?

That is the next chapter.

---

What you built in this chapter:

Component	What It Does
Messages API client	Clean, typed interface to Claude with streaming and error handling
Tool-use loop	Agent loop that detects tool calls, executes tools, and feeds results back
Tool Runner	Built-in agent loop that handles the observe-think-act cycle automatically
Managed Agents	Persistent, server-hosted agents that survive process boundaries
Computer Use agent	Agent that sees screenshots and controls mouse/keyboard
Prompt caching	85%+ reduction in input token costs for repeated prompt content
Complete agent	All features combined: Tool Runner, caching, streaming, error handling

Key takeaways:

• The Anthropic SDK hits the sweet spot: thin enough to understand, powerful enough to be productive.
• Tool Runner handles the agent loop for you. Use it for standard patterns. Build your own loop when you need custom control.
• Managed Agents give you persistent, always-on agents hosted by Anthropic. Use them for long-running tasks and shared team agents.
• Computer Use lets Claude see and control a computer. Run it in a sandbox. Validate destructive actions.
• Prompt caching is a free lunch. Cache your system prompt and tool definitions on every agent you build.
• The SDK is a tool, not a crutch. You can always drop down to the raw Messages API when you need more control.