Claude & Anthropic · Chapter C2

The Anthropic API — Messages, System Prompts & Tool Use

Q: Why must you iterate over resp.content and check block.type instead of reading resp.content[0].text directly?

Because content is a list of blocks (text, thinking, tool_use), so the first block isn't guaranteed to be text — a thinking or refusal block can come first and break a blind content[0].text read.

Q: In the manual tool-use loop, what tells your code Claude wants a tool, and what must each returned tool_result carry?

A response with stop_reason == "tool_use" signals Claude wants a tool; each tool_result must include the matching tool_use_id from the tool_use block, and parallel results go back in a single user message.

Everything Claude does over the API goes through one endpoint: POST /v1/messages. Master its shape — the request, the response, streaming, and tool use — and every advanced feature is just another field on the same call.

⏱️ ~70 min🧪 5 labs🎯 Beginner→Intermediate

⚙️ To run this for realThe code here is complete and correct as written. To actually execute it you'll need:

an Anthropic API key (ANTHROPIC_API_KEY) + pip install anthropic

Reading and learning works without any of this — run when you're ready. IDs/ARNs in examples are placeholders; swap in your own.

Learning objectives

Install the SDK, manage keys safely, and make an authenticated call.
Dissect a request (model, max_tokens, system, messages) and a response (content, stop_reason, usage).
Stream a response and get the final message safely.
Run a manual tool-use loop by hand — the pattern every agent is built on.
Handle errors and stop reasons like production code.

☁️ Also available on AWSThe same Claude models run on Amazon Bedrock, billed through AWS with auth, logging, and data residency handled for you. Everything in this chapter — system, messages, tokens, tool use — maps directly. See W2 · Bedrock & Claude for the Bedrock front door.

Lab C2.1 · Install & authenticate essential

Example code for learning — review, test, and adapt it before running against real or production systems. Commands can create, change, or delete resources. See the Terms & Disclaimer.

Lab C2.1

Install the SDK.

shellpip install anthropic python-dotenv

Store your key in .env — never hard-code it, never commit it.
```
.envANTHROPIC_API_KEY=sk-ant-...
```

Construct the client. With the key in the environment, the zero-arg constructor just works.

client.pyfrom dotenv import load_dotenv
from anthropic import Anthropic
load_dotenv()
client = Anthropic()   # reads ANTHROPIC_API_KEY from the environment

▶ How this works

Before you can talk to Claude you need a client — the object that carries your secret API key and knows how to reach Anthropic's servers. These three lines set that up once, and every later example reuses the client they create.

load_dotenv() reads a file named .env sitting next to your code and copies the values inside it (like ANTHROPIC_API_KEY=sk-ant-...) into the program's environment. Keeping the key in a file — not in your code — means you can't accidentally publish it.
client = Anthropic() builds the client with no arguments. It quietly looks in the environment for ANTHROPIC_API_KEY and picks it up on its own — that's why the key never appears anywhere in this file.
From here on, client is your door to the API: every request in this lesson is a method call on this one object (client.messages.create(...) and friends).

Try this: Make sure your .env file lists ANTHROPIC_API_KEY=sk-ant-... and add .env to .gitignore. Then in a Python shell run these three lines — if no error appears, your key was found and the client is ready.

Never commit keysAdd .env to .gitignore before your first commit. A leaked key is a billable liability — rotate it immediately in the console if it ever lands in git history or a log.

Anatomy of a request & response essential

Every call is the same four required-or-common fields in, one structured object out.

One endpoint, one shape. Tools, thinking, structured output, caching — every advanced feature is just another field on this same request. Learn the skeleton once and everything else slots in.

🗺️ How to read this diagram

This picture is the whole API on one slide: on the left is what you send, in the middle is the single network call that carries it, and on the right is what comes back. Every feature you'll ever use is just another field in one of those two boxes.

