Model recipes: Bedrock, Azure OpenAI, Mistral, Groq, OpenRouter

Each recipe wraps the vendor's own SDK in a CallableModel. The adapter adds no dependency: you install the SDK, and the recipe imports it only when you build the model. Every recipe turns the vendor's usage into Usage, so the ledger prices it from your Prices table, and turns SDK errors into typed model errors.

flowchart LR
  L["run_tool_loop<br/>or executor.model(port)"] -->|LLMRequest| C["CallableModel(call)"]
  C --> T["to vendor kwargs"] --> S["vendor SDK<br/>converse · chat.complete · chat.completions.create"]
  S -->|"reply"| F["from vendor reply<br/>text · tool calls · Usage"] --> C
  S -.->|"SDK error"| E["typed_error<br/>ModelRateLimited · ModelTimeout …"] -.-> C
  C -->|LLMResponse| L

Vendor	SDK	Recipe	Shape
Amazon Bedrock	boto3	examples/models_bedrock.py	Converse API, own converter
Azure OpenAI	openai	examples/models_openai_compatible.py	Chat Completions
Groq	groq	examples/models_openai_compatible.py	Chat Completions
OpenRouter	openai	examples/models_openai_compatible.py	Chat Completions
Mistral	mistralai	examples/models_mistral.py	Chat Completions with Mistral's names

Amazon Bedrock (Converse)

def converse_model(client: Any, model_id: str) -> CallableModel:
    """A model port over a `bedrock-runtime` client's `converse` call."""

    def call(request: LLMRequest) -> dict[str, Any]:
        model = request.model or model_id
        try:
            raw = client.converse(**to_converse(request, model))  # sync: runs in a worker thread
        except Exception as exc:
            error = typed_error(exc, provider="bedrock", model=model)
            if error is None:
                raise
            raise error from exc
        return from_converse(raw)

    return CallableModel(call, provider="bedrock", model=model_id)


def bedrock_model(model_id: str, region: str = "us-east-1") -> CallableModel:
    """Bedrock with the default AWS credential chain; boto3 is imported on first use."""
    import boto3

    return converse_model(boto3.client("bedrock-runtime", region_name=region), model_id)

The converter keeps the system prompt apart, sends tool results as one user turn, and maps stopReason to a finish reason:

STOP = {"end_turn": "stop", "stop_sequence": "stop", "tool_use": "tool_calls",
        "max_tokens": "length", "guardrail_intervened": "content_filter",
        "content_filtered": "content_filter"}  # fmt: skip


def _turn(message: Message) -> dict[str, Any]:
    content: list[dict[str, Any]] = [{"text": message.content}] if message.content else []
    content += [
        {"toolUse": {"toolUseId": call.id, "name": call.name, "input": call.arguments}}
        for call in message.tool_calls
    ]
    return {"role": message.role, "content": content}


def to_converse(request: LLMRequest, model_id: str) -> dict[str, Any]:
    """`converse(**kwargs)` for a request: system apart, tool results as user turns."""
    turns: list[dict[str, Any]] = []
    for message in request.messages:
        if message.role == "system":
            continue
        if message.role != "tool":
            turns.append(_turn(message))
            continue
        result = {
            "toolResult": {
                "toolUseId": message.tool_call_id,
                "content": [{"text": message.content}],
            }
        }
        if turns and turns[-1]["role"] == "user" and "toolResult" in turns[-1]["content"][0]:
            turns[-1]["content"].append(result)  # results of one turn share one user message
        else:
            turns.append({"role": "user", "content": [result]})
    kwargs: dict[str, Any] = {"modelId": model_id, "messages": turns}
    system = [{"text": m.content} for m in request.messages if m.role == "system"]
    if system:
        kwargs["system"] = system
    config = {"maxTokens": request.max_output_tokens, "temperature": request.temperature,
              "topP": request.top_p, "stopSequences": list(request.stop) or None}  # fmt: skip
    if any(value is not None for value in config.values()):
        kwargs["inferenceConfig"] = {k: v for k, v in config.items() if v is not None}
    if request.tools and request.tool_choice != "none":  # Converse has no "none": send no tools
        specs = [
            {
                "toolSpec": {
                    "name": t.name,
                    "description": t.description,
                    "inputSchema": {"json": t.parameters},
                }
            }
            for t in request.tools
        ]
        choice: dict[str, Any] = {"auto": {"auto": {}}, "required": {"any": {}}}.get(
            request.tool_choice, {"tool": {"name": request.tool_choice}}
        )
        kwargs["toolConfig"] = {"tools": specs, "toolChoice": choice}
    return kwargs


