Hosted APIs in production

Calling a hosted model API from a script is easy. Running it behind a feature that thousands of people use every day is a different job: requests time out, you hit rate limits during a traffic spike, the provider has a bad hour, a prompt change quietly makes answers worse, and the bill doubles without anyone noticing. This lesson covers the engineering that turns an API call into a dependable production dependency: where the call lives, how to handle errors and retries, staying inside rate limits, degrading gracefully, serving streamed responses, and the observability and release practices that let you operate it. (The alternative path in this roadmap covers running open models on your own GPUs instead.)

#Put every call behind one gateway

Two architectural rules come first.

Call the model only from your backend. An API key in a mobile app or browser JavaScript is public, and anyone can extract it and spend your money. The client talks to your server; your server authenticates the user, applies limits and calls the model.

Route every call through one module. Instead of client.messages.create scattered across your codebase, write a small internal gateway, llm.py, that every feature uses. It is the one place to set timeouts, handle errors, fall back, log usage and compute cost. When you need to add a budget check or change a model, you change one file.

Configuration belongs in code and environment, not in each call site:

import anthropic

# Reads ANTHROPIC_API_KEY from the environment. Never hard-code keys.
client = anthropic.Anthropic(
    timeout=anthropic.Timeout(60.0, connect=5.0),  # the default is 10 minutes: far too long for a web request
    max_retries=2,                                 # the default; retries use exponential backoff
)

MODELS = {
    "support_chat": "claude-sonnet-5-5",
    "ticket_tagging": "claude-haiku-5-5",
    "contract_review": "claude-opus-5-5",
}

Model aliases such as claude-sonnet-5-5 carry no date suffix. Treat a model change like any dependency upgrade: change the configuration, run your evals, then deploy.

#Errors, retries and timeouts

The API returns standard HTTP status codes, and the Python SDK raises a typed exception for each:

The SDK already retries connection errors, 408, 409, 429 and 5xx responses with exponential backoff (twice by default). You configure that with max_retries, per client or per call with client.with_options(max_retries=5). Do not wrap it in a second retry loop of your own, or one failing request becomes nine.

Set timeouts deliberately. The default request timeout is 10 minutes, which suits long batch jobs but not a web request where the user gives up after 30 seconds. Use per-call overrides such as client.with_options(timeout=20.0) for latency-sensitive paths. For long outputs, stream (covered below): the SDK refuses non-streaming requests that it estimates could exceed about ten minutes.

A successful HTTP response is not necessarily a usable answer. Always check stop_reason before reading content:

• "end_turn": finished normally.
• "max_tokens": cut off. Treat as a failure or continue; do not show half an answer as if complete.
• "refusal": the model declined for safety reasons. Show a graceful message.
• "tool_use": the model wants a tool run (agent and tool-use flows).

Every response also has response._request_id. Log it: it is what the provider's support team needs to investigate a specific failure.

#Staying inside rate limits

Your organisation has limits per model on requests per minute and on input and output tokens per minute. Exceed them and you get 429s; the SDK backs off and retries, but under sustained overload requests start failing. Rate-limit headers on every response tell you how much of your limit remains, and a 429 includes retry-after with the number of seconds to wait.

The best defence is not to burst in the first place. Bound concurrency in your own code:

import asyncio
import anthropic

aclient = anthropic.AsyncAnthropic(max_retries=4)
in_flight = asyncio.Semaphore(8)   # at most 8 concurrent requests from this process

async def summarise(doc: str) -> str:
    async with in_flight:
        response = await aclient.messages.create(
            model="claude-haiku-5-5",
            max_tokens=1024,
            output_config={"effort": "low"},
            messages=[{"role": "user", "content": f"Summarise in two sentences:\n\n{doc}"}],
        )
    return "".join(b.text for b in response.content if b.type == "text")

async def summarise_all(docs: list[str]) -> list[str]:
    return await asyncio.gather(*(summarise(d) for d in docs))

Other measures, by workload:

• Background jobs go through a queue with a fixed number of workers, so a backlog waits in the queue instead of hammering the API. Work nobody waits for can use the Message Batches API, which has its own limits and costs half as much.
• Prompt caching helps twice: cached input tokens are cheaper, and on most models cache reads do not count toward input-token rate limits.
• Per-user limits in your own app stop one heavy user consuming the organisation's whole budget.
• Request a higher limit before a launch, based on measured traffic, rather than discovering the ceiling on launch day.

#Fallbacks and graceful degradation

Sooner or later the API will be slow or unavailable for a while. Decide in advance what each feature does then.

Fall back to another model. If the primary model is overloaded, a different model may not be. A fallback should be a model you have actually evaluated on the task, and you may need a separate prompt for it.

Degrade the feature instead of failing the page. A search page can show plain results without the AI summary. A support widget can offer the contact form. A drafting tool can show "Suggestions are unavailable right now" while the rest of the editor works.

