Chapter 1 ended with a three-line provider contract and a promise to replace it. This chapter does that. The goal is simple to state: no code outside one folder should ever mention a
vendor. The conversation manager, the channels, the persona and the cost controls of later chapters will all talk to a Provider and get back a Reply, whether the model
behind it is Claude, a Grok model reached through xAI's OpenAI-compatible API, OpenAI itself, or a model running on your own machine.
What was run, and what was not
Everything in this chapter was run on Python 3.14.0 with httpx 0.28.1, and the test results shown are real. No request was sent to Anthropic, xAI, OpenAI or any real model
(this machine has no API keys). Instead, the two HTTP providers were tested against a small local HTTP server that answers with responses shaped like each vendor's documented format. That proves
the providers build the requests they should and read the responses and errors they should. It does not prove a vendor's server accepts them today: that needs a real key, and those steps are
labelled "not run here". Anthropic's formats below come from its API documentation, which I read; the OpenAI-style formats are the widely implemented Chat Completions convention, and I could not fetch
OpenAI's own reference page (it returned an access error), so check them against your provider's documentation.
What Actually Differs Between Providers
The easiest way to design an interface is to put the real differences side by side. These are the ones that matter to a chatbot, for Anthropic's Messages API (from its documentation) and for the
Chat Completions style that OpenAI, xAI, Ollama and many other servers implement:
Anthropic Messages API
OpenAI-compatible Chat Completions
Endpoint
POST https://api.anthropic.com/v1/messages
POST {base_url}/chat/completions (OpenAI https://api.openai.com/v1, xAI https://api.x.ai/v1, Ollama http://localhost:11434/v1)
Authentication
x-api-key header (the docs list Authorization: Bearer as primary and x-api-key as a still-supported fallback); anthropic-version header is required, e.g. 2023-06-01
Authorization: Bearer <key>; a local server may need none
System prompt
A top-level system field. The docs say there is no "system" role for input messages (newer models can accept a later mid-conversation system message, never as the first entry)
A first message with "role": "system" (some newer OpenAI models use a developer role; not checked here)
Output limit
max_tokens, required
max_tokens (OpenAI marks it deprecated and not usable on its o-series in favour of max_completion_tokens; Ollama's examples use max_tokens)
The answer
content: a list of blocks; text blocks have "type": "text"
retry-after header on a 429 (the docs say a spend-cap 429 has none)
usually a retry-after header (not verified here)
Request id
request-id header on every response
varies by server (x-request-id is common)
Anthropic's error page also lists the status codes and their meanings: 400 invalid_request_error, 401 authentication_error, 402 billing_error, 403
permission_error, 404 not_found_error, 413 request_too_large, 429 rate_limit_error, 500 api_error, 504 timeout_error
and 529 overloaded_error. Two practical notes from the same documentation: its official SDK retries transient failures twice by default, honouring retry-after; and
the sampling parameters temperature, top_p and top_k are not supported on Claude 4.7 and later models, so the provider in this chapter sends none of them.
A moving target to design around
xAI's documentation, as shown by a search result I did not open in full, now recommends its newer Responses API and calls Chat Completions legacy, though still working; OpenAI also has a
Responses API. Chat Completions remains the common language of local servers such as Ollama and llama.cpp, which is why this course builds against it. Because everything vendor-specific lives in
one file per provider, adding a /responses-style provider later means writing one new class, not changing the bot. That is the payoff of the interface.
The Contract: Messages In, a Reply or a Typed Error Out
One housekeeping change first: chapter 1's single file bot/providers.py is now a package, bot/providers/, with base.py (the contract), fake.py,
anthropic.py, openai_compat.py and factory.py, and bot/providers/__init__.py re-exports the public names so from bot.providers import Provider
still works. Bot now calls provider.complete(...) and accepts an optional system prompt.
The interface has three decisions in it. A provider takes messages and an optional system prompt (so the system prompt is a separate argument, not a message with a special role, which is the
one thing the two vendors disagree about). It returns a Reply that carries the text and the token usage, because chapter 7's cost control needs the usage and will
not want to know which vendor's field names produced it. And every failure is raised as one of a small family of typed errors that say whether retrying could help.
from dataclasses import dataclass
from typing import Protocol
Message = dict[str, str] # {"role": "user" | "assistant", "content": "..."}
@dataclass(frozen=True)
class Usage:
input_tokens: int = 0
output_tokens: int = 0
@dataclass(frozen=True)
class Reply:
text: str
usage: Usage = Usage()
model: str = ""
stop_reason: str = ""
request_id: str | None = None
class ProviderError(Exception):
"""Every provider failure becomes one of these, whatever the vendor said."""
retryable = False
def __init__(self, message: str, *, status: int | None = None,
retry_after: float | None = None, request_id: str | None = None) -> None:
super().__init__(message)
self.status = status
self.retry_after = retry_after
self.request_id = request_id
class AuthError(ProviderError):
"""Bad, missing or unauthorised key, or a billing problem (401, 402, 403)."""
class BadRequest(ProviderError):
"""Our request was wrong (400, 404, 413): retrying the same request will not help."""
class RateLimited(ProviderError):
"""Too many requests or tokens (429). May carry retry_after in seconds."""
retryable = True
class ServerError(ProviderError):
"""The provider failed or is overloaded (5xx, including Anthropic's 529)."""
retryable = True
class ProviderTimeout(ProviderError):
"""No answer within our timeout."""
retryable = True
class ProviderUnavailable(ProviderError):
"""Could not connect at all (DNS, refused, network down)."""
retryable = True
def error_from_status(status: int, message: str, *, retry_after: float | None = None,
request_id: str | None = None) -> ProviderError:
kw = dict(status=status, retry_after=retry_after, request_id=request_id)
if status in (401, 402, 403):
return AuthError(message, **kw)
if status == 429:
return RateLimited(message, **kw)
if status >= 500:
return ServerError(message, **kw)
return BadRequest(message, **kw)
def parse_retry_after(value: str | None) -> float | None:
"""The Retry-After header as seconds, or None if absent or not a plain number."""
if value is None:
return None
try:
return float(value)
except ValueError:
return None
class Provider(Protocol):
"""Anything that can turn a conversation into the next reply."""
name: str
async def complete(self, messages: list[Message], *, system: str | None = None,
max_tokens: int = 1024) -> Reply: ...
The status mapping in error_from_status is the whole taxonomy, and it is deliberately small. For twelve status codes (most are from Anthropic's list above; 502 and 503 are the usual gateway errors), run for real:
The one flag the rest of the bot cares about is retryable: rate limits, server errors, timeouts and connection failures might succeed if tried again later; a bad key or a malformed request
will not, however often you retry. Chapter 7 builds the retry logic on that single property.
The Fake Provider, Version 2
The fake grows up: its script may now contain text, a complete Reply, or an exception to raise, it can add a delay, and it records the system prompt and token limit it was given as well as the
messages. Its "tokens" are words, which is crude but deterministic, so tests can assert on exact numbers.
import asyncio
from bot.providers.base import Message, Reply, Usage
def count_words(text: str) -> int:
return len(text.split())
class EchoProvider:
"""The dumbest possible provider: repeats the last thing the user said."""
name = "echo"
async def complete(self, messages: list[Message], *, system: str | None = None,
max_tokens: int = 1024) -> Reply:
return Reply("You said: " + messages[-1]["content"], model="echo")
class FakeProvider:
"""A scripted stand-in for a model.
script entries are used in order: a str becomes the reply text, a Reply is returned
as it is, and an exception instance is raised. When the script runs out it describes
what it was sent. "Tokens" are counted as words, which is crude but deterministic.
"""
name = "fake"
def __init__(self, script: list | None = None, *, delay: float = 0.0) -> None:
self.script = list(script or [])
self.delay = delay
self.calls: list[list[Message]] = []
self.systems: list[str | None] = []
self.max_tokens_seen: list[int] = []
async def complete(self, messages: list[Message], *, system: str | None = None,
max_tokens: int = 1024) -> Reply:
self.calls.append(list(messages))
self.systems.append(system)
self.max_tokens_seen.append(max_tokens)
if self.delay:
await asyncio.sleep(self.delay)
if self.script:
item = self.script.pop(0)
if isinstance(item, BaseException):
raise item
if isinstance(item, Reply):
return item
text = item
else:
text = f"(fake) I was sent {len(messages)} message(s); the last was: {messages[-1]['content']!r}"
used = sum(count_words(m["content"]) for m in messages) + (count_words(system) if system else 0)
return Reply(text, Usage(used, count_words(text)), model="fake", stop_reason="end_turn")
The Claude Provider
This provider speaks the Messages API directly over HTTP with httpx (install it with pip install httpx). I chose raw HTTP rather than the official SDK so that every field in the
table above is visible in the code, and so the provider can be tested against a local server; the SDK is a fine alternative and is covered in Advanced AI Workflows & the Claude API 1.
import httpx
from bot.providers.base import (
Message, ProviderTimeout, ProviderUnavailable, Reply, Usage,
error_from_status, parse_retry_after,
)
class ClaudeProvider:
"""Talks to the Anthropic Messages API directly over HTTP."""
name = "claude"
def __init__(self, api_key: str, model: str, *, base_url: str = "https://api.anthropic.com",
timeout: float = 60.0, transport: httpx.AsyncBaseTransport | None = None) -> None:
self.model = model
self._client = httpx.AsyncClient(
base_url=base_url,
timeout=timeout,
transport=transport,
headers={
"x-api-key": api_key,
"anthropic-version": "2023-06-01",
"content-type": "application/json",
},
)
async def complete(self, messages: list[Message], *, system: str | None = None,
max_tokens: int = 1024) -> Reply:
body: dict = {"model": self.model, "max_tokens": max_tokens, "messages": messages}
if system:
body["system"] = system # top-level field: there is no "system" role in messages
try:
response = await self._client.post("/v1/messages", json=body)
except httpx.ConnectTimeout as e: # could not even connect: unavailable, not "slow"
raise ProviderUnavailable(f"could not connect: {e!r}") from e
except httpx.TimeoutException as e:
raise ProviderTimeout(f"no answer in time: {e!r}") from e
except httpx.TransportError as e:
raise ProviderUnavailable(f"could not connect: {e!r}") from e
request_id = response.headers.get("request-id")
if response.status_code != 200:
message = response.text
try:
message = response.json()["error"]["message"]
except (ValueError, KeyError, TypeError):
pass
raise error_from_status(response.status_code, message,
retry_after=parse_retry_after(response.headers.get("retry-after")),
request_id=request_id)
data = response.json()
text = "".join(block.get("text", "") for block in data["content"] if block.get("type") == "text")
usage = data.get("usage", {})
return Reply(text, Usage(usage.get("input_tokens", 0), usage.get("output_tokens", 0)),
model=data.get("model", self.model), stop_reason=data.get("stop_reason") or "",
request_id=request_id)
async def aclose(self) -> None:
await self._client.aclose()
Points worth noticing:
The system prompt goes in the top-level system field, and only when there is one.
The model name is a constructor argument. This course never hard-codes a model ID: the documentation's own examples change from page to page and release to release, so you read the current list from the provider and put the ID in configuration.
Network problems are turned into our errors at the edge. httpx.ConnectTimeout is caught before the general TimeoutException, for a reason explained in the box further down.
The reply's text is built from the blocks of type text, since content is a list. Other block types (such as tool calls) are ignored for now.
The request-id header and the retry-after header are kept on the error, so logs and retry logic can use them.
The OpenAI-Compatible Provider
import httpx
from bot.providers.base import (
Message, ProviderTimeout, ProviderUnavailable, Reply, Usage,
error_from_status, parse_retry_after,
)
class OpenAICompatProvider:
"""Talks to any server that implements the Chat Completions API.
That covers OpenAI itself, xAI (https://api.x.ai/v1), and local servers such as Ollama
(http://localhost:11434/v1). Only the base URL, key and model change.
"""
def __init__(self, base_url: str, model: str, api_key: str | None = None, *, name: str = "openai-compatible",
max_tokens_field: str = "max_tokens", timeout: float = 60.0,
transport: httpx.AsyncBaseTransport | None = None) -> None:
self.name = name
self.model = model
self.max_tokens_field = max_tokens_field # some newer OpenAI models want "max_completion_tokens"
headers = {"content-type": "application/json"}
if api_key:
headers["authorization"] = f"Bearer {api_key}"
self._client = httpx.AsyncClient(base_url=base_url.rstrip("/"), timeout=timeout,
transport=transport, headers=headers)
async def complete(self, messages: list[Message], *, system: str | None = None,
max_tokens: int = 1024) -> Reply:
wire = ([{"role": "system", "content": system}] if system else []) + list(messages)
body = {"model": self.model, "messages": wire, self.max_tokens_field: max_tokens}
try:
response = await self._client.post("/chat/completions", json=body)
except httpx.ConnectTimeout as e: # could not even connect: unavailable, not "slow"
raise ProviderUnavailable(f"could not connect: {e!r}") from e
except httpx.TimeoutException as e:
raise ProviderTimeout(f"no answer in time: {e!r}") from e
except httpx.TransportError as e:
raise ProviderUnavailable(f"could not connect: {e!r}") from e
request_id = response.headers.get("x-request-id")
if response.status_code != 200:
message = response.text
try:
message = response.json()["error"]["message"]
except (ValueError, KeyError, TypeError):
pass
raise error_from_status(response.status_code, message,
retry_after=parse_retry_after(response.headers.get("retry-after")),
request_id=request_id)
data = response.json()
choice = data["choices"][0]
usage = data.get("usage") or {}
return Reply(choice["message"].get("content") or "",
Usage(usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0)),
model=data.get("model", self.model), stop_reason=choice.get("finish_reason") or "",
request_id=request_id)
async def aclose(self) -> None:
await self._client.aclose()
The same class serves OpenAI, xAI's Grok and local servers: only base_url, model and api_key change. The max_tokens_field argument exists because of the
one difference that does bite in practice: OpenAI's newer models want max_completion_tokens while others (and Ollama's documented examples) use max_tokens. An earlier version of Ollama's documentation called its OpenAI compatibility "experimental"; the current page (checked while writing chapter 3) has no such warning, but it does list request fields that are not supported (tool_choice, n and user, for example). Either way, compatibility layers differ at the edges, which is a good reason to keep each behind our interface.
Configuration: Choosing a Provider Without Touching Code
A bot should pick its provider from configuration. Note that the config names the environment variable that holds the key, never the key itself, so a config file can be committed or shown without
leaking anything:
import os
from dataclasses import dataclass
from bot.providers.anthropic import ClaudeProvider
from bot.providers.base import Provider
from bot.providers.fake import EchoProvider, FakeProvider
from bot.providers.openai_compat import OpenAICompatProvider
@dataclass(frozen=True)
class ProviderConfig:
kind: str # "fake", "echo", "claude", "openai-compatible"
model: str = ""
base_url: str = ""
api_key_env: str = "" # NAME of the environment variable, never the key itself
max_tokens_field: str = "max_tokens"
def make_provider(cfg: ProviderConfig, env: dict[str, str] | None = None) -> Provider:
env = os.environ if env is None else env
key = env.get(cfg.api_key_env) if cfg.api_key_env else None
if cfg.kind == "fake":
return FakeProvider()
if cfg.kind == "echo":
return EchoProvider()
if cfg.kind == "claude":
if not key:
raise RuntimeError(f"set the {cfg.api_key_env or 'API key'} environment variable")
return ClaudeProvider(key, cfg.model, **({"base_url": cfg.base_url} if cfg.base_url else {}))
if cfg.kind == "openai-compatible":
if cfg.api_key_env and not key:
raise RuntimeError(f"set the {cfg.api_key_env} environment variable")
return OpenAICompatProvider(cfg.base_url, cfg.model, key, max_tokens_field=cfg.max_tokens_field)
raise ValueError(f"unknown provider kind: {cfg.kind!r}")
Here is every provider from the planned course, built from configuration with no keys set. The hosted ones refuse politely (and say which variable to set); the local one needs no key; the
fake is always available. The model names are placeholders: look up a current one for each service. The last two lines show the system prompt reaching the provider through the new Bot:
claude -> RuntimeError: set the ANTHROPIC_API_KEY environment variable
grok -> RuntimeError: set the XAI_API_KEY environment variable
openai -> RuntimeError: set the OPENAI_API_KEY environment variable
ollama -> OpenAICompatProvider (name='openai-compatible')
fake -> FakeProvider (name='fake')
(fake) I was sent 1 message(s); the last was: 'hello'
system the provider saw: ['Be brief.']
Contract Tests: One Test Suite, Every Provider
Having an interface only helps if every implementation really behaves the same way. A contract test is a set of tests written once against the interface and run against every
implementation. Each concrete test class only says how to stage a situation ("make the next call succeed with this text", "make it fail with a rate limit") and how to see what was sent:
class ProviderContract:
"""Every provider, however it talks to its model, must pass these tests."""
provider = None
exact_usage = True # False for providers that only estimate tokens
# --- each concrete test class says how to stage a situation and how to see what was sent
def arrange_ok(self, text, input_tokens, output_tokens): raise NotImplementedError
def arrange_error(self, kind, retry_after=None): raise NotImplementedError
def sent_messages(self): raise NotImplementedError
def sent_system(self): raise NotImplementedError
async def test_returns_text_and_usage(self):
self.arrange_ok("Hello Ada!", 12, 6)
reply = await self.provider.complete([{"role": "user", "content": "hi"}])
self.assertEqual(reply.text, "Hello Ada!")
if self.exact_usage:
self.assertEqual(reply.usage, Usage(12, 6))
async def test_sends_the_conversation_in_order(self):
self.arrange_ok("ok", 1, 1)
await self.provider.complete(CONVERSATION)
self.assertEqual(self.sent_messages(), CONVERSATION)
async def test_passes_the_system_prompt(self):
self.arrange_ok("ok", 1, 1)
await self.provider.complete(CONVERSATION, system="Be brief.")
self.assertEqual(self.sent_system(), "Be brief.")
async def test_no_system_prompt_sends_none(self):
self.arrange_ok("ok", 1, 1)
await self.provider.complete(CONVERSATION)
self.assertIsNone(self.sent_system())
async def test_rate_limit_is_normalised(self):
self.arrange_error("rate_limit", retry_after=7)
with self.assertRaises(RateLimited) as cm:
await self.provider.complete(CONVERSATION)
self.assertEqual(cm.exception.retry_after, 7.0)
self.assertTrue(cm.exception.retryable)
async def test_bad_key_is_normalised(self):
self.arrange_error("auth")
with self.assertRaises(AuthError) as cm:
await self.provider.complete(CONVERSATION)
self.assertFalse(cm.exception.retryable)
async def test_server_failure_is_normalised(self):
self.arrange_error("server")
with self.assertRaises(ServerError) as cm:
await self.provider.complete(CONVERSATION)
self.assertTrue(cm.exception.retryable)
async def test_bad_request_is_normalised(self):
self.arrange_error("bad_request")
with self.assertRaises(BadRequest) as cm:
await self.provider.complete(CONVERSATION)
self.assertFalse(cm.exception.retryable)
For the two HTTP providers the "server" is a small scripted HTTP server from the standard library, started on a free local port, which answers with a queue of canned responses and records every request
it receives (path, headers, JSON body). It is about fifty lines and is shown here in full because the exercises use it:
tests/support.py: the scripted local server
import json
import socketserver
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
class QuickServer(ThreadingHTTPServer):
def server_bind(self):
# HTTPServer.server_bind calls socket.getfqdn(), which can take ~1s on Windows
socketserver.TCPServer.server_bind(self)
self.server_port = self.server_address[1]
class ScriptedServer:
"""A tiny local HTTP server that answers with a queue of canned responses.
Each queued item is (status, headers, body, delay_seconds). Every request received is
recorded in .requests as (path, headers, parsed JSON body).
"""
def __init__(self) -> None:
self.queue: list[tuple[int, dict, object, float]] = []
self.requests: list[tuple[str, dict, dict]] = []
outer = self
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
self.do_POST()
def do_POST(self):
length = int(self.headers.get("content-length", 0))
body = json.loads(self.rfile.read(length) or b"{}")
outer.requests.append((self.path, {k.lower(): v for k, v in self.headers.items()}, body))
status, headers, payload, delay = outer.queue.pop(0)
if delay:
time.sleep(delay)
data = json.dumps(payload).encode() if not isinstance(payload, bytes) else payload
try:
self.send_response(status)
self.send_header("content-type", "application/json")
self.send_header("content-length", str(len(data)))
for k, v in headers.items():
self.send_header(k, v)
self.end_headers()
self.wfile.write(data)
except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError):
pass
def log_message(self, *args):
pass
self._server = QuickServer(("127.0.0.1", 0), Handler)
self.port = self._server.server_address[1]
self.url = f"http://127.0.0.1:{self.port}"
self._thread = threading.Thread(target=lambda: self._server.serve_forever(poll_interval=0.01), daemon=True)
def __enter__(self):
self._thread.start()
return self
def __exit__(self, *exc):
self._server.shutdown()
self._server.server_close()
def respond(self, status=200, payload=None, headers=None, delay=0.0):
self.queue.append((status, headers or {}, payload if payload is not None else {}, delay))
The whole suite, including chapter 1's four tests, with the number of tests per class:
----------------------------------------------------------------------
Ran 38 tests in 11.785s
OK
CoreTests 4 (chapter 1)
FakeProviderTests 8 the contract
ClaudeProviderTests 13 the contract + 2 HTTP checks + 3 Claude-specific
OpenAICompatProviderTests 13 the contract + 2 HTTP checks + 3 OpenAI-specific
The contract covers eight behaviours for every provider: the reply carries text and usage; the conversation is sent in order; the system prompt is passed when given; no system prompt means none is sent;
and rate-limit, auth, server and bad-request failures each become the right typed error. The HTTP providers also have to turn a slow server into ProviderTimeout and a missing server into
ProviderUnavailable, and each has tests for its own wire format (the paths and headers in the table above, the max_completion_tokens option, and Anthropic's 529).
And the typed errors in action, from a real run against the local server (the rate-limit response carried retry-after: 7):
A refused connection was reported as "slow". My first run had two failing tests: connecting to a closed port raised httpx.ConnectTimeout, which is a kind ofTimeoutException, so my first mapping reported ProviderTimeout instead of ProviderUnavailable. The two mean different things (the server is slow versus there is no server), so the fix was to catch ConnectTimeout first. The takeaway: exception classes inherit from each other, so order the except clauses from most to least specific, and write a test for the unhappy case, which is how this was found.
A guess about slow tests that was wrong, then two measured fixes. The suite took about 37 seconds. I first suspected the standard library's HTTPServer, which looks up the machine's full name when it starts, and patched around that: the suite went from 24.5 s to 23.3 s, which is noise, so the guess was wrong. Timing the pieces found the real causes. The server's serve_forever polls for a shutdown request every half second by default, so every test paid up to 0.5 s to stop it; setting poll_interval=0.01 took the suite from about 37 s to 24 s. And httpx.AsyncClient() takes about 0.2 s to create (measured 0.22 s and 0.19 s) while my set-up built three per test; creating the two extra clients only in the two tests that need them took it from 24 s to about 12 s. The takeaway is to measure before optimising, and to create a client once and reuse it in a real bot.
A test that only some providers could pass. The contract originally checked isinstance(provider, FakeProvider) to skip the exact-token-count assertion. That put knowledge of one provider into the shared test. It is now a class flag, exact_usage, which any provider that only estimates tokens sets to False.
What Is Still Missing
Nothing here retries a failed call, limits how often it calls, or counts cost: it only classifies failures and reports usage, which is the groundwork. The providers also read the whole reply at
once; streaming into a chat window is a channel concern covered in chapter 8. And nothing has talked to a real model yet. When you have a key, the smallest real check is to build the provider from
configuration, send one message, and print the reply, usage and request id. That step is not run here.
Hands-On Exercises
Exercise 1
Write a new provider class (any behaviour you like, for example one that replies with the last message in capitals) and run the whole contract against it by subclassing ProviderContract. You will need to implement the four hooks. Which contract test makes you decide something about token usage, and what did you choose?
Print the error class and retryable flag that error_from_status gives for 400, 401, 402, 403, 404, 413, 429, 500, 502, 503, 504 and 529. Pick one status where you disagree with the mapping, say what you would change and what other code would be affected.
Run a Bot on an OpenAICompatProvider that points at the scripted local server, with no API key, and a system prompt. Print the answers, then print for each request the path, whether an Authorization header was sent, and the roles of the messages on the wire. This is the exact code path a local Ollama server will use in the next chapter.
Errors: AuthError (401/402/403), BadRequest (400/404/413), RateLimited (429, has retry_after), ServerError (5xx incl. 529), ProviderTimeout, ProviderUnavailable; retryable is true for the last four
Anthropic: POST /v1/messages, headers x-api-key + anthropic-version: 2023-06-01, top-level system, required max_tokens, answer in content[] text blocks, usage input_tokens/output_tokens
OpenAI-compatible: POST {base_url}/chat/completions, Authorization: Bearer, system as a first message, choices[0].message.content, usage prompt_tokens/completion_tokens; max_tokens versus max_completion_tokens is configurable
Same class for OpenAI, xAI Grok (https://api.x.ai/v1) and local servers (Ollama http://localhost:11434/v1); the xAI docs now prefer a Responses API, so keep each vendor behind its own class
Configuration names the environment variable for the key; never put keys in code or config files; never hard-code model IDs
Contract tests: write the tests once against the interface, run them against every provider; the scripted local server proves request and response handling without a real model
Exceptions inherit from each other: catch ConnectTimeout before TimeoutException; creating an httpx client costs about 0.2 s, so create one and reuse it
Coming next
Build a Chatbot: From API Call to Deployed Bot 3 adds a model running on your own machine behind this same interface: why you might want one, what an OpenAI-compatible local server such as
Ollama or llama.cpp involves, how model size meets hardware reality, and how to compare it honestly with a hosted model. That chapter needs a local runtime installed, which this machine does not
have, so it will be clear about what was run and what comes from documentation.