def from_converse(raw: dict[str, Any]) -> dict[str, Any]:
    """A `CallableModel` reply mapping from a Converse response."""
    blocks = raw.get("output", {}).get("message", {}).get("content", [])
    calls = [
        ToolCall(
            id=b["toolUse"]["toolUseId"],
            name=b["toolUse"]["name"],
            arguments=b["toolUse"].get("input") or {},
        )
        for b in blocks
        if "toolUse" in b
    ]
    usage = raw.get("usage", {})
    reason = raw.get("stopReason", "")
    return {
        "text": "".join(b["text"] for b in blocks if "text" in b),
        "tool_calls": calls,
        "finish_reason": STOP.get(reason, "other"),
        "raw_finish_reason": reason,
        "usage": Usage(  # Converse counts cache reads and writes apart from inputTokens
            input_tokens=usage.get("inputTokens", 0),
            cached_input_tokens=usage.get("cacheReadInputTokens", 0),
            cache_write_input_tokens=usage.get("cacheWriteInputTokens", 0),
            output_tokens=usage.get("outputTokens", 0),
        ),
    }

Converse has no tool_choice="none", so the recipe sends no tools then. Credentials come from the default AWS chain (environment, profile or role).

Azure OpenAI, Groq and OpenRouter

All three speak Chat Completions, so one function covers them. It reuses build_chat_request and parse_chat_completion from agent_adapter.models.openai:

def chat_model(
    create: Callable[..., Any], *, provider: str, model: str, plain_tools: bool = False
) -> CallableModel:
    """A model port over any OpenAI-shaped `chat.completions.create` of a vendor SDK."""

    def call(request: LLMRequest) -> dict[str, Any]:
        kwargs = build_chat_request(request, model=request.model or model, strict_tools=False)
        if plain_tools:  # send tool schemas without the OpenAI-only "strict" flag
            for tool in kwargs.get("tools", []):
                tool["function"].pop("strict", None)
        try:
            raw = create(**kwargs)  # a sync SDK call: CallableModel runs it in a worker thread
        except Exception as exc:
            error = typed_error(exc, provider=provider, model=model)
            if error is None:
                raise
            raise error from exc
        reply = parse_chat_completion(raw)  # SDK object or dict -> text, tool calls, usage
        return {
            "id": reply.id,
            "text": reply.text,
            "tool_calls": reply.tool_calls,
            "finish_reason": reply.finish_reason,
            "raw_finish_reason": reply.raw_finish_reason,
            "usage": reply.usage,  # prompt tokens split into uncached, cached and written
        }

    return CallableModel(call, provider=provider, model=model)

def azure_model(endpoint: str, api_key: str, deployment: str) -> CallableModel:
    """Azure OpenAI: `model` is your deployment name; the SDK is imported on first use."""
    from openai import AzureOpenAI

    client = AzureOpenAI(azure_endpoint=endpoint, api_key=api_key, api_version="2024-10-21")
    return chat_model(client.chat.completions.create, provider="azure", model=deployment)


def groq_model(api_key: str, model: str = "llama-3.3-70b-versatile") -> CallableModel:
    """Groq's own SDK (`pip install groq`)."""
    from groq import Groq

    client = Groq(api_key=api_key)
    return chat_model(
        client.chat.completions.create, provider="groq", model=model, plain_tools=True
    )


OPENROUTER = "https://openrouter.ai/api/v1"


def openrouter_model(api_key: str, model: str = "openai/gpt-5.4") -> CallableModel:
    """OpenRouter through the openai SDK; `model` is OpenRouter's "vendor/model" id."""
    from openai import OpenAI

    client = OpenAI(base_url=OPENROUTER, api_key=api_key)
    return chat_model(client.chat.completions.create, provider="openrouter", model=model)

Vendor	Notes
Azure OpenAI	model is your deployment name; use the API version your resource supports
Groq	tool schemas go without the OpenAI-only strict flag
OpenRouter	ids are vendor/model; one pattern row ("openai/*") prices a whole vendor

Azure also works with the built-in port: OpenAIModel("shop-gpt", client=AsyncAzureOpenAI(...)).

Mistral

def complete_model(client: Any, model: str) -> CallableModel:
    """A model port over a `Mistral` client's `chat.complete`."""

    def call(request: LLMRequest) -> dict[str, Any]:
        name = request.model or model
        try:
            raw = client.chat.complete(**to_mistral(request, name))  # sync: a worker thread
        except Exception as exc:
            error = typed_error(exc, provider="mistral", model=name)
            if error is None:
                raise
            raise error from exc
        return from_mistral(raw)

    return CallableModel(call, provider="mistral", model=model)


def mistral_model(api_key: str, model: str = "mistral-large-latest") -> CallableModel:
    """Mistral's API; the SDK is imported on first use."""
    from mistralai import Mistral

    return complete_model(Mistral(api_key=api_key), model)

