LLM gateways

Algorithms never talk to a provider. They emit aaa.llm.requested, and the LLMHandler you attach to the engine answers by calling an LLMClient. Swapping providers or gateways means changing one line.

from aaa.llm import LLMHandler, build_client, gateways

client = build_client(gateways.portkey("gpt-4o-mini", provider="openai"))
LLMHandler(client).attach(engine)

Native providers: Anthropic and Gemini

Besides OpenAI-compatible gateways and LiteLLM, AAA speaks two vendors' own APIs directly, over httpx (no vendor SDK, no extra install):

build_client(LLMConfig(provider="anthropic", model="claude-sonnet-5-5", api_key="${ANTHROPIC_API_KEY}", params={"max_tokens": 2048}))
build_client(LLMConfig(provider="gemini", model="gemini-2.5-flash", api_key="${GEMINI_API_KEY}"))

"""One agent, two native wire formats: Anthropic's Messages API and Google's Generative Language API. Offline (mock servers)."""

import asyncio
import json

import httpx

from aaa.algos import ToolLoop
from aaa.llm import LLMHandler
from aaa.llm.anthropic import AnthropicClient
from aaa.llm.config import LLMConfig
from aaa.llm.gemini import GeminiClient
from aaa.runtime import Engine
from aaa.tools import ToolHandler, ToolRegistry, tool


@tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b


def anthropic_server(request: httpx.Request) -> httpx.Response:
    body = json.loads(request.content)
    if "tool_result" in json.dumps(body):
        content, stop = [{"type": "text", "text": "The sum is 5."}], "end_turn"
    else:
        content, stop = (
            [{"type": "tool_use", "id": "toolu_1", "name": "add", "input": {"a": 2, "b": 3}}],
            "tool_use",
        )
    usage = {"input_tokens": 20, "output_tokens": 8}
    return httpx.Response(
        200,
        json={
            "id": "m",
            "type": "message",
            "role": "assistant",
            "model": "claude",
            "content": content,
            "stop_reason": stop,
            "usage": usage,
        },
    )


def gemini_server(request: httpx.Request) -> httpx.Response:
    body = json.loads(request.content)
    if "functionResponse" in json.dumps(body):
        parts = [{"text": "The sum is 5."}]
    else:
        parts = [{"functionCall": {"name": "add", "args": {"a": 2, "b": 3}}}]
    cand = {"content": {"role": "model", "parts": parts}, "finishReason": "STOP"}
    return httpx.Response(
        200,
        json={
            "candidates": [cand],
            "usageMetadata": {
                "promptTokenCount": 20,
                "candidatesTokenCount": 8,
                "totalTokenCount": 28,
            },
        },
    )


async def run(name: str, client) -> None:
    registry = ToolRegistry([add])
    algo = ToolLoop(registry)
    async with Engine() as engine:
        LLMHandler(client).attach(engine)
        ToolHandler(registry).attach(engine)
        outcome = await engine.run(algo, {"task": "What is 2 + 3?"})
    print(f"{name}: {outcome.value['text']}")


async def main() -> None:
    http = lambda handler: httpx.AsyncClient(transport=httpx.MockTransport(handler))
    # In production: LLMConfig(provider="anthropic", model="claude-sonnet-5-5", api_key="${ANTHROPIC_API_KEY}")
    #                LLMConfig(provider="gemini", model="gemini-2.5-flash", api_key="${GEMINI_API_KEY}")  -> build_client(config)
    await run(
        "anthropic",
        AnthropicClient(
            LLMConfig(provider="anthropic", model="claude", api_key="k"),
            http_client=http(anthropic_server),
        ),
    )
    await run(
        "gemini",
        GeminiClient(
            LLMConfig(provider="gemini", model="gemini", api_key="k"),
            http_client=http(gemini_server),
        ),
    )


asyncio.run(main())

anthropic: The sum is 5.
gemini: The sum is 5.

	Anthropic (aaa.llm.anthropic)	Gemini (aaa.llm.gemini)
