Calling LLM APIs in Production

A large language model (LLM) answered correctly in your local notebook. In the deployed application, the exact same request stalls, times out, or throws an HTTP 429. Did the remote server finish? Did it charge your credit balance? Should your client retry right away, or will repeating the call knock down an already struggling provider? What should the user see while you don't know?

In Prompt Engineering Fundamentals, a model cited a policy line while proposing an unsupported 90-minute deadline. We'll carry that lesson into a fictional operational security task: deciding whether a stale service-account key on account acct_10234 meets rotation policy P-7. The trusted key age in our fixture is 45 days; its policy threshold is 30 days. These fixture values support the lesson; set real credential lifetimes from organizational policy and threat models. The model can propose an eligibility candidate, while the server checks it against trusted facts and keeps rotation-job creation in a separately authorized operation.

Every example below runs locally, through a mocked HTTP transport, with no live credentials and no billable calls. Ordinary code could compare 45 days with 30. We still ask the model for a candidate, because the failure worth catching is a fluent answer that disagrees with the account record. The same checks apply when a later workflow summarizes or extracts unstructured text.

View larger

Keep credentials on the server

A hosted model call needs an API key or an IAM role token. That key authorizes billable requests against your organization's quota. Client-side browser bundles, mobile apps, git repositories, and build artifacts are dangerous places for it. Keep provider keys strictly on the server.