Mistral calls the output limit max_tokens and the seed random_seed, and "any" forces a tool. It may send content as a list of chunks and arguments as an object:

def to_mistral(request: LLMRequest, model: str) -> dict[str, Any]:
    """`chat.complete(**kwargs)`: Chat Completions kwargs with Mistral's own names."""
    kwargs = build_chat_request(request, model=model, strict_tools=False)
    if "max_completion_tokens" in kwargs:
        kwargs["max_tokens"] = kwargs.pop("max_completion_tokens")
    if "seed" in kwargs:
        kwargs["random_seed"] = kwargs.pop("seed")
    kwargs.pop("reasoning_effort", None)  # not a Mistral chat parameter
    if kwargs.get("tool_choice") == "required":
        kwargs["tool_choice"] = "any"
    return kwargs


def _text(content: Any) -> str | None:
    """Mistral may send content as a list of chunks; keep the text ones."""
    if isinstance(content, list):
        return "".join(str(c.get("text", "")) for c in content if c.get("type") == "text")
    return None if content is None else str(content)


def from_mistral(raw: Any) -> dict[str, Any]:
    """A `CallableModel` reply mapping from a `ChatCompletionResponse` (or its dict)."""
    data = raw.model_dump() if hasattr(raw, "model_dump") else dict(raw)
    for choice in data.get("choices") or []:
        message = choice.get("message") or {}
        message["content"] = _text(message.get("content"))
        for call in message.get("tool_calls") or []:
            arguments = call["function"].get("arguments")
            if isinstance(arguments, dict):  # Mistral may send arguments as an object
                call["function"]["arguments"] = json.dumps(arguments)
    reply = parse_chat_completion(data)
    return {
        "id": reply.id,
        "text": reply.text,
        "tool_calls": reply.tool_calls,
        "finish_reason": reply.finish_reason,
        "raw_finish_reason": reply.raw_finish_reason,
        "usage": reply.usage,
    }

Prices

Each recipe loads its own rows. The rates in the examples are placeholders: copy yours from the vendor's price page.

PRICES = Prices.load(  # example rates: copy yours from each vendor's price page
    {
        "version": "mine-2026-11",
        "models": {
            "shop-gpt": {"provider": "azure", "input": 2.50, "cached_input": 0.25, "output": 15.0},
            "llama-3.3-70b-versatile": {"provider": "groq", "input": 0.60, "output": 0.80},
            "openai/*": {"provider": "openrouter", "input": 2.50, "output": 15.00},
        },
    }
)

Errors

def typed_error(exc: Exception, *, provider: str, model: str) -> ModelError | None:
    """The SDK error as a `ModelError`, or None for one that is not a provider error."""
    # Typed errors let retries, budgets and FallbackModel tell a rate limit from a bug.
    status = getattr(exc, "status_code", None)
    response = getattr(exc, "response", None)
    if status is None and isinstance(response, dict):  # botocore's ClientError
        status = response.get("ResponseMetadata", {}).get("HTTPStatusCode")
    if isinstance(status, int):
        cls = with_overflow(class_for_status(status), exc, getattr(exc, "body", None))
        return model_error(cls, exc, provider=provider, model=model, status_code=status)
    name = type(exc).__name__  # SDKs name their transport errors alike
    if "Timeout" in name:
        return model_error(ModelTimeout, exc, provider=provider, model=model)
    if "Connection" in name:
        return model_error(ModelUnavailable, exc, provider=provider, model=model)
    return None

SDK error	Becomes
a status code (status_code, or botocore's ResponseMetadata)	ModelRateLimited (429), ModelAuthError (401, 403), ModelTimeout (408), ModelUnavailable (5xx), ModelBadRequest (4xx) or ModelContextOverflow
a timeout or connection error	ModelTimeout, ModelUnavailable
anything else	propagates unchanged

FallbackModel falls over on rate limits, outages and timeouts, so the typed errors make a second vendor a one-line change (Models).

Output

Each example runs offline on a fake SDK client:

bedrock stop 2 Ticket T-1: waiting for parts.
amazon.nova-pro-v1:0 ['0.000800', '0.000800']

The line under each answer is the price of each model call from the ledger.

Files

File	Does
examples/models_bedrock.py	Bedrock Converse: converter, model port, prices
examples/models_openai_compatible.py	Azure OpenAI, Groq, OpenRouter over one Chat Completions port
examples/models_mistral.py	Mistral: converter, model port, prices
examples/models_common.py	typed_error and the shared demo run
tests/docs/test_model_recipes.py	runs each recipe on fake SDK modules

More on wrapping a model: Bring your own model.