Anthropic API Fundamentals

Documentation tells you what a request looks like. Running one tells you what it feels like — the latency, the response shape, what stop_reason actually means when you read it live.

Get your key

Go to console.anthropic.com, sign up, and create an API key under the API Keys section. The console shows the key exactly once — copy it immediately. Store it in an environment variable, not in your source code.

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

Add that line to your shell profile (~/.zshrc or ~/.bashrc) so it survives a new terminal session. Every example in this course reads from that variable. Nothing in the codebase ever holds the key string itself — not a config file, not a comment, not a test fixture.

Make the call

curl https://api.anthropic.com/v1/messages \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -H "content-type: application/json" \
  -d '{
    "model": "claude-opus-4-5",
    "max_tokens": 256,
    "messages": [
      {"role": "user", "content": "Say hello."}
    ]
  }'

Three headers, one JSON body. The anthropic-version header is required — without it the API returns a 400. The version string pins the response contract; Anthropic does not break old versions silently, so you can leave this value fixed until you deliberately upgrade.

Read the response

{
  "id": "msg_01XFDUDYJgAACzvnptvVoYEL",
  "type": "message",
  "role": "assistant",
  "content": [{"type": "text", "text": "Hello! How can I help you today?"}],
  "model": "claude-opus-4-5",
  "stop_reason": "end_turn",
  "usage": {"input_tokens": 10, "output_tokens": 12}
}

The text lives at content[0].text. The usage block tells you the exact token cost — input_tokens for what you sent, output_tokens for the reply. Store both when you log requests in production. The stop_reason field tells you why generation stopped: end_turn means the model finished naturally; max_tokens means you hit the cap and the response was cut short.

The same call in Python

Once you have a working curl command you understand what is on the wire. Now install the SDK and verify the two produce the same result:

# scripts/hello.py
import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from environment

response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=256,
    messages=[{"role": "user", "content": "Say hello."}],
)

print(response.content[0].text)
print(f"stop_reason: {response.stop_reason}")
print(f"input_tokens: {response.usage.input_tokens}")
print(f"output_tokens: {response.usage.output_tokens}")

Run pip install anthropic first. The SDK reads ANTHROPIC_API_KEY automatically — you do not pass it to the constructor. The response object gives you the same fields as the JSON above, typed. response.content[0].text is the string. response.usage.input_tokens is an integer. The SDK is the curl call with serialization handled.

Why no library first

The SDK wraps these three headers and the JSON serialization. Starting with raw curl means you know exactly what is on the wire, which means you can debug any SDK problem by stepping down to curl. That is worth ten minutes now.

If you get stuck

• 401 authentication_error: the key is wrong, missing, or the shell variable was not exported. Run echo $ANTHROPIC_API_KEY to confirm it is set. If it prints nothing, you need export, not just assignment (ANTHROPIC_API_KEY=... without export sets a local shell variable that child processes cannot see).
• 400 invalid_request_error: the JSON is malformed or a required field is missing. Paste the body into jq . to check syntax. Confirm model, max_tokens, and messages are all present.
• 400 with a message about anthropic-version: you omitted the version header. Add -H "anthropic-version: 2023-06-01".
• Response truncated mid-sentence: max_tokens is too low. Raise it and re-run.
• Python ModuleNotFoundError: run pip install anthropic. If you are in a virtual environment, confirm it is activated before installing.

Cross-reference

l2 (Messages, system, and the request shape) breaks down each field in the request body in detail. l4 (Errors, retries, and timeouts) covers what to do when the API returns 429 or 529 in production.

Deliverable

A 200 response with "role": "assistant" and a non-empty content[0].text from both the curl command and the Python script. Note the input_tokens value from both runs — they should match.

Further reading

• "One File for All My Keys" — names the convention that keeps ANTHROPIC_API_KEY out of every project's .env and into a single canonical file you source.

The request shape is small — four fields cover 90% of production use. Getting them wrong in subtle ways produces errors that look unrelated to the actual problem.

The three required fields

Every call to POST /v1/messages needs model, max_tokens, and messages.

model names which Claude version to use. Use claude-opus-4-5 for this course unless you have a reason to deviate. Model names are stable; Anthropic does not silently reassign them.

max_tokens is a hard ceiling on output length. The API stops generating when it hits this limit and returns stop_reason: "max_tokens". Set it deliberately — a value of 4096 on a route that only needs short answers wastes money and inflates latency. A value of 128 on a summarizer will produce truncated summaries.

messages is a list of turn objects. Each object has role ("user" or "assistant") and content (a string, or a list of content blocks — more on that in the tool-use course).

What system actually does

system is a separate top-level field, not part of messages. Anthropic processes it before any user turn. Use it for persistent instructions that should frame the entire conversation: persona, output format, language, constraints.

curl https://api.anthropic.com/v1/messages \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -H "content-type: application/json" \
  -d '{
    "model": "claude-opus-4-5",
    "max_tokens": 512,
    "system": "You are a terse technical assistant. Reply in plain text only, no markdown.",
    "messages": [
      {"role": "user", "content": "Explain what an API key is."}
    ]
  }'

