Your First Production LLM API Call: A 2026 Wrapper Pattern for Claude Opus 4.7

Last month I sat in on an incident review. A small AI feature that had shipped six weeks earlier was throwing 500s on roughly four percent of requests. The on-call engineer had no logs from the failing calls. No token counts. No correlation. No model id. Just stack traces from a one-liner inside a route handler that called the Anthropic SDK directly.

The fix took an afternoon. The cost of not having done it the first time, in support hours and customer trust, was about three weeks. This article is about the file that should have existed on day one.

The Problem

A production LLM call is not the same as a working LLM call. The SDK request looks identical. What differs is everything that surrounds it: how errors are mapped to your domain, how retries are policed, how the call is observable in logs and metrics, how token usage rolls into your cost dashboard, how a single user-facing operation can be traced through every component it touches.

In 2026, the gap matters more than it did even a year ago. Claude Opus 4.7 shipped on April 16 with API behaviour that breaks code written for Opus 4.6 (temperature, top_p, and top_k now return HTTP 400 if non-default; the new tokenizer produces up to 35% more tokens; thinking content is omitted by default). Anthropic’s SDK is on 0.97.0 as of April 23. OpenAI’s SDK is on 2.32.0. The version notes are real. The wrapper that abstracts your team away from those notes pays for itself the first time you migrate a model.

Image prompt for this article

Medium featured image

The Solution

One file in your codebase owns the boundary to every language-model API. Every call goes through it. Every caller sees a stable interface. The five responsibilities of that file are non-negotiable.

1. Typed errors with stable names

SDK exception classes change names across major versions. Application code that catches RateLimitError directly will need to be rewritten the next time Anthropic restructures its exception hierarchy. The wrapper catches every SDK exception and re-raises a small set of stable, domain-named errors: rate_limited, overloaded, timeout, connection_error, apierror<status>. Upstream code branches on those.

One specific case to handle: HTTP 529 (overloaded_error) is a capacity signal from Anthropic’s platform, separate from a 429 rate limit on your account. The SDK surfaces it as APIStatusError. Treat it as a distinct condition; a 429 means slow down, a 529 means try a different model.

2. Retries the SDK already does

The Anthropic Python SDK retries 408, 409, 429, 5xx, and connection errors twice by default with exponential backoff and jitter. It honours the retry-after and retry-after-ms headers. The OpenAI SDK has equivalent behaviour. A second retry loop on top of the SDK is a thundering-herd generator. State max_retries explicitly so future readers can audit it. Then leave it.

3. Five-field structured logging

Every successful call emits one structured log line with five fields: correlation_id, model, input_tokens, output_tokens, latency_ms. Add stop_reason as a sixth and you can spot truncations. Add request_id as a seventh and Anthropic support can find your call. Do not log prompt or response bodies; user data ends up in places it should not.

4. Token accounting at the call site

message.usage carries the canonical token counts. Read it, return it on the typed response, and do not approximate. Opus 4.7’s new tokenizer makes any client-side estimate wrong; billing happens on the model’s count, not yours.

5. Correlation that survives

A twelve-character hex ID generated at the entry point, threaded through every layer that touches the call. When something fails, one grep tells the whole story.

Code Walkthrough

Here is the wrapper. It is the only place in the codebase that imports from anthropic. Every other call site goes through call_llm.

# llm_client.py
# A production-grade wrapper around the Anthropic Messages API.
# Verified against anthropic SDK 0.97.0 (released April 23, 2026)
# and Claude Opus 4.7 / Sonnet 4.6 / Haiku 4.5.

from __future__ import annotations

import logging
import os
import time
import uuid
from dataclasses import dataclass
from typing import Any

import anthropic
from anthropic import (
    Anthropic,
    APIConnectionError,
    APIStatusError,
    APITimeoutError,
    RateLimitError,
)

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class LLMResponse:
    """Typed response object returned to upstream callers."""
    text: str
    model: str
    input_tokens: int
    output_tokens: int
    stop_reason: str | None
    request_id: str | None
    latency_ms: int
    correlation_id: str


class LLMClientError(Exception):
    """Domain error wrapping any failure from the LLM call site."""


# Build the SDK client once at import time. The SDK is internally async-safe
# and reuses an httpx connection pool, so a module-level instance is the
# correct production shape. Key precedence is: explicit kwarg > env var.
_CLIENT = Anthropic(
    # api_key omitted intentionally; SDK reads ANTHROPIC_API_KEY from env.
    # max_retries=2 is the SDK default; we keep it explicit for auditability.
    max_retries=2,
    # 60 seconds is enough for non-streaming completions up to ~4K tokens;
    # raise this only when generating long structured output.
    timeout=60.0,
)


