Using Sonnet 5 in production: a Claude API guide

A production-grade guide to calling Claude Sonnet 5 - streaming, retry and backoff, error handling, prompt caching, batch jobs, and keeping token spend predictable.

Putting Claude Sonnet 5 into production takes about eight things the quickstart leaves out: streaming with real event handling, retries with exponential backoff and jitter, idempotency, structured output through tool use, token budgeting before the call, caching a long system prompt, batching overnight work, and logging enough to explain the bill afterwards. This page is those eight, in runnable Python.

It assumes you already have a key and a first successful request. If you do not, start at the Claude API reference, which covers keys, the rate card, rate-limit tiers and the parameters that now return a 400. Everything here uses claude-sonnet-5 and the official anthropic SDK, on the 1M-token context window that is the default on Sonnet 5.

#How do you stream a Sonnet 5 response properly?

Use the SDK's streaming context manager rather than assembling raw SSE by hand. It reconstructs the final message for you while giving you deltas as they arrive, which is what you want for both a user-facing UI and a long generation that would otherwise sit behind a proxy timeout.

import anthropic

client = anthropic.Anthropic()

def stream_answer(prompt: str) -> anthropic.types.Message:
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=4096,
        system="You are a precise technical editor. Never invent citations.",
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for event in stream:
            if event.type == "content_block_start":
                if event.content_block.type == "thinking":
                    print("[thinking]", end="", flush=True)
            elif event.type == "content_block_delta":
                if event.delta.type == "text_delta":
                    print(event.delta.text, end="", flush=True)
            elif event.type == "message_delta":
                # stop_reason and final output token count arrive here
                pass
        return stream.get_final_message()

Three things to handle that people skip. First, thinking blocks come down the same stream - Sonnet 5 thinks by default, so filter on block type instead of assuming every delta is user-visible text. Second, stop_reason only becomes final at the end; read it from get_final_message(), not from a mid-stream event. Third, a dropped connection mid-stream leaves you with a partial answer that you have already been billed for - so persist the partial text before retrying, and decide deliberately whether a retry re-asks the whole question or resumes from what you have.

If you only need the text and nothing else, stream.text_stream yields exactly the text deltas and nothing else. The verbose loop above is for when you need to distinguish thinking, tool calls and text.

#How should you retry Claude API errors?

Retry only the statuses that can succeed on a second attempt, back off exponentially, and add jitter so that a fleet of workers does not synchronise into a thundering herd. Honour retry-after when the server sends it - your own backoff curve is a guess, and the header is not.

import random, time
import anthropic
from anthropic import APIStatusError, APIConnectionError

client = anthropic.Anthropic(max_retries=0)  # we handle retries ourselves

RETRYABLE = {429, 500, 529}

def call_with_backoff(*, max_attempts=6, base=1.0, cap=60.0, **kwargs):
    for attempt in range(max_attempts):
        try:
            return client.messages.create(model="claude-sonnet-5", **kwargs)

        except APIStatusError as err:
            status = err.status_code
            if status not in RETRYABLE or attempt == max_attempts - 1:
                raise

            retry_after = err.response.headers.get("retry-after")
            if retry_after:
                delay = float(retry_after)
            elif status == 429 and "enforced_spend_limit_reached" in str(err):
                raise  # a spend cap will not clear on its own
            else:
                # full jitter: uniform over [0, min(cap, base * 2**attempt)]
                delay = random.uniform(0, min(cap, base * (2 ** attempt)))

            time.sleep(delay)

        except APIConnectionError:
            if attempt == max_attempts - 1:
                raise
            time.sleep(random.uniform(0, min(cap, base * (2 ** attempt))))

    raise RuntimeError("unreachable")

Full jitter - a uniform draw over the whole window rather than a fixed delay plus noise - is what actually de-synchronises workers. Cap the delay so a 529 storm does not park a request for twenty minutes. And keep max_attempts small: six attempts with a 60-second cap is already two or three minutes of waiting, which is longer than most callers will tolerate.

The SDKs do retry automatically by default. That is fine for a script; for a service, set max_retries=0 and own the policy, because the default has no idea about your queue depth, your deadline, or the fact that this particular request has already been paid for once.

#Idempotency and request IDs

The Messages API is not idempotent - the same body sent twice generates twice and bills twice. A retry after a connection error, where the response may or may not have been produced, is the dangerous case. Two habits fix it:

• Key the work, not the call. Derive a stable job ID from your input (a hash of the document, the ticket ID, the row primary key), write the result under that key, and check for an existing result before calling. Deduplication belongs in your storage layer, not in the HTTP client.
• Log the request ID. Every response carries one, exposed as message._request_id in the Python SDK and on err.request_id for errors. It is the only identifier Anthropic support can trace, and without it a report of "some requests failed yesterday" is unactionable.

msg = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "..."}],
)
log.info("claude_ok", extra={"request_id": msg._request_id, "job_id": job_id})

#How do you get reliable structured output from Sonnet 5?

Through tool use, with tool_choice forcing the tool. The old approach - prefilling the assistant turn with an opening brace so the model had to continue in JSON - returns a 400 on Sonnet 5, along with every other 4.6-and-later model. Nor can you fall back on temperature: 0; that parameter is gone too.

EXTRACT_TOOL = {
    "name": "record_invoice",
    "description": "Record the structured fields extracted from an invoice.",
    "input_schema": {
        "type": "object",
        "properties": {
            "invoice_number": {"type": "string"},
            "issued_on": {"type": "string", "description": "ISO 8601 date"},
            "total_cents": {"type": "integer"},
            "currency": {"type": "string", "enum": ["USD", "EUR", "GBP"]},
            "line_items": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "description": {"type": "string"},
                        "amount_cents": {"type": "integer"},
                    },
                    "required": ["description", "amount_cents"],
                },
            },
        },
        "required": ["invoice_number", "issued_on", "total_cents", "currency"],
    },
}

def extract(document_text: str) -> dict:
    msg = call_with_backoff(
        max_tokens=2048,
        tools=[EXTRACT_TOOL],
        tool_choice={"type": "tool", "name": "record_invoice"},
        messages=[{"role": "user", "content": document_text}],
    )
    for block in msg.content:
        if block.type == "tool_use" and block.name == "record_invoice":
            return block.input
    raise ValueError(f"no tool_use block; stop_reason={msg.stop_reason}")

Forcing the tool guarantees the shape but does not guarantee the content - an enum still needs validating, and a required field can come back empty. Run the result through a schema validator such as Pydantic and treat a validation failure as a retryable application error, not an API error. Also budget max_tokens generously: a truncated tool call arrives as stop_reason: "max_tokens" with unparseable JSON, and that failure looks exactly like a model error while being entirely your fault. Prompt-side technique for extraction tasks is on the prompt engineering page.

#How do you count tokens before you spend them?

Use the token-counting endpoint. It takes the same request shape as messages.create and returns the input token count without generating anything, so you can reject an oversized job, pick a model, or check a budget before you commit.

PRICE = {  # USD per token
    "claude-sonnet-5": {"in": 2 / 1e6, "out": 10 / 1e6},
    "claude-opus-5":   {"in": 5 / 1e6, "out": 25 / 1e6},
    "claude-haiku-4-5": {"in": 1 / 1e6, "out": 5 / 1e6},
}

def estimate_cost(messages, system, model="claude-sonnet-5", expected_output=1500):
    counted = client.messages.count_tokens(
        model=model, system=system, messages=messages,
    )
    p = PRICE[model]
    return {
        "input_tokens": counted.input_tokens,
        "usd": counted.input_tokens * p["in"] + expected_output * p["out"],
    }

This matters more on Sonnet 5 than it did on earlier models. Sonnet 5 uses the tokenizer introduced with Opus 4.7, which produces roughly 30% more tokens for the same English text than the one in Sonnet 4.6. Any max_tokens value, context guard or cost model carried over from a 4.6 deployment is now wrong in the same direction. Recount before you migrate, and remember that thinking tokens bill as output and count against max_tokens - so lowering effort to low on simple classification work is a real cost lever, not a tuning nicety. How the effort ladder maps onto reasoning depth is covered under adaptive and extended thinking. If a cheaper model would do the job - or if you need to escalate to Opus 5 for the hard 5% - the model comparison puts the tiers side by side.

#How do you cache a long system prompt across many requests?

Pass system as a list of content blocks and put a cache_control breakpoint on the last stable block. Everything before the breakpoint is cached; everything after it is fresh input. The prefix must be byte-identical on every request and at least 1,024 tokens long on Sonnet 5, otherwise the cache silently never fires.

