100%

Page 2 of 22 · 498 min left

Chapter 1 · LLM APIs and a Provider-Neutral Client

Adapters, Configuration, and Error Handling

Page 2 of 22
21 min read

The last page gave EvidenceDesk a provider-neutral client. Three questions stayed open: where do the model name and the key live, what happens when a call fails, and how do adapters stay honest as providers change? This page answers them with config.py, errors.py and a mini-lab: a summariser that works with any provider and passes its tests with none.

What you will learn

How to keep model IDs, timeouts and provider choice in validated settings, and secrets out of source code.
How to tell retryable failures from invalid requests, authentication failures and exhausted quotas, and retry with a hard bound.
How to compare providers, test with no credentials, pin versions and keep fixtures.

Record model and adapter configuration

A model ID, a timeout or an if provider == ... check inside business code makes every change touch that code. Keep them in one validated object built at start-up. load_settings takes the environment as a Mapping, so a test passes a dictionary and never touches os.environ. The default is provider="fake", so the project runs offline before anything is configured. A real provider must name its model; the file holds no model ID, because IDs change.

The provider is a Literal, so a typo fails at start-up, and the mini-lab picks an adapter from a registry, not from scattered conditionals.

Python

# file: evidencedesk/config.py
"""Validated settings. The environment is passed in as a mapping, so tests never touch os.environ."""
from __future__ import annotations

from collections.abc import Mapping
from pathlib import Path
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, SecretStr, model_validator

PREFIX = "EVIDENCEDESK_"
NEEDS_KEY = ("openai", "anthropic", "gemini")  # "local" and "fake" need no key


