AI & ML / Inference & serving / 11_claude_messages_api_in_practice.md

The Claude Messages API in practice

Updated 4 interview angles 8 min read source
On this page11
  1. One endpoint, one shape
  2. Prompt design that survives production
  3. Structured outputs
  4. Tool use: the loop is yours
  5. Streaming
  6. Prompt caching
  7. Batches and token counting
  8. Errors, retries, concurrency
  9. Where it breaks in production
  10. Related
  11. Interview angle

The Claude Messages API in practice

Job ads that say “LLM APIs, in particular Anthropic’s Claude (prompt design, structured outputs, tool/function calling)” are asking for one endpoint known cold: what a request looks like, what comes back, how the tool loop and the JSON guarantee work, and the operational habits — caching, streaming, retries, batches, token counting — that separate a demo from a service.

The chat API surface is the provider-neutral view; this is the Anthropic surface, Python SDK.

One endpoint, one shape

python 
import anthropic

client = anthropic.Anthropic()          # key from ANTHROPIC_API_KEY, never code

resp = client.messages.create(
    model="claude-opus-5",
    max_tokens=16000,                   # a cap, not a target - do not lowball
    system="You are a support assistant for Acme. Answer from the docs given.",
    messages=[
        {"role": "user", "content": "How do I rotate an API key?"},
    ],
)
for block in resp.content:              # a list of typed blocks, never a str
    if block.type == "text":
        print(block.text)
print(resp.stop_reason, resp.usage.input_tokens, resp.usage.output_tokens)

Everything is POST /v1/messages. Tools, images, PDFs, structured output and thinking are parameters of this call, not other endpoints. Three things to know about the shape:

  • system is separate from messages. Operator instructions go in system; messages alternates user and assistant, starts with user, and each message’s content is a string or a list of blocks (text, image, document, tool_use, tool_result, thinking).
  • The API is stateless. Every request carries the whole history. That is your responsibility, and the mistake that follows is appending only the text of a reply — dropping tool_use and thinking blocks that the next turn needs. Append resp.content, the whole list.
  • stop_reason is the control flow: end_turn, max_tokens (you truncated it — raise the cap), tool_use (run the tools), refusal (check stop_details), pause_turn (re-send to continue).

Prompt design that survives production

  • Stable first, variable last. Instructions, examples and reference documents in system; the user’s turn at the end. This is both the clearer prompt and the cacheable one (below).
  • Structure the input, not just the instructions. Wrap documents and user data in labelled tags — <document>, <ticket> — so the model can tell content from instruction; that is also half of your injection defence (Prompt injection).
  • Say what the output is for, and give one worked example of the shape when the shape matters. Do not ask for JSON in prose when structured output exists.
  • Assistant prefill is gone on current models (a 400); format control is output_config, not a half-written answer.
  • Thinking is on by default on the current Opus; thinking={"type": "adaptive"} makes it explicit and output_config={"effort": "low"} turns spend down for simple routes. The old budget_tokens is rejected on current models.

The prompt-engineering craft — few-shot, decomposition, evaluation — is in Prompt Engineering.

Structured outputs

python 
from pydantic import BaseModel

class Ticket(BaseModel):
    category: str
    priority: int
    needs_human: bool
    summary: str

resp = client.messages.parse(              # SDK helper: schema in, model out
    model="claude-opus-5",
    max_tokens=16000,
    messages=[{"role": "user", "content": f"<ticket>{text}</ticket>"}],
    output_format=Ticket,
)
ticket: Ticket = resp.parsed_output

# The raw form on create(): output_config, not the deprecated output_format
resp = client.messages.create(
    ..., output_config={"format": {"type": "json_schema",
                                   "schema": Ticket.model_json_schema()}},
)

The guarantee is that the text block parses and validates against the schema — a grammar constraint at decode time, not a strong tendency. What it does not guarantee is that the values are true, or that constraints the grammar cannot express (ranges, formats, cross-field rules) hold; run Pydantic after.

Schemas need additionalProperties: false and every property in required; express optionality as a union with null. The mechanism and its limits: Function calling and structured output.

Tool use: the loop is yours

python 
TOOLS = [{
    "name": "get_order",
    "description": "Look up an order by its ID. Use when the user cites one.",
    "input_schema": {"type": "object",
                     "properties": {"order_id": {"type": "string"}},
                     "required": ["order_id"], "additionalProperties": False},
    "strict": True,                          # arguments validate exactly
}]

messages = [{"role": "user", "content": "Where is order A-1043?"}]
while True:
    resp = client.messages.create(model="claude-opus-5", max_tokens=16000,
                                  tools=TOOLS, messages=messages)
    messages.append({"role": "assistant", "content": resp.content})
    if resp.stop_reason != "tool_use":
        break
    results = []
    for block in resp.content:
        if block.type == "tool_use":         # block.input is already a dict
            try:
                out = run_tool(block.name, block.input)
                results.append({"type": "tool_result",
                                "tool_use_id": block.id, "content": out})
            except ToolError as e:
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": str(e), "is_error": True})
    messages.append({"role": "user", "content": results})   # ALL results, one message

What the model emits

The model does not call your function. It emits a tool_use block — an id, a name, an input dict — and stops; you run the code, append the assistant turn as-is, and reply with a user message of tool_result blocks whose tool_use_ids match.

Two rules that are easy to break: a turn may contain several tool_use blocks and every result goes back in one user message (splitting them teaches the model to stop parallelising); and a failed tool returns is_error: true, never nothing. tool_choice can force a tool where the model still allows it; for most code {"type": "auto"} plus a clear description is the right answer.

