Environment & Your First API Call
Before any RAG or agent, you need a rock-solid understanding of the single API call — the atom every larger system is built from. By the end you'll have a working environment and will understand every field of a request and response.
- an Anthropic API key (
ANTHROPIC_API_KEY) +pip install anthropic
Learning objectives
- Install the SDK and configure credentials the way production teams do.
- Make a first request and read back token usage and stop reasons.
- Explain every field in a request:
model,max_tokens,system,messages,thinking,effort. - Stream responses and handle errors like a real service.
Why we start with one call essential
It's tempting to jump straight to agents. Resist that. A RAG pipeline is many single calls plus retrieval; an agent is many single calls in a loop. If you don't deeply understand one request — what you pay for, why it stops, how it streams, how it fails — you'll be debugging blind later. This chapter is the foundation the whole course stands on.
Lab 1.1 · Set up the environment essential
- Create an isolated project. A virtual environment keeps this course's packages away from the rest of your system.
terminal
mkdir llm-course && cd llm-course python3 -m venv .venv source .venv/bin/activate # Windows: .venv\Scripts\activate - Install the SDK (plus a few packages we'll use in later chapters).
terminal
pip install anthropic pydantic python-dotenv - Verify it imported.
terminal
python -c "import anthropic; print(anthropic.__version__)"If that prints a version number, you're ready.
Lab 1.2 · API keys, the safe way essential
Your API key is a credential that spends real money. Never hardcode it in a .py file, never commit it to git. The standard pattern is an environment variable loaded from a .env file that's git-ignored.
- Get a key from your provider's console and copy it.
- Create a
.envfile in your project root:.env
ANTHROPIC_API_KEY=sk-ant-your-key-here - Git-ignore it so it never leaves your machine:
.gitignore
.env .venv/ __pycache__/ - Load it in Python. The SDK reads
ANTHROPIC_API_KEYautomatically once it's in the environment.config.py
from dotenv import load_dotenv load_dotenv() # reads .env into os.environ
client = Anthropic(api_key="sk-ant-abc123...") — a hardcoded key gets committed, ends up in logs, and leaks. Also: never put a key in a prompt or message; prompts are logged.Lab 1.3 · Your first call intermediate
Create first_call.py and type it out (don't paste — typing builds memory):
first_call.pyfrom dotenv import load_dotenv
from anthropic import Anthropic
load_dotenv()
client = Anthropic() # reads the key from the env
resp = client.messages.create(
model="claude-opus-4-8",
max_tokens=300,
system="You are a concise technical tutor.",
messages=[
{"role": "user", "content": "Explain what an API token is in 2 sentences."},
],
)
for block in resp.content: # content is a LIST of blocks
if block.type == "text":
print(block.text)
print("---")
print("stop reason:", resp.stop_reason)
print("input tokens:", resp.usage.input_tokens)
print("output tokens:", resp.usage.output_tokens)
Run it:
terminalpython first_call.py
Expected shape of the output:
An API token is a unit of text... (a couple of sentences)
---
stop reason: end_turn
input tokens: 24
output tokens: 41
This is the whole shape of every call you'll ever make to a language model: build a client, send a request describing what you want, then read the reply. If you understand these ~15 lines, you understand the API.
load_dotenv()loads your secret API key from the.envfile into the environment.client = Anthropic()creates the object that talks to the API — it picks up the key automatically, so the key never appears in your code.client.messages.create(...)sends the request.modelpicks which Claude to use;max_tokenscaps how long the reply can be;systemsets the assistant's role;messagesis the conversation — here one user message.- The reply's
resp.contentis a list of blocks, not a plain string. We loop over it and print only thetextblocks. (Later, tool-use replies add other block types — that's why it's a list.) resp.stop_reasontells you why the model stopped (normally"end_turn"), andresp.usagereports how many tokens you were charged for — input and output separately.
What the output means: The model's 2-sentence explanation prints, then a separator, then the stop reason and the input/output token counts you'll use to compute cost.
Try this: Change the user content to your own question and re-run. Then lower max_tokens to 20 and watch stop_reason become "max_tokens" — the model was cut off, not finished.
contentis a list of blocks, not a string. Always loop and checkblock.type— later there will bethinkingandtool_useblocks too.stop_reason: "end_turn"means the model finished naturally. Other values matter — see §Reading the response.- You're billed on input + output tokens. That 24/41 split is your cost model.
Anatomy of a request intermediate
messages, so a multi-turn chat means appending each reply and resending the whole transcript. That statelessness is why "memory" is something you manage (Ch 4).
This picture explains the single most surprising thing about the API: it has no memory. Each call is independent.
- The left box (REQUEST) is everything you send: the model name, limits, your system prompt, and the
messageslist — the entire conversation so far. - The arrow is the network call (
POST /v1/messages) — your request going up to Anthropic's servers. - The right box (RESPONSE) is what comes back: the content blocks, the stop reason, and token usage.
- The key idea in the caption: to continue a conversation you must append the reply and resend the whole thing next time. The server didn't remember your last turn — you carry the history.
In short: "Stateless" means you own the conversation history, not the server. Every multi-turn chat you build is just this loop: send history → get reply → add it to history → send again.
Every field you sent, explained:
| Field | What it is | Guidance |
|---|---|---|
model | Which model serves the request | Use the exact ID string. Frontier for reasoning, small/fast for classification. |
max_tokens | Hard cap on output length | Too low → truncated mid-sentence. Default ~4096 for chat; higher needs streaming. |
system | The role/rules the model follows | Stable instructions. Kept separate from the conversation. |
messages | The conversation so far | Alternating user/assistant. First must be user. |
thinking | Whether the model reasons before answering | {"type":"adaptive"} lets it decide depth. Great default for hard tasks. |
output_config.effort | How hard the model works | low/medium/high/max. Trades cost/latency for quality. |
The messages array is a transcript
To have a multi-turn conversation, you append to messages and resend the whole thing:
multi_turn.pymessages = [
{"role": "user", "content": "My name is Sam."},
{"role": "assistant", "content": "Nice to meet you, Sam!"},
{"role": "user", "content": "What's my name?"}, # model sees the history
]
# The API is stateless — this whole list is sent every call.
Reading the response — stop reasons intermediate
Before you use content, check why the model stopped. This one habit prevents a whole class of production bugs.
stop_reason | Meaning | What to do |
|---|---|---|
end_turn | Finished naturally | Use the content. |
max_tokens | Hit the output cap — truncated | Raise max_tokens or stream; don't treat as complete. |
tool_use | Model wants to call a tool | Execute it, send the result back (Chapter 4). |
refusal | Declined for safety | Don't read content[0] blindly — it may be empty. Handle gracefully. |
pause_turn | Long server-side tool run paused | Re-send to resume. |
if resp.stop_reason == "refusal": ... and == "max_tokens" branches from day one. Code that assumes content[0].text always exists will crash on refusals.Lab 1.4 · Streaming advanced
For anything that produces more than a sentence or two, stream. It shows tokens as they arrive (better UX) and avoids HTTP timeouts on long outputs.
streaming.pyfrom dotenv import load_dotenv
from anthropic import Anthropic
load_dotenv()
client = Anthropic()
with client.messages.stream(
model="claude-opus-4-8",
max_tokens=500,
messages=[{"role":"user", "content":"Write a haiku about databases, then explain it."}],
) as stream:
for text in stream.text_stream: # yields text chunks as they arrive
print(text, end="", flush=True)
final = stream.get_final_message() # full Message once done
print("\n---\ntokens:", final.usage.output_tokens)
Streaming shows the reply as it's generated, word by word, instead of waiting for the whole thing. It's the difference between a chatbot that types live and one that freezes then dumps a paragraph.
with client.messages.stream(...) as stream:opens a streaming connection. Thewithblock makes sure the connection is closed cleanly when you're done.for text in stream.text_stream:receives small chunks of text as the model produces them.print(text, end="", flush=True)prints each chunk immediately with no newline, so they flow together like live typing.stream.get_final_message()gives you the complete assembled reply once streaming finishes — handy for reading the total token usage.
What the output means: You see the haiku and its explanation appear progressively, then a token count on the final line.
Try this: Remove flush=True on some terminals and the text may appear in bursts instead of smoothly — flush forces each chunk to the screen right away.
max_tokens. Use get_final_message() to still get token counts and stop reason even while streaming.Lab 1.5 · Handling errors like a service advanced
Networks fail, rate limits trip, servers overload. The SDK auto-retries 429/5xx, but you should still catch and classify errors.
safe_call.pyimport anthropic
from anthropic import Anthropic
client = Anthropic()
def ask(prompt: str) -> str:
try:
resp = client.messages.create(
model="claude-opus-4-8", max_tokens=500,
messages=[{"role":"user","content":prompt}],
)
except anthropic.RateLimitError as e: # 429
return "Busy right now — try again shortly."
except anthropic.APIStatusError as e: # other non-2xx
if e.status_code >= 500:
return "Service issue — please retry."
raise
except anthropic.APIConnectionError: # network
return "Network problem — check your connection."
if resp.stop_reason == "refusal":
return "I can't help with that request."
return next((b.text for b in resp.content if b.type=="text"), "")
print(ask("Give me one tip for learning to code."))
Real services fail sometimes — rate limits, server hiccups, dropped network. This function wraps a call so that instead of crashing, it returns a friendly message for each kind of failure. This is the difference between a demo and something you'd let real users touch.
try:attempts the API call. If it succeeds, we skip all theexceptblocks and go to the checks at the bottom.- Each
exceptcatches a specific error:RateLimitError(too many requests, HTTP 429),APIStatusError(other bad responses — we only soften 500-level server errors and re-raisethe rest), andAPIConnectionError(network problem). Catching specific errors, not a blanketexcept, is good practice. - After a successful call,
if resp.stop_reason == "refusal"handles the model declining, and the finalnext((b.text ...), "")safely pulls the text out, defaulting to an empty string if there's no text block.
What the output means: For a normal prompt you get the model's tip. If the API were rate-limited, you'd get "Busy right now — try again shortly." instead of a stack trace.
Try this: This try/except pattern (from P4) is used all over the course's production code. Notice it catches named errors — you always want to know which failure happened.
except blocks most-specific → least-specific. A single broad except Exception hides retryable-vs-fatal distinctions you need in production.Exercises advanced
Exercise 1.1 — Token accountant
Context: Token counts are only useful once you turn them into money. Pricing the call is the smallest possible step from “it works” to “I know what it costs.”
Your task: Modify first_call.py to print the estimated cost of the call, assuming input costs $5 per 1M tokens and output $25 per 1M tokens.
Requirements:
- Read both
resp.usage.input_tokensandresp.usage.output_tokens - Apply
(in*5 + out*25) / 1_000_000 - Print the cost formatted to enough decimals to see a fractional-cent figure
- Keep input and output priced at their separate rates
💡 Hint: Divide by 1_000_000 once at the end; an f-string like f"${cost:.6f}" shows the small numbers clearly.
Show solution
Illustrative fragment — defines demo values / files are needed before this runs standalone.
cost = (resp.usage.input_tokens * 5 +
resp.usage.output_tokens * 25) / 1_000_000
print(f"cost: ${cost:.6f}")
Exercise 1.2 — Break it on purpose
Context: The fastest way to internalise stop_reason is to force a truncation on purpose and watch what the API reports — a lesson in never assuming a reply is complete.
Your task: Set max_tokens=10 and ask for a long essay, then inspect resp.stop_reason. Report what you get and what it implies for defensive coding.
Requirements:
- Use a deliberately tiny
max_tokensagainst a long request - Observe
stop_reasonbecome"max_tokens"with text cut off mid-sentence - State the lesson: check the stop reason before using content downstream
- Note that a truncated reply must not be treated as a complete answer
💡 Hint: The cut-off text plus stop_reason == "max_tokens" is the whole point — the model was interrupted, not finished.
Show solution
stop_reason will be "max_tokens" and the text cuts off mid-sentence. Lesson: never assume a response is complete — check the stop reason before using the content downstream.
Exercise 1.3 — Mini chat loop
Context: A stateless API means “memory” is just you re-sending history. Building a terminal chat loop makes that trick concrete: append every turn and resend the whole list.
Your task: Build a terminal chat that keeps the conversation going — append each user input and each assistant reply to a messages list and resend it every turn.
Requirements:
- Loop reading user input until a sentinel like
quitends it - Append each user message as
{"role":"user",...}before the call - Resend the full
messageslist every turn — the API remembers nothing - Extract the assistant text safely from the block list and print it
- Append the assistant reply back so it's part of the next turn's history
💡 Hint: The “memory” is entirely in your list: send history → get reply → append it → send again. Needs ANTHROPIC_API_KEY to actually run.
Show solution
Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.
messages = []
while True:
user = input("you> ")
if user == "quit": break
messages.append({"role":"user","content":user})
resp = client.messages.create(model="claude-opus-4-8",
max_tokens=500, messages=messages)
reply = next(b.text for b in resp.content if b.type=="text")
print("bot>", reply)
messages.append({"role":"assistant","content":reply})
Notice memory = you resending history. That's the whole trick.
🪜 Practice ladder beginner → industry
Six graded exercises, easy to real-world. Try each before opening its solution.
Context: Every LLM call is billed on input plus output tokens, and the response object carries both counts. Reading them is the first habit toward understanding what a call actually costs.
Your task: Given a response object resp, print the input tokens, the output tokens, and their total, using the resp.usage fields from the lesson.
Requirements:
- Read
resp.usage.input_tokensandresp.usage.output_tokens - Total is the sum of the two
- Keep it runnable with a tiny stand-in
usageobject (no API key) - Print all three values with clear labels
💡 Hint: You don't need a live call — a small class exposing input_tokens/output_tokens is enough to exercise the logic.
Show solution
The lesson shows resp.usage.input_tokens and resp.usage.output_tokens. Total is just their sum. Runnable with a tiny stand-in for the usage object:
class Usage:
input_tokens = 24
output_tokens = 41
class Resp:
usage = Usage()
resp = Resp()
print("input:", resp.usage.input_tokens)
print("output:", resp.usage.output_tokens)
print("total:", resp.usage.input_tokens + resp.usage.output_tokens)
Context: Two lesson facts combine here: resp.content is a list of blocks (not a string), and usage lets you price the call. Getting text out safely is what keeps code from crashing on tool-use or refusal replies.
Your task: Write a function that returns the concatenated text of all text blocks in resp.content, then print the call's cost at $5/1M input and $25/1M output tokens.
Requirements:
- Iterate
resp.contentand keep only blocks whereblock.type == "text" - Never index
content[0]blindly — join all text blocks - Apply the cost formula:
(in*5 + out*25) / 1_000_000 - Print the extracted text and the cost formatted to a few decimals
- Runnable offline with fake block/usage objects
💡 Hint: A one-line generator inside "".join(...) filtered on block.type does the extraction; reuse Exercise 1.1's cost formula.
Show solution
Loop the content list and keep only block.type == "text" (never assume content[0]). Then apply the cost formula from Exercise 1.1. Runnable with fakes:
class Block:
def __init__(self, type, text=""):
self.type = type; self.text = text
class Usage:
input_tokens = 24; output_tokens = 41
class Resp:
content = [Block("text", "An API token is a chunk of text. "),
Block("text", "Models are billed per token.")]
usage = Usage()
def text_of(resp):
return "".join(b.text for b in resp.content if b.type == "text")
resp = Resp()
print(text_of(resp))
cost = (resp.usage.input_tokens * 5 + resp.usage.output_tokens * 25) / 1_000_000
print(f"cost: ${cost:.6f}")
Context: Before touching content, production code checks why the model stopped — this one habit prevents a whole class of bugs, from crashing on refusals to treating truncated output as complete.
Your task: Write handle(resp) that returns the right action string for each of the five stop_reason values (end_turn, max_tokens, tool_use, refusal, pause_turn) and does NOT read content[0].text on a refusal.
Requirements:
- Branch on
resp.stop_reasonfirst, before reading any content max_tokensmust be flagged as truncated, not treated as completerefusalpath must not touch content (it may be empty)- Only pull text on the safe paths, via a helper that defaults to
"" - Include a fallback for an unknown stop reason
- Runnable with fake response objects, including one with empty content
💡 Hint: A first_text() helper using next((b.text ...), "") lets the refusal and empty-content cases pass through without an IndexError.
Show solution
Branch on the stop reason first, then only touch content on the safe paths. Note the max_tokens case must flag the text as truncated, not complete:
def first_text(resp):
return next((b.text for b in resp.content if b.type == "text"), "")
def handle(resp):
sr = resp.stop_reason
if sr == "end_turn":
return "use: " + first_text(resp)
if sr == "max_tokens":
return "TRUNCATED (raise max_tokens or stream): " + first_text(resp)
if sr == "tool_use":
return "run the requested tool, send result back"
if sr == "refusal":
return "handle refusal gracefully (content may be empty)"
if sr == "pause_turn":
return "re-send to resume"
return "unknown stop reason"
class B:
def __init__(self, t, x=""): self.type=t; self.text=x
class R:
def __init__(self, sr, content): self.stop_reason=sr; self.content=content
print(handle(R("end_turn", [B("text","done.")])))
print(handle(R("max_tokens", [B("text","half a sen")])))
print(handle(R("refusal", []))) # note: empty content, must not crash
Context: Because the API is stateless, a multi-turn chat is you re-sending the whole transcript — and the API rejects a first non-user message or two same-role turns in a row. Enforcing that invariant yourself catches bugs before the network does.
Your task: Write append_turn(messages, role, text) that appends a turn while asserting the roles strictly alternate user/assistant and that the first message is user — the invariant the lesson's messages table requires.
Requirements:
- Assert
roleis one ofuser/assistant - First appended message must be
user - Reject a turn whose role equals the previous message's role (no repeats)
- Append
{"role": role, "content": text}and return the list - Demonstrate both a valid transcript and a rejected first-assistant turn
💡 Hint: Two asserts cover it: an empty list requires user; otherwise role != messages[-1]["role"].
Show solution
The subtle correctness point: the API rejects a first non-user message and rejects two same-role messages in a row. Enforce it yourself:
def append_turn(messages, role, text):
assert role in ("user", "assistant")
if not messages:
assert role == "user", "first message must be user"
else:
assert role != messages[-1]["role"], "roles must alternate"
messages.append({"role": role, "content": text})
return messages
m = []
append_turn(m, "user", "My name is Sam.")
append_turn(m, "assistant", "Nice to meet you, Sam!")
append_turn(m, "user", "What's my name?")
print(len(m), "turns; first role =", m[0]["role"])
for bad in [("assistant","x")]: # first must be user
try:
append_turn([], *bad)
except AssertionError as e:
print("rejected:", e)
Context: Networks fail, rate limits trip, servers overload. Production code wraps the SDK call in an explicit, observable layer that retries transient failures and still guards against refusals and truncation before trusting the text.
Your task: Harden the lesson's safe_call.py: wrap client.messages.create so it retries transient failures (429 / 5xx) with exponential backoff, catches each error type specifically, and checks stop_reason == "refusal" before reading text. (Needs an API key to run.)
Requirements:
- Bounded retry loop with exponential backoff (e.g.
2 ** attempt) - Catch
RateLimitError,APIStatusError, andAPIConnectionErrorspecifically, most- to least-specific - Retry 429 / 5xx; re-
raise4xx as fatal rather than hiding it - Return a friendly message on refusal and log/flag a
max_tokenstruncation - Extract text safely, defaulting to
""; add structured logging
💡 Hint: Treat only 429, 5xx, and connection errors as retryable; a 4xx is a caller bug you want to surface, not swallow.
Show solution
Real SDK code (the SDK already retries internally; this adds an explicit, observable layer). Needs an API key to run — it is not offline-runnable:
import time, logging
import anthropic
from anthropic import Anthropic
client = Anthropic()
log = logging.getLogger("ask")
def ask(prompt: str, retries: int = 3) -> str:
for attempt in range(retries):
try:
resp = client.messages.create(
model="claude-opus-4-8", max_tokens=500,
messages=[{"role": "user", "content": prompt}],
)
if resp.stop_reason == "refusal":
return "I can't help with that request."
if resp.stop_reason == "max_tokens":
log.warning("truncated output")
return next((b.text for b in resp.content if b.type == "text"), "")
except anthropic.RateLimitError: # 429 — retryable
wait = 2 ** attempt
log.warning("rate limited; backoff %ss", wait); time.sleep(wait)
except anthropic.APIStatusError as e: # other non-2xx
if e.status_code >= 500 and attempt < retries - 1:
time.sleep(2 ** attempt); continue # 5xx — retryable
raise # 4xx — fatal, don't hide
except anthropic.APIConnectionError: # network — retryable
time.sleep(2 ** attempt)
return "Service unavailable after retries."
print(ask("Give me one tip for learning to code."))Key production points: specific exceptions ordered most->least specific, bounded retries with backoff, structured logging for observability, and the refusal / truncation guards from the lesson so downstream code never trusts an incomplete answer.
Context: A team sharing one ANTHROPIC_API_KEY can't answer “what did RAG-search cost last month?” without per-call accounting. The single call is the unit of cost, so a thin meter at that seam turns usage into a finance report.
Your task: Design a wrapper around messages.create that tags every call with a feature label and accumulates input/output tokens and dollar cost per label. Provide a runnable accounting core and describe how you'd wire it to the real SDK.
Requirements:
- Keep a per-
featureledger of input tokens, output tokens, and USD - Take price rates per model as input — do not hard-code a single model's price
- Compute cost from
resp.usageafter each call:(in*pin + out*pout) / 1_000_000 - Provide a
report()that lists features, sorted by spend - Show the real SDK seam (a
metered_callthat readsresp.usage) as commented code needing a key - Accounting core runs offline; note that rates drift so versions/dates should be stamped
💡 Hint: Make the wrapper the only sanctioned entry point so every caller must pass a feature; a defaultdict keyed by label is the whole ledger.
Show solution
Design. The single call is the unit of cost, so the meter lives at exactly that seam. Wrap create(), read resp.usage after each call, and fold the numbers into a per-label ledger keyed by feature. Persist the ledger (DB / metrics backend) keyed by month; expose a report. Model tier is a per-call price so the meter must take rates as input, not hard-code one model.
The accounting core is pure Python and runnable; the commented line shows the real SDK seam (needs an API key there):
from collections import defaultdict
# price per 1M tokens, per model: (input, output)
PRICES = {
"claude-opus-4-8": (5.0, 25.0),
"claude-haiku-4-5": (1.0, 5.0),
}
class CostMeter:
def __init__(self):
self.ledger = defaultdict(lambda: {"in": 0, "out": 0, "usd": 0.0})
def record(self, feature, model, in_tok, out_tok):
pin, pout = PRICES[model]
usd = (in_tok * pin + out_tok * pout) / 1_000_000
row = self.ledger[feature]
row["in"] += in_tok; row["out"] += out_tok; row["usd"] += usd
return usd
def report(self):
for feat, r in sorted(self.ledger.items(), key=lambda kv: -kv[1]["usd"]):
print(f"{feat:16} {r['in']:>7} in {r['out']:>7} out ${r['usd']:.4f}")
# def metered_call(self, feature, **kw):
# resp = client.messages.create(**kw) # real SDK — needs API key
# self.record(feature, kw["model"],
# resp.usage.input_tokens, resp.usage.output_tokens)
# return resp
m = CostMeter()
m.record("rag-search", "claude-haiku-4-5", 120_000, 30_000)
m.record("rag-search", "claude-haiku-4-5", 80_000, 20_000)
m.record("chat", "claude-opus-4-8", 40_000, 60_000)
m.report()Tradeoffs. Tagging at the wrapper is cheap and accurate but relies on every caller passing a feature; enforce it by making the wrapper the only sanctioned entry point. Prices drift, so store the rate table with a version/date and stamp each ledger row so old months re-cost correctly. For high volume, emit each call as a metric (feature, model, tokens) to your observability stack and aggregate there instead of an in-process dict.
✓ Checkpoint — you can move on when you can…
- Make a streaming and non-streaming call from a clean environment.
- Explain
model,max_tokens,system,messages,thinking,effortwithout looking. - Name what
end_turn,max_tokens, andrefusalmean and handle each. - Explain why the API being stateless means you manage conversation history.
kubectl get or terraform plan, it's one call like first_call.py. And checking stop_reason before acting? That habit becomes critical when the "output" is a decision to touch infrastructure. See the build-along plan →Knowledge check check yourself
The lesson calls an LLM call a 'stateless function.' What practical consequence does that statelessness have for building a multi-turn conversation?
Show answer
Why does the chapter insist you inspect stop_reason before using resp.content, and what bug does assuming content[0].text always exists cause?