SYSTEM = [
    {"type": "text", "text": POLICY_HANDBOOK},          # ~18k tokens, stable
    {"type": "text", "text": CLASSIFICATION_TAXONOMY,   # ~4k tokens, stable
     "cache_control": {"type": "ephemeral", "ttl": "1h"}},
]

def triage(ticket_text: str):
    msg = call_with_backoff(
        max_tokens=1024,
        system=SYSTEM,
        messages=[{"role": "user", "content": ticket_text}],  # volatile, not cached
    )
    u = msg.usage
    log.info(
        "triage",
        extra={
            "cache_write": u.cache_creation_input_tokens,
            "cache_read": u.cache_read_input_tokens,
            "fresh_in": u.input_tokens,
            "out": u.output_tokens,
        },
    )
    return msg

The arithmetic on a 22,000-token prefix at Sonnet 5 rates: uncached it costs $0.044 per request. A one-hour cache write costs $0.088 once, and each read costs $0.0044 - so the write pays for itself on the second read, and 1,000 requests drop from $44 to about $4.50. There is a second, less obvious win: cache reads do not count toward your input-tokens-per-minute limit, so caching buys throughput headroom as well as money.

Two failure modes to watch. Anything volatile above the breakpoint - a timestamp, a user name, a request ID - invalidates every hit; keep volatile content in the user turn. And a five-minute TTL is refreshed by each read, so it survives steady traffic but not a quiet Sunday; use the one-hour TTL when traffic is bursty or when the prefix feeds a batch job.

#How do you run an overnight batch job end to end?

Batch usage is half price, results are retained for 29 days, and one batch takes up to 100,000 requests or 256 MB. Nothing that runs while a human waits belongs here; everything else probably does.

from anthropic.types.messages.batch_create_params import Request

requests = [
    Request(
        custom_id=f"doc-{doc.id}",          # your key, echoed back on the result
        params={
            "model": "claude-sonnet-5",
            "max_tokens": 2048,
            "system": SYSTEM,               # 1h cache TTL - batches outlive 5 minutes
            "messages": [{"role": "user", "content": doc.text}],
        },
    )
    for doc in documents
]

batch = client.messages.batches.create(requests=requests)
print(batch.id, batch.processing_status)

Then poll, and stream the results rather than loading them into memory:

import time

while True:
    batch = client.messages.batches.retrieve(batch.id)
    if batch.processing_status == "ended":
        break
    time.sleep(60)

for result in client.messages.batches.results(batch.id):
    kind = result.result.type            # succeeded | errored | canceled | expired
    if kind == "succeeded":
        store(result.custom_id, result.result.message)
    elif kind == "errored":
        requeue(result.custom_id, result.result.error)
    else:
        log.warning("batch_%s", kind, extra={"custom_id": result.custom_id})

Four rules that save a rerun. Make custom_id a stable business key, because it is the only thing tying a result back to its input. Handle all four result statuses - expired is not an error and is not billed, but it does mean the work did not happen. Do not set stream: true inside a batch request; it is rejected. And if a single generation needs more than 128,000 output tokens, the output-300k-2026-03-24 beta header raises the batch ceiling to 300,000 on Sonnet 5, at standard batch pricing - expect a generation that size to take over an hour. Scheduling batches around other automated work is covered on the automation page.

#What should you log on every Claude call?

Enough to answer three questions after the fact: what did it cost, why did it stop, and which request was it. Anthropic's usage dashboard aggregates; it will not tell you which of your code paths caused a spike.

Alert on the ratio of cache_creation_input_tokens to cache_read_input_tokens. If writes stop being rare, someone has edited the system prompt into something volatile and your bill has quietly gone up tenfold with no error to show for it. Watch stop_reason distribution too: a rise in max_tokens means answers are being cut off mid-sentence, and a rise in model_context_window_exceeded means inputs have outgrown their guard.

For agent-shaped workloads where the loop itself is the thing you are operating, the trade-offs shift - see MCP and Claude agents for tool orchestration, and Claude for developers if the job is a coding workflow rather than a service. Rate-limit tiers, server-tool pricing and platform differences are all on the API reference, and the per-token rates sit alongside subscription costs on the pricing page.

#Frequently asked questions