The SDK’s beta tool runner (@beta_tool on a typed function, then client.beta.messages.tool_runner(...)) writes this loop for you and generates the schema from the docstring — a Google-style docstring, which is also what the ad’s documentation line means. The loop above is what to be able to draw on a whiteboard.

Streaming

python 
aclient = anthropic.AsyncAnthropic()

async def relay(ws, messages):               # e.g. inside a FastAPI websocket
    async with aclient.messages.stream(model="claude-opus-5", max_tokens=64000,
                                       messages=messages) as stream:
        async for text in stream.text_stream:   # text deltas as they arrive
            await ws.send_text(text)
        return stream.get_final_message()       # usage, stop_reason, blocks

Streaming changes the felt latency, not the total — the first token arrives in under a second instead of after the whole answer. The events are message_start, content_block_start, content_block_delta (with text_delta, thinking_delta or input_json_delta for tool arguments), content_block_stop, message_delta (carries stop_reason and usage), message_stop. Use text_stream unless you need the events.

The SDK refuses a non-streaming call whose max_tokens implies a

10-minute response, so long generations stream by construction. Relaying to a browser is an SSE or WebSocket problem: Server-Sent Events (SSE).

Prompt caching

python 
resp = client.messages.create(
    model="claude-opus-5", max_tokens=16000,
    system=[{"type": "text", "text": BIG_STABLE_PROMPT,
             "cache_control": {"type": "ephemeral"}}],   # 5 min; "ttl": "1h"
    messages=messages,
)
resp.usage.cache_read_input_tokens        # zero every time => something varies

Caching is a prefix match over tools → system → messages; one changed byte before the breakpoint invalidates it. So: no timestamp, user ID or unsorted json.dumps in the system prompt, a fixed tool list in a fixed order, and dynamic context in the user turn.

Up to four breakpoints; prefixes shorter than the model’s minimum (hundreds to a few thousand tokens) silently do not cache. Cache reads cost about a tenth of an uncached token, which on a long system prompt is most of the bill.

Batches and token counting

python 
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

batch = client.messages.batches.create(requests=[
    Request(custom_id=doc.id, params=MessageCreateParamsNonStreaming(
        model="claude-opus-5", max_tokens=4000, messages=[...]))
    for doc in docs])
# poll client.messages.batches.retrieve(batch.id).processing_status == "ended"
for r in client.messages.batches.results(batch.id):   # any order: key by id
    if r.result.type == "succeeded":
        store(r.custom_id, r.result.message)

n = client.messages.count_tokens(model="claude-opus-5", system=system,
                                 messages=messages).input_tokens

Batches run asynchronously at half price, up to 100,000 requests, usually within the hour — the right shape for nightly classification, backfills and evals. count_tokens is the only correct tokenizer; tiktoken is a different model’s and undercounts.

Errors, retries, concurrency

python 
try:
    resp = client.with_options(timeout=30.0, max_retries=3).messages.create(...)
except anthropic.RateLimitError as e:          # 429: honour retry-after
    wait = int(e.response.headers.get("retry-after", "10"))
except anthropic.APIStatusError as e:          # other non-2xx; 529 = overloaded
    log.error("claude %s %s req=%s", e.status_code, e.message, e.request_id)
except anthropic.APIConnectionError:           # never got a response
    ...

The SDK already retries 408, 409, 429 and 5xx with backoff (max_retries default 2); a hand-rolled loop on top double-retries. Catch a chain, most specific first — a single broad except loses the difference between “back off” and “fix the request”. For throughput use anthropic.AsyncAnthropic with asyncio.gather under a semaphore sized to your rate limit, and log resp._request_id on every failure you report.

Where it breaks in production

  • History rebuilt from text. Tool-use and thinking blocks dropped on replay; the next turn cannot see what it called. Append resp.content.
  • max_tokens set to 500 “to save money”. The answer truncates with stop_reason == "max_tokens" and the retry costs more than the cap saved.
  • A date in the system prompt. Zero cache hits, full price on a 6k-token prefix, and nobody notices for a quarter.
  • Model IDs from memory. A date-suffixed or retired ID is a 404 in production; pin the exact current string from the models list, in config.
  • Secrets and content in logs. The API key in a repo, and user conversations in the logging platform at full fidelity. Env vars, and redact before logging.
  • One broad except. 400s retried, 429s not backed off, and the on-call cannot tell a bad request from an outage.

Interview angle 4

  • “Walk me through a tool call with Claude.” - the model returns stop_reason: tool_use with tool_use blocks (id, name, input); I run the code, append the assistant content unchanged, and reply with a user message holding one tool_result per call, matched by tool_use_id, errors flagged with is_error. Loop until end_turn. Follow-up: what if it calls three tools at once? Run them concurrently and return all three results in the same user message.
  • “How do you get reliable JSON out of it?” - output_config.format with a JSON Schema (or messages.parse with a Pydantic model): the output is grammar-constrained to parse. Then Pydantic for ranges and cross-field rules, because the grammar guarantees shape, not truth.
  • “Your token bill doubled - where do you look first?” - cache hits. usage.cache_read_input_tokens at zero means the prefix changes per request; the usual culprit is a timestamp or user ID in the system prompt. Fix the ordering, put a breakpoint on the stable block, and the long prefix costs a tenth.
  • “How do you handle rate limits?” - the SDK retries 429 and 5xx with backoff already; I set max_retries and a timeout per call, honour retry-after when I do retry myself, bound concurrency with a semaphore on the async client, and move anything not latency-sensitive to the Batches API at half price.
Previous
Determinism and reproducible LLM tests