Local code can read a credential from an environment variable. A deployed service can inject it from a secret manager or read it through an authorized server-side integration. Frameworks expose selected environment values to frontend code: Next.js inlines referenced NEXT_PUBLIC_ values, and Vite exposes VITE_ values through import.meta.env. Never use those prefixes for provider secrets.^{[1]Reference 1How to use environment variables in Next.jshttps://nextjs.org/docs/app/guides/environment-variables[2]Reference 2Env Variables and Modeshttps://vite.dev/guide/env-and-mode} Browser-delivered credentials are readable by the client.

The lab supplies a placeholder through a local test dictionary. Production code reads os.environ; the startup check fails fast when the deployment secret is missing.

The following check rejects an absent or blank configuration value. The remote provider verifies model permissions. The sample output omits the key itself.

import os

def require_server_key(environ) -> str:
    api_key = environ.get("LLM_API_KEY", "")
    if not isinstance(api_key, str) or not api_key.strip():
        raise RuntimeError("LLM_API_KEY is not set")
    return api_key

try:
    require_server_key({})
except RuntimeError:
    print({"missing_rejected": True})

api_key = require_server_key({"LLM_API_KEY": "dev_key_not_a_real_secret"})
print({"loaded": True})

{'missing_rejected': True}
{'loaded': True}

The check verifies that a value is present. Verify server isolation separately and ask the provider to confirm model permissions. Both sample branches omit the value. A production service calls require_server_key(os.environ) during startup, keeps separate credentials for staging and production, and revokes or rotates leaked credentials promptly.

A task contract for the model

A server-side secret protects the credential from theft. Typed task contracts also give application routes a consistent interface across vendor SDKs.

When a browser client asks whether account acct_10234 needs key rotation, the frontend identifies the account and transmits the user's intent. After authenticating the caller and checking tenant access, the backend loads the trusted key age and rotation policy from its database. Client-supplied status flags or timestamps must never become authoritative merely because they arrived in a JSON payload.

The boundary separates several core responsibilities:

The model turns supplied text into a candidate response. Your application enforces the contract. That split is what lets you reject a fluent answer that contradicts the account record.

Name the application's request

A vendor request format changes often: Anthropic uses messages with specific system parameters, OpenAI uses messages or responses formats, and local inference engines use OpenAI-compatible HTTP schemas. If your route handlers call vendor SDKs directly, vendor-specific parameters bleed into your business logic.

This internal contract encapsulates intent, evidence IDs, deadline tiers, a trace ID, and a prompt version. It stays independent of any vendor SDK; the application still authorizes rotation jobs separately. Python's dataclass annotations document the internal contract clearly.

from dataclasses import dataclass
from typing import Literal

@dataclass(frozen=True)
class LLMTask:
    task: Literal["decide_rotation", "summarize_scan", "draft_reply"]
    account_id: str
    policy_line_ids: tuple[str, ...]
    customer_question: str
    trace_id: str
    prompt_version: str
    connect_timeout_seconds: float = 2.0
    read_timeout_seconds: float = 8.0
    overall_deadline_seconds: float = 20.0
    max_retries: int = 2

task = LLMTask(
    task="decide_rotation",
    account_id="acct_10234",
    policy_line_ids=("P-7",),
    customer_question="Can I rotate the stale service-account key?",
    trace_id="trace_7e3b",
    prompt_version="rotation_decision@2",
)

print(task.task, task.account_id, task.policy_line_ids)

decide_rotation acct_10234 ('P-7',)

LLMTask stores application intent. Each adapter maps it to the provider's wire schema and supported features. Dataclass annotations and Literal describe intended types; validate external requests, allowed tasks, account access, finite positive timeouts, and nonnegative retry counts before constructing it.

Notice that LLMTask separates the read-only eligibility inquiry from the write action. Deciding eligibility is a safe read; creating a rotation job is an authorized mutation.

overall_deadline_seconds specifies the total latency budget for the end-to-end task. If the wrapper retries after a transient network drop, each subsequent attempt receives only the time that remains. Three 20-second attempts must stay within the shared budget instead of consuming a full minute of user waiting time.

What belongs in the wrapper

The wrapper encapsulates operational policies that must remain consistent across all application routes:

If individual route handlers call vendor SDKs directly, these policies drift. One handler retries five times on HTTP 500 while another fails immediately; one route logs raw customer emails while another logs nothing. A centralized wrapper makes operational behavior testable and uniform.

A provider adapter translates that domain contract into provider wire format. Here, required_response_fields makes our validation requirements explicit before mapping them to a concrete SDK:

import json

policy_lines = {"P-7": "Stale service-account keys at least 30 days old require rotation."}
task = {
    "account_id": "acct_10234",
    "credential_age_days": 45,
    "key_stale": True,
    "policy_line_ids": ["P-7"],
}

context = "\n".join(
    f"{line_id}: {policy_lines[line_id]}" for line_id in task["policy_line_ids"]
)
payload = {
    "messages": [
        {"role": "system", "content": "Decide rotation eligibility only from supplied policy lines. Return JSON."},
        {"role": "user", "content": f"{context}\nAccount: {json.dumps(task, sort_keys=True)}"},
    ],
    "required_response_fields": ["decision", "rotation_window_days", "source_line_ids"],
}

assert "P-7" in payload["messages"][1]["content"]
assert "acct_10234" in payload["messages"][1]["content"]
print(json.dumps(payload, indent=2))

{
  "messages": [
    {
      "role": "system",
      "content": "Decide rotation eligibility only from supplied policy lines. Return JSON."
    },
    {
      "role": "user",
      "content": "P-7: Stale service-account keys at least 30 days old require rotation.\nAccount: {\"account_id\": \"acct_10234\", \"credential_age_days\": 45, \"key_stale\": true, \"policy_line_ids\": [\"P-7\"]}"
    }
  ],
  "required_response_fields": [
    "decision",
    "rotation_window_days",
    "source_line_ids"
  ]
}

The adapter injects policy text and trusted facts so the model can interpret the decision. This internal payload's required_response_fields describes application requirements; the adapter must map and enforce them for each provider. Here, the legacy field rotation_window_days means a minimum key age of 30 days under P-7. A new contract should use a clearer name such as minimum_key_age_days.

What the wrapper does

Follow one request through the wrapper. Tool calling lets a model propose a tool and arguments; application code decides whether to execute it.^{[3]Reference 3Function callinghttps://developers.openai.com/api/docs/guides/function-calling} Structured Outputs constrains the output to a supported response schema. Strict function arguments can use schema constraints too.^{[4]Reference 4Structured model outputshttps://developers.openai.com/api/docs/guides/structured-outputs} Both mechanisms still produce an untrusted candidate; checking source facts and authorization remains the server's responsibility.

Every model invocation serves two distinct audiences:

1. The product caller requires a validated typed object or a structured failure it can present to the user.
2. The operations team requires a telemetry record capturing latency, token usage, and status, even when the model fails or returns garbage.

Validate a candidate before returning it

Start with a fixed candidate so validation mechanics stay transparent. We verify the root dictionary, required keys, strict types, cited source IDs, and whether the proposed decision matches trusted database facts. Notice that Python's bool is a subclass of int (isinstance(True, int) evaluates to True), so we verify type(data["rotation_window_days"]) is int to reject booleans pretending to be integers.

import json
import time
from dataclasses import dataclass
from typing import Any

@dataclass(frozen=True)
class RotationDecision:
    decision: str
    rotation_window_days: int
    source_line_ids: tuple[str, ...]

def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate JSON field")
        result[key] = value
    return result

def reject_constant(_value):
    raise ValueError("nonstandard JSON number")

def fake_provider_call(prompt: str) -> str:
    return json.dumps({
        "decision": "eligible",
        "rotation_window_days": 30,
        "source_line_ids": ["P-7"],
    })

def parse_decision(
    raw: str,
    policy_windows: dict[str, int],
    credential_facts: dict[str, object],
) -> RotationDecision:
    if not isinstance(raw, str):
        raise ValueError("response must be text")
    threshold = policy_windows.get("P-7")
    if type(threshold) is not int or threshold < 0:
        raise ValueError("trusted policy threshold is missing or invalid")
    age = credential_facts.get("credential_age_days")
    stale = credential_facts.get("key_stale")
    if type(age) is not int or age < 0 or type(stale) is not bool:
        raise ValueError("trusted credential facts are missing or invalid")
    try:
        data: Any = json.loads(raw, object_pairs_hook=unique_object, parse_constant=reject_constant)
    except RecursionError as exc:
        raise ValueError("response nesting is excessive") from exc
    if not isinstance(data, dict) or set(data) != {"decision", "rotation_window_days", "source_line_ids"}:
        raise ValueError("response must be an object with exactly the required fields")
    if not isinstance(data["decision"], str) or data["decision"] not in {"eligible", "not_eligible"}:
        raise ValueError("decision is outside allowed enum")
    if type(data["rotation_window_days"]) is not int:
        raise ValueError("rotation_window_days must be an integer")
    if not isinstance(data["source_line_ids"], list) or not all(isinstance(x, str) for x in data["source_line_ids"]):
        raise ValueError("source_line_ids must be a list of strings")
    cited = tuple(data["source_line_ids"])
    if cited != ("P-7",) or data["rotation_window_days"] != threshold:
        raise ValueError("decision is unsupported by supplied policy")
    expected_decision = "eligible" if stale and age >= threshold else "not_eligible"
    if data["decision"] != expected_decision:
        raise ValueError("decision conflicts with trusted credential facts")
    return RotationDecision(
        decision=data["decision"],
        rotation_window_days=data["rotation_window_days"],
        source_line_ids=cited,
    )

def decide_rotation(account_id: str, trace_id: str, now_fn=time.perf_counter) -> RotationDecision:
    start = now_fn()
    trusted_credentials = {
        "acct_10234": {"credential_age_days": 45, "key_stale": True},
    }
    status = "failed"
    try:
        credential_facts = trusted_credentials[account_id]
        raw = fake_provider_call(f"Decide stale-key rotation eligibility for {account_id} using P-7.")
        decision = parse_decision(raw, {"P-7": 30}, credential_facts)
        status = "ok"
        return decision
    finally:
        latency_ms = (now_fn() - start) * 1000
        print({"trace_id": trace_id, "status": status, "latency_ms": round(latency_ms, 2)})

fake_times = iter([100.0, 100.00005])
decision = decide_rotation(
    "acct_10234",
    "trace_7e3b",
    now_fn=lambda: next(fake_times),
)
print(decision)

{'trace_id': 'trace_7e3b', 'status': 'ok', 'latency_ms': 0.05}
RotationDecision(decision='eligible', rotation_window_days=30, source_line_ids=('P-7',))

The caller receives a validated RotationDecision with typed fields. Duplicate fields and Python's permissive NaN/Infinity parsing are rejected.^{[5]Reference 5json: JSON encoder and decoderhttps://docs.python.org/3/library/json.html} Missing or mistyped trusted facts produce a failure status; they never become an not_eligible decision. An unknown account fails lookup before the fake provider is called, and finally emits a status on both success and failure.

This example has one fixed policy and integer-day fixture. Its fake provider ignores its prompt; it cannot test grounding, permissions, real latency, or model accuracy. The real SDK adapter below sends the evidence. Bound response sizes in the wrapper and map failures to stable application error codes without logging raw customer payloads.

What happens to telemetry on the rejected path? Replace the fake provider temporarily with a contradictory response, then try an unknown account. Both attempts must record failure; only the known account should reach the provider stub.

original_provider = fake_provider_call
provider_calls = []

def wrong_provider(prompt):
    provider_calls.append(prompt)
    return json.dumps({
        "decision": "not_eligible", "rotation_window_days": 30,
        "source_line_ids": ["P-7"],
    })

fake_provider_call = wrong_provider
try:
    for account in ["acct_10234", "unknown_account"]:
        times = iter([200.0, 200.001])
        try:
            decide_rotation(account, "trace_failure", now_fn=lambda: next(times))
        except (ValueError, KeyError):
            print({"result": "rejected"})
finally:
    fake_provider_call = original_provider
print({"provider_calls": len(provider_calls)})

{'trace_id': 'trace_failure', 'status': 'failed', 'latency_ms': 1.0}
{'result': 'rejected'}
{'trace_id': 'trace_failure', 'status': 'failed', 'latency_ms': 1.0}
{'result': 'rejected'}
{'provider_calls': 1}

The wrapper can reject a bad candidate and record the failure. Next, add a deadline and define how it handles provider 429 responses.

Timeouts, rate limits, and resilience

Model latency depends on the workload and provider. A stalled synchronous call can occupy a worker; an asynchronous call can still hold sockets, memory, and concurrency slots. Bound resource use and distinguish temporary rate limits from failures requiring intervention.

Tiered timeouts: connect, TTFT, read, and total budget

Choose timeout values from the product's latency target and observed workload, including long prompts and reasoning. Each timeout below covers a separate stage; set its duration for that stage's latency budget:

1. Connect timeout: Bounds connection establishment under the transport's semantics. DNS, TLS, proxies, and pooled connections affect what this timer covers; timing spans for those phases can pinpoint the delay.
2. First visible text deadline: Bounds the wait for useful streamed output. TTFT includes network and serving delays and can include hidden reasoning. A slow first token can result from network or serving delays; use scheduler telemetry to diagnose backlog. An SSE heartbeat or metadata event isn't a text token.
3. Read timeout: Bounds waiting for a received data chunk, whose cadence can differ from model-token cadence. HTTPX also has write and pool-acquisition timeouts.^{[6]Reference 6Timeoutshttps://www.python-httpx.org/advanced/timeouts/} Regular heartbeats can satisfy a read timer while useful text remains stalled.
4. Overall task deadline: Bounds the application budget across retrieval, attempts, backoff, and validation. In a hypothetical 20-second budget, an attempt consuming 14 seconds leaves at most 6 seconds for everything remaining. Compute elapsed time with a monotonic clock.

Enforce deadlines around an asynchronous provider call using asyncio.wait_for:

import asyncio

async def slow_provider() -> str:
    await asyncio.sleep(0.02)
    return "late decision"

async def call_with_deadline(timeout_seconds: float) -> dict[str, str]:
    try:
        text = await asyncio.wait_for(slow_provider(), timeout=timeout_seconds)
        return {"status": "ok", "text": text}
    except TimeoutError:
        return {"status": "timeout", "next_step": "show_retry_option"}

print(asyncio.run(call_with_deadline(0.001)))

{'status': 'timeout', 'next_step': 'show_retry_option'}

asyncio.wait_for requests cancellation and waits for it, so actual elapsed time can exceed its timeout during cleanup.^{[7]Reference 7Coroutines and Taskshttps://docs.python.org/3/library/asyncio-task.html} It can't interrupt blocking synchronous code or confirm that remote generation stopped. Use a cooperative async transport, configure its timeouts, and measure deadline overruns. A local timeout leaves remote generation and billing status unknown; reconcile with provider usage or request state when available.

Rate limit dynamics: RPM, TPM, and 429 response handling

Providers can enforce several rate and usage limits. Two common dimensions are:

• Requests Per Minute (RPM): Expresses request capacity per minute. Providers may enforce it over shorter intervals or replenish an allowance continuously.
• Tokens Per Minute (TPM): A token-capacity limit whose accounting depends on the provider, model, and endpoint.

Either limit can bind first. Under a simplified input-only budget, five 16,000-token requests consume 80,000 tokens while using just five request slots. Real accounting can reserve output capacity or estimate tokens before completion. OpenAI documents estimated request size and output-limit accounting, additional daily limits, shared model limits, and shorter enforcement intervals.^{[8]Reference 8Rate Limitshttps://developers.openai.com/api/docs/guides/rate-limits} Billing token counts needn't equal the amount charged against a rate limiter.

Inspect status, structured error body, and response headers before deciding whether to retry. Provider-specific adapters normalize them to application error categories:

1. Temporary throttling (often HTTP 429): Retry a read-only call only within the attempt/deadline budget, respecting available retry guidance. The limiter may be smoothing a traffic surge over a short interval.
2. Billing or enforced usage limits (provider-specific status/code): OpenAI documents credit_balance_exhausted, organization_spend_limit_exceeded, and project_spend_limit_exceeded.^{[9]Reference 9API Error Codeshttps://developers.openai.com/api/docs/guides/error-codes} Blind immediate retries won't fix these. Report an actionable failure; an operator can change funding/limits, or an enforced monthly limit can reset. Other providers expose their own codes and labels.
3. Response headers to inspect:
   • Retry-After: A requested wait, expressed as delay-seconds or an HTTP date. This wrapper honors a valid value as a retry floor.
   • x-ratelimit-remaining-requests and x-ratelimit-reset-requests: Remaining request allowance and replenishment window.
   • x-ratelimit-remaining-tokens and x-ratelimit-reset-tokens: Remaining token balance and reset timing.^{[8]Reference 8Rate Limitshttps://developers.openai.com/api/docs/guides/rate-limits}

The classifier below consumes normalized application codes, including rate_limit. An adapter must map actual provider codes such as OpenAI's slow_down and rate-limit error details before calling it. Unknown 429 responses stop for inspection; retry only after mapping their provider-specific code.

BILLING_OR_QUOTA = {
    "insufficient_quota",
    "credit_balance_exhausted",
    "organization_spend_limit_exceeded",
    "project_spend_limit_exceeded",
    "organization_usage_limit_exceeded",
}

def retry_decision(status_code: int, error_code: str = "") -> str:
    if error_code in {
        "content_filter",
        "refusal",
        "incomplete_output",
        "validation_reject",
    }:
        return "fail_and_inspect"
    if status_code == 429:
        if error_code in BILLING_OR_QUOTA:
            return "fail_and_fix_quota"
        if error_code == "rate_limit":
            return "retry_with_backoff"
        return "fail_and_inspect"
    if status_code in {408, 500, 502, 503, 504}:
        return "retry_with_backoff"
    if status_code in {400, 401, 403}:
        return "fail_and_fix_request"
    return "fail_and_inspect"

cases = [
    (429, "rate_limit"),
    (429, "credit_balance_exhausted"),
    (429, ""),
    (503, ""),
    (401, ""),
    (400, ""),
    (200, "refusal"),
    (200, "incomplete_output"),
    (200, "validation_reject"),
]
for status, error_code in cases:
    print(status, error_code or "-", retry_decision(status, error_code))

429 rate_limit retry_with_backoff
429 credit_balance_exhausted fail_and_fix_quota
429 - fail_and_inspect
503 - retry_with_backoff
401 - fail_and_fix_request
400 - fail_and_fix_request
200 refusal fail_and_inspect
200 incomplete_output fail_and_inspect
200 validation_reject fail_and_inspect

Exponential backoff with full and decorrelated jitter

Immediate retries can amplify an outage. In a simplified example, 500 workers fail together and use the same backoff delays: 1s, then 2s, then 4s. Ignoring request durations, their retries cluster at elapsed times 1s, 3s, and 7s. Randomizing waits spreads traffic; recovery still depends on the provider and workload.

View larger

Jitter math: full jitter vs decorrelated jitter

AWS compares several backoff strategies. We'll work through full and decorrelated jitter to spread synchronized retry waves:^{[10]Reference 10Exponential Backoff And Jitterhttps://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/}

1. Full Jitter: Sample uniformly between zero and the current exponential delay ceiling: $\text{delay} = \text{random}(0, \min(\text{cap}, \text{base} \times 2^{\text{attempt}-1}))$ Independent draws reduce synchronization. Finite timer resolution and chance can still put workers in the same time bucket.
2. Decorrelated Jitter: Instead of calculating powers of two from the attempt index, sample uniformly between the base delay and three times the prior sleep interval: $\text{sleep}_{i+1} = \min(\text{cap}, \text{random}(\text{base}, \text{sleep}_i \times 3))$ This changes the next randomization window using the previous wait. Workers can still coincide, especially when many draws reach the same capped delay.

Compare both algorithms side by side:

The fixed draw 0.5 makes this lab's arithmetic reproducible. Production workers need independent random draws. This example sleeps with full jitter, prints the decorrelated alternative, and caps attempts. A production wrapper also needs a task-wide deadline.

import time
import random

class RetryableModelError(Exception):
    pass

def flaky_call(attempt: int) -> str:
    if attempt < 2:
        raise RetryableModelError("temporary rate limit")
    return "ok"

def full_jitter_delay(attempt: int, base: float = 1.0, cap: float = 8.0, rand_fn=random.random) -> float:
    ceiling = min(cap, base * (2 ** (attempt - 1)))
    return rand_fn() * ceiling

def decorrelated_jitter_delay(prev_sleep: float, base: float = 1.0, cap: float = 8.0, rand_fn=random.random) -> float:
    return min(cap, base + rand_fn() * max(0.0, (prev_sleep * 3.0) - base))

def call_with_retry(max_retries: int = 2, sleep_fn=time.sleep, rand_fn=random.random) -> str:
    attempts = 0
    prev_sleep = 1.0
    while True:
        try:
            return flaky_call(attempts)
        except RetryableModelError as exc:
            if attempts >= max_retries:
                print({"status": "failed", "reason": str(exc)})
                raise
            attempts += 1
            full_delay = full_jitter_delay(attempts, rand_fn=rand_fn)
            decorr_delay = decorrelated_jitter_delay(prev_sleep, rand_fn=rand_fn)
            prev_sleep = decorr_delay
            print({
                "status": "retrying",
                "attempt": attempts,
                "full_jitter_s": full_delay,
                "decorr_jitter_s": round(decorr_delay, 2),
            })
            sleep_fn(full_delay)

result = call_with_retry(sleep_fn=lambda _seconds: None, rand_fn=lambda: 0.5)
print(result)

{'status': 'retrying', 'attempt': 1, 'full_jitter_s': 0.5, 'decorr_jitter_s': 2.0}
{'status': 'retrying', 'attempt': 2, 'full_jitter_s': 1.0, 'decorr_jitter_s': 3.5}
ok

Respecting Retry-After and deadline caps

For this wrapper, a valid Retry-After becomes a minimum wait. Parse its delay-seconds or HTTP-date form first; an expired date becomes zero delay, and an invalid header needs a defined fallback.^{[11]Reference 11HTTP Semantics, Section 10.2.3: Retry-Afterhttps://www.rfc-editor.org/rfc/rfc9110.html#section-10.2.3} The function below takes that already-parsed number and adds jitter above the floor. If the requested wait exceeds the remaining task budget, stop and return a timeout while preserving the server's minimum wait:

import math
import random

def retry_wait_s(
    attempt: int,
    remaining_s: float,
    retry_after_s: float | None = None,
    rand_fn=random.random,
) -> float:
    if type(attempt) is not int or attempt < 1:
        raise ValueError("attempt must be a positive retry number")
    floor = 0.0 if retry_after_s is None else retry_after_s
    if not math.isfinite(floor) or floor < 0:
        raise ValueError("Retry-After must be finite and nonnegative")
    if not math.isfinite(remaining_s) or remaining_s <= 0:
        raise TimeoutError("task deadline exhausted")
    ceiling = 2 ** min(attempt - 1, 3)
    wait = floor + rand_fn() * ceiling
    if wait >= remaining_s:
        raise TimeoutError("retry would miss the task deadline")
    return wait

print(retry_wait_s(attempt=1, remaining_s=20.0, rand_fn=lambda: 0.5))
print(retry_wait_s(attempt=1, remaining_s=20.0, retry_after_s=2.0, rand_fn=lambda: 0.5))
try:
    retry_wait_s(attempt=1, remaining_s=1.2, retry_after_s=2.0)
except TimeoutError as exc:
    print({"status": "timeout", "reason": str(exc)})

0.5
2.5
{'status': 'timeout', 'reason': 'retry would miss the task deadline'}

With 1.2 seconds left in the task budget and a 2.0-second wait requested, the wrapper stops immediately and emits a deadline timeout. No further attempt can fit within the remaining time.

This function decides whether a wait fits. Recompute remaining time after sleeping, then reserve enough time for and enforce the next attempt. Let one layer own retries: an SDK with its own retry loop beneath this wrapper can multiply attempts and exceed the budget. The SDK lab explicitly sets max_retries=0.^{[12]Reference 12OpenAI Python API libraryhttps://github.com/openai/openai-python}

Circuit breakers, fallback providers, and multi-region routing

During persistent upstream failure, retrying every incoming request adds load and latency. A circuit breaker can stop attempts to the affected dependency while it recovers.^{[13]Reference 13Circuit Breaker patternhttps://learn.microsoft.com/en-us/azure/architecture/patterns/circuit-breaker} Scope it to the failure domain. Authentication errors or one tenant's quota exhaustion shouldn't trip a global provider breaker; repeated 429s can require local admission control instead of cross-provider failover.

View larger

The three-state circuit breaker

A circuit breaker transitions across three well-defined states:

1. Closed: Primary calls are allowed. In a consecutive-failure design, a success resets the counter. Other breakers use error rates in a rolling window; track these signals to assess provider health.
2. Open (Tripped): When the failure count or error rate breaches a configured threshold (such as 5 consecutive 5xx errors), the breaker trips to Open. Incoming calls don't touch the primary provider; they fail fast or route immediately to a secondary fallback provider. This protects the primary provider from load so it can recover.
3. Half-Open: After cooldown, admit a bounded number of recovery probes. Set a concurrency cap as well as any percentage rule, especially during traffic surges. This lab admits one probe: success closes the breaker; failure immediately reopens it.

Fallback providers and multi-region routing

When the breaker opens, a model router can divert requests along two dimensions:

• Multi-region routing: Use distinct regional deployments where the service exposes and supports them. openai/us-east and openai/us-west aren't public OpenAI model IDs or selectable region endpoints. Regions and availability zones are different failure domains.
• Cross-provider fallback: Use a separately evaluated secondary model. Capabilities, schemas, context, safety behavior, latency, and pricing can differ. Check spare capacity and shared account limits before moving traffic.

Require the same typed contract and semantic verification on fallback results. Also check data residency, retention, and customer permissions before sending data to another provider. A fallback can be unavailable or unapproved; a named failure is then appropriate.

Exercise a small sequential routing model that selects a target label. A production router also needs provider calls, atomic admission/state transitions, bounded probe lifetimes, and handling for late results from older breaker generations.

import time
from typing import Literal

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 2, cooldown_seconds: float = 10.0):
        self.failure_threshold = failure_threshold
        self.cooldown_seconds = cooldown_seconds
        self.state: Literal["closed", "open", "half_open"] = "closed"
        self.failure_count = 0
        self.last_failure_time = 0.0
        self.probe_in_flight = False

    def record_success(self) -> None:
        if self.state == "open":
            return
        self.failure_count = 0
        self.state = "closed"
        self.probe_in_flight = False

    def record_failure(self, now: float) -> None:
        was_probe = self.state == "half_open"
        self.failure_count += 1
        self.last_failure_time = now
        self.probe_in_flight = False
        if was_probe or self.failure_count >= self.failure_threshold:
            self.state = "open"

    def allow_request(self, now: float) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if now - self.last_failure_time >= self.cooldown_seconds:
                self.state = "half_open"
                self.probe_in_flight = True
                return True
            return False
        return False  # The single half-open probe is already in flight.

def route_request(breaker: CircuitBreaker, now: float) -> str:
    if breaker.allow_request(now):
        target = "primary_provider" if breaker.state != "half_open" else "primary_canary"
    else:
        target = "fallback_provider"
    return f"{breaker.state} -> {target}"

cb = CircuitBreaker(failure_threshold=2, cooldown_seconds=10.0)
print(route_request(cb, now=100.0))
cb.record_failure(now=101.0)
print(route_request(cb, now=102.0))
cb.record_failure(now=103.0)
print(route_request(cb, now=104.0))
print(route_request(cb, now=114.0))
print(route_request(cb, now=114.0))  # No second probe while the first is pending.
cb.record_failure(now=115.0)
print(route_request(cb, now=116.0))
print(route_request(cb, now=126.0))
cb.record_success()
print(route_request(cb, now=127.0))

closed -> primary_provider
closed -> primary_provider
open -> fallback_provider
half_open -> primary_canary
half_open -> fallback_provider
open -> fallback_provider
half_open -> primary_canary
closed -> primary_provider

Idempotency and side-effect isolation

A timeout describes local uncertainty: your client gave up waiting for a response, but the remote provider or downstream worker might still have processed the request.

Checking rotation eligibility is a read operation: retrying it carries no risk of corrupting database records. In contrast, creating a rotation job or triggering an AWS IAM key deactivation is an action with real side effects. Retrying an unconfirmed action without an idempotency key can spawn duplicate rotation tickets, confuse operations teams, or trigger multiple credential revokes in production.

Always enforce durable idempotency keys on mutating operations. The following in-memory demonstration records arguments alongside each idempotency key:

created_rotation_jobs: dict[str, tuple[str, str]] = {}

def create_rotation_job_once(account_id: str, idempotency_key: str) -> tuple[str, bool]:
    if idempotency_key in created_rotation_jobs:
        saved_account, job_id = created_rotation_jobs[idempotency_key]
        if saved_account != account_id:
            raise ValueError("idempotency key reused with different arguments")
        return job_id, False
    job_id = f"job_{len(created_rotation_jobs) + 1:03d}"
    created_rotation_jobs[idempotency_key] = (account_id, job_id)
    return job_id, True

key = "request_7e3b"
first = create_rotation_job_once("acct_10234", key)
retry_after_timeout = create_rotation_job_once("acct_10234", key)

assert first[0] == retry_after_timeout[0]
assert len(created_rotation_jobs) == 1
print({"first": first, "retry_after_timeout": retry_after_timeout})

{'first': ('job_001', True), 'retry_after_timeout': ('job_001', False)}

This dictionary demonstrates sequential retries only; it loses state on restart and isn't safe for concurrent workers. Durable database storage with an atomic unique constraint on (tenant_id, idempotency_key) can claim local ownership, with arguments and pending/completed/failed states recorded. If the action is remote, a crash after dispatch still leaves its outcome uncertain: the remote service must enforce the same action key, or you must reconcile its status before dispatching again. A local unique constraint alone doesn't close that gap.

Scope the key to one authorized user intent; an account ID alone would block future legitimate rotations. Before any action, recheck authorization and current policy/state. Idempotency prevents duplicate execution of a particular intent; enforce permission and decision correctness separately.

From text to typed objects

Free-form natural language is fine for human reading, but programmatic software requires typed structures.

For key rotation, downstream systems need exact fields: decision, rotation_window_days, and source_line_ids. OpenAI's strict JSON schema feature constrains successful structured output to a supported schema subset. Recursive structures need more general machinery than a simple finite-state machine; the previous lesson explains grammar-guided decoding. JSON mode (json_object) targets JSON syntax without enforcing your particular keys or types. Refusals and incomplete-output cases must be handled separately.^{[4]Reference 4Structured model outputshttps://developers.openai.com/api/docs/guides/structured-outputs}

Both modes need application checks for truth and policy compliance. Validate the schema, then verify business invariants before creating a RotationDecision:

Schema checks vs semantic checks

Run the parser from minimal-wrapper.py against five candidates. Two have invalid shape or types; two have plausible fields but contradict trusted facts or policy. Only one passes. Then test parser ambiguity and trusted-input boundaries separately:

import json

policy_windows = {"P-7": 30}
trusted_facts = {"credential_age_days": 45, "key_stale": True}
candidates = [
    '{"decision":"eligible","rotation_window_days":30,"source_line_ids":["P-7"]}',
    '{"decision":"not_eligible","rotation_window_days":30,"source_line_ids":["P-7"]}',
    '{"decision":"eligible","rotation_window_days":90,"source_line_ids":["P-7"]}',
    '[]',
    '{"decision":"eligible","rotation_window_days":true,"source_line_ids":["P-7"]}',
]

def validate(raw: str) -> str:
    try:
        parse_decision(raw, policy_windows, trusted_facts)
    except ValueError as exc:
        return f"REJECT {exc}"
    return "ACCEPT grounded decision"

for candidate in candidates:
    print(validate(candidate))

valid = json.loads(candidates[0])
checks = [
    (candidates[0], {"P-7": 30}, {"credential_age_days": 30, "key_stale": True}, True),
    (candidates[0], {"P-7": 30}, {"credential_age_days": 29, "key_stale": True}, False),
    (json.dumps({**valid, "decision": "not_eligible"}), {"P-7": 30}, {"credential_age_days": 29, "key_stale": True}, True),
    (json.dumps({**valid, "decision": "not_eligible"}), {"P-7": 30}, {"credential_age_days": 45, "key_stale": False}, True),
    ('{"decision":"not_eligible","decision":"eligible","rotation_window_days":30,"source_line_ids":["P-7"]}', {"P-7": 30}, trusted_facts, False),
    ('{"decision":"eligible","rotation_window_days":NaN,"source_line_ids":["P-7"]}', {"P-7": 30}, trusted_facts, False),
    (candidates[0], {"P-7": 30}, {"credential_age_days": True, "key_stale": True}, False),
    (candidates[0], {"P-7": 30}, {"credential_age_days": 45, "key_stale": "false"}, False),
    (candidates[0], {"P-7": 30}, {"key_stale": True}, False),
    (candidates[0], {}, trusted_facts, False),
    (candidates[0], {"P-7": True}, trusted_facts, False),
]
for raw, windows, facts, expected in checks:
    try:
        parse_decision(raw, windows, facts)
        accepted = True
    except ValueError:
        accepted = False
    assert accepted == expected
print({"additional_checks_passed": len(checks)})

ACCEPT grounded decision
REJECT decision conflicts with trusted credential facts
REJECT decision is unsupported by supplied policy
REJECT response must be an object with exactly the required fields
REJECT rotation_window_days must be an integer
{'additional_checks_passed': 11}

The second candidate cites P-7 but claims not_eligible for the 45-day-old stale key. The third invents a 90-day threshold. JSON parsing checks syntax; schema checks enforce the permitted shape and values; the policy/fact comparison checks this fixture's decision. A write still requires separate authorization.

Exercise a real SDK without spending tokens

OpenAI's Responses API accepts a strict schema in text.format; its SDK exposes generated text through response.output_text.^{[12]Reference 12OpenAI Python API libraryhttps://github.com/openai/openai-python} The adapter below maps our contract to those fields, verifying completion status and checking for refusals before calling parse_decision.

As checked October 10, 2026, OpenAI Python 3.16.2 uses HTTPX2 by default. We use its native httpx2.MockTransport; the older HTTPX client is only a temporary compatibility path in this SDK.^{[14]Reference 14Migrating to HTTPX2, OpenAI Python 3.16.2https://github.com/openai/openai-python/blob/v3.16.2/httpx2.md} The API key and model ID are placeholders. Four fixtures exercise success, an unsupported policy threshold, refusal, and token-limit truncation. All requests remain local. Run this cell in the same session as the earlier parser and policy cases, with openai==3.16.2 and httpx2==2.13.0 installed.

import httpx2
from openai import OpenAI

DECISION_SCHEMA = {
    "type": "object",
    "properties": {
        "decision": {"type": "string", "enum": ["eligible", "not_eligible"]},
        "rotation_window_days": {"type": "integer"},
        "source_line_ids": {"type": "array", "items": {"type": "string"}},
    },
    "required": ["decision", "rotation_window_days", "source_line_ids"],
    "additionalProperties": False,
}

def request_decision(client, model, account_id, facts):
    response = client.responses.create(
        model=model,
        instructions="Use only the supplied policy and facts. Decide eligibility; do not execute any action.",
        input=json.dumps({
            "policy": {"P-7": "Stale service-account keys at least 30 days old require rotation."},
            "account_id": account_id,
            "trusted_facts": facts,
        }),
        text={"format": {
            "type": "json_schema", "name": "rotation_decision",
            "strict": True, "schema": DECISION_SCHEMA,
        }},
        max_output_tokens=512,
        store=False,
    )
    if response.status != "completed":
        raise ValueError("incomplete_output")
    if any(
        part.type == "refusal"
        for item in response.output if item.type == "message"
        for part in item.content
    ):
        raise ValueError("refusal")
    if not response.output_text:
        raise ValueError("empty_output")
    return parse_decision(response.output_text, {"P-7": 30}, facts)

def fixture(*, window=30, refusal=False, incomplete=False):
    part = (
        {"type": "refusal", "refusal": "Mock refusal"}
        if refusal else
        {"type": "output_text", "annotations": [], "text": json.dumps({
            "decision": "eligible", "rotation_window_days": window,
            "source_line_ids": ["P-7"],
        })}
    )
    return {
        "id": "resp_mock", "object": "response", "created_at": 0,
        "model": "model-under-test", "status": "incomplete" if incomplete else "completed",
        "error": None,
        "incomplete_details": {"reason": "max_output_tokens"} if incomplete else None,
        "output": [{
            "id": "msg_mock", "type": "message", "role": "assistant",
            "status": "incomplete" if incomplete else "completed", "content": [part],
        }],
    }

replies = iter([fixture(), fixture(window=90), fixture(refusal=True), fixture(incomplete=True)])

def mock_http(request):
    payload = json.loads(request.content)
    assert request.url.path == "/v1/responses"
    assert payload["text"]["format"]["strict"] is True
    assert json.loads(payload["input"])["trusted_facts"]["credential_age_days"] == 45
    return httpx2.Response(200, json=next(replies), headers={"x-request-id": "req_mock"})

with OpenAI(
    api_key="mock-key-not-a-secret",
    max_retries=0,
    timeout=5.0,
    http_client=httpx2.Client(transport=httpx2.MockTransport(mock_http)),
) as client:
    for label in ["valid", "wrong_threshold", "refused", "truncated"]:
        try:
            result = request_decision(client, "model-under-test", "acct_10234", {
                "credential_age_days": 45, "key_stale": True,
            })
            print(label, result.decision)
        except ValueError as exc:
            print(label, "REJECT", str(exc))

valid eligible
wrong_threshold REJECT decision is unsupported by supplied policy
refused REJECT refusal
truncated REJECT incomplete_output

Only the first candidate reaches application code as a validated RotationDecision. The other three return explicit failures; the adapter never fabricates fallback decisions.

The mock verifies request mapping and handling of constructed replies. Test the chosen live model and hosted adapter for schema support, policy adherence, and latency. In production, load server credentials, reuse connection pools, and configure one retry owner plus transport and overall deadlines. Here timeout=5.0 is a transport timeout; configure a separate deadline for the full task.

HTTPX2 also changes the default certificate trust to the operating-system store. Check container CA certificates and corporate proxy trust when migrating; the mock doesn't exercise TLS.^{[14]Reference 14Migrating to HTTPX2, OpenAI Python 3.16.2https://github.com/openai/openai-python/blob/v3.16.2/httpx2.md}

As checked October 10, 2026, store=False opts out of Responses application-state storage for this foreground text request. It doesn't itself disable abuse-monitoring logs, approve Zero Data Retention, or change separate tool/cache retention rules. Check the endpoint and organization's actual data controls before sending sensitive data.^{[15]Reference 15Data controls in the OpenAI platformhttps://developers.openai.com/api/docs/guides/your-data}

Server-Sent Events streaming vs buffered responses

You can buffer a response or receive incremental events over HTTP using Server-Sent Events (SSE).^{[16]Reference 16Using server-sent eventshttps://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events[17]Reference 17HTML Standard: Server-sent eventshttps://html.spec.whatwg.org/multipage/server-sent-events.html} An event can contain metadata, text, tool arguments, or a terminal status; it needn't correspond to one model token.

SSE defines UTF-8 event framing with fields such as event:, data:, and id:, separated by blank lines. Browser EventSource uses GET and can reconnect automatically; provider SDKs can read SSE from POST responses. [DONE] isn't an SSE standard. The abbreviated wire example below illustrates Chat Completions text deltas and its sentinel; finish-reason/usage chunks are omitted. The Responses API used in our SDK lab instead exposes typed events such as response.output_text.delta and response.completed.^{[18]Reference 18Streaming API responseshttps://developers.openai.com/api/docs/guides/streaming-responses}

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

data: {"choices": [{"delta": {"content": "Account"}}]}

data: {"choices": [{"delta": {"content": " acct_10234"}}]}

data: {"choices": [{"delta": {"content": " is eligible."}}]}

data: [DONE]

TTFT vs inter-token latency

Streaming splits response time into two distinct metrics:

1. Time-to-first visible text: Duration from dispatch to the first nonempty text delta, often reported as client TTFT. Define the milestone explicitly; a first HTTP byte or metadata event measures something else.
2. Inter-delta latency: Time between received text deltas. True inter-token latency (ITL) concerns individual tokens; deltas may batch several tokens or be split by transport buffering. You can't infer token-level ITL from event spacing alone.

Streaming can let a person start reading before generation ends. Total completion time and TTFT depend on the model, serving stack, and workload, so measure them separately. Automated pipelines can also benefit from early progress, cancellation, or independently validated records, while a single eligibility object is simpler to buffer. Validate the complete JSON object before making a decision.

View larger

The streaming client state machine

A streaming client that can stop safely moves through distinct states:

1. Connect: Request the provider's streaming mode and parse SSE with the SDK or a compliant parser. Network chunks don't necessarily align with events or characters.
2. First text milestone: Measure the first nonempty text delta separately from HTTP bytes, metadata, and heartbeats. Enforce the overall deadline even if the connection remains active.
3. Buffer deltas: Accumulate the intended output item, with memory limits and read timeouts. Real responses can have multiple items and non-text events.
4. Terminal status: Follow the provider protocol. Responses distinguishes completed, incomplete, and failed events; Chat Completions requires finish-reason checks as well as transport framing. Approve a result only after validating terminal status.
5. Application validation: Check refusals, complete output, schema, policy, fresh facts, and authorization. A separate action uses its durable idempotency key.

Test a small normalized, single-text-output event stream. This fixture covers event handling after normalization; wire-level SSE parsing and model quality need separate tests:

from collections.abc import Iterator

def fake_stream(fail: bool = False, terminal: bool = True) -> Iterator[dict]:
    yield {"type": "response.created"}  # Metadata isn't first visible text.
    chunks = ["Account ", "acct_10234 ", "is eligible ", "under P-7."]
    for index, chunk in enumerate(chunks):
        if fail and index == 2:
            raise ConnectionError("mock stream dropped")
        yield {"type": "response.output_text.delta", "delta": chunk}
    if terminal:
        yield {"type": "response.completed"}

def render_draft(events: Iterator[dict]) -> None:
    buffer = ""
    try:
        for event in events:
            if event["type"] == "response.output_text.delta":
                buffer += event["delta"]
                print({"event": "delta", "preview": buffer})
            elif event["type"] == "response.completed":
                print({"event": "completed", "display_only": True, "text": buffer})
                return
            elif event["type"] in {"response.incomplete", "response.failed", "error"}:
                raise ConnectionError("stream did not complete successfully")
        raise ConnectionError("missing terminal event")
    except ConnectionError:
        print({"event": "failed", "display_only": True, "partial_text": buffer})
        return

render_draft(fake_stream())
render_draft(fake_stream(fail=True))
render_draft(fake_stream(terminal=False))

{'event': 'delta', 'preview': 'Account '}
{'event': 'delta', 'preview': 'Account acct_10234 '}
{'event': 'delta', 'preview': 'Account acct_10234 is eligible '}
{'event': 'delta', 'preview': 'Account acct_10234 is eligible under P-7.'}
{'event': 'completed', 'display_only': True, 'text': 'Account acct_10234 is eligible under P-7.'}
{'event': 'delta', 'preview': 'Account '}
{'event': 'delta', 'preview': 'Account acct_10234 '}
{'event': 'failed', 'display_only': True, 'partial_text': 'Account acct_10234 '}
{'event': 'delta', 'preview': 'Account '}
{'event': 'delta', 'preview': 'Account acct_10234 '}
{'event': 'delta', 'preview': 'Account acct_10234 is eligible '}
{'event': 'delta', 'preview': 'Account acct_10234 is eligible under P-7.'}
{'event': 'failed', 'display_only': True, 'partial_text': 'Account acct_10234 is eligible under P-7.'}

The third run receives all text but no successful terminal event, so it still fails. These previews can contain incorrect or sensitive model text; choose whether the product should display them at all. The completed draft remains a preview until validated as a RotationDecision. A real adapter must inspect the terminal response's status and refusal content too.

Production telemetry and cost attribution

After each attempt, record the requested and reported model, latency, provider request ID when available, usage when reported, and outcome. A timeout may leave usage and billing unknown; don't record invented zero-token success. Keep the overall task's retries and validation outcome distinct from individual provider attempts.

The OpenTelemetry GenAI convention snapshot pinned for this lesson is Development. Pin a convention/instrumentation version; don't assume every library emits the same fields.^{[19]Reference 19Semantic conventions for generative client AI spanshttps://github.com/open-telemetry/semantic-conventions-genai/blob/c88d504ab3d9879f8e50d3cc87e69775e11db234/docs/gen-ai/gen-ai-spans.md}

• gen_ai.provider.name: Provider identifier, replacing older examples using gen_ai.system.
• gen_ai.operation.name: Operation, such as chat.
• gen_ai.request.model and gen_ai.response.model: Requested and reported model IDs, when available.
• gen_ai.usage.input_tokens and gen_ai.usage.output_tokens: Reported token usage. These aren't a complete bill; cached, reasoning, media, and tool usage need the provider's accounting rules.
• gen_ai.response.finish_reasons: Terminal reason (e.g. stop, length, content_filter).

One useful trace groups retrieval, each model attempt, and validation under the application request:

[Root Span: POST /api/v1/rotation-decision]
  ├── [Child Span: db.fetch_credential_facts]
  ├── [Child Span: chat provider-model-id (input: 250, output: 80)]
  └── [Child Span: validation.check_p7_policy]

Token accounting and cost attribution

Text-model pricing often separates input and output, with additional provider-specific categories. Prompt caching can reduce repeated-prefix cost and prefill work. Eligibility, minimum lengths, cache-write charges, and rates vary by provider and model; verify usage and current rates in telemetry when estimating savings.^{[20]Reference 20Prompt cachinghttps://developers.openai.com/api/docs/guides/prompt-caching}

At illustrative baseline rates of $2.00 per million input tokens and $6.00 per million output tokens:

These fictional rates support arithmetic practice. At 200,000 identical calls daily for 30 days, the illustrated base cost is $5,880 before retries, cache savings, or other charges. Attribute costs by authorized tenant and task, and reconcile estimates with provider billing.

Cached usage details are subsets of totals. OpenAI's cache-accounting example subtracts cached and cache-write tokens from total input before pricing ordinary input.^{[20]Reference 20Prompt cachinghttps://developers.openai.com/api/docs/guides/prompt-caching} Reasoning-token usage is included in output token usage.^{[21]Reference 21Reasoning modelshttps://developers.openai.com/api/docs/guides/reasoning} Confirm these relationships for each provider adapter.

Here is a constructed text-only usage record with fictional rates, including a separate cache-write rate. Predict which counts should be subtracted and which should stay within the output total:

input_tokens, cached_tokens, cache_write_tokens = 2000, 1200, 100
output_tokens, reasoning_tokens = 200, 80
ordinary_input = input_tokens - cached_tokens - cache_write_tokens
assert ordinary_input >= 0 and 0 <= reasoning_tokens <= output_tokens

# Fictional USD rates per million tokens for arithmetic practice.
cost = (
    ordinary_input * 2.0
    + cached_tokens * 0.5
    + cache_write_tokens * 2.5
    + output_tokens * 6.0
) / 1_000_000
print("ordinary input:", ordinary_input)
print("visible output:", output_tokens - reasoning_tokens)
print(f"estimated cost: ${cost:.6f}")

ordinary input: 700
visible output: 120
estimated cost: $0.003450

Adding 80 reasoning tokens to the 200 output tokens would charge them twice. Pricing all 2,000 input tokens at the ordinary rate and then adding cache charges would also double-count subsets. These assumptions cover this text-only fixture; tools and media can introduce other charges. If a timed-out attempt has no usage record, retain an unknown estimate and reconcile later instead of applying this arithmetic to invented zeros.

Redacting customer text

Prompts can contain customer names, proprietary code, or personal data. OWASP's Top 10 for LLM Applications highlights sensitive information disclosure and prompt injection as top vulnerabilities.^{[22]Reference 22OWASP Top 10 for LLM Applications 2025https://genai.owasp.org/resource/owasp-top-10-for-llm-applications-2025/} Copying raw prompts into log collectors turns your monitoring platform into a high-risk data store.^{[23]Reference 23Handling sensitive datahttps://opentelemetry.io/docs/security/handling-sensitive-data/}

A keyed HMAC can correlate identical input text without storing that text. This lab fingerprints the customer question and omits the rest of the delivered prompt. Keep the real fingerprint key in separate secret storage and scope correlation to its intended tenant/environment. The resulting 128-bit fingerprint remains linkable data, and truncation can introduce collisions.

import hmac
from hashlib import sha256

trace_id = "trace_7e3b"
customer_question = "Can I rotate the stale service-account key for account acct_10234?"
fingerprint_key = b"dev-only-fingerprint-key"
usage = {"input_tokens": 250, "output_tokens": 80}
rates_per_million = {"input": 2.0, "output": 6.0}
model = "provider-model-id"
latency_ms = 840
retry_count = 0
cached_input_tokens = 0
cost_usd = (
    usage["input_tokens"] * rates_per_million["input"]
    + usage["output_tokens"] * rates_per_million["output"]
) / 1_000_000
log_record = {
    "trace_id": trace_id,
    "task": "decide_rotation",
    "model": model,
    "prompt_version": "rotation_decision@2",
    "latency_ms": latency_ms,
    "retry_count": retry_count,
    "question_fingerprint": hmac.new(fingerprint_key, customer_question.encode(), sha256).hexdigest()[:32],
    "cached_input_tokens": cached_input_tokens,
    **usage,
    "estimated_cost_usd": round(cost_usd, 6),
    "status": "ok",
}

assert customer_question not in str(log_record)
print(log_record)

{'trace_id': 'trace_7e3b', 'task': 'decide_rotation', 'model': 'provider-model-id', 'prompt_version': 'rotation_decision@2', 'latency_ms': 840, 'retry_count': 0, 'question_fingerprint': '85ef707a963bf1208e6621899bd3c995', 'cached_input_tokens': 0, 'input_tokens': 250, 'output_tokens': 80, 'estimated_cost_usd': 0.00098, 'status': 'ok'}

Choosing the execution shape

The server boundary supports different execution patterns depending on user expectations and latency constraints. Caching is an optimization usable with several patterns:

Drafts can benefit from streaming, one eligibility object can be buffered, and nightly scans can use a queue or supported batch API. The separate action gate must handle authorization and idempotency atomically; schema validation checks request shape.

Carry the boundary into an app

Production systems require an authenticated backend environment. Before exposing /rotation-decision to the web, authenticate the user session, verify tenant permissions, and apply rate limiting at your API gateway. Hiding a provider key on the server doesn't protect you if an unprotected public endpoint lets anyone burn your budget through your proxy.

The local labs exercise configuration checks, parsing, constructed failures, retry arithmetic, bounded recovery probes, event handling, and a mocked SDK adapter. Live reliability and the deployed secret boundary require deployment tests. Before exposing the route, test the assembled wrapper under slow connections, cancellation, concurrency, and provider-specific failures. The user-facing result also needs a complete HTTP path, a trace written before the provider call, and a message that distinguishes a lost response from a policy outcome.