Endpoint	POST /v1/messages, x-api-key, anthropic-version: 2023-06-01	POST /models/{model}:generateContent, x-goog-api-key
System prompt	the separate system field	systemInstruction
Tools	tool_use / tool_result blocks; consecutive results merged into one user turn, before any user text	functionCall / functionResponse parts, results ordered like their calls; a call id from the API is echoed back
Structured output (response_schema)	a forced tool carrying the schema; its input is returned as the reply text	responseMimeType: application/json + responseJsonSchema
Attachments	images, PDFs, plain-text documents (no audio)	images, audio, files (inlineData / fileData)
Usage	cache reads and writes count as prompt tokens; cached_tokens = cache reads	thinking tokens count as output; cached_tokens from cachedContentTokenCount
Extras	cache=True adds prompt-caching breakpoints; max_tokens is required by the API (default 4096)	embeddings via batchEmbedContents (Anthropic has no embeddings API)
Streaming	yes (SSE)	yes (alt=sse)

Errors are mapped onto AAA's classes (RateLimited with retry-after, TransientError for 5xx and Anthropic's 529 overloaded, AuthError, ContextOverflow, ContentFiltered for blocked prompts and replies), so retries, circuit breakers and the engine behave as with any other client. Parameters the provider does not know are dropped with a warning log instead of being sent (both APIs reject unknown fields); Anthropic extended-thinking blocks are neither requested nor round-tripped.

Verified. Requests, replies and streaming events are validated against the vendors' own SDK types (anthropic, google-genai) in CI, and a full tool-loop runs through each client against a mock server speaking the real wire format. Nothing was sent to a live API; model names, new model-specific parameters and server-side behaviour are the things this cannot catch.

The LLMClient protocol

Anything with these three async methods is an LLM backend:

class LLMClient(Protocol):
    async def complete(self, request: LLMRequest) -> LLMResponse: ...
    async def embed(self, texts: list[str], model: str | None = None) -> EmbedResponse: ...
    async def aclose(self) -> None: ...

A client may also implement stream(request) -> AsyncIterator[StreamChunk]; the handler detects it and streams when a request asks for it (Streaming).

Requests and responses are normalized, whatever the provider:

Type	Fields
LLMRequest	model (None = client default), messages, tools, response_schema (JSON Schema, for structured output), params (temperature, max_tokens, stop, ...), stream
Message	role (system/user/assistant/tool), content, tool_calls, tool_call_id, name
ToolCall	id, name, arguments (parsed dict), raw_arguments (set only when the model emitted invalid JSON)
LLMResponse	message, usage, cost, model, finish_reason

Configuring a client

LLMConfig is declarative. Secrets are references, never values: ${ENV_VAR} is resolved when the client is built, and a missing variable is a loud LLMConfigError.