Stop calling a dependency that is clearly down. A circuit breaker counts recent failures; past a threshold it skips the call entirely for a cool-down period and goes straight to the fallback, so users do not each wait through the full timeout.

Cache whole responses for deterministic tasks with repeated inputs, such as classifying the same product title, keyed by a hash of model, prompt version and input.

Here is the core of a gateway function with fallback, logging and cost:

import json
import logging
import time

log = logging.getLogger("llm")

PRICES = {  # USD per million tokens (input, output)
    "claude-opus-5-5": (4.00, 20.00),
    "claude-sonnet-5-5": (2.00, 10.00),
    "claude-haiku-5-5": (0.10, 0.50),
}
FALLBACK = {"claude-sonnet-5-5": "claude-haiku-5-5"}
RETRYABLE = (anthropic.APIConnectionError, anthropic.RateLimitError, anthropic.InternalServerError,
             anthropic.OverloadedError)

class LLMUnavailable(Exception): ...
class LLMBadResponse(Exception): ...

def estimate_cost(model: str, usage) -> float:
    inp, out = PRICES[model]
    written = usage.cache_creation_input_tokens or 0
    read = usage.cache_read_input_tokens or 0
    # Approximate: cache writes 1.25x, cache reads about 0.1x input (the read rate varies by model)
    return (usage.input_tokens * inp + written * inp * 1.25 + read * inp * 0.1
            + usage.output_tokens * out) / 1_000_000

def complete(feature: str, system: str, messages: list, max_tokens: int = 4096) -> str:
    model = MODELS[feature]
    for candidate in (model, FALLBACK.get(model)):
        if candidate is None:
            break
        start = time.perf_counter()
        try:
            response = client.messages.create(
                model=candidate, max_tokens=max_tokens, system=system, messages=messages
            )
        except RETRYABLE as exc:  # the SDK has already retried; try the fallback
            log.warning(json.dumps({"feature": feature, "model": candidate, "error": type(exc).__name__}))
            continue
        log.info(json.dumps({
            "feature": feature,
            "model": candidate,
            "fallback": candidate != model,
            "request_id": response._request_id,
            "latency_s": round(time.perf_counter() - start, 3),
            "stop_reason": response.stop_reason,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "cache_read_tokens": response.usage.cache_read_input_tokens,
            "cost_usd": round(estimate_cost(candidate, response.usage), 6),
        }))
        if response.stop_reason in ("refusal", "max_tokens"):
            raise LLMBadResponse(response.stop_reason)
        return "".join(b.text for b in response.content if b.type == "text")
    raise LLMUnavailable(feature)

Non-retryable errors (400, 401, 404) are not caught: they mean a bug or misconfiguration, and should fail loudly in your error tracker rather than be papered over by a fallback.

#Streaming endpoints

For chat-style features, stream the response to the browser so the first words appear in a second or two. Server-Sent Events (SSE) are the simplest transport. With FastAPI:

import json
import anthropic
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()
aclient = anthropic.AsyncAnthropic()

class ChatIn(BaseModel):
    message: str

@app.post("/chat")
async def chat(body: ChatIn, request: Request):
    async def events():
        async with aclient.messages.stream(
            model="claude-sonnet-5-5",
            max_tokens=8000,
            messages=[{"role": "user", "content": body.message}],
        ) as stream:
            async for text in stream.text_stream:
                if await request.is_disconnected():
                    return  # leaving the context manager closes the upstream stream
                yield f"data: {json.dumps({'text': text})}\n\n"
            final = await stream.get_final_message()
            yield f"data: {json.dumps({'done': True, 'stop_reason': final.stop_reason})}\n\n"

    return StreamingResponse(events(), media_type="text/event-stream")

Stopping when the client disconnects matters: otherwise you keep paying for tokens nobody will read. Send the final stop_reason so the UI can tell a complete answer from a truncated or refused one. (Real code also loads conversation history from your database, authenticates the user and applies limits before streaming.)

#Observability and safe releases

You cannot operate what you cannot see. Log one structured record per model call, like the gateway above does:

Build dashboards and alerts on: error rate and fallback rate, p95 latency, cost per day per feature, cache hit ratio, and the share of max_tokens and refusal stop reasons. A sudden drop in cache reads or a jump in output tokens usually means a prompt change went wrong. For agents, record a trace per run (each turn's tool calls and results) using your tracing system, OpenTelemetry for example. Redact personal data before it reaches logs, and set a retention period.

Finally, release changes to prompts and models the way you release code:

• Version prompts in the repository, reviewed in pull requests, with the version in every log line.
• Run evals in CI on every prompt or model change, and block merges that regress.
• Roll out gradually behind a feature flag: 5% of traffic, compare metrics, then widen.
• Shadow-test model migrations: send a sample of real requests to the new model in parallel, compare outputs offline, and switch only when the evidence supports it.

#Practice

Build llm.py, a gateway module for two features (support_chat on Sonnet 5.5, ticket_tagging on Haiku 5.5) with:

1. Per-model timeouts (30 s for Sonnet, 15 s for Haiku) applied with client.with_options.
2. Fallback from Sonnet to Haiku on retryable errors.
3. A JSON log line per call with feature, model, fallback flag, request id, latency, tokens, stop reason and estimated cost.
4. A per-feature daily budget in USD: once exceeded, complete raises BudgetExceeded without calling the API.

Test the fallback by temporarily setting Sonnet's timeout to 0.001 seconds.

import datetime as dt
import json
import logging
import time
from collections import defaultdict

import anthropic

logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("llm")

client = anthropic.Anthropic(max_retries=2)

MODELS = {"support_chat": "claude-sonnet-5-5", "ticket_tagging": "claude-haiku-5-5"}
TIMEOUTS = {"claude-sonnet-5-5": 30.0, "claude-haiku-5-5": 15.0}
FALLBACK = {"claude-sonnet-5-5": "claude-haiku-5-5"}
PRICES = {"claude-sonnet-5-5": (2.00, 10.00), "claude-haiku-5-5": (0.10, 0.50)}
DAILY_BUDGET_USD = {"support_chat": 200.0, "ticket_tagging": 20.0}
RETRYABLE = (anthropic.APIConnectionError, anthropic.RateLimitError, anthropic.InternalServerError,
             anthropic.OverloadedError)

spent: dict[tuple[str, dt.date], float] = defaultdict(float)  # use Redis or a DB across processes

class BudgetExceeded(Exception): ...
class LLMUnavailable(Exception): ...
class LLMBadResponse(Exception): ...

def estimate_cost(model, usage) -> float:
    inp, out = PRICES[model]
    written = usage.cache_creation_input_tokens or 0
    read = usage.cache_read_input_tokens or 0
    return (usage.input_tokens * inp + written * inp * 1.25 + read * inp * 0.1
            + usage.output_tokens * out) / 1_000_000

def complete(feature: str, system: str, messages: list, max_tokens: int = 4096) -> str:
    key = (feature, dt.date.today())
    if spent[key] >= DAILY_BUDGET_USD[feature]:
        raise BudgetExceeded(f"{feature} spent ${spent[key]:.2f} today")
    primary = MODELS[feature]
    for model in (primary, FALLBACK.get(primary)):
        if model is None:
            break
        start = time.perf_counter()
        try:
            response = client.with_options(timeout=TIMEOUTS[model]).messages.create(
                model=model, max_tokens=max_tokens, system=system, messages=messages
            )
        except RETRYABLE as exc:
            log.warning(json.dumps({"feature": feature, "model": model, "error": type(exc).__name__}))
            continue
        cost = estimate_cost(model, response.usage)
        spent[key] += cost
        log.info(json.dumps({
            "feature": feature, "model": model, "fallback": model != primary,
            "request_id": response._request_id,
            "latency_s": round(time.perf_counter() - start, 3),
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "stop_reason": response.stop_reason, "cost_usd": round(cost, 6),
        }))
        if response.stop_reason in ("refusal", "max_tokens"):
            raise LLMBadResponse(response.stop_reason)
        return "".join(b.text for b in response.content if b.type == "text")
    raise LLMUnavailable(feature)

if __name__ == "__main__":
    TIMEOUTS["claude-sonnet-5-5"] = 0.001  # force the primary to time out
    print(complete("support_chat", "You are a concise support assistant.",
                   [{"role": "user", "content": "How do I reset my password?"}]))

{"feature": "support_chat", "model": "claude-sonnet-5-5", "error": "APITimeoutError"}
{"feature": "support_chat", "model": "claude-haiku-5-5", "fallback": true, "request_id": "req_...", "latency_s": 1.84, "input_tokens": 24, "output_tokens": 212, "stop_reason": "end_turn", "cost_usd": 0.000108}
Go to Settings > Security and choose "Reset password"...

#Key takeaways

• Call models only from your backend and route every call through one gateway module that owns timeouts, error handling, fallbacks, logging and cost.
• The SDK retries connection errors, 429 and 5xx with backoff; configure max_retries and set timeouts well below the 10-minute default for user-facing paths. Never retry 400-class errors.
• Check stop_reason (max_tokens, refusal) before using content, log _request_id, and make side effects idempotent.
• Avoid rate limits by bounding concurrency, queueing background work, batching what can wait and caching prompts; degrade features gracefully when the API is unavailable.
• Stream to the browser over SSE and stop on disconnect; log structured per-call records, alert on cost, latency, errors and cache hit rate, and release prompt and model changes behind evals and gradual rollouts.