The left box (request) lists the fields you fill in: model (which Claude), max_tokens (how long the reply may be), system (the assistant's standing instructions), and messages[] (the conversation). tools? / thinking? have a ? because they're optional.
The arrow is the actual network request: POST /v1/messages. "POST" just means "here is some data, please process it" — your request travels up to Anthropic once per call.
The right box (response) is the structured reply: content[] (the answer, as a list of blocks), stop_reason (why it stopped), usage (tokens you were billed for), plus the model and a unique id.
Notice both content and messages end in [] — they are lists, not single strings. That detail matters in the code below.

In short: Learn this skeleton once — request in, one structured object out. Tool use, streaming, structured output and caching are all just extra fields on this exact same call.

Request field	What it is
`model`	Exact model ID, e.g. `claude-opus-4-8`
`max_tokens`	Hard cap on output tokens. Too low → truncated mid-thought.
`system`	The system prompt — role, rules, tools-usage, examples (Chapter 2's five-part pattern).
`messages`	The conversation so far: a list of `{"role","content"}`. The API is stateless — you resend the whole history each call.

Response field	What it is
`content`	A list of blocks (text, thinking, tool_use). Check `block.type` before reading `.text`.
`stop_reason`	Why it stopped: `end_turn`, `max_tokens`, `tool_use`, `refusal`, `pause_turn`.
`usage`	Token counts — input, output, and cache reads/writes. Your cost meter.

Lab C2.2 · A complete call, dissected essential

Lab C2.2

Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.

basic_call.pyresp = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    system="You are a concise assistant for software engineers.",
    messages=[{"role":"user","content":"What is the capital of France?"}],
)

# content is a LIST of blocks — never assume content[0] is text
for block in resp.content:
    if block.type == "text":
        print(block.text)

print(resp.stop_reason)                 # "end_turn"
print(resp.usage.input_tokens, resp.usage.output_tokens)

Paris.
end_turn
24 4

▶ How this works

This is a complete, real call — the shape of every request you'll make. You build a request describing what you want, send it, then read the reply apart field by field. Understand these few lines and you understand the API.

client.messages.create(...) sends the request. model chooses which Claude; max_tokens caps the reply length; system sets the assistant's role; messages is the conversation — here just one user turn asking a question.
A message is a small dictionary with two keys: "role" (who is speaking — user, assistant, or set separately, system) and "content" (what they said).
resp.content is a list of blocks, not a plain string — so we for-loop over it and print only the blocks whose block.type is "text". The comment warns you never to grab content[0] blindly.
resp.stop_reason tells you why the model stopped (here "end_turn" — it finished naturally), and resp.usage reports the input and output token counts you're charged for.

What the output means: The three printed lines: the answer Paris., then end_turn (a clean finish), then 24 4 — 24 input tokens and 4 output tokens for this call.

Try this: Change the question in content and re-run. Then set max_tokens to 2 and watch stop_reason flip to max_tokens — the reply was cut off, not finished.

Guard the content blocksReading resp.content[0].text blindly breaks the moment a thinking block or a refusal comes first. Always iterate and check block.type.

Multi-turn: the API is stateless essential

The API keeps no memory between calls. To continue a conversation you append the assistant's reply to your messages list and send the whole thing again.

Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.

conversation.pymessages = [{"role":"user","content":"My name is Alice."}]
r1 = client.messages.create(model="claude-opus-4-8", max_tokens=256, messages=messages)

reply = next(b.text for b in r1.content if b.type == "text")
messages.append({"role":"assistant","content": reply})
messages.append({"role":"user","content":"What's my name?"})

r2 = client.messages.create(model="claude-opus-4-8", max_tokens=256, messages=messages)
# Claude answers "Alice" — because you resent the history, not because it remembered

First message must be userThe conversation always opens with a user turn. This resend-the-history model is exactly why prompt caching (Chapter 6) matters so much for long conversations.

▶ How this works

The API has no memory — each call is independent. So to hold a conversation, you keep the history in a list and resend the whole thing every turn. This block proves it: Claude only "remembers" Alice's name because we mailed the earlier turns back to it.

messages starts as a list with one user turn ("My name is Alice."). The first create call sends it and gets reply r1.
next(b.text for b ... if b.type == "text") pulls the first text block out of the reply — the same guard-the-blocks pattern as before, written compactly.
We then append two things to the list: the assistant's reply (role assistant) and our new question (role user). The list now holds the full history of the chat.
The second create call sends that entire grown list. Because the earlier turns are physically in the request, Claude can answer "Alice" — the comment stresses it's not memory, it's you resending.

Try this: Comment out the two append lines, then ask "What's my name?" — Claude can't answer, because without the history in messages the server has no idea who Alice is. That is what stateless means.

Lab C2.3 · Streaming intermediate

For anything with a long output, stream — it shows tokens as they arrive and avoids HTTP timeouts on big responses.

Lab C2.3

Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.

stream.pywith client.messages.stream(
    model="claude-opus-4-8", max_tokens=1024,
    messages=[{"role":"user","content":"Write a haiku about databases."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()   # full Message once done
print(final.usage.output_tokens)

▶ How this works

Streaming shows the reply as it is being written, chunk by chunk, instead of waiting for the whole answer. It's the difference between a chatbot that types live in front of you and one that freezes, then dumps a paragraph all at once.

with client.messages.stream(...) as stream: opens a live connection. The with block guarantees the connection is closed cleanly when the block ends, even if something goes wrong.
for text in stream.text_stream: receives small pieces of the answer as they're produced. print(text, end="", flush=True) prints each piece immediately with no line break — flush=True forces it to the screen at once, so it flows like typing.
stream.get_final_message() hands you the complete assembled reply once streaming ends — the same object a normal call returns — so you can read totals like final.usage.output_tokens.

What the output means: You watch the haiku appear word-by-word on screen, then a single number on the last line — the total output token count from the finished message.

Try this: Remove flush=True; on some terminals the text now arrives in bursts instead of smoothly. Streaming is recommended for any large output — it also avoids the timeout that big non-streaming replies can hit.

Default to streaming for big outputsFor large max_tokens (above ~16K), non-streaming calls can hit the SDK's timeout guard. messages.stream() plus get_final_message() gives you both live output and the complete object.

Structured output (recap) intermediate

You met this in Chapter 2 — it belongs to the same endpoint. When another program consumes the output, constrain it to a schema with messages.parse() so you get a typed object instead of parsing free text.

Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.

parse.pyfrom pydantic import BaseModel
from typing import Literal

class Ticket(BaseModel):
    category: Literal["bug","feature","billing","other"]
    priority: Literal["low","medium","high"]

resp = client.messages.parse(
    model="claude-opus-4-8", max_tokens=256,
    messages=[{"role":"user","content":"Charged twice, no reply in 3 days!"}],
    output_format=Ticket,
)
t = resp.parsed_output                    # a real Ticket instance
print(t.category, t.priority)             # billing high

See Chapter 2 for the full treatmentThe five-part system prompt, few-shot examples, and schema-constrained JSON all live in Chapter 2 · Prompting & structured output. This chapter is the API mechanics underneath them.

▶ How this works

Sometimes you don't want a paragraph — you want clean, typed data another program can use directly. messages.parse() forces the model's answer to match a shape you define, and hands you back a real Python object instead of text you'd have to parse by hand.

class Ticket(BaseModel) uses Pydantic to declare the exact shape you want: a category that must be one of four words and a priority that must be one of three. Literal[...] is what pins it to those allowed values — this is your contract.
client.messages.parse(...) is like create, but with an extra output_format=Ticket. That tells the API: don't reply in prose, reply as data that fits this Ticket shape.
resp.parsed_output is the result already turned into a Ticket object, so you read t.category and t.priority like normal attributes — no JSON parsing, no guesswork.

What the output means: For the angry billing message, t.category is billing and t.priority is high — structured values you could store in a database or branch on with an if.

Try this: Change the input to something calm like "Small typo on the pricing page." and watch priority drop to low. The shape is guaranteed either way — only the values change.

Lab C2.4 · Tool use — the manual loop intermediate

Tools let Claude call your code. You describe a tool with a JSON schema; when Claude wants it, the response comes back with stop_reason == "tool_use", you run the function, feed the result back, and loop. This hand-written loop is the beating heart of every agent (Chapter 4).

Round-trip until done. Claude never executes anything itself — it asks, your harness runs the tool and returns a tool_result, and the loop repeats until stop_reason is end_turn.

🗺️ How to read this diagram

This diagram animates the loop from the code above: a round-trip between your program (left) and Claude (right) that repeats until Claude is finished. Read it as a back-and-forth conversation, not a one-shot call.

The top arrow (left → right) is you sending messages + tools: the conversation so far plus the list of tools Claude is allowed to ask for.
The bottom arrow (right → left) is Claude replying with stop_reason: tool_use — it isn't answering yet, it's asking your code to run a tool.
The left labels show your side of the deal: your harness runs the function and sends a tool_result back. The right labels show Claude reading that result and either answering or calling again.
The two arrows form a cycle. It keeps spinning until Claude's stop_reason becomes end_turn instead of tool_use — that's the loop's exit.

In short: Claude only ever asks; your code is the only thing that actually runs tools. That gap is exactly where an agent adds safety checks, logging, or a human approval step before acting.

Lab C2.4

Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.

tool_loop.pytools = [{
    "name": "get_weather",
    "description": "Get current weather for a city. Call when the user asks about weather.",
    "input_schema": {
        "type": "object",
        "properties": {"city": {"type": "string"}},
        "required": ["city"],
    },
}]

def get_weather(city): return f"18°C and clear in {city}"

messages = [{"role":"user","content":"What's the weather in Paris?"}]

while True:
    resp = client.messages.create(
        model="claude-opus-4-8", max_tokens=1024,
        tools=tools, messages=messages,
    )
    if resp.stop_reason != "tool_use":
        break                              # Claude is done — final answer is in resp.content

    messages.append({"role":"assistant","content": resp.content})
    results = []
    for block in resp.content:
        if block.type == "tool_use":
            out = get_weather(**block.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,   # MUST match the tool_use block
                "content": out,
            })
    messages.append({"role":"user","content": results})

print(next(b.text for b in resp.content if b.type == "text"))

It's currently 18°C and clear in Paris.

▶ How this works

Tools let Claude run your code. You describe a function to Claude; when it decides it needs that function it asks for it, you run it, hand the answer back, and repeat. This hand-written loop is the beating heart of every agent — Claude never runs anything itself, it only asks.

tools describes one function to Claude as data: its name, a plain description of when to use it, and an input_schema saying it takes a city string. Claude reads this to know the tool exists.
get_weather(city) is the real Python function that does the work. Here it just returns a canned string, but in real life it might call a weather service.
The while True loop calls the API. If stop_reason is not "tool_use", Claude is finished and we break. Otherwise Claude wants a tool, so we keep going.
We append Claude's request to messages, then loop over its content for any tool_use block, run get_weather(**block.input), and package the answer as a tool_result. The tool_use_id must match the block's id so Claude knows which request this answers.
We send all results back as one user message and loop again. Now Claude has the weather and can write the final sentence, which the last print extracts.

What the output means: After one round-trip, Claude has the tool's answer and prints the natural reply: It's currently 18°C and clear in Paris.

Try this: Add a print(resp.stop_reason) right after the create call. You'll see tool_use on the first pass (Claude asks for the tool) and end_turn on the second (Claude is done) — the loop's two states, made visible.

The SDK can drive the loop for youThe manual loop is worth writing once so you understand it. In production the SDK's tool runner (client.beta.messages.tool_runner(...) with the @beta_tool decorator) handles the round-trips automatically. Use the manual loop when you need human-in-the-loop approval, custom logging, or a safety gate before each call — exactly what the capstone does.

Two rules that bite everyoneEvery tool_use block needs exactly one matching tool_result with the same tool_use_id. And when Claude calls several tools at once, return all results in a single user message — splitting them trains the model to stop calling tools in parallel.

Lab C2.5 · Errors & stop reasons advanced

Production code branches on both exceptions and stop reasons.

Lab C2.5

Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.

robust.pyimport anthropic

try:
    resp = client.messages.create(
        model="claude-opus-4-8", max_tokens=1024,
        messages=[{"role":"user","content": user_input}],
    )
except anthropic.RateLimitError as e:      # 429 — SDK already retries; back off further if needed
    ...
except anthropic.APIStatusError as e:      # other non-2xx
    print(e.status_code, e.message)
else:
    if resp.stop_reason == "refusal":       # safety decline — don't retry the same prompt
        handle_refusal(resp)
    elif resp.stop_reason == "max_tokens":  # truncated — raise max_tokens or stream
        handle_truncation(resp)

▶ How this works

Real services fail sometimes — rate limits, server hiccups, dropped networks — and even a successful call can come back truncated or refused. Production code branches on both: the exceptions that mean "the call failed" and the stop_reason that describes how a successful call ended.

try: attempts the API call. If it works, Python skips the except blocks and jumps to the else: checks at the bottom.
Each except catches a specific, named error: RateLimitError (HTTP 429 — too many requests; the SDK already retried) and APIStatusError (any other bad HTTP response). Catching named errors, not a blanket except, means you always know which failure happened.
The else: runs only when the call succeeded, and inspects stop_reason: "refusal" means the model declined for safety (don't just retry the same prompt), and "max_tokens" means the answer was cut off (raise the cap or stream).

Try this: The golden rule shown here: catch typed exceptions (anthropic.RateLimitError) — never match on error message text, which can change. And always check stop_reason even on a call that didn't throw.

`stop_reason`	Meaning & what to do
`end_turn`	Finished naturally. Use the output.
`max_tokens`	Hit the output cap. Raise `max_tokens` or stream; output is incomplete.
`tool_use`	Wants a tool. Run it, return the result, loop.
`refusal`	Declined for safety. Surface it; don't blindly retry the same prompt.
`pause_turn`	Server-side tool paused. Resend to resume.

The SDK retries for youThe client auto-retries 429 and 5xx with exponential backoff (default 2 retries). Don't hand-roll retry logic unless you need behavior beyond that. Catch the typed exceptions — never string-match error messages.

Common pitfalls advanced

Pitfall	Fix
Reading `resp.content[0].text` blindly	Iterate blocks; check `block.type`
Expecting the API to remember the chat	It's stateless — resend the full `messages` history
Mismatched / missing `tool_use_id`	Every `tool_use` needs one matching `tool_result`
Splitting parallel tool results across messages	Return all results in one user message
Low `max_tokens` silently truncating	Default ~1024–16K; stream for large outputs; check `stop_reason`
Hard-coding the API key	Load from env; `.gitignore` the `.env`

Exercises advanced

Exercise C2.1 — Add a second tool

Context: Real agents carry more than one tool, and Claude can ask for several at once. Adding a second tool proves your loop batches every result back correctly.

Your task: Extend Lab C2.4 with a get_time(city) tool, ask "What's the weather and time in Tokyo?", and confirm Claude calls both tools and you return both results in one user message.

Requirements:

Add the second tool definition to the tools list
Handle a response that contains multiple tool_use blocks
Run each requested tool and collect its result
Return all results in a single user message
Note that splitting results across messages breaks parallel tool calls

💡 Hint: Loop over the content blocks collecting results, then send them once — one tool_use, one tool_result, all in the same turn.

Show hint

Claude may emit two tool_use blocks in one response. Loop over resp.content, run each, and append all tool_result blocks to a single {"role":"user","content":[...]}.

Exercise C2.2 — Cost meter

Context: The habit that keeps a real system's bill predictable is logging usage on every single call — runaway cost is caught by the meter you built, not the invoice.

Your task: Wrap messages.create so it prints usage.input_tokens and usage.output_tokens after every call, plus an estimated cost using the per-token rates from C1.

Requirements:

Read usage.input_tokens and usage.output_tokens after each call
Multiply by the per-token rates from Chapter C1
Print input, output and estimated cost every call
Make it a wrapper so the habit is automatic, not remembered
Track it to catch runaway expense early

💡 Hint: A thin wrapper around create means the meter runs whether or not anyone remembered to think about cost that day.

Exercise C2.3 — Refusal handling

Context: A refusal dressed up as a normal answer is a bug that ships silently. A wrapper that inspects stop_reason makes the decline explicit to everything downstream.

Your task: Write a wrapper that inspects stop_reason and returns a structured result: {"ok": True, "text": ...} on end_turn, and {"ok": False, "reason": "refusal"} otherwise.

Requirements:

Return an ok result with the text on end_turn
Return a not-ok result carrying the reason on anything else (refusal, max_tokens, ...)
Never hide a refusal inside the text output
Make the caller check ok before trusting text
Surface the reason so downstream code can react

💡 Hint: Force the caller to look at ok first — a structured result makes "was this actually answered?" impossible to skip.

🪜 Practice ladder beginner → industry

Six graded exercises, easy to real-world. Try each before opening its solution.

Exercise 1 · Construct the client and make your first callBeginner

Context: Every Claude program starts the same way: load the key from the environment, build the client, send a Messages request, and read the response's content blocks — not a naive content[0].text.

Your task: Load your API key from .env, construct the Anthropic() client, ask Claude "What is the capital of France?" with model="claude-opus-4-8", and print only the text blocks.

Requirements:

Read the key from the environment (e.g. load_dotenv()); never hard-code it
Send model, max_tokens, an optional system prompt, and a messages list
Treat content as a list of blocks and print only those with type == "text"
Note stop_reason and usage for later cost tracking
Needs an API key to run

💡 Hint: Iterating the content blocks and checking each block's type is the habit that survives thinking and tool_use blocks appearing alongside text.

Show solution

from dotenv import load_dotenv
from anthropic import Anthropic

load_dotenv()
client = Anthropic()   # reads ANTHROPIC_API_KEY from the environment

resp = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    system="You are a concise assistant for software engineers.",
    messages=[{"role": "user", "content": "What is the capital of France?"}],
)

# content is a LIST of blocks — never assume content[0] is text
for block in resp.content:
    if block.type == "text":
        print(block.text)

The zero-arg Anthropic() constructor picks up ANTHROPIC_API_KEY from the environment, so the key never appears in code. Because resp.content is a list of blocks, you iterate and check block.type == "text" rather than reading content[0].text blindly. Needs an API key to run.

Exercise 2 · Continue a multi-turn conversation (stateless API)Intermediate

Context: The Messages API keeps no memory between calls. A two-turn conversation only works because you resend the whole history — the mechanic behind why prompt caching matters later.

Your task: Hold a two-turn conversation: tell Claude "My name is Alice.", then ask "What's my name?" — proving Claude only answers because you resend the history.

Requirements:

Each call is independent; the server holds no conversation state
Append the assistant's reply and the new user turn to messages, then resend the whole list
The conversation must open with a user turn
Show Claude answers "Alice" only because the earlier turns are physically in the request
Needs an API key to run

💡 Hint: If you dropped the first turn from the second request, the answer would vanish — that's the proof the memory lives in your list, not the server.

Show solution

messages = [{"role": "user", "content": "My name is Alice."}]
r1 = client.messages.create(
    model="claude-opus-4-8", max_tokens=256, messages=messages,
)

reply = next(b.text for b in r1.content if b.type == "text")
messages.append({"role": "assistant", "content": reply})
messages.append({"role": "user", "content": "What's my name?"})

r2 = client.messages.create(
    model="claude-opus-4-8", max_tokens=256, messages=messages,
)
print(next(b.text for b in r2.content if b.type == "text"))   # "Alice"

The Messages endpoint is stateless: to continue a conversation you append the assistant's reply and the new user turn to the messages list and resend the whole history. Claude answers "Alice" because the earlier turns are physically in the request, not because the server remembered. The conversation must always open with a user turn. Needs an API key to run.

Exercise 3 · Stream a long response and get the final messageAdvanced

Context: For long outputs you stream tokens as they're produced — better UX and it dodges the SDK's timeout guard — then still recover the complete assembled message afterward.

Your task: Stream a haiku about databases with client.messages.stream(...), print each chunk live, then retrieve the complete assembled message and print its output token count.

Requirements:

Open the stream in a with block so it closes cleanly
Iterate stream.text_stream and print chunks live (no line breaks, flushed)
After streaming, call stream.get_final_message() for the complete Message
Read usage.output_tokens off that final message
Needs an API key to run

💡 Hint: The final message is the same object a non-streaming call returns — streaming changes the delivery, not the shape of what you get back.

Show solution

with client.messages.stream(
    model="claude-opus-4-8", max_tokens=1024,
    messages=[{"role": "user", "content": "Write a haiku about databases."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()   # full Message once done
print()
print(final.usage.output_tokens)

The with block guarantees the live connection closes cleanly. stream.text_stream yields chunks as they arrive; flush=True forces each to the screen so it flows like typing. get_final_message() returns the same complete Message object a normal call would, so you can read totals like final.usage.output_tokens. Streaming is recommended for large outputs (above ~16K max_tokens) to avoid the SDK timeout guard. Needs an API key to run.

Exercise 4 · Write the manual tool-use loopExpert

Context: Claude never runs your functions; it only asks to. The manual tool-use loop is your harness driving that conversation — the pattern every agent in the course is built on.

Your task: Reproduce Lab C2.4: define a get_weather tool with a JSON input_schema, run the manual loop that breaks when stop_reason != "tool_use", and return each tool_result with the matching tool_use_id.

Requirements:

Define the tool as a dict with name, a when-to-use description, and an input_schema
Loop calling messages.create(..., tools=tools) until stop_reason is no longer tool_use
Append the assistant turn verbatim, then run each tool_use block's function
Return every tool_result carrying the matching tool_use_id, all in one user message
Remember the harness runs the function — Claude only requests it
Needs an API key to run

💡 Hint: The description tells Claude when to call the tool and the tool_use_id pairing keeps request and result matched — both are contracts, not decoration.

Show solution

tools = [{
    "name": "get_weather",
    "description": "Get current weather for a city. Call when the user asks about weather.",
    "input_schema": {
        "type": "object",
        "properties": {"city": {"type": "string"}},
        "required": ["city"],
    },
}]

def get_weather(city):
    return f"18°C and clear in {city}"

messages = [{"role": "user", "content": "What's the weather in Paris?"}]

while True:
    resp = client.messages.create(
        model="claude-opus-4-8", max_tokens=1024,
        tools=tools, messages=messages,
    )
    if resp.stop_reason != "tool_use":
        break                               # Claude is done

    messages.append({"role": "assistant", "content": resp.content})
    results = []
    for block in resp.content:
        if block.type == "tool_use":
            out = get_weather(**block.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,     # MUST match the tool_use block
                "content": out,
            })
    messages.append({"role": "user", "content": results})

print(next(b.text for b in resp.content if b.type == "text"))

Claude never runs anything itself — a response with stop_reason == "tool_use" means it is asking your code to run a tool. You append Claude's request, run each tool_use block's function, and return every tool_result in a single user message, each carrying the matching tool_use_id. The loop exits when stop_reason becomes end_turn. Needs an API key to run.

Exercise 5 · Robust error and stop-reason handlingProfessional

Context: Production code catches the SDK's typed exceptions and branches on stop_reason — because a refusal that reads as a valid answer is a silent, dangerous bug.

Your task: Wrap a call in production-grade handling: catch typed exceptions (anthropic.RateLimitError, anthropic.APIStatusError) and, on success, branch on stop_reason for refusal and max_tokens.

Requirements:

Catch typed exceptions — never string-match error messages
On RateLimitError, don't hand-roll extra retries (the SDK already retried)
On other API status errors, log the status code and message
On success, branch on stop_reason: surface a refusal, raise the cap or stream on max_tokens, use the output on end_turn
Return a structured result (ok/text or ok=false/reason); never let a refusal masquerade as an answer
Needs an API key to run

💡 Hint: Treat stop_reason as a fork in the road, not a footnote — the same text field means "good answer", "declined", or "cut off".

Show solution

import anthropic

def robust_call(user_input):
    try:
        resp = client.messages.create(
            model="claude-opus-4-8", max_tokens=1024,
            messages=[{"role": "user", "content": user_input}],
        )
    except anthropic.RateLimitError as e:      # 429 — SDK already retried
        return {"ok": False, "reason": "rate_limit"}
    except anthropic.APIStatusError as e:      # other non-2xx
        print(e.status_code, e.message)
        return {"ok": False, "reason": "api_error"}
    else:
        if resp.stop_reason == "refusal":       # don't retry same prompt
            return {"ok": False, "reason": "refusal"}
        if resp.stop_reason == "max_tokens":    # truncated — raise cap / stream
            return {"ok": False, "reason": "truncated"}
        text = next((b.text for b in resp.content if b.type == "text"), "")
        return {"ok": True, "text": text}

Production code branches on both failures (exceptions) and how a successful call ended (stop_reason). Catch typed exceptions — never string-match error messages, which can change. The SDK already auto-retries 429 and 5xx with exponential backoff (default 2 retries), so don't hand-roll retries. A refusal must never masquerade as a valid answer, and max_tokens means the output is incomplete. Needs an API key to run.

Exercise 6 · Schema-constrained structured output with a cost meterIndustry scenario

Context: Two production habits combine here: force the reply into a typed schema so downstream code can trust its shape, and log token usage after every call so the bill stays predictable.

Your task: Use client.messages.parse() with a Pydantic output_format to classify a support message into a typed Ticket, and log usage tokens after the call.

Requirements:

Define a Pydantic model with Literal[...] fields to pin the allowed values
Call messages.parse(..., output_format=Ticket) so the reply matches the schema
Read parsed_output as a real typed instance — no manual JSON parsing
Log usage.input_tokens and usage.output_tokens after the call
The shape is guaranteed; only the field values vary
Needs an API key to run

💡 Hint: Literal fields do double duty — they document the allowed values and constrain the model to them at the same time.

Show solution

from pydantic import BaseModel
from typing import Literal

class Ticket(BaseModel):
    category: Literal["bug", "feature", "billing", "other"]
    priority: Literal["low", "medium", "high"]

def classify(text):
    resp = client.messages.parse(
        model="claude-opus-4-8", max_tokens=256,
        messages=[{"role": "user", "content": text}],
        output_format=Ticket,
    )
    u = resp.usage
    print(f"in={u.input_tokens} out={u.output_tokens}")   # cost meter
    return resp.parsed_output

t = classify("Charged twice, no reply in 3 days!")
print(t.category, t.priority)   # billing high

messages.parse() is like create but takes an output_format Pydantic model, forcing the reply to match the schema; resp.parsed_output is a real Ticket instance, so no JSON parsing is needed. The Literal[...] fields pin the allowed values — the shape is guaranteed, only the values change. Printing usage.input_tokens and usage.output_tokens after every call is the habit that keeps a real system's bill predictable. Needs an API key to run.

✓ Checkpoint — you can move on when you can…

Make an authenticated call with a key loaded from the environment.
Name every field in a request and a response and say what it's for.
Continue a multi-turn conversation by resending history.
Stream a response and retrieve the final message.
Write the manual tool-use loop and explain the tool_use_id contract.
Branch correctly on stop_reason and typed exceptions.

🏗️ Toward the capstoneThe DevOps agent is this tool loop, hardened: each proposed tool call passes through a safety gate before your harness runs it, every call is logged, and risky actions wait for human approval. You'll write that manual loop — not the auto-runner — precisely so you can intercept each step. Build the agent loop in Chapter 4 →

Knowledge check check yourself

✓ Knowledge check

Why must you iterate over resp.content and check block.type instead of reading resp.content[0].text directly?

Show answer

Because content is a list of blocks (text, thinking, tool_use), so the first block isn't guaranteed to be text — a thinking or refusal block can come first and break a blind content[0].text read.

✓ Knowledge check

In the manual tool-use loop, what tells your code Claude wants a tool, and what must each returned tool_result carry?

Show answer

A response with stop_reason == "tool_use" signals Claude wants a tool; each tool_result must include the matching tool_use_id from the tool_use block, and parallel results go back in a single user message.

Educational content, provided as-is and without warranty. Code samples are examples — review, test, and adapt them before using in production. See the Terms of Use & Disclaimer. Use at your own risk.

Learning objectives

Lab C2.1 · Install & authenticate essential

Anatomy of a request & response essential

Lab C2.2 · A complete call, dissected essential

Multi-turn: the API is stateless essential

Lab C2.3 · Streaming intermediate

Structured output (recap) intermediate

Lab C2.4 · Tool use — the manual loop intermediate

Lab C2.5 · Errors & stop reasons advanced

Common pitfalls advanced

Exercises advanced

Exercise C2.1 — Add a second tool

Exercise C2.2 — Cost meter

Exercise C2.3 — Refusal handling

🪜 Practice ladder beginner → industry

✓ Checkpoint — you can move on when you can…

Knowledge check check yourself

More in Claude & Anthropic