class Settings(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")
    provider: Literal["fake", "openai", "anthropic", "gemini", "local"] = "fake"
    model: str = "fake-model"
    timeout_s: float = Field(30.0, gt=0, le=600)
    max_retries: int = Field(2, ge=0, le=5)
    api_key: SecretStr | None = None  # prints as ********** in repr and logs

    @model_validator(mode="after")
    def _check_provider(self) -> Settings:
        if self.provider != "fake" and self.model == "fake-model":
            raise ValueError(f"set {PREFIX}MODEL: copy the model id from the provider's documentation")
        if self.provider in NEEDS_KEY and self.api_key is None:
            raise ValueError(f"provider {self.provider!r} needs {PREFIX}API_KEY or {PREFIX}API_KEY_FILE")
        return self


def load_settings(env: Mapping[str, str]) -> Settings:
    """Build Settings from an environment mapping. An empty value counts as not set."""
    data: dict[str, str] = {}
    for field in ("provider", "model", "timeout_s", "max_retries", "api_key"):
        if env.get(PREFIX + field.upper()):
            data[field] = env[PREFIX + field.upper()]
    key_file = env.get(PREFIX + "API_KEY_FILE")
    if "api_key" not in data and key_file:
        data["api_key"] = Path(key_file).read_text(encoding="utf-8").strip()
    return Settings(**data)

An empty environment, a complete one and two broken ones (errors as field path and message only):

Python

from pydantic import ValidationError

from evidencedesk.config import load_settings

print(load_settings({}))
secret = "demo-key-not-real"
ready = load_settings({"EVIDENCEDESK_PROVIDER": "openai", "EVIDENCEDESK_MODEL": "model-id-from-the-docs",
                       "EVIDENCEDESK_API_KEY": secret})
print(ready)
print(secret in repr(ready), ready.api_key.get_secret_value() == secret)
for env in ({"EVIDENCEDESK_TIMEOUT_S": "0"}, {"EVIDENCEDESK_PROVIDER": "gemini", "EVIDENCEDESK_MODEL": "m1"}):
    try:
        load_settings(env)
    except ValidationError as error:
        print([(e["loc"], e["msg"]) for e in error.errors()])

Output

provider='fake' model='fake-model' timeout_s=30.0 max_retries=2 api_key=None
provider='openai' model='model-id-from-the-docs' timeout_s=30.0 max_retries=2 api_key=SecretStr('**********')
False True
[(('timeout_s',), 'Input should be greater than 0')]
[((), "Value error, provider 'gemini' needs EVIDENCEDESK_API_KEY or EVIDENCEDESK_API_KEY_FILE")]

Notice that the empty environment is valid, SecretStr hides the key while get_secret_value() returns it, a bad value is reported against its field, and a rule spanning fields has an empty path.

Keep secrets out of source code

A committed key is a leaked key: history keeps it even after a deleting commit. Read the key at run time from an environment variable (EVIDENCEDESK_API_KEY) or from a file whose path is in the environment (EVIDENCEDESK_API_KEY_FILE, like container secret mounts). Keep .env out of git, commit an .env.example with empty values, never print a key.

Output

# file: .gitignore
.env
.env.*
!.env.example
*.key
secrets/
.venv/
__pycache__/

Rules are not enough, so the tests include a key detector, a regular expression for key-shaped strings. One test shows it fires on a made-up key; another scans our source, tests and fixtures. It is a safety net, not a secret scanner. If a real key is ever committed, rotate it first and clean history second.

Handle timeouts, rate limits and provider errors

Retrying a wrong request never fixes it, and a spent quota only costs time, so the reaction depends on the kind:

Kind	Typical signal	Retry?
`timeout`	no reply in time; HTTP 408, 504	yes, bounded
`rate_limited`	HTTP 429, often with `Retry-After`	yes, after the wait
`server`	HTTP 500 and above	yes, bounded
`auth`	HTTP 401, 403	no: alert the operator
`invalid_request`	HTTP 400, 404, 413, 422	no: fix the request
`quota_exhausted`	HTTP 402, or a 429 whose body names credit or a spend limit	no: alert the owner
`unknown`	anything else	no: log, fail loudly

The hard row is 429: usually a rate limit, which clears, but some providers also use it for used-up credit, which does not. The body tells them apart, so from_status looks for marker strings. That list is our heuristic; an unmarked 429 counts as a rate limit and costs at most a few bounded attempts.

Provider note (as of 2026-10-12): the SDKs of the first two providers raise one exception class per status, plus connection and timeout classes. At the time of writing both retry 408, 409, 429, 500 and above and connection errors twice by default, time out after 10 minutes and accept max_retries. The first advises honouring Retry-After, else backoff with jitter and a cap, and says a 429 can mean used-up credit or a spend limit (credit_balance_exhausted): do not retry that. The second documents 402 as billing, 529 as overloaded, 504 as timeout, and a spend-cap 429 without retry-after. The third documents 429, 503, 504 and no rate-limit headers; its exception names and default timeout, and the local runtime's timeout and retries, are unconfirmed: assume no built-in retry. Check the current documentation.

Normalise responses and errors

Adapters return one LLMResponse; failure gets one LLMError with a kind, retryable and an optional retry_after. from_status(status, headers, body) uses only HTTP facts, so it serves every provider. The retry is:

Bounded: max_attempts counts every try; then the last error is raised.
Exponential backoff with jitter: 0.5, 1, 2 ... seconds up to max_delay, each cut by a random share so clients do not retry in step.
The server wins: a longer Retry-After is used; one beyond max_delay means give up, not hammer.
Injected sleep and rng: tests pass a list's append and a seeded random.Random.

RetryingClient wraps any LLMClient and is one itself.

Python

# file: evidencedesk/errors.py
from __future__ import annotations

import random
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass

from evidencedesk.llm import LLMClient, LLMRequest, LLMResponse

RETRYABLE = {"timeout", "rate_limited", "server"}
QUOTA_MARKERS = ("credit_balance_exhausted", "spend_limit", "quota_exceeded")  # our heuristic


class LLMError(Exception):
    def __init__(self, kind: str, message: str = "", retry_after: float | None = None):
        super().__init__(f"{kind}: {message}" if message else kind)
        self.kind, self.retry_after, self.attempts = kind, retry_after, 1

    @property
    def retryable(self) -> bool:
        return self.kind in RETRYABLE


def from_status(status: int, headers: Mapping[str, str] | None = None, body: str = "") -> LLMError:
    lowered = {k.lower(): v for k, v in (headers or {}).items()}
    try:
        retry_after = max(0.0, float(lowered["retry-after"]))  # seconds only; a date is ignored
    except (KeyError, ValueError):
        retry_after = None
    if status == 429 and any(marker in body for marker in QUOTA_MARKERS):
        kind = "quota_exhausted"
    elif status in (408, 504):
        kind = "timeout"
    elif status == 429:
        kind = "rate_limited"
    elif status == 402:
        kind = "quota_exhausted"
    elif status in (401, 403):
        kind = "auth"
    elif status in (400, 404, 413, 422):
        kind = "invalid_request"
    elif status >= 500:
        kind = "server"
    else:
        kind = "unknown"
    return LLMError(kind, f"HTTP {status}", retry_after)


@dataclass(frozen=True)
class RetryPolicy:
    max_attempts: int = 3  # every try counts, the first too
    base_delay: float = 0.5
    factor: float = 2.0
    max_delay: float = 8.0  # the longest wait we accept
    jitter: float = 0.5  # share of each wait that is random

    def wait_for(self, attempt: int, error: LLMError, rng: random.Random) -> float | None:
        backoff = min(self.max_delay, self.base_delay * self.factor ** (attempt - 1))
        wait = backoff * (1 - self.jitter * rng.random())
        wait = max(wait, error.retry_after or 0.0)  # the server's word beats our guess
        return wait if wait <= self.max_delay else None


def call_with_retry(fn: Callable, policy: RetryPolicy = RetryPolicy(),
                    sleep: Callable[[float], None] = time.sleep, rng: random.Random | None = None):
    rng = rng or random.Random()
    for attempt in range(1, policy.max_attempts + 1):
        try:
            return fn()
        except LLMError as error:
            error.attempts = attempt
            wait = policy.wait_for(attempt, error, rng) if error.retryable else None
            if wait is None or attempt == policy.max_attempts:
                raise
            sleep(wait)
    raise ValueError("max_attempts must be at least 1")


class RetryingClient:
    """Wraps any LLMClient and is one itself."""

    def __init__(self, inner: LLMClient, policy: RetryPolicy = RetryPolicy(),
                 sleep: Callable[[float], None] = time.sleep, rng: random.Random | None = None) -> None:
        self.inner, self.policy, self.sleep, self.rng = inner, policy, sleep, rng

    def complete(self, request: LLMRequest) -> LLMResponse:
        return call_with_retry(lambda: self.inner.complete(request), self.policy, self.sleep, self.rng)

Two 429s, a call that fails twice, then one that never recovers. Sleeping only appends to a list:

Python

import random

from evidencedesk.errors import LLMError, RetryPolicy, call_with_retry, from_status

print(from_status(429, {"Retry-After": "3"}).kind, from_status(429, {}, "credit_balance_exhausted").kind)
tries, waits = [], []


def flaky():
    tries.append(1)
    if len(tries) < 3:
        raise from_status(503)
    return "ok"


print(call_with_retry(flaky, RetryPolicy(), waits.append, random.Random(7)), len(tries), [round(w, 2) for w in waits])


def always_down():
    raise from_status(500)


more = []
try:
    call_with_retry(always_down, RetryPolicy(max_attempts=3), more.append, random.Random(7))
except LLMError as error:
    print("gave up:", error, "after", error.attempts, "attempts;", len(more), "waits")

Output

rate_limited quota_exhausted
ok 3 [0.42, 0.92]
gave up: server: HTTP 500 after 3 attempts; 2 waits

Notice that the bodies tell the 429s apart, that 0.42 and 0.92 are the 0.5 and 1.0 second backoffs cut by seeded jitter, and that the hopeless call stopped after three attempts.

Use a provider capability matrix

A shared feature name can hide differences. This matrix is qualitative: "documented" means the documentation describes it, not that behaviour or limits match.

Capability	First	Second	Third	Local runtime
Structured output	documented; a strict subset of JSON Schema	documented; another subset	documented; another subset	documented; validate anyway
Streaming, tool calls	documented; tool arguments as a JSON string	documented; tool arguments as an object	documented	documented; tool support depends on the model [VERIFY]
Embeddings	yes	no first-party model	yes	yes
Usage metadata	own names; in a stream [VERIFY]	own names; cumulative in a stream	names differ per API path; stream [VERIFY]	on the plain reply; stream [VERIFY]

So the shared contract covers text in, text and usage out; the rest are optional capabilities that adapters declare and the application checks.

Provider note (as of 2026-10-12): the second provider has no first-party embedding model and points to a partner service, so later embedding adapters cover the first and third providers and a local runtime. The third has two API paths (the Interactions API, which its documentation leads with, and the older generateContent) with different usage field names. Check the current documentation.

Test the no-credential path

With an empty environment the application builds and every test passes.

Python

# file: tests/test_config.py
import re
from pathlib import Path

import pytest
from pydantic import ValidationError

from evidencedesk.config import load_settings

ROOT = Path(__file__).resolve().parent.parent
KEY_PATTERN = re.compile(r"sk-[A-Za-z0-9_-]{16,}|AIza[0-9A-Za-z_-]{20,}|-----BEGIN [A-Z ]*PRIVATE KEY")
REAL = {"EVIDENCEDESK_PROVIDER": "anthropic", "EVIDENCEDESK_MODEL": "m1"}


def fields(env):
    with pytest.raises(ValidationError) as info:
        load_settings(env)
    return [".".join(map(str, e["loc"])) for e in info.value.errors()]


def test_empty_environment_means_offline_fake():
    s = load_settings({})
    assert (s.provider, s.model, s.api_key, s.max_retries) == ("fake", "fake-model", None, 2)


def test_the_key_comes_from_env_or_file_and_never_appears_in_repr(tmp_path):
    secret = "sk-" + "a" * 24  # made up
    (tmp_path / "key.txt").write_text(secret, encoding="utf-8")
    for source in ({"EVIDENCEDESK_API_KEY": secret}, {"EVIDENCEDESK_API_KEY_FILE": str(tmp_path / "key.txt")}):
        s = load_settings({**REAL, **source})
        assert secret not in repr(s) + str(s) + s.model_dump_json()
        assert s.api_key.get_secret_value() == secret


def test_bad_settings_are_reported_by_field():
    assert fields({"EVIDENCEDESK_PROVIDER": "nope"}) == ["provider"]
    assert fields({"EVIDENCEDESK_TIMEOUT_S": "0"}) == ["timeout_s"]
    assert fields(REAL) == [""]  # a real provider without a key
    assert fields({"EVIDENCEDESK_PROVIDER": "local"}) == [""]  # a real provider without a model


def test_detector_finds_a_fake_key_and_the_source_tree_has_none():
    assert KEY_PATTERN.search("API_KEY = 'sk-" + "b" * 20 + "'")
    for path in [*ROOT.glob("evidencedesk/*.py"), *ROOT.glob("tests/*.py"), *ROOT.glob("fixtures/*.json")]:
        assert not KEY_PATTERN.search(path.read_text(encoding="utf-8")), path.name

Python

# file: tests/test_errors.py
import pytest

from evidencedesk.errors import LLMError, RetryPolicy, call_with_retry, from_status

NO_JITTER = RetryPolicy(max_attempts=4, jitter=0)


@pytest.mark.parametrize("status, headers, body, kind, retryable, retry_after", [
    (400, {}, "", "invalid_request", False, None),
    (401, {}, "", "auth", False, None),
    (429, {"Retry-After": "3"}, "", "rate_limited", True, 3.0),
    (429, {}, "credit_balance_exhausted", "quota_exhausted", False, None),
    (500, {}, "", "server", True, None),
])
def test_status_mapping(status, headers, body, kind, retryable, retry_after):
    error = from_status(status, headers, body)
    assert (error.kind, error.retryable, error.retry_after) == (kind, retryable, retry_after)


class Flaky:
    def __init__(self, *failures):
        self.failures, self.calls = list(failures), 0

    def __call__(self):
        self.calls += 1
        if self.failures:
            raise self.failures.pop(0)
        return "ok"


def run(fn, policy=NO_JITTER):
    slept = []
    try:
        return call_with_retry(fn, policy, slept.append), slept
    except LLMError as error:
        return error, slept


def test_recovers_and_never_really_sleeps():
    fn = Flaky(from_status(503), from_status(503))
    assert run(fn) == ("ok", [0.5, 1.0]) and fn.calls == 3  # base 0.5, doubled once


def test_attempts_are_bounded():
    fn = Flaky(*[from_status(500)] * 10)
    error, slept = run(fn)
    assert fn.calls == 4 and len(slept) == 3 and error.attempts == 4


def test_permanent_errors_are_not_retried():
    fn = Flaky(from_status(401), from_status(401))
    assert run(fn)[1] == [] and fn.calls == 1


def test_server_wait_wins_but_only_up_to_the_cap():
    assert run(Flaky(from_status(429, {"retry-after": "2"})))[1] == [2.0]
    fn = Flaky(from_status(429, {"retry-after": "60"}))
    assert run(fn)[1] == [] and fn.calls == 1  # 60 s is over max_delay: give up, do not hammer

Bash

python -m pytest -q tests/test_config.py tests/test_errors.py

Output

.............                                                            [100%]
13 passed

Wall-clock seconds are stripped from pytest lines on this page (sed 's/ in [0-9.]*s//'). Nothing needed a key, a network or a real sleep.

Understand model and SDK version drift

SDKs change names and defaults; models are renamed and retired. Pin what you tested and say what you did not. Isolate version-sensitive code: field names live only in the small from_... functions and adapters of the mini-lab, so an SDK change touches one place. Never hard-code a model ID; it comes from settings.

Output

# file: requirements.txt
# Tested here with Python 3.13; the target is Python 3.12.
pydantic==2.14.1
pytest==9.1.1
# openai==3.29.0  (optional adapters: latest on 2026-10-12, NOT tested here)
# anthropic==1.13.0

Provider note (as of 2026-10-12): latest on PyPI that day: openai 3.29.0, anthropic 1.13.0, google-genai 2.29.0, ollama 0.6.3. Version 1 of the second SDK removed temperature, top_p and top_k (passing them raises TypeError); the third SDK's README says to pin below 3.0.0 because a future release changes automatic function calling. The second provider documents at least 60 days' notice before retiring a model. Check the current documentation.

Create recorded provider fixtures

A provider fixture is a small response shaped like a provider's, so mapping code can be tested without the provider. Keep it small, strip anything personal (ids, keys) and say how it was made. These were written by hand after the fields our adapters read; nothing was captured from a real service and the values are invented. Replace them with sanitised real captures once an adapter has run for real.

JSON

# file: fixtures/provider_samples.json
{
  "_note": "Hand-written. Not captured from a real service.",
  "first": {"status": "completed", "output_text": "Returns: 30 days, full refund.",
            "usage": {"input_tokens": 52, "output_tokens": 9}},
  "second": {"stop_reason": "end_turn",
             "content": [{"type": "text", "text": "Returns: 30 days, "}, {"type": "text", "text": "full refund."}],
             "usage": {"input_tokens": 52, "output_tokens": 9}},
  "errors": [
    {"name": "rate limit with a wait", "status": 429, "headers": {"Retry-After": "3"},
     "body": "{\"error\": {\"type\": \"rate_limit_error\"}}", "kind": "rate_limited"},
    {"name": "credit used up", "status": 429, "headers": {},
     "body": "{\"error\": {\"code\": \"credit_balance_exhausted\"}}", "kind": "quota_exhausted"},
    {"name": "overloaded", "status": 529, "headers": {}, "body": "overloaded_error", "kind": "server"}
  ]
}

Build a minimal API comparison report

Record functional differences, operational constraints and unknowns; mark every unknown [VERIFY].

Area	Finding	Status
Functional	System text goes in `instructions` (first), a top-level `system` (second) or a message (local); stop reasons use different words	documented
Operational	First and second SDKs retry twice by default (third: secondary source); rate-limit headers for first and second only; prices and limits change	partly [VERIFY]
Unknown	Usage in a stream; third SDK exceptions and timeout; local timeout; local tool support	[VERIFY]

Mini-lab: Provider-Neutral Text Summariser

summary.py holds summarise, one normaliser per provider shape, guard (SDK failure into LLMError) and a registry-based build_client that adds a RetryingClient. summarise enforces its word limit in code: a prompt is not a guardrail.

Python

# file: evidencedesk/summary.py
from __future__ import annotations

from collections.abc import Callable

from evidencedesk.config import Settings
from evidencedesk.errors import LLMError, RetryingClient, RetryPolicy, from_status
from evidencedesk.llm import FakeClient, LLMClient, LLMRequest, LLMResponse, Message, Usage

DEMO_TEXT = ("You can return a new, unridden bike within 30 days of delivery for a full refund. We refund the "
             "price of the bike; the delivery fee is refunded only if the bike arrived damaged.")
DEMO_FIXTURES = {DEMO_TEXT: {"text": "A new, unridden bike can be returned within 30 days for a full refund; "
                                     "the delivery fee is refunded only if it arrived damaged.",
                             "usage": {"input_tokens": 52, "output_tokens": 24}}}


def summarise(client: LLMClient, text: str, max_words: int = 30) -> str:
    """The word limit is enforced here, in code, not only requested in the prompt."""
    text = text.strip()
    if not text:
        raise ValueError("nothing to summarise")
    system = f"Summarise the text in at most {max_words} words. Use only the text."
    response = client.complete(LLMRequest((Message("system", system), Message("user", text))))
    words = response.text.split()
    return " ".join(words[:max_words]) + (" ..." if len(words) > max_words or response.stop_reason == "length" else "")


# Version-sensitive: only these functions know a provider's field names.
def split_system(request: LLMRequest) -> tuple[str, str]:
    def join(role: str) -> str:
        return "\n".join(m.content for m in request.messages if m.role == role)

    return join("system"), join("user")


def from_first(r, model: str) -> LLMResponse:  # r: an object with .output_text, .usage, .status
    stop = "length" if r.status == "incomplete" else "end" if r.status == "completed" else "other"
    return LLMResponse(r.output_text, model, Usage(r.usage.input_tokens, r.usage.output_tokens), stop)


def from_second(r, model: str) -> LLMResponse:  # r: an object with .content blocks, .usage, .stop_reason
    text = "".join(b.text for b in r.content if b.type == "text")
    return LLMResponse(text, model, Usage(r.usage.input_tokens, r.usage.output_tokens),
                       {"end_turn": "end", "max_tokens": "length"}.get(r.stop_reason, "other"))


def guard(call: Callable[[], object], timeout_type: type[Exception]):
    """Run an SDK call and translate its failures."""
    try:
        return call()
    except timeout_type as e:
        raise LLMError("timeout", str(e)) from e
    except Exception as e:
        if getattr(e, "status_code", None) is None:
            raise
        raise from_status(e.status_code, dict(e.response.headers), e.response.text) from e


REGISTRY: dict[str, Callable[[Settings], LLMClient]] = {"fake": lambda s: FakeClient(DEMO_FIXTURES)}


def build_client(settings: Settings) -> LLMClient:
    if settings.provider not in REGISTRY:
        raise ValueError(f"no adapter registered for {settings.provider!r}")
    return RetryingClient(REGISTRY[settings.provider](settings), RetryPolicy(max_attempts=settings.max_retries + 1))

The tests use fakes, the fixtures and an empty environment (the fake summary is text a person wrote, not model output); obj gives a fixture attributes, as an SDK result has:

Python

# file: tests/test_summariser.py
import json
from pathlib import Path
from types import SimpleNamespace

import pytest

from evidencedesk.config import load_settings
from evidencedesk.errors import LLMError, from_status
from evidencedesk.llm import FakeClient, LLMClient
from evidencedesk.summary import DEMO_TEXT, build_client, from_first, from_second, guard, summarise

SAMPLES = json.loads((Path(__file__).parent.parent / "fixtures" / "provider_samples.json").read_text(encoding="utf-8"))


def obj(d):  # dictionary -> object with attributes, like an SDK result
    return json.loads(json.dumps(d), object_hook=lambda x: SimpleNamespace(**x))


def test_empty_environment_gives_a_working_offline_summariser():
    client = build_client(load_settings({}))
    assert isinstance(client, LLMClient)
    assert summarise(client, DEMO_TEXT).startswith("A new, unridden bike")


def test_word_limit_is_enforced_in_code():
    long = {"text": "one two three four five six", "usage": {"input_tokens": 1, "output_tokens": 6}}
    assert summarise(FakeClient({"x": long}), "x", max_words=4) == "one two three four ..."


def test_two_provider_shapes_become_one_result():
    results = [from_first(obj(SAMPLES["first"]), "m"), from_second(obj(SAMPLES["second"]), "m")]
    assert results[0] == results[1]  # same text, usage and stop reason from two different shapes
    assert results[0].text == "Returns: 30 days, full refund." and results[0].usage.total_tokens == 61


@pytest.mark.parametrize("case", SAMPLES["errors"], ids=lambda c: c["name"])
def test_recorded_failures_map_to_the_expected_kind(case):
    assert from_status(case["status"], case["headers"], case["body"]).kind == case["kind"]


def test_guard_translates_sdk_failures():
    class Boom(Exception):
        status_code, response = 429, SimpleNamespace(headers={"retry-after": "2"}, text="")

    def fail():
        raise Boom()

    with pytest.raises(LLMError) as info:
        guard(fail, TimeoutError)
    assert (info.value.kind, info.value.retry_after) == ("rate_limited", 2.0)

Bash

python -m pytest -q tests/test_summariser.py

Output

.......                                                                  [100%]
7 passed

Two optional adapters, at most 12 lines each. To use one, add it to REGISTRY and set EVIDENCEDESK_PROVIDER, _MODEL and the key. They cannot run here, so there is no output.

Python

# not run here: needs the SDK, a key and internet
import openai
from evidencedesk.summary import from_first, guard, split_system

class FirstAdapter:
    def __init__(self, s):  # max_retries=0: our RetryingClient retries
        self.model = s.model
        self.api = openai.OpenAI(api_key=s.api_key.get_secret_value(), timeout=s.timeout_s, max_retries=0)
    def complete(self, request):
        system, user = split_system(request)
        call = lambda: self.api.responses.create(model=self.model, instructions=system, input=user)
        return from_first(guard(call, openai.APITimeoutError), self.model)

Python

# not run here: needs the SDK, a key and internet
import anthropic
from evidencedesk.summary import from_second, guard, split_system

class SecondAdapter:
    def __init__(self, s):
        self.model = s.model
        self.api = anthropic.Anthropic(api_key=s.api_key.get_secret_value(), timeout=s.timeout_s, max_retries=0)
    def complete(self, request):
        system, user = split_system(request)
        ask = dict(model=self.model, max_tokens=request.max_tokens, system=system, messages=[{"role": "user", "content": user}])
        return from_second(guard(lambda: self.api.messages.create(**ask), anthropic.APITimeoutError), self.model)

Provider note (as of 2026-10-12): the sketches use responses.create(model=..., instructions=..., input=...) for the first provider and messages.create(model=..., max_tokens=..., system=..., messages=[...]) for the second, and pass max_retries=0. The timeout argument name, and whether the second SDK's status exception has status_code and response like the first, were not confirmed [VERIFY]. Copy a current model ID from the provider's model page. Check the current documentation.

What these tests cannot show: that a real provider still answers in the shape of a hand-written fixture. They catch mapping bugs, not provider changes, so run each adapter once for real. The next page builds what goes into complete: instructions, user text and retrieved evidence.

Where this shows up at work

An overnight batch hits a rate limit and either waits and finishes or dies half done. An invoice jumps because an unbounded retry loop ran through a provider incident.

Common mistakes

Retrying every error. A bad key or invalid request fails the same way each time and you pay for each try. Retry only timeouts, rate limits and server errors.
No bound, no delay. An endless or instant retry loop worsens an outage. Bound, back off, add jitter.
Stacked retries. SDK and code both retry: one failure becomes nine calls. One layer owns the policy.
Secrets in code, logs or tests. Use the environment and SecretStr; rotate any key ever committed.

Try it yourself

Easy. Show that EVIDENCEDESK_MAX_RETRIES=9 is rejected, and which field the error names.
Medium. Revisit Provider-Neutral LLM Clients. Copy its ShapeBFake so it fails once with a 429 and Retry-After: 1, wrap it in RetryingClient, and check that answer_question works and the recorded wait is one second.
Hard. Revisit Git, APIs and Backend for AI: Beginner. Your API must not pass a provider status through (a bad provider key is not the caller's 401). Write and test public_status(error), mapping each kind to your service's status.

Answers

Easy: load_settings({"EVIDENCEDESK_MAX_RETRIES": "9"}) raises a ValidationError for max_retries (the limit is 5). Medium and Hard:

Python

# file: tests/test_exercises_p02.py
from test_llm import Q_RETURN, ShapeBFake

from evidencedesk.errors import RetryingClient, RetryPolicy, from_status
from evidencedesk.llm import answer_question


class OnceLimited(ShapeBFake):
    todo = [from_status(429, {"Retry-After": "1"})]

    def complete(self, request):
        if self.todo:
            raise self.todo.pop()
        return super().complete(request)


def test_medium_retry_wrapper_around_an_earlier_fake():
    slept = []
    client = RetryingClient(OnceLimited(), RetryPolicy(jitter=0), slept.append)
    assert answer_question(client, Q_RETURN).startswith("You can return") and slept == [1.0]


def public_status(error):  # never pass a provider 401 on: it is our key, not the caller's
    return {"timeout": 504, "rate_limited": 503, "server": 503, "auth": 502, "quota_exhausted": 502}.get(error.kind, 500)


def test_hard_provider_errors_become_our_own_http_statuses():
    assert [public_status(from_status(s)) for s in (401, 429, 400)] == [502, 503, 500]

Run python -m pytest -q tests/test_exercises_p02.py; two tests should pass. Pass checklist:

Easy: the error names max_retries; config.py is unchanged.
Medium: nothing really sleeps; the wait is 1.0 because the server's value beats our 0.5 second guess.
Hard: auth and quota failures become a server-side status (not 401); other defensible mappings are fine if you can explain them.

Summary

Keep provider, model, timeout and retry limits in validated settings built from a mapping; the fake is the offline default.
Keep secrets in the environment or a secret file, hide them (SecretStr), ignore the files, test for key-shaped strings.
Normalise failures into one LLMError: retry timeouts, rate limits and server errors with bounded, jittered backoff; never retry invalid requests, authentication failures or exhausted quotas.
Compare providers with a capability matrix and a report; mark unknowns [VERIFY].
Pin tested versions, isolate version-sensitive mapping, keep honest fixtures, test with no credentials.

Next: Context Assembly and Trust Boundaries. You will build the model input from named components and keep trusted instructions apart from untrusted text.

AI Engineering with LLMs: Beginner

100%

Page 2 of 22 · 498 min left

Chapter 1 · LLM APIs and a Provider-Neutral Client

Adapters, Configuration, and Error Handling

Page 2 of 22
21 min read

What you will learn

How to keep model IDs, timeouts and provider choice in validated settings, and secrets out of source code.
How to tell retryable failures from invalid requests, authentication failures and exhausted quotas, and retry with a hard bound.
How to compare providers, test with no credentials, pin versions and keep fixtures.

Record model and adapter configuration

The provider is a Literal, so a typo fails at start-up, and the mini-lab picks an adapter from a registry, not from scattered conditionals.

Python

# file: evidencedesk/config.py
"""Validated settings. The environment is passed in as a mapping, so tests never touch os.environ."""
from __future__ import annotations

from collections.abc import Mapping
from pathlib import Path
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, SecretStr, model_validator

PREFIX = "EVIDENCEDESK_"
NEEDS_KEY = ("openai", "anthropic", "gemini")  # "local" and "fake" need no key


class Settings(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")
    provider: Literal["fake", "openai", "anthropic", "gemini", "local"] = "fake"
    model: str = "fake-model"
    timeout_s: float = Field(30.0, gt=0, le=600)
    max_retries: int = Field(2, ge=0, le=5)
    api_key: SecretStr | None = None  # prints as ********** in repr and logs

    @model_validator(mode="after")
    def _check_provider(self) -> Settings:
        if self.provider != "fake" and self.model == "fake-model":
            raise ValueError(f"set {PREFIX}MODEL: copy the model id from the provider's documentation")
        if self.provider in NEEDS_KEY and self.api_key is None:
            raise ValueError(f"provider {self.provider!r} needs {PREFIX}API_KEY or {PREFIX}API_KEY_FILE")
        return self


def load_settings(env: Mapping[str, str]) -> Settings:
    """Build Settings from an environment mapping. An empty value counts as not set."""
    data: dict[str, str] = {}
    for field in ("provider", "model", "timeout_s", "max_retries", "api_key"):
        if env.get(PREFIX + field.upper()):
            data[field] = env[PREFIX + field.upper()]
    key_file = env.get(PREFIX + "API_KEY_FILE")
    if "api_key" not in data and key_file:
        data["api_key"] = Path(key_file).read_text(encoding="utf-8").strip()
    return Settings(**data)

An empty environment, a complete one and two broken ones (errors as field path and message only):

Python

from pydantic import ValidationError

from evidencedesk.config import load_settings

print(load_settings({}))
secret = "demo-key-not-real"
ready = load_settings({"EVIDENCEDESK_PROVIDER": "openai", "EVIDENCEDESK_MODEL": "model-id-from-the-docs",
                       "EVIDENCEDESK_API_KEY": secret})
print(ready)
print(secret in repr(ready), ready.api_key.get_secret_value() == secret)
for env in ({"EVIDENCEDESK_TIMEOUT_S": "0"}, {"EVIDENCEDESK_PROVIDER": "gemini", "EVIDENCEDESK_MODEL": "m1"}):
    try:
        load_settings(env)
    except ValidationError as error:
        print([(e["loc"], e["msg"]) for e in error.errors()])

Output

provider='fake' model='fake-model' timeout_s=30.0 max_retries=2 api_key=None
provider='openai' model='model-id-from-the-docs' timeout_s=30.0 max_retries=2 api_key=SecretStr('**********')
False True
[(('timeout_s',), 'Input should be greater than 0')]
[((), "Value error, provider 'gemini' needs EVIDENCEDESK_API_KEY or EVIDENCEDESK_API_KEY_FILE")]

Notice that the empty environment is valid, SecretStr hides the key while get_secret_value() returns it, a bad value is reported against its field, and a rule spanning fields has an empty path.

Keep secrets out of source code

Output

# file: .gitignore
.env
.env.*
!.env.example
*.key
secrets/
.venv/
__pycache__/

Handle timeouts, rate limits and provider errors

Retrying a wrong request never fixes it, and a spent quota only costs time, so the reaction depends on the kind:

Kind	Typical signal	Retry?
`timeout`	no reply in time; HTTP 408, 504	yes, bounded
`rate_limited`	HTTP 429, often with `Retry-After`	yes, after the wait
`server`	HTTP 500 and above	yes, bounded
`auth`	HTTP 401, 403	no: alert the operator
`invalid_request`	HTTP 400, 404, 413, 422	no: fix the request
`quota_exhausted`	HTTP 402, or a 429 whose body names credit or a spend limit	no: alert the owner
`unknown`	anything else	no: log, fail loudly

Provider note (as of 2026-10-12): the SDKs of the first two providers raise one exception class per status, plus connection and timeout classes. At the time of writing both retry 408, 409, 429, 500 and above and connection errors twice by default, time out after 10 minutes and accept max_retries. The first advises honouring Retry-After, else backoff with jitter and a cap, and says a 429 can mean used-up credit or a spend limit (credit_balance_exhausted): do not retry that. The second documents 402 as billing, 529 as overloaded, 504 as timeout, and a spend-cap 429 without retry-after. The third documents 429, 503, 504 and no rate-limit headers; its exception names and default timeout, and the local runtime's timeout and retries, are unconfirmed: assume no built-in retry. Check the current documentation.

Normalise responses and errors

Bounded: max_attempts counts every try; then the last error is raised.
Exponential backoff with jitter: 0.5, 1, 2 ... seconds up to max_delay, each cut by a random share so clients do not retry in step.
The server wins: a longer Retry-After is used; one beyond max_delay means give up, not hammer.
Injected sleep and rng: tests pass a list's append and a seeded random.Random.

RetryingClient wraps any LLMClient and is one itself.

Python

# file: evidencedesk/errors.py
from __future__ import annotations

import random
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass

from evidencedesk.llm import LLMClient, LLMRequest, LLMResponse

RETRYABLE = {"timeout", "rate_limited", "server"}
QUOTA_MARKERS = ("credit_balance_exhausted", "spend_limit", "quota_exceeded")  # our heuristic


class LLMError(Exception):
    def __init__(self, kind: str, message: str = "", retry_after: float | None = None):
        super().__init__(f"{kind}: {message}" if message else kind)
        self.kind, self.retry_after, self.attempts = kind, retry_after, 1

    @property
    def retryable(self) -> bool:
        return self.kind in RETRYABLE


def from_status(status: int, headers: Mapping[str, str] | None = None, body: str = "") -> LLMError:
    lowered = {k.lower(): v for k, v in (headers or {}).items()}
    try:
        retry_after = max(0.0, float(lowered["retry-after"]))  # seconds only; a date is ignored
    except (KeyError, ValueError):
        retry_after = None
    if status == 429 and any(marker in body for marker in QUOTA_MARKERS):
        kind = "quota_exhausted"
    elif status in (408, 504):
        kind = "timeout"
    elif status == 429:
        kind = "rate_limited"
    elif status == 402:
        kind = "quota_exhausted"
    elif status in (401, 403):
        kind = "auth"
    elif status in (400, 404, 413, 422):
        kind = "invalid_request"
    elif status >= 500:
        kind = "server"
    else:
        kind = "unknown"
    return LLMError(kind, f"HTTP {status}", retry_after)


@dataclass(frozen=True)
class RetryPolicy:
    max_attempts: int = 3  # every try counts, the first too
    base_delay: float = 0.5
    factor: float = 2.0
    max_delay: float = 8.0  # the longest wait we accept
    jitter: float = 0.5  # share of each wait that is random

    def wait_for(self, attempt: int, error: LLMError, rng: random.Random) -> float | None:
        backoff = min(self.max_delay, self.base_delay * self.factor ** (attempt - 1))
        wait = backoff * (1 - self.jitter * rng.random())
        wait = max(wait, error.retry_after or 0.0)  # the server's word beats our guess
        return wait if wait <= self.max_delay else None


def call_with_retry(fn: Callable, policy: RetryPolicy = RetryPolicy(),
                    sleep: Callable[[float], None] = time.sleep, rng: random.Random | None = None):
    rng = rng or random.Random()
    for attempt in range(1, policy.max_attempts + 1):
        try:
            return fn()
        except LLMError as error:
            error.attempts = attempt
            wait = policy.wait_for(attempt, error, rng) if error.retryable else None
            if wait is None or attempt == policy.max_attempts:
                raise
            sleep(wait)
    raise ValueError("max_attempts must be at least 1")


class RetryingClient:
    """Wraps any LLMClient and is one itself."""

    def __init__(self, inner: LLMClient, policy: RetryPolicy = RetryPolicy(),
                 sleep: Callable[[float], None] = time.sleep, rng: random.Random | None = None) -> None:
        self.inner, self.policy, self.sleep, self.rng = inner, policy, sleep, rng

    def complete(self, request: LLMRequest) -> LLMResponse:
        return call_with_retry(lambda: self.inner.complete(request), self.policy, self.sleep, self.rng)

Two 429s, a call that fails twice, then one that never recovers. Sleeping only appends to a list:

Python

import random

from evidencedesk.errors import LLMError, RetryPolicy, call_with_retry, from_status

print(from_status(429, {"Retry-After": "3"}).kind, from_status(429, {}, "credit_balance_exhausted").kind)
tries, waits = [], []


def flaky():
    tries.append(1)
    if len(tries) < 3:
        raise from_status(503)
    return "ok"


print(call_with_retry(flaky, RetryPolicy(), waits.append, random.Random(7)), len(tries), [round(w, 2) for w in waits])


def always_down():
    raise from_status(500)


more = []
try:
    call_with_retry(always_down, RetryPolicy(max_attempts=3), more.append, random.Random(7))
except LLMError as error:
    print("gave up:", error, "after", error.attempts, "attempts;", len(more), "waits")

Output

rate_limited quota_exhausted
ok 3 [0.42, 0.92]
gave up: server: HTTP 500 after 3 attempts; 2 waits

Notice that the bodies tell the 429s apart, that 0.42 and 0.92 are the 0.5 and 1.0 second backoffs cut by seeded jitter, and that the hopeless call stopped after three attempts.

Use a provider capability matrix

A shared feature name can hide differences. This matrix is qualitative: "documented" means the documentation describes it, not that behaviour or limits match.

Capability	First	Second	Third	Local runtime
Structured output	documented; a strict subset of JSON Schema	documented; another subset	documented; another subset	documented; validate anyway
Streaming, tool calls	documented; tool arguments as a JSON string	documented; tool arguments as an object	documented	documented; tool support depends on the model [VERIFY]
Embeddings	yes	no first-party model	yes	yes
Usage metadata	own names; in a stream [VERIFY]	own names; cumulative in a stream	names differ per API path; stream [VERIFY]	on the plain reply; stream [VERIFY]

So the shared contract covers text in, text and usage out; the rest are optional capabilities that adapters declare and the application checks.

Provider note (as of 2026-10-12): the second provider has no first-party embedding model and points to a partner service, so later embedding adapters cover the first and third providers and a local runtime. The third has two API paths (the Interactions API, which its documentation leads with, and the older generateContent) with different usage field names. Check the current documentation.

Test the no-credential path

With an empty environment the application builds and every test passes.

Python

# file: tests/test_config.py
import re
from pathlib import Path

import pytest
from pydantic import ValidationError

from evidencedesk.config import load_settings

ROOT = Path(__file__).resolve().parent.parent
KEY_PATTERN = re.compile(r"sk-[A-Za-z0-9_-]{16,}|AIza[0-9A-Za-z_-]{20,}|-----BEGIN [A-Z ]*PRIVATE KEY")
REAL = {"EVIDENCEDESK_PROVIDER": "anthropic", "EVIDENCEDESK_MODEL": "m1"}


def fields(env):
    with pytest.raises(ValidationError) as info:
        load_settings(env)
    return [".".join(map(str, e["loc"])) for e in info.value.errors()]


def test_empty_environment_means_offline_fake():
    s = load_settings({})
    assert (s.provider, s.model, s.api_key, s.max_retries) == ("fake", "fake-model", None, 2)


def test_the_key_comes_from_env_or_file_and_never_appears_in_repr(tmp_path):
    secret = "sk-" + "a" * 24  # made up
    (tmp_path / "key.txt").write_text(secret, encoding="utf-8")
    for source in ({"EVIDENCEDESK_API_KEY": secret}, {"EVIDENCEDESK_API_KEY_FILE": str(tmp_path / "key.txt")}):
        s = load_settings({**REAL, **source})
        assert secret not in repr(s) + str(s) + s.model_dump_json()
        assert s.api_key.get_secret_value() == secret


def test_bad_settings_are_reported_by_field():
    assert fields({"EVIDENCEDESK_PROVIDER": "nope"}) == ["provider"]
    assert fields({"EVIDENCEDESK_TIMEOUT_S": "0"}) == ["timeout_s"]
    assert fields(REAL) == [""]  # a real provider without a key
    assert fields({"EVIDENCEDESK_PROVIDER": "local"}) == [""]  # a real provider without a model


def test_detector_finds_a_fake_key_and_the_source_tree_has_none():
    assert KEY_PATTERN.search("API_KEY = 'sk-" + "b" * 20 + "'")
    for path in [*ROOT.glob("evidencedesk/*.py"), *ROOT.glob("tests/*.py"), *ROOT.glob("fixtures/*.json")]:
        assert not KEY_PATTERN.search(path.read_text(encoding="utf-8")), path.name

Python

# file: tests/test_errors.py
import pytest

from evidencedesk.errors import LLMError, RetryPolicy, call_with_retry, from_status

NO_JITTER = RetryPolicy(max_attempts=4, jitter=0)


@pytest.mark.parametrize("status, headers, body, kind, retryable, retry_after", [
    (400, {}, "", "invalid_request", False, None),
    (401, {}, "", "auth", False, None),
    (429, {"Retry-After": "3"}, "", "rate_limited", True, 3.0),
    (429, {}, "credit_balance_exhausted", "quota_exhausted", False, None),
    (500, {}, "", "server", True, None),
])
def test_status_mapping(status, headers, body, kind, retryable, retry_after):
    error = from_status(status, headers, body)
    assert (error.kind, error.retryable, error.retry_after) == (kind, retryable, retry_after)


class Flaky:
    def __init__(self, *failures):
        self.failures, self.calls = list(failures), 0

    def __call__(self):
        self.calls += 1
        if self.failures:
            raise self.failures.pop(0)
        return "ok"


def run(fn, policy=NO_JITTER):
    slept = []
    try:
        return call_with_retry(fn, policy, slept.append), slept
    except LLMError as error:
        return error, slept


def test_recovers_and_never_really_sleeps():
    fn = Flaky(from_status(503), from_status(503))
    assert run(fn) == ("ok", [0.5, 1.0]) and fn.calls == 3  # base 0.5, doubled once


def test_attempts_are_bounded():
    fn = Flaky(*[from_status(500)] * 10)
    error, slept = run(fn)
    assert fn.calls == 4 and len(slept) == 3 and error.attempts == 4


def test_permanent_errors_are_not_retried():
    fn = Flaky(from_status(401), from_status(401))
    assert run(fn)[1] == [] and fn.calls == 1


def test_server_wait_wins_but_only_up_to_the_cap():
    assert run(Flaky(from_status(429, {"retry-after": "2"})))[1] == [2.0]
    fn = Flaky(from_status(429, {"retry-after": "60"}))
    assert run(fn)[1] == [] and fn.calls == 1  # 60 s is over max_delay: give up, do not hammer

Bash

python -m pytest -q tests/test_config.py tests/test_errors.py

Output

.............                                                            [100%]
13 passed

Wall-clock seconds are stripped from pytest lines on this page (sed 's/ in [0-9.]*s//'). Nothing needed a key, a network or a real sleep.

Understand model and SDK version drift

Output

# file: requirements.txt
# Tested here with Python 3.13; the target is Python 3.12.
pydantic==2.14.1
pytest==9.1.1
# openai==3.29.0  (optional adapters: latest on 2026-10-12, NOT tested here)
# anthropic==1.13.0

Provider note (as of 2026-10-12): latest on PyPI that day: openai 3.29.0, anthropic 1.13.0, google-genai 2.29.0, ollama 0.6.3. Version 1 of the second SDK removed temperature, top_p and top_k (passing them raises TypeError); the third SDK's README says to pin below 3.0.0 because a future release changes automatic function calling. The second provider documents at least 60 days' notice before retiring a model. Check the current documentation.

Create recorded provider fixtures

JSON

# file: fixtures/provider_samples.json
{
  "_note": "Hand-written. Not captured from a real service.",
  "first": {"status": "completed", "output_text": "Returns: 30 days, full refund.",
            "usage": {"input_tokens": 52, "output_tokens": 9}},
  "second": {"stop_reason": "end_turn",
             "content": [{"type": "text", "text": "Returns: 30 days, "}, {"type": "text", "text": "full refund."}],
             "usage": {"input_tokens": 52, "output_tokens": 9}},
  "errors": [
    {"name": "rate limit with a wait", "status": 429, "headers": {"Retry-After": "3"},
     "body": "{\"error\": {\"type\": \"rate_limit_error\"}}", "kind": "rate_limited"},
    {"name": "credit used up", "status": 429, "headers": {},
     "body": "{\"error\": {\"code\": \"credit_balance_exhausted\"}}", "kind": "quota_exhausted"},
    {"name": "overloaded", "status": 529, "headers": {}, "body": "overloaded_error", "kind": "server"}
  ]
}

Build a minimal API comparison report

Record functional differences, operational constraints and unknowns; mark every unknown [VERIFY].

Area	Finding	Status
Functional	System text goes in `instructions` (first), a top-level `system` (second) or a message (local); stop reasons use different words	documented
Operational	First and second SDKs retry twice by default (third: secondary source); rate-limit headers for first and second only; prices and limits change	partly [VERIFY]
Unknown	Usage in a stream; third SDK exceptions and timeout; local timeout; local tool support	[VERIFY]

Mini-lab: Provider-Neutral Text Summariser

Python

# file: evidencedesk/summary.py
from __future__ import annotations

from collections.abc import Callable

from evidencedesk.config import Settings
from evidencedesk.errors import LLMError, RetryingClient, RetryPolicy, from_status
from evidencedesk.llm import FakeClient, LLMClient, LLMRequest, LLMResponse, Message, Usage

DEMO_TEXT = ("You can return a new, unridden bike within 30 days of delivery for a full refund. We refund the "
             "price of the bike; the delivery fee is refunded only if the bike arrived damaged.")
DEMO_FIXTURES = {DEMO_TEXT: {"text": "A new, unridden bike can be returned within 30 days for a full refund; "
                                     "the delivery fee is refunded only if it arrived damaged.",
                             "usage": {"input_tokens": 52, "output_tokens": 24}}}


def summarise(client: LLMClient, text: str, max_words: int = 30) -> str:
    """The word limit is enforced here, in code, not only requested in the prompt."""
    text = text.strip()
    if not text:
        raise ValueError("nothing to summarise")
    system = f"Summarise the text in at most {max_words} words. Use only the text."
    response = client.complete(LLMRequest((Message("system", system), Message("user", text))))
    words = response.text.split()
    return " ".join(words[:max_words]) + (" ..." if len(words) > max_words or response.stop_reason == "length" else "")


# Version-sensitive: only these functions know a provider's field names.
def split_system(request: LLMRequest) -> tuple[str, str]:
    def join(role: str) -> str:
        return "\n".join(m.content for m in request.messages if m.role == role)

    return join("system"), join("user")


def from_first(r, model: str) -> LLMResponse:  # r: an object with .output_text, .usage, .status
    stop = "length" if r.status == "incomplete" else "end" if r.status == "completed" else "other"
    return LLMResponse(r.output_text, model, Usage(r.usage.input_tokens, r.usage.output_tokens), stop)


def from_second(r, model: str) -> LLMResponse:  # r: an object with .content blocks, .usage, .stop_reason
    text = "".join(b.text for b in r.content if b.type == "text")
    return LLMResponse(text, model, Usage(r.usage.input_tokens, r.usage.output_tokens),
                       {"end_turn": "end", "max_tokens": "length"}.get(r.stop_reason, "other"))


def guard(call: Callable[[], object], timeout_type: type[Exception]):
    """Run an SDK call and translate its failures."""
    try:
        return call()
    except timeout_type as e:
        raise LLMError("timeout", str(e)) from e
    except Exception as e:
        if getattr(e, "status_code", None) is None:
            raise
        raise from_status(e.status_code, dict(e.response.headers), e.response.text) from e


REGISTRY: dict[str, Callable[[Settings], LLMClient]] = {"fake": lambda s: FakeClient(DEMO_FIXTURES)}


def build_client(settings: Settings) -> LLMClient:
    if settings.provider not in REGISTRY:
        raise ValueError(f"no adapter registered for {settings.provider!r}")
    return RetryingClient(REGISTRY[settings.provider](settings), RetryPolicy(max_attempts=settings.max_retries + 1))

The tests use fakes, the fixtures and an empty environment (the fake summary is text a person wrote, not model output); obj gives a fixture attributes, as an SDK result has:

Python

# file: tests/test_summariser.py
import json
from pathlib import Path
from types import SimpleNamespace

import pytest

from evidencedesk.config import load_settings
from evidencedesk.errors import LLMError, from_status
from evidencedesk.llm import FakeClient, LLMClient
from evidencedesk.summary import DEMO_TEXT, build_client, from_first, from_second, guard, summarise

SAMPLES = json.loads((Path(__file__).parent.parent / "fixtures" / "provider_samples.json").read_text(encoding="utf-8"))


def obj(d):  # dictionary -> object with attributes, like an SDK result
    return json.loads(json.dumps(d), object_hook=lambda x: SimpleNamespace(**x))


def test_empty_environment_gives_a_working_offline_summariser():
    client = build_client(load_settings({}))
    assert isinstance(client, LLMClient)
    assert summarise(client, DEMO_TEXT).startswith("A new, unridden bike")


def test_word_limit_is_enforced_in_code():
    long = {"text": "one two three four five six", "usage": {"input_tokens": 1, "output_tokens": 6}}
    assert summarise(FakeClient({"x": long}), "x", max_words=4) == "one two three four ..."


def test_two_provider_shapes_become_one_result():
    results = [from_first(obj(SAMPLES["first"]), "m"), from_second(obj(SAMPLES["second"]), "m")]
    assert results[0] == results[1]  # same text, usage and stop reason from two different shapes
    assert results[0].text == "Returns: 30 days, full refund." and results[0].usage.total_tokens == 61


@pytest.mark.parametrize("case", SAMPLES["errors"], ids=lambda c: c["name"])
def test_recorded_failures_map_to_the_expected_kind(case):
    assert from_status(case["status"], case["headers"], case["body"]).kind == case["kind"]


def test_guard_translates_sdk_failures():
    class Boom(Exception):
        status_code, response = 429, SimpleNamespace(headers={"retry-after": "2"}, text="")

    def fail():
        raise Boom()

    with pytest.raises(LLMError) as info:
        guard(fail, TimeoutError)
    assert (info.value.kind, info.value.retry_after) == ("rate_limited", 2.0)

Bash

python -m pytest -q tests/test_summariser.py

Output

.......                                                                  [100%]
7 passed

Two optional adapters, at most 12 lines each. To use one, add it to REGISTRY and set EVIDENCEDESK_PROVIDER, _MODEL and the key. They cannot run here, so there is no output.

Python

# not run here: needs the SDK, a key and internet
import openai
from evidencedesk.summary import from_first, guard, split_system

class FirstAdapter:
    def __init__(self, s):  # max_retries=0: our RetryingClient retries
        self.model = s.model
        self.api = openai.OpenAI(api_key=s.api_key.get_secret_value(), timeout=s.timeout_s, max_retries=0)
    def complete(self, request):
        system, user = split_system(request)
        call = lambda: self.api.responses.create(model=self.model, instructions=system, input=user)
        return from_first(guard(call, openai.APITimeoutError), self.model)

Python

# not run here: needs the SDK, a key and internet
import anthropic
from evidencedesk.summary import from_second, guard, split_system

class SecondAdapter:
    def __init__(self, s):
        self.model = s.model
        self.api = anthropic.Anthropic(api_key=s.api_key.get_secret_value(), timeout=s.timeout_s, max_retries=0)
    def complete(self, request):
        system, user = split_system(request)
        ask = dict(model=self.model, max_tokens=request.max_tokens, system=system, messages=[{"role": "user", "content": user}])
        return from_second(guard(lambda: self.api.messages.create(**ask), anthropic.APITimeoutError), self.model)

Provider note (as of 2026-10-12): the sketches use responses.create(model=..., instructions=..., input=...) for the first provider and messages.create(model=..., max_tokens=..., system=..., messages=[...]) for the second, and pass max_retries=0. The timeout argument name, and whether the second SDK's status exception has status_code and response like the first, were not confirmed [VERIFY]. Copy a current model ID from the provider's model page. Check the current documentation.

Where this shows up at work

An overnight batch hits a rate limit and either waits and finishes or dies half done. An invoice jumps because an unbounded retry loop ran through a provider incident.

Common mistakes

Retrying every error. A bad key or invalid request fails the same way each time and you pay for each try. Retry only timeouts, rate limits and server errors.
No bound, no delay. An endless or instant retry loop worsens an outage. Bound, back off, add jitter.
Stacked retries. SDK and code both retry: one failure becomes nine calls. One layer owns the policy.
Secrets in code, logs or tests. Use the environment and SecretStr; rotate any key ever committed.

Try it yourself

Easy. Show that EVIDENCEDESK_MAX_RETRIES=9 is rejected, and which field the error names.
Medium. Revisit Provider-Neutral LLM Clients. Copy its ShapeBFake so it fails once with a 429 and Retry-After: 1, wrap it in RetryingClient, and check that answer_question works and the recorded wait is one second.
Hard. Revisit Git, APIs and Backend for AI: Beginner. Your API must not pass a provider status through (a bad provider key is not the caller's 401). Write and test public_status(error), mapping each kind to your service's status.

Answers

Easy: load_settings({"EVIDENCEDESK_MAX_RETRIES": "9"}) raises a ValidationError for max_retries (the limit is 5). Medium and Hard:

Python

# file: tests/test_exercises_p02.py
from test_llm import Q_RETURN, ShapeBFake

from evidencedesk.errors import RetryingClient, RetryPolicy, from_status
from evidencedesk.llm import answer_question


class OnceLimited(ShapeBFake):
    todo = [from_status(429, {"Retry-After": "1"})]

    def complete(self, request):
        if self.todo:
            raise self.todo.pop()
        return super().complete(request)


def test_medium_retry_wrapper_around_an_earlier_fake():
    slept = []
    client = RetryingClient(OnceLimited(), RetryPolicy(jitter=0), slept.append)
    assert answer_question(client, Q_RETURN).startswith("You can return") and slept == [1.0]


def public_status(error):  # never pass a provider 401 on: it is our key, not the caller's
    return {"timeout": 504, "rate_limited": 503, "server": 503, "auth": 502, "quota_exhausted": 502}.get(error.kind, 500)


def test_hard_provider_errors_become_our_own_http_statuses():
    assert [public_status(from_status(s)) for s in (401, 429, 400)] == [502, 503, 500]

Run python -m pytest -q tests/test_exercises_p02.py; two tests should pass. Pass checklist:

Easy: the error names max_retries; config.py is unchanged.
Medium: nothing really sleeps; the wait is 1.0 because the server's value beats our 0.5 second guess.
Hard: auth and quota failures become a server-side status (not 401); other defensible mappings are fine if you can explain them.

Summary

Keep provider, model, timeout and retry limits in validated settings built from a mapping; the fake is the offline default.
Keep secrets in the environment or a secret file, hide them (SecretStr), ignore the files, test for key-shaped strings.
Normalise failures into one LLMError: retry timeouts, rate limits and server errors with bounded, jittered backoff; never retry invalid requests, authentication failures or exhausted quotas.
Compare providers with a capability matrix and a report; mark unknowns [VERIFY].
Pin tested versions, isolate version-sensitive mapping, keep honest fixtures, test with no credentials.

Next: Context Assembly and Trust Boundaries. You will build the model input from named components and keep trusted instructions apart from untrusted text.