Field	Default
provider	"openai_compat"	"openai_compat", "litellm", or any name you registered.
model	required	Default model; a request can override it.
base_url	""	Required for openai_compat.
api_key	None	Sent as Authorization: Bearer .... Use "${OPENAI_API_KEY}".
headers	{}	Extra headers; values may use ${VAR}.
chat_path	"/chat/completions"	Appended to base_url. Use "" when base_url already is the full endpoint.
embeddings_path	"/embeddings"
timeout_s	60.0	Per-request timeout. Raise it for slow reasoning models.
params	{}	Default request params (merged under each request's own).
price_in_per_mtok, price_out_per_mtok	None	Enables cost accounting and Budget(max_cost=...).

from aaa.llm import LLMConfig, build_client

client = build_client(
    LLMConfig(
        model="gemma4:latest",
        base_url="http://localhost:11434/v1",  # Ollama's OpenAI-compatible endpoint
        timeout_s=240,
    )
)

Gateway presets

All of these gateways speak the OpenAI chat-completions dialect, so a preset is only a base URL plus the auth and routing headers that gateway documents. Every field can be overridden by passing it as a keyword.

"""Pointing AAA at an LLM gateway is configuration. This prints each preset; no network is used."""

import os

from aaa.llm import LLMConfig, OpenAICompatClient, build_client, gateways, register_provider

# Secrets are written as ${ENV_VAR} and resolved when the client is built. Dummy values here.
for name in (
    "PORTKEY_API_KEY",
    "PROVIDER_API_KEY",
    "CLOUDFLARE_API_TOKEN",
    "AI_GATEWAY_API_KEY",
    "LITELLM_API_KEY",
):
    os.environ.setdefault(name, "dummy")

presets = {
    "LiteLLM proxy": gateways.litellm_proxy("gpt-4o-mini"),
    "Portkey": gateways.portkey("gpt-4o-mini", provider="openai"),
    "Cloudflare AI Gateway": gateways.cloudflare("openai/gpt-4o-mini", account_id="ACCT"),
    "Vercel AI Gateway": gateways.vercel("openai/gpt-4o-mini"),
    "Bifrost": gateways.bifrost("openai/gpt-4o-mini"),
    "Kong AI Gateway": gateways.kong("gpt-4o-mini", route_url="http://kong:8000/chat"),
    "MLflow AI Gateway": gateways.mlflow("my-endpoint"),
}
for label, cfg in presets.items():
    client = build_client(cfg)  # a real OpenAI-compatible client; construction makes no request
    url = cfg.base_url.rstrip("/") + cfg.chat_path
    print(f"{label:22} POST {url}  headers={sorted(cfg.headers)}")
    assert isinstance(client, OpenAICompatClient)

# Anything else that speaks the OpenAI dialect (vLLM, Ollama, Azure-compatible proxies, ...):
local = LLMConfig(model="gemma4:latest", base_url="http://localhost:11434/v1")


# A fully custom backend: implement `complete()`, `embed()`, `aclose()` and register it by name.
class MyClient:
    def __init__(self, config: LLMConfig) -> None:
        self.config = config

    async def complete(self, request):
        raise NotImplementedError  # call your backend, return an LLMResponse

    async def embed(self, texts, model=None):
        raise NotImplementedError

    async def aclose(self) -> None:
        return None


register_provider("mine", MyClient)
assert isinstance(build_client(LLMConfig(provider="mine", model="x")), MyClient)
print("custom provider registered")

Preset	Base URL	Auth and routing	Model naming
gateways.litellm_proxy(model)	http://localhost:4000/v1	Bearer ${LITELLM_API_KEY} (api_key=None for none)	whatever the proxy routes
gateways.portkey(model, provider="openai")	https://api.portkey.ai/v1	x-portkey-api-key: ${PORTKEY_API_KEY}, x-portkey-provider; Bearer ${PROVIDER_API_KEY} is the upstream key	provider's model name
gateways.cloudflare(model, account_id, gateway_id="default")	https://gateway.ai.cloudflare.com/v1/{account}/{gateway}/compat	cf-aig-authorization: Bearer ${CLOUDFLARE_API_TOKEN} (authenticated=False to omit); Bearer ${PROVIDER_API_KEY}	provider/model
gateways.vercel(model)	https://ai-gateway.vercel.sh/v1	Bearer ${AI_GATEWAY_API_KEY}	creator/model, case-sensitive
gateways.bifrost(model)	http://localhost:8080/v1	optional api_key=	provider/model
gateways.kong(model, route_url)	your route URL, used verbatim (chat_path="")	optional api_key=	the model your ai-proxy allows
gateways.mlflow(endpoint)	http://localhost:5000/gateway/mlflow/v1	optional api_key=	the gateway endpoint name is the model

Verify your gateway before relying on a preset

Preset URLs and headers follow each vendor's documentation, but only a local Ollama has been exercised live. Two presets depend on the gateway version: Bifrost's path (/v1 vs /openai) and MLflow's base path (/gateway/mlflow/v1 in current docs, /v1 in older 3.x). Override base_url if yours differs. Run the smoke test against your gateway before production; it takes a minute and a few thousand tokens.

Model choice per call

Every algorithm accepts model=; it is passed in each LLMRequest. A request with model=None uses the client's default. There is one LLMHandler per engine, so to route different models to different backends, give the handler a small client that dispatches on request.model (see Extending AAA).

What the OpenAI-compatible client does

• Tool calls: your ToolSpecs become tools=[{"type":"function",...}]; calls come back as parsed ToolCalls. Arguments that are not valid JSON are kept in raw_arguments and reported back to the model as an error instead of guessed at.
• Structured output: response_schema becomes response_format={"type":"json_schema","json_schema":{...,"strict":true}}. Strict mode is demanding on some providers (all properties required, additionalProperties: false); built-in schemas follow those rules, but verify yours against your provider.
• Streaming: server-sent events with stream_options.include_usage. Not every gateway returns usage in the stream.
• Cost: computed from the prices in your config, if set.

Error mapping

HTTP and transport failures become typed errors so retry logic can be correct:

Condition	Error	Retryable
429	RateLimited (carries retry_after from the Retry-After header)	yes
408, 409, 425, any 5xx; timeouts; connection errors; non-JSON bodies	TransientError	yes
401, 403	AuthError	no
content_filter / content_policy in the body, or finish reason content_filter	ContentFiltered	no
400/413 mentioning context length / too many tokens	ContextOverflow	no
anything else 4xx	FatalError	no
the breaker is open	CircuitOpen	no

LLMHandler converts a raised LLMError into an aaa.llm.failed event (with error.type, error.message, error.retryable). The built-in loops end the run as failed on it; write your own algorithm to react differently (shrink the prompt on ContextOverflow, fall back to a cheaper model on RateLimited).

The OpenAI Responses API

provider="openai_responses" speaks POST /v1/responses instead of chat completions. Its point is that the provider can hold the conversation (store=True): the LLMHandler then sends only the new messages on each turn. It is a different wire format (flat function tools, function_call and function_call_output items, previous_response_id), mapped for you.

client = build_client(LLMConfig(provider="openai_responses", model="gpt-4o", base_url="https://api.openai.com/v1", api_key="${OPENAI_API_KEY}"))

Set OpenAIResponsesClient(config, store=False) for stateless use. Embeddings are not part of this API: use an openai_compat client for them. The data retention, erasure and fallback behavior is described in Chat sessions and provider-side state. Both clients report Usage.cached_tokens, and LLMConfig.price_cached_in_per_mtok prices cache hits.

Retries and circuit breaking

Opt-in wrappers, applied around any client:

from aaa.llm import CircuitBreakerClient, RetryingClient, build_client

client = RetryingClient(
    CircuitBreakerClient(build_client(cfg), failure_threshold=5, cooldown_s=30),
    max_attempts=3,
    base_delay=0.5,
    max_delay=20,
)

• RetryingClient retries only retryable errors, with exponential backoff and full jitter, and honors Retry-After.
• CircuitBreakerClient opens after failure_threshold consecutive transient failures, fails fast with CircuitOpen for cooldown_s, then lets one probe through (half-open). Caller mistakes (auth, bad request) never trip it.
• Streaming passes straight through both wrappers (no retry once bytes may have been seen).

Do not stack retries

If your gateway already retries and fails over (Portkey, LiteLLM, Cloudflare), leave RetryingClient out or set max_attempts=1, or one outage becomes a retry storm.

LiteLLM in-process

build_client(
    LLMConfig(provider="litellm", model="anthropic/claude-sonnet-4-5")
)  # pip install "...[litellm]"

Uses the LiteLLM SDK instead of HTTP, maps its exceptions onto the same error types, and reports LiteLLM's computed cost. Not exercised live in this repository's test suite.

The LLMHandler

LLMHandler(client, concurrency=8, emit_deltas=True).attach(engine)

• concurrency bounds simultaneous requests per worker.
• It rewrites each tool call's id to be unique per call (call_<request id>_<n>): providers may omit or repeat ids.
• It reconstructs full conversations for delta requests from a small cache, falling back to the log after a restart.
• Failures are events; the handler never raises into the bus.

Custom backends

from aaa.llm import LLMConfig, register_provider


class MyClient:
    def __init__(self, config: LLMConfig) -> None: ...
    async def complete(self, request): ...   # call your backend, return an LLMResponse
    async def embed(self, texts, model=None): ...
    async def aclose(self): ...


register_provider("mine", MyClient)
client = build_client(LLMConfig(provider="mine", model="x"))

Raise the typed errors from aaa.llm.errors so retries and algorithms behave correctly.

Fakes for tests

FakeLLMClient(script) takes a list of responses (or exceptions to raise) or a function of the request, and records the requests it saw. ReplayLLMClient(events) serves the answers recorded in a run's log. See Testing your agents.