def call_llm(
    prompt: str,
    *,
    system: str | None = None,
    model: str = "claude-sonnet-4-6",
    max_tokens: int = 1024,
    correlation_id: str | None = None,
) -> LLMResponse:
    """Production entry point for a single-turn Messages call.

    Returns a typed LLMResponse. Raises LLMClientError on any unrecoverable
    failure. Caller is expected to log the correlation_id.
    """
    cid = correlation_id or uuid.uuid4().hex[:12]
    started = time.monotonic()

    # Note on model choice: claude-opus-4-7 rejects temperature/top_p/top_k
    # if non-default, so we do NOT pass them by default. If you must control
    # sampling on Sonnet/Haiku, branch on model id and pass them only there.
    request: dict[str, Any] = {
        "model": model,
        "max_tokens": max_tokens,
        "messages": [{"role": "user", "content": prompt}],
    }
    if system is not None:
        request["system"] = system

    try:
        # The SDK retries APIConnectionError, 408, 409, 429, and 5xx by
        # default with exponential backoff and jitter. We layer logging
        # around it; we do NOT add a second retry loop on top.
        message = _CLIENT.messages.create(**request)

    except RateLimitError as e:
        # 429: request-rate or token-rate quota exceeded. The retry-after
        # header is the source of truth; the SDK already honours it on
        # the inner retries, so reaching here means we are out of budget.
        logger.warning(
            "llm_rate_limited cid=%s model=%s status=%s",
            cid, model, e.status_code,
        )
        raise LLMClientError("rate_limited") from e

    except APIStatusError as e:
        # 529 overloaded_error is a capacity signal, distinct from 429.
        # Surface it explicitly so upstream can fall back to another model.
        if e.status_code == 529:
            logger.warning("llm_overloaded cid=%s model=%s", cid, model)
            raise LLMClientError("overloaded") from e
        logger.error(
            "llm_api_error cid=%s model=%s status=%s body=%s",
            cid, model, e.status_code, getattr(e, "body", None),
        )
        raise LLMClientError(f"api_error_{e.status_code}") from e

    except APITimeoutError as e:
        logger.error("llm_timeout cid=%s model=%s", cid, model)
        raise LLMClientError("timeout") from e

    except APIConnectionError as e:
        logger.error("llm_connection_error cid=%s model=%s", cid, model)
        raise LLMClientError("connection_error") from e

    latency_ms = int((time.monotonic() - started) * 1000)

    # The Messages API returns a list of content blocks. For a normal
    # text reply, the first block is type="text". We concatenate text
    # blocks defensively so that future model behaviour does not silently
    # drop content (for example, when summarized thinking blocks appear).
    text = "".join(
        block.text for block in message.content if block.type == "text"
    )

    response = LLMResponse(
        text=text,
        model=message.model,
        input_tokens=message.usage.input_tokens,
        output_tokens=message.usage.output_tokens,
        stop_reason=message.stop_reason,
        request_id=getattr(message, "_request_id", None),
        latency_ms=latency_ms,
        correlation_id=cid,
    )

    # Structured success log. Always include correlation_id, model,
    # token counts, latency, and stop_reason. These five fields are the
    # minimum viable observability surface for a production LLM call.
    logger.info(
        "llm_call_ok cid=%s rid=%s model=%s input_tokens=%d output_tokens=%d "
        "latency_ms=%d stop_reason=%s",
        cid, response.request_id, response.model,
        response.input_tokens, response.output_tokens,
        response.latency_ms, response.stop_reason,
    )

    # stop_reason="max_tokens" means the model was cut off mid-thought.
    # Do not silently return truncated output; let the caller decide.
    if response.stop_reason == "max_tokens":
        logger.warning(
            "llm_truncated cid=%s model=%s output_tokens=%d max_tokens=%d",
            cid, model, response.output_tokens, max_tokens,
        )

    return response


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
    )
    out = call_llm(
        prompt="In one sentence, what is a production-grade LLM API call?",
        system="You are a precise senior engineer. Reply concisely.",
        model="claude-sonnet-4-6",
        max_tokens=200,
    )
    print(out.text)

Three design choices in this file are worth calling out.

The SDK client is constructed once at import time. The httpx connection pool inside the SDK is reused across requests; per-call client construction would add TLS handshake overhead to every request. max_retries=2 and timeout=60.0 are stated explicitly even though they match the SDK defaults; future readers can audit the policy without reading SDK source.

Errors are caught in three tiers. RateLimitError (429) is a quota signal. APIStatusError covers everything else from the API, including 529. APITimeoutError and APIConnectionError are the network-side errors. Each maps to a stable domain code. The catch order matters: RateLimitError is a subclass of APIStatusError, so it must come first.

Text extraction concatenates only blocks where block.type == “text”. This is forward-compatible with Opus 4.7’s thinking-block behaviour (where summarized thinking blocks can appear before text blocks) and any future content-block types.

Production Gotchas

Here is the same call written naively. Every line is wrong; every wrongness is a class of incident.

# DO NOT SHIP THIS
# Every line in this file is a future production incident.

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-api03-XXXXXXXXXXXX")

def ask(prompt):
    response = client.messages.create(
        model="claude-3-opus-20240229",        # Deprecated model id
        max_tokens=4096,
        temperature=0.7,                       # 400 error on Opus 4.7
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text             # Crashes if the first
                                                # block is not text.

print(ask("Hello"))

Naive vs production

Hardcoded API key vs environment variable: a single secret leak undoes everything.

Deprecated model id vs config-driven id: a model deprecation should be a config bump, not a deploy.

Default sampling parameters vs no sampling parameters: temperature=0.7 returns 400 on Opus 4.7.

No error handling vs five typed domain errors: every failure mode is uncatchable in the naive case.

No logging vs five structured fields: incidents are uninvestigable in the naive case.

Index-based content access vs type-filtered concatenation: brittle to any future content-block type.

Summary

The shape of your first production LLM call decides the shape of your production system. One file owns the boundary. That file accepts a prompt and a model, returns a typed response with token counts and correlation, raises stable domain errors, emits one structured log line per call, and works correctly across the Claude 4.x family. Every other section in this chapter (token economics, streaming, error handling, multi-provider abstraction, observability) builds on the shape established here.

Cost: forty extra minutes the first time. Savings: every incident week you do not have to spend reverse-engineering an unobservable call site.