The system instruction holds for every turn in the conversation. A common mistake is putting persistent instructions in the first user message. That works until the user's next message overrides the framing, or until the model weighs the user instruction against itself and produces inconsistent output. Put the constraint in system and it stays out of the turn-weighting.

Multi-turn shape

The API is stateless. You maintain the conversation history yourself by building up the messages array across calls.

# scripts/multiturn.py
import anthropic, os

client = anthropic.Anthropic()
history = []

def turn(user_text):
    history.append({"role": "user", "content": user_text})
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=512,
        system="You are a concise assistant.",
        messages=history,
    )
    reply = response.content[0].text
    history.append({"role": "assistant", "content": reply})
    return reply

print(turn("My name is Dana."))
print(turn("What is my name?"))

Run this. The second reply should use "Dana" — because the assistant turn containing that information is in history when the second call fires. Now remove the line that appends the assistant reply and run again:

# broken version — do not append assistant reply
def turn_broken(user_text):
    history.append({"role": "user", "content": user_text})
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=512,
        system="You are a concise assistant.",
        messages=history,
    )
    # missing: history.append({"role": "assistant", ...})
    return response.content[0].text

The second call will not know the name. More than that, the messages array now contains two consecutive user turns, which violates the alternating-roles requirement and will return a 400. That is the entire statefulness model of the API — you own the array, you own the state.

The content field in detail

Content can be a plain string: "content": "Hello." That is the shorthand. The full form is a list of content blocks: "content": [{"type": "text", "text": "Hello."}]. Both are valid. The API always returns the list form in responses. When you build tools or handle images, you will use the list form for the request too. For now, the string shorthand is fine for text-only requests.

If you get stuck

• 400 with a message about messages: the array is malformed, empty, or the last message has role: "assistant" instead of "user". The final message in the array must have role: "user".
• 400 about alternating roles: the messages array must strictly alternate user / assistant. You cannot send two user turns in a row without an assistant turn between them. The most common cause is forgetting to append the assistant reply before the next user message.
• System instruction not holding: confirm it is in the top-level system field, not inside messages. A system-style instruction placed as the first user message will have lower weight than a true system prompt.
• The model ignores max_tokens: max_tokens is an output ceiling, not a target. The model stops when it finishes naturally or hits the cap — whichever comes first. You cannot force longer output by raising max_tokens.

Cross-reference

l5 (Rate limits and the cost of context) shows how the usage.input_tokens field grows as you extend the history array. The cost compounds faster than most people expect. l3 (Streaming vs. blocking) builds directly on the request shape — everything shown here applies unchanged.

Deliverable

A two-turn conversation where the second user message refers to something said in the first turn, and the assistant correctly uses that context. Confirm usage.input_tokens is larger on the second call than the first.

Blocking and streaming use the same endpoint and the same request body. One field changes. The experience for a human reading the output changes completely.

Blocking mode — the full response arrives at once

Everything you have written so far uses blocking mode. The connection stays open until the model finishes, then the full JSON response arrives. Your code reads content[0].text and moves on.

This is the right choice when the response feeds downstream work: a database write, another API call, a test assertion. You cannot do any of that work on partial text.

Streaming mode — tokens arrive as they are generated

Add "stream": true to the request body. The endpoint switches from returning a single JSON object to returning a stream of server-sent events (SSE).

curl https://api.anthropic.com/v1/messages \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -H "content-type: application/json" \
  -d '{
    "model": "claude-opus-4-5",
    "max_tokens": 512,
    "stream": true,
    "messages": [{"role": "user", "content": "Count to ten slowly, one word per line."}]
  }'

The terminal will print raw SSE lines as they arrive. You will see events like event: content_block_delta with delta.text containing the next few characters, followed eventually by event: message_stop.

Handling streaming in Python

# scripts/stream.py
import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-opus-4-5",
    max_tokens=512,
    messages=[{"role": "user", "content": "Write a two-sentence summary of HTTP."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    print()  # newline after stream completes
    final = stream.get_final_message()
    print(f"\ninput_tokens={final.usage.input_tokens} output_tokens={final.usage.output_tokens}")

The text_stream iterator yields each partial text delta. get_final_message() returns the complete message object — including usage — after message_stop. You cannot access usage mid-stream because token counts are not finalized until generation ends.

Handling streaming in Node.js

Many web apps handle Claude responses in a JavaScript backend. The pattern is the same, just spelled differently:

// scripts/stream.mjs
import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

const stream = await client.messages.stream({
  model: "claude-opus-4-5",
  max_tokens: 512,
  messages: [{ role: "user", content: "Write a two-sentence summary of HTTP." }],
});

for await (const chunk of stream) {
  if (
    chunk.type === "content_block_delta" &&
    chunk.delta.type === "text_delta"
  ) {
    process.stdout.write(chunk.delta.text);
  }
}

const final = await stream.finalMessage();
console.log(`\ninput_tokens=${final.usage.input_tokens}`);

Run with node scripts/stream.mjs after npm install @anthropic-ai/sdk. The chunk shape mirrors the Python SDK: content_block_delta events carry the text, message_stop terminates the stream.

When to choose each mode

Streaming makes a ten-second response feel interactive because the first tokens arrive in under a second. For chat interfaces, search results, and any output a human reads live, streaming is the better choice.

Blocking is simpler to reason about. The response is either there or it is not. Use it for batch processing, pipelines, and any situation where the full response is needed before the next step runs.

The message_stop event matters

In a raw SSE implementation (not using the SDK's messages.stream helper), you must detect event: message_stop to know the stream is done. Stopping early — closing the connection before this event — means you silently miss the end of the response. The SDK handles this for you, but if you are implementing streaming in a language without an official SDK, this is the failure mode to watch for.

If you get stuck

• Terminal prints nothing and hangs: the stream: true flag may not have been parsed correctly. Check that your JSON is valid and that the field is a boolean, not the string "true".
• Stream ends mid-sentence: max_tokens was hit. Check stop_reason in the final message event — it will say max_tokens instead of end_turn.
• usage is all zeros: you are reading usage from a partial event rather than from the final message. Call get_final_message() (Python) or finalMessage() (Node) after the stream completes.
• SSE lines look garbled: each event is prefixed with data: — parse the JSON after that prefix, not the whole line.
• Node: for await loop exits before the stream finishes: confirm you are using stream (the async iterable) not stream.on('data', ...). The SDK exposes both; the for await form is safer.

Cross-reference

l7 (Wrap a call in a small CLI) shows how to wire streaming output to stdout cleanly via the --stream flag. l4 (Errors, retries, and timeouts) covers what happens when a streaming connection drops mid-response.

Deliverable

Run the same prompt in both blocking and streaming mode. Print the elapsed wall time for each. Confirm that the accumulated streaming text matches the blocking response text character-for-character.

Every production API integration eventually hits an error. The question is whether your code handles it gracefully or whether it surfaces a stack trace to the user.

The five error classes

The API returns standard HTTP status codes. Each has a distinct cause and the right response to it.

400 invalid_request_error — the request body is malformed or missing a required field. This is a bug in your code, not a transient condition. Do not retry. Fix the request.

401 authentication_error — the API key is wrong, missing, or expired. Do not retry. Fix the key.

403 permission_error — the key is valid but cannot access the requested model or feature. Check the console for model access. Do not retry.

429 rate_limit_error — you have exceeded your token-per-minute or request-per-minute quota. Retry with backoff after the wait period indicated in the Retry-After header.

529 overloaded_error — Anthropic's servers are temporarily at capacity. This is not your fault. Retry with backoff.

429 versus 529 — different causes, same response

A 429 means you are moving too fast relative to your tier's limits. A 529 means Anthropic's infrastructure is temporarily saturated. Both errors respond to exponential backoff, but the 529 sometimes needs a longer initial wait before the first retry.

Check the response body to confirm which error you have. Both return JSON with {"type": "error", "error": {"type": "rate_limit_error", ...}} or "overloaded_error".

Exponential backoff with jitter

# scripts/retry.py
import anthropic, time, random

client = anthropic.Anthropic()

def call_with_retry(messages, max_retries=3):
    delay = 1.0
    for attempt in range(max_retries + 1):
        try:
            return client.messages.create(
                model="claude-opus-4-5",
                max_tokens=256,
                messages=messages,
            )
        except anthropic.RateLimitError as e:
            if attempt == max_retries:
                raise
            jitter = random.uniform(0, delay * 0.1)
            print(f"429 rate limit, retry {attempt + 1}/{max_retries}, waiting {delay + jitter:.1f}s", flush=True)
            time.sleep(delay + jitter)
            delay *= 2
        except anthropic.APIStatusError as e:
            if e.status_code == 529 and attempt < max_retries:
                jitter = random.uniform(0, delay * 0.1)
                print(f"529 overloaded, retry {attempt + 1}/{max_retries}, waiting {delay + jitter:.1f}s", flush=True)
                time.sleep(delay + jitter)
                delay *= 2
            else:
                raise

result = call_with_retry([{"role": "user", "content": "Say hello."}])
print(result.content[0].text)

The jitter — random.uniform(0, delay * 0.1) — prevents a stampede when many clients fail at the same moment and all wake up at the same time. Without jitter, they hit the API simultaneously and trigger another 429.

Timeouts — the silent failure

The Anthropic SDK does not set a default request timeout. A long-running request can hang indefinitely if the connection is dropped silently. Set a timeout explicitly:

client = anthropic.Anthropic(timeout=30.0)  # seconds

On timeout, the SDK raises anthropic.APITimeoutError. Treat it like a 529 — retryable with backoff. Add it to the except chain:

except anthropic.APITimeoutError:
    if attempt < max_retries:
        time.sleep(delay)
        delay *= 2
    else:
        raise

What errors to surface to users

Never surface raw error messages from the API to end users. Map them: 429 or 529 becomes "The assistant is busy — try again in a moment." A 400 or 401 means something is broken in your backend — return 500 to the caller and log the real error server-side.

If you get stuck

• Your retry loop retries a 400: 400 is not retryable. Add a guard that only retries on 429, 529, and APITimeoutError. Retrying a 400 wastes quota and time.
• The backoff delay is not increasing between retries: confirm delay *= 2 runs after time.sleep, not before. If it is at the wrong indentation level it may reset each iteration.
• APIStatusError is not catching 429: use anthropic.RateLimitError for 429 specifically — it is a subclass with its own except clause. Catching APIStatusError alone will catch it, but then you need to inspect e.status_code to distinguish 429 from other status codes.
• You do not see the Retry-After header: the SDK surfaces it on the exception object. Print e.response.headers.get("retry-after") to read it.

Cross-reference

l5 (Rate limits and the cost of context) explains why you hit 429 — understanding your tier limits gives you the right ceiling to build against. l3 (Streaming) mentions what happens when a streaming connection drops mid-response; wrap streaming calls in the same retry logic.

Deliverable

A call_with_retry function that handles 429 and 529 with exponential backoff and jitter, logs each retry attempt to stderr, and raises after three failed attempts. Test it by temporarily setting a wrong key to trigger 401, and confirm the function does not retry on that error class.

Rate limits and token costs are not independent problems. Understanding one without the other will leave you with surprises in production.

Your tier's limits

Anthropic assigns accounts to tiers based on usage history and prepayment amount. Each tier has two limits: tokens per minute (TPM) and requests per minute (RPM). You can see your current tier at console.anthropic.com under Settings > Limits.

The limits reset every 60 seconds. A single large request can exhaust most of your TPM quota for that window. If your application sends bursts — ten requests simultaneously at startup — you can hit the RPM limit even on a generous tier.

Input tokens are the cost that compounds

Every token in the messages array costs money and counts against your TPM limit — not just the new user message, but every prior turn you include. This is where most production cost overruns originate.

Run this and watch:

# scripts/measure_context.py
import anthropic

client = anthropic.Anthropic()
history = []

topics = [
    "What is an HTTP header?",
    "What is JSON?",
    "What is REST?",
    "What is idempotency?",
    "What is a webhook?",
]

for i, topic in enumerate(topics, 1):
    history.append({"role": "user", "content": topic})
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=256,
        messages=history,
    )
    reply = response.content[0].text
    history.append({"role": "assistant", "content": reply})
    print(f"Turn {i}: input_tokens={response.usage.input_tokens}, output_tokens={response.usage.output_tokens}")

Each turn's input_tokens will be meaningfully larger than the last. The fifth call sends the first four exchanges — question and answer both — plus the fifth question. If the assistant replies verbosely, the compounding accelerates.

Three patterns that keep cost predictable

First: trim history. For most applications, the model does not need every prior turn — it needs the last few. A sliding window is the simplest approach:

MAX_TURNS = 4  # keep last 4 message objects (2 user + 2 assistant)

def trimmed_history(history):
    return history[-MAX_TURNS * 2:]

Add this and re-run measure_context.py. Compare input_tokens at turn 5 before and after. The trimmed version stops compounding after the window fills.

Second: move stable context to prompt caching (covered in l6). If you send the same system prompt and document on every request, caching it costs roughly 10% of the normal input price on cache hits.

Third: set max_tokens to the minimum the use case requires. Output tokens cost more than input tokens on most models — do not request 4096 tokens for a route that produces 100-word answers.

Reading the usage block

The usage object appears in every response:

"usage": {
  "input_tokens": 847,
  "output_tokens": 134
}

Log both fields for every production request. After a week of real traffic, query your logs to find which routes are sending the largest payloads. That is where optimization has the highest return. A route that sends a 2,000-token system prompt on every call and handles 500 requests per day is spending a million input tokens per day on that prefix alone — the same prefix that prompt caching would serve at 10% of the cost.

If you get stuck

• input_tokens is not growing between turns: confirm you are appending the assistant reply to history before the next call. If only user messages accumulate, the count grows but not as fast as when both sides are included.
• TPM limit hit on a single large request: reduce the document size, enable caching for the stable prefix, or request a tier upgrade from the console.
• Cost higher than expected: run the sliding window and compare input_tokens before and after. The difference is almost always the conversation history. Check whether verbose assistant replies are being included in full on every subsequent call.
• usage field missing from streaming response: call get_final_message() after the stream completes to get the finalized usage block.

Cross-reference

l6 (Prompt caching) shows how to attach a cache_control marker to make the stable prefix of every request dramatically cheaper. l4 (Errors, retries, and timeouts) covers 429 rate_limit_error, which is what you will see when TPM or RPM is exhausted.

Deliverable

Run measure_context.py as written, then add the sliding window and run it again. Compare the input_tokens values at turn 5 between the two runs. The trimmed version should show meaningfully fewer input tokens on the later turns.

Further reading

• "The Pipeline Rate-Limits Itself, Or It Eats Itself" — what happens when a multi-step Claude pipeline ignores TPM and stampedes its own quota, and the operator's fix.

Prompt caching is the single highest-leverage cost optimization available in the Anthropic API. It is also frequently misconfigured in ways that look correct but produce zero savings.

What gets cached

Prompt caching stores a prefix of your request on Anthropic's servers. When the next request begins with an identical prefix, the cached tokens are read at roughly 10% of the normal input token price instead of being reprocessed. The cache lives for five minutes — the TTL resets on each cache hit.

The prefix can include the system block, large documents, few-shot examples, or any stable content you send on every request. The marker tells Anthropic where the cacheable prefix ends.

Where to place the cache marker

Add "cache_control": {"type": "ephemeral"} to the last content block in the prefix you want cached. The key word is last — the marker must be at the boundary where the stable content ends and the dynamic content begins.

System prompt with caching:

# scripts/cache_demo.py
import anthropic

client = anthropic.Anthropic()

SYSTEM = [
    {
        "type": "text",
        "text": "You are a technical writing assistant. " * 200,  # artificially long for demo
        "cache_control": {"type": "ephemeral"},
    }
]

def ask(question):
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=256,
        system=SYSTEM,
        messages=[{"role": "user", "content": question}],
    )
    u = response.usage
    print(
        f"input={u.input_tokens} "
        f"cache_create={getattr(u, 'cache_creation_input_tokens', 0)} "
        f"cache_read={getattr(u, 'cache_read_input_tokens', 0)} "
        f"output={u.output_tokens}"
    )
    return response.content[0].text

ask("What is an API?")   # first call — cache miss, creates entry
ask("What is REST?")     # second call — cache hit
ask("What is JSON?")     # third call — cache hit

First call: cache_creation_input_tokens is non-zero, cache_read_input_tokens is zero. Second and third calls: cache_creation_input_tokens is zero, cache_read_input_tokens is the size of the cached prefix.

Why the five-minute TTL matters for your architecture

Cache entries expire after five minutes of inactivity. Applications with consistent, frequent traffic — a support bot handling dozens of requests per minute — maintain warm caches and pay the low cache-read price continuously.

Batch jobs with irregular timing — something that runs once an hour — will cache-miss on every batch run. Caching is still worth enabling because the batch itself may run many requests within five minutes, but the savings model is different. A batch of 50 requests in a two-minute window will hit the cache on requests 2 through 50.

The most common misconfiguration

Placing the cache marker inside dynamic content. If the user's current message is part of the cached prefix, the prefix changes every turn and the cache never hits.

Wrong — marks the user message as the cache boundary:

messages=[{
    "role": "user",
    "content": [{"type": "text", "text": user_input, "cache_control": {"type": "ephemeral"}}]
}]

Every new message produces a different prefix. Zero hits. The marker belongs on the system prompt or the last stable document block, not on the user's message.

Minimum token threshold

Caching only activates when the cached prefix exceeds 1024 tokens for Claude Sonnet and Haiku and 2048 tokens for Claude Opus. A short system prompt will not populate the cache regardless of the marker. In the demo above, the system string is repeated to hit this threshold. In production, a real system prompt plus reference documents will easily clear it.

If you get stuck

• cache_read_input_tokens is always zero: the most likely cause is a prefix shorter than the minimum threshold. Check the token count. Second cause: the prefix is not identical between calls — a timestamp or per-user variable in the cached section breaks the cache key.
• cache_creation_input_tokens is zero on the first call: the prefix may be below the minimum threshold, or the marker is missing. Confirm the system string is long enough by counting tokens using the Anthropic tokenizer or a rough 4-characters-per-token estimate.
• Cost is not dropping after adding caching: confirm you are reading cache_read_input_tokens from the actual response, not from the first call. The savings only appear on the second and subsequent calls within the TTL window.
• Fields not present on the usage object: access them with getattr(u, 'cache_read_input_tokens', 0) as shown above — older SDK versions may not surface them as direct attributes.

Cross-reference

l5 (Rate limits and the cost of context) shows where the input token costs come from. Caching is the primary tool for reducing those costs when the stable prefix is large. l8 (Add it to a real app) is where caching should be enabled — any route that sends a long system prompt or document on every request is a candidate.

Deliverable

Run cache_demo.py. Confirm the second and third calls show cache_read_input_tokens greater than zero. Calculate the percentage savings on input tokens compared to a call without caching.

A script that runs when you execute a file is a draft. A tool reads its input from the caller, reads secrets from the environment, prints only what is needed, and exits with a code the caller can act on.

The full CLI

# scripts/ask.py
import anthropic
import argparse
import os
import sys

def main():
    parser = argparse.ArgumentParser(description="Ask Claude a question.")
    parser.add_argument("--prompt", required=True, help="The user message.")
    parser.add_argument("--system", default="", help="Optional system prompt.")
    parser.add_argument("--max-tokens", type=int, default=512)
    parser.add_argument("--stream", action="store_true", help="Stream output.")
    args = parser.parse_args()

    key = os.getenv("ANTHROPIC_API_KEY")
    if not key:
        print("Error: ANTHROPIC_API_KEY is not set.", file=sys.stderr)
        sys.exit(1)

    client = anthropic.Anthropic()
    messages = [{"role": "user", "content": args.prompt}]

    if args.stream:
        with client.messages.stream(
            model="claude-opus-4-5",
            max_tokens=args.max_tokens,
            system=args.system,
            messages=messages,
        ) as stream:
            for text in stream.text_stream:
                print(text, end="", flush=True)
        print()
    else:
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=args.max_tokens,
            system=args.system,
            messages=messages,
        )
        print(response.content[0].text)

if __name__ == "__main__":
    main()

What each choice does

required=True on --prompt means the parser exits with a readable usage message if the argument is missing — before any network call is made.

The key check runs before the API client is initialized. If the variable is missing, the error is clear and immediate: Error: ANTHROPIC_API_KEY is not set. The sys.exit(1) instead of sys.exit(0) lets a shell script or CI job detect the failure.

print(text, end="", flush=True) in the streaming branch suppresses the newline between chunks so tokens accumulate on one line in the terminal, and flush=True pushes each chunk to the terminal immediately rather than buffering.

Test the exit codes

# Normal operation
python scripts/ask.py --prompt "What is entropy?" --system "Reply in one sentence."
echo $?  # → 0

# Missing key
ANTHROPIC_API_KEY="" python scripts/ask.py --prompt "Hello"
echo $?  # → 1

# Streaming
python scripts/ask.py --prompt "Count to five." --stream

Exit codes matter when this tool is called from a Makefile, a CI job, or another script. A silent 0 on failure silently corrupts the pipeline.

Composing with pipes

Because the tool writes to stdout and exits cleanly, it composes with standard Unix tools:

# Pipe a file into Claude
cat README.md | python scripts/ask.py --prompt "$(cat -)" --system "Summarize this in two sentences."

# Feed the output into another command
python scripts/ask.py --prompt "List five HTTP status codes." | grep "^[0-9]"

# Use it in a Makefile target
# Makefile
.PHONY: changelog
changelog:
    git log --oneline -20 | python scripts/ask.py \
      --prompt "$(shell git log --oneline -20)" \
      --system "Turn these git commits into a changelog entry. Plain text, no markdown." \
      > CHANGELOG_ENTRY.txt

The pipe pattern works because the tool does not print anything to stdout except the Claude response. No "Calling Claude..." prefix, no status lines, no token counts on stdout. Status and debug output go to stderr.

Make it executable

Add a shebang line at the top of the file:

#!/usr/bin/env python3

Then run chmod +x scripts/ask.py. Now ./scripts/ask.py --prompt "Hello" works without the python prefix. Add the scripts/ directory to your PATH and you have a system-wide tool.

If you get stuck

• argparse exits with code 2 on bad input: that is correct behavior. Code 2 conventionally means usage error in Unix tools.
• Streaming output is buffered in a terminal: confirm flush=True is set. You can also run with python -u to disable Python's output buffering globally.
• The shebang line is ignored: the file may not be executable. Run chmod +x scripts/ask.py and confirm with ls -l scripts/ask.py.
• anthropic.Anthropic() ignores the key: the SDK reads ANTHROPIC_API_KEY from the environment automatically. You do not need to pass it to the constructor — just confirm the variable is set before instantiating.
• Pipes truncate output: confirm nothing before the streaming loop is writing to stdout. A stray print() before the loop can break the pipe contract.

Cross-reference

l3 (Streaming vs. blocking) explains the streaming event model the --stream flag builds on. l8 (Add it to a real app) takes the same call and embeds it in an HTTP route with structured output and error handling.

Deliverable

python scripts/ask.py --prompt "What is entropy?" --system "Reply in one sentence." returns a single sentence and exits 0. Running with ANTHROPIC_API_KEY="" exits 1 with a message on stderr and makes no network call. Piping the output into wc -w prints a word count without error.

This lesson has a wider time range than the others because the actual duration depends on what application you have at hand. If you have a running web app, the integration takes 25-35 minutes. If you need to scaffold a new route from scratch in a framework you know well, allow 45 minutes.

The exercise

Pick one route in an application you maintain — a webhook handler, an admin endpoint, an internal tool. Add one Claude-powered feature: a summarizer, a classifier, a one-line draft generator. The constraint is that it must live behind a real HTTP endpoint that accepts user input and returns structured JSON. Not a notebook. Not a script you run by hand.

The route contract

The endpoint accepts a JSON body, calls Claude, and returns:

{
  "result": "non-empty string from Claude",
  "tokens_used": 134,
  "model": "claude-opus-4-5"
}

The tokens_used field is response.usage.input_tokens + response.usage.output_tokens. This forces you to read the usage block on every call — good practice that pays off during cost debugging.

A minimal Flask example

# app.py
from flask import Flask, request, jsonify
import anthropic
import os

app = Flask(__name__)
client = anthropic.Anthropic()

@app.route("/summarize", methods=["POST"])
def summarize():
    body = request.get_json(silent=True)
    if not body or not body.get("text"):
        return jsonify({"error": "missing 'text' field"}), 400

    text = body["text"]
    if len(text) > 10_000:
        return jsonify({"error": "text too long"}), 400

    try:
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=256,
            system="Summarize the following text in two sentences. Plain text only, no markdown.",
            messages=[{"role": "user", "content": text}],
        )
    except anthropic.APIStatusError as e:
        return jsonify({"error": "upstream API error", "code": e.status_code}), 502

    return jsonify({
        "result": response.content[0].text,
        "tokens_used": response.usage.input_tokens + response.usage.output_tokens,
        "model": response.model,
    })

if __name__ == "__main__":
    app.run(port=8000)

The one test

# tests/test_summarize.py
import pytest
from app import app as flask_app

@pytest.fixture
def client():
    flask_app.config["TESTING"] = True
    with flask_app.test_client() as c:
        yield c

def test_summarize_returns_result(client):
    response = client.post(
        "/summarize",
        json={"text": "The Anthropic API is an HTTP interface. It accepts JSON and returns JSON."},
        content_type="application/json",
    )
    assert response.status_code == 200
    data = response.get_json()
    assert isinstance(data["result"], str) and len(data["result"]) > 0
    assert isinstance(data["tokens_used"], int) and data["tokens_used"] > 0

def test_missing_text_returns_400(client):
    response = client.post("/summarize", json={})
    assert response.status_code == 400

The test does not assert what Claude said — that would be brittle because the model's exact wording varies. It asserts the shape: result is a non-empty string, tokens_used is a positive integer. The test fails if the route is broken, the API key is missing from the environment, or the response parsing is wrong.

Handling the API error path correctly

The except anthropic.APIStatusError block returns a 502 — "bad gateway" — rather than a 500. The distinction matters: 500 conventionally means your code is broken; 502 means an upstream dependency failed. Callers, monitoring, and on-call alerts treat these differently.

Do not let the raw Anthropic error message reach the HTTP response body in production. Map it to a generic message and log the real error server-side.

Enable caching before you ship

If your route sends a long system prompt on every request, add the cache marker before you deploy:

system=[{
    "type": "text",
    "text": "Your system prompt here...",
    "cache_control": {"type": "ephemeral"},
}]

A route handling ten requests per minute pays the cache-creation price once every five minutes and the cache-read price on the remaining requests. At scale this is significant.

If you get stuck

• ANTHROPIC_API_KEY not available in tests: the test runner must have the variable in its environment. For local runs, export ANTHROPIC_API_KEY=... before running pytest. For CI, add it as a repository secret.
• The test passes locally but fails in CI: the key is likely missing in the CI environment. Check the CI secret configuration.
• 502 in production: anthropic.APIStatusError was raised. Log e.status_code and e.message server-side to see which error type triggered it.
• Response body is truncated: max_tokens is too low for the input length. Increase it or add a check that rejects oversized input before calling the API (as the length check in the example does).
• Test is brittle because it asserts on exact Claude output: remove the string assertion and replace it with len(data["result"]) > 0. The test is checking your code, not the model.

Cross-reference

l6 (Prompt caching) shows the exact syntax for the cache marker — enable it before this route goes to production. l4 (Errors, retries, and timeouts) covers the retry logic you may want to wrap around client.messages.create if this endpoint handles high-value requests where transient failures should not surface to users.

Deliverable

A route behind a real HTTP endpoint with a passing test that runs against the live Anthropic API using ANTHROPIC_API_KEY from the environment. The test asserts shape, not exact text. A missing key produces a clear error before any network call is attempted.

Further reading

• "What Agents Actually Need From Your API" — what the response shape, error codes, and operationIds need to look like when the caller is a model rather than a human.

Your authenticated endpoint works for users who are logged in. It does nothing for the AI agent that has never visited your site, has no token, and cannot get one — because you cannot pre-share a token with every prospect's AI client. The conceptual layer is covered in "Closing the Agent Discovery Loop"; this lesson is the builder's guide.

Why auth-only fails

The failure mode: you have a route that returns your membership catalog. It checks for a valid token. A prospective customer asks their AI assistant what plans your site offers. The agent hits your site with no credentials. Your route returns 401. The agent reports back: nothing found.

Auth did its job. That is the problem.

The less obvious failure comes later, when you add a public endpoint and protect it only with an auth toggle. You route anonymous callers to a different handler, but the handler calls the same data-fetching function and the same formatter as the admin endpoint. The formatter knows everything — member counts, access rules, internal notes. The toggle says "skip the auth check." The formatter still runs in full. One misconfigured flag and you are returning admin-shaped data to an anonymous caller.

The three layers

The fix has three parts. All three must be present for the loop to close:

1. A public read-only endpoint. No token required. Per-IP rate limited (60 req/min is a reasonable floor). Cache-Control: public, max-age=300 so CDNs and agents can cache it. Expose only the catalog reads you want public — in MemberPress AI Foundation that was four tools: list_memberships, get_membership, list_courses, get_course.

2. A public-safe formatter. This is the layer most builders skip, and it is the load-bearing one. The formatter strips dangerous fields — member counts, access rules, custom meta — and adds the fields that are safe and useful to a prospective caller, like the signup URL. It does this as a separate code path, not as a conditional inside the existing formatter.

3. A /.well-known/mcp.json discovery file. When the public catalog is enabled, a GET to that path returns the endpoint URL and metadata. When disabled, it returns 404 — not an empty object, a 404. The endpoint's existence should not be discoverable when you have not turned it on.

Walk-through

Think of the formatter like a security door, not a locked cabinet. A locked cabinet relies on one control point — the lock. If the lock fails, everything inside is exposed. A security door is structural: even if you prop it open, the things behind it are still in separate rooms. The public_detail() formatter is not a toggle on the existing formatter; it is a different room. The fields it returns are listed explicitly. Admin fields are not present in the list, so they cannot appear in the output, regardless of what the auth check does upstream.

# api/formatters.py  (illustrative — adapt to your ORM/serializer)
class MembershipFormatter:

    def admin_detail(self, obj):
        return {
            "id": obj.id,
            "title": obj.title,
            "price": obj.price,
            "member_count": obj.member_count,       # internal
            "access_rules": obj.access_rules,       # internal
            "custom_meta": obj.custom_meta,         # internal
            "signup_url": obj.signup_url,
        }

    def public_detail(self, obj):
        # Fields are enumerated explicitly.
        # Omission is the defense — admin fields are not present in this method.
        return {
            "id": obj.id,
            "title": obj.title,
            "price": obj.price,
            "signup_url": obj.signup_url,
        }

The public endpoint calls public_detail() directly. It does not call admin_detail() and strip fields after the fact. Stripping after the fact is the wrong pattern — a new field added to admin_detail() is live in the public response until someone remembers to add it to the strip list.

# api/routes.py  (illustrative)
import os
from flask import Flask, jsonify, abort, make_response

app = Flask(__name__)
formatter = MembershipFormatter()

@app.route("/.well-known/mcp.json")
def well_known_mcp():
    if not os.getenv("PUBLIC_CATALOG_ENABLED"):
        abort(404)
    payload = {
        "name": "Example Site",
        "protocol": "mcp",
        "version": "2025-06-18",
        "endpoints": {
            "public": "https://example.com/api/v1/public-mcp"
        },
        "auth_required": False,
        "transport": "http",
    }
    response = make_response(jsonify(payload))
    response.headers["Cache-Control"] = "public, max-age=300"
    return response

@app.route("/api/v1/public-mcp/memberships")
def public_memberships():
    memberships = Membership.query.filter_by(public=True).all()
    return jsonify([formatter.public_detail(m) for m in memberships])

The .well-known JSON shape above is illustrative. The MCP spec is moving; check the current draft before shipping. The field names, version string, and transport values have all changed between drafts and will change again.

If you get stuck

• You return an empty object from /.well-known/mcp.json when the feature is off. An agent that receives {} may cache it or report "no tools available" instead of treating the endpoint as absent. Return 404 so agents know the endpoint does not exist in this state.
• Your public_detail() calls the same internal query as admin_detail(). The formatter strips fields at the output layer, but the query itself still fetches restricted data and runs any access-rule logic. Make the public query explicit — only select the columns you intend to expose.
• Credentials appear in the .well-known payload. The payload is public by definition. Any token, key, or internal URL in that payload is now public. The auth_required: false field signals that no credentials are needed — leave all credential material out of the discovery file.

Cross-reference

l8 (Add it to a real app) is the route this lesson extends — the public endpoint follows the same Flask pattern with the same 502 error handling convention. The conceptual framing for why agents need this shape is in "Closing the Agent Discovery Loop".

Deliverable

Two observable checks. First: curl https://yoursite.com/.well-known/mcp.json returns a JSON payload with the public endpoint URL when PUBLIC_CATALOG_ENABLED is set, and returns 404 when it is unset. Second: compare the JSON output of your public endpoint against your admin endpoint for the same membership object. The public response must be missing at least member_count, access_rules, and custom_meta — run diff on the two outputs and confirm those keys are absent.

Further reading

• "Closing the Agent Discovery Loop" — the conceptual layer: why the three-part loop exists, what breaks when any part is missing, and the six-point checklist for any commerce-adjacent service.
• "What Agents Actually Need From Your API" — the caller's perspective: what response shape, error codes, and operationIds an agent caller needs to work reliably with your endpoint.