Connect a model provider

Choose an adapter

Every adapter implements the same protocol, so swapping one for another does not touch the rest of your code.

Adapter	Talks to	You supply
OpenAIProvider	OpenAI Chat Completions or Responses	Model name and API key
AnthropicProvider	Anthropic messages API	Model name and API key
GoogleProvider	Google GenAI	Model name and API key
DeepSeekProvider	DeepSeek Chat Completions or Responses	Model name and API key
OllamaProvider	A running Ollama server	Installed model (and server URL if not localhost)

The adapter normalizes messages, tools, and events, not the capabilities of every model. Pick an endpoint and model that support your workflow. In particular, images, visible reasoning, hosted tools, and tool search are not available on every endpoint. This runtime does not expose a general response_format or structured-final-output setting; for validated data, use a typed tool or validate the final answer in your application.

Configure your chosen provider

Read credentials and model names from your environment. Every tab creates a model you can pass to create_agent; close it with await model.aclose() when finished. These snippets configure real services and are not offline examples.

OpenAI

import os
from zett_agent.providers.openai import OpenAIProvider

model = OpenAIProvider(
    model=os.environ["OPENAI_MODEL"],
    api_key=os.environ["OPENAI_API_KEY"],
    response=True,
)

response=True selects the Responses API. Omit it for Chat Completions (the default). Choose a protocol supported by your endpoint; enabling this flag does not add Responses support to a gateway.

Anthropic

import os
from zett_agent.providers.anthropic import AnthropicProvider

model = AnthropicProvider(
    model=os.environ["ANTHROPIC_MODEL"],
    api_key=os.environ["ANTHROPIC_API_KEY"],
)

Uses the Messages API. response=True is not supported.

Google

import os
from zett_agent.providers.google import GoogleProvider

model = GoogleProvider(
    model=os.environ["GOOGLE_MODEL"],
    api_key=os.environ["GOOGLE_API_KEY"],
)

Uses Google GenAI with an API key. base_url points it at a proxy or a local test server. The built-in constructor does not expose Vertex AI project/location configuration.

DeepSeek

import os
from zett_agent.providers.deepseek import DeepSeekProvider

model = DeepSeekProvider(
    model=os.environ["DEEPSEEK_MODEL"],
    api_key=os.environ["DEEPSEEK_API_KEY"],
)

Uses Chat Completions by default. response=True selects a Responses-compatible endpoint when your service supports it.

Ollama

import os
from zett_agent.providers.ollama import OllamaProvider

model = OllamaProvider(
    model=os.environ["OLLAMA_MODEL"],
    base_url="http://127.0.0.1:11434",
)

Start Ollama and install the selected model separately. Construction does not start a server or download a model. No API key is required for a local service; this adapter uses Ollama’s chat protocol only.

A complete tool-calling request

This program makes a real network request and may incur charges. Export the variables in your shell rather than pasting keys into source or logs. The program registers an addition tool, streams the response, and closes the provider even if the request fails.

"""Opt-in network example. Requires credentials and may incur provider charges."""

import asyncio
import os

from zett_agent.client import create_agent
from zett_agent.dispatcher import AgentEventDispatcher
from zett_agent.events import AgentEvent
from zett_agent.model import RetryOptions
from zett_agent.providers.openai import OpenAIProvider
from zett_agent.tools.base import tool


@tool(guidelines="Use for exact integer addition.")
def add(left: int, right: int) -> int:
    """Add two integers exactly."""
    return left + right


class ConsoleEvents(AgentEventDispatcher):
    async def on_text_delta_event(self, event: AgentEvent) -> None:
        print(event.delta, end="", flush=True)


async def main() -> None:
    model = OpenAIProvider(
        model=os.environ["OPENAI_MODEL"],
        api_key=os.environ["OPENAI_API_KEY"],
        retry=RetryOptions(base_delay=0.5, max_delay=8, max_retries=3),
    )
    try:
        client = await create_agent(model, tools=[add], event_dispatcher=ConsoleEvents())
        await client.run("Use the add tool to calculate 20 plus 22, then explain the result.")
        print()
    finally:
        await model.aclose()


if __name__ == "__main__":
    asyncio.run(main())

Download real_provider.py.

export OPENAI_MODEL="your-tool-capable-model"
export OPENAI_API_KEY="your-api-key"
uv run python docs/_examples/real_provider.py

From an installed package, download the file and run python real_provider.py. The exact wording and whether the model obeys the tool request depend on the selected model. To verify the same tool round trip deterministically, run Stream a tool round trip.

Connect a compatible gateway

Use base_url for a service implementing an existing protocol instead of writing a new adapter:

model = OpenAIProvider(
    model=os.environ["GATEWAY_MODEL"],
    api_key=os.environ["GATEWAY_API_KEY"],
    base_url=os.environ["GATEWAY_BASE_URL"],
    send_prompt_cache_key=False,
)

The URL is the API root, for example a service’s .../v1, not its .../chat/completions route. Match the protocol to the gateway. Some gateways reject optional fields such as prompt_cache_key or particular reasoning levels; configure only supported options and test your combination. The adapter does not auto-detect model capabilities from an endpoint URL.

Send extra headers, query parameters, or body fields

Gateways often route, bill, or version on something the adapters do not model. Every adapter accepts extra_headers; the OpenAI-style and Anthropic adapters also take extra_query and extra_body, Google takes extra_body and base_url, and Ollama takes extra_query.

model = OpenAIProvider(
    model=os.environ["GATEWAY_MODEL"],
    api_key=os.environ["GATEWAY_API_KEY"],
    base_url=os.environ["GATEWAY_BASE_URL"],
    extra_headers={"X-Tenant": "acme"},
    extra_query={"api-version": "2025-01-01"},
    extra_body={"service_tier": "flex"},
)

Headers and query parameters go on every request after the adapter’s own, so one given here replaces the SDK default of the same name (use it for User-Agent or an anthropic-beta flag). extra_body is merged into the request body last: it adds options the adapter does not model and may also override ones it does, such as DeepSeek’s thinking. The values are copied and validated when the provider is built, and extra_body must be JSON.

Reasoning effort

Set a provider-neutral level on the run instead of vendor-specific options:

from zett_agent.model import ReasoningEffort

reply = await client.run("Plan the migration.", reasoning_effort=ReasoningEffort.HIGH)

Level	Meaning
MINIMAL	Request a minimal reasoning budget when the model supports it.
LOW / MEDIUM	Lower or moderate effort; MEDIUM is the runtime default.
HIGH / XHIGH	Request more effort for difficult multi-step tasks.
MAX / ULTRA	Highest requested levels; an adapter may map them to the same supported setting.
OFF	Omit or disable explicit reasoning configuration according to the adapter.

The current Responses adapter maps XHIGH, MAX, and ULTRA to high. Budget-based adapters also group the highest levels. OpenAI Chat Completions forwards named levels, so a model that rejects one can still raise an endpoint error. The runtime does not negotiate supported levels.

OFF is not a universal guarantee of zero reasoning: OpenAI adapters omit explicit effort configuration, while some other adapters disable thinking. The endpoint’s default may still reason. Raising the level never guarantees visible reasoning text, a better answer, or a particular price. Set a default for the whole agent with create_agent(model, reasoning_effort=...) and override it per request.

Retries and proxies

RetryOptions retries only before the first streamed event, so a failure halfway through an answer never replays visible text or tool fragments.

OpenAIProvider(
    model=os.environ["OPENAI_MODEL"],
    api_key=os.environ["OPENAI_API_KEY"],
    retry=RetryOptions(base_delay=0.5, max_delay=8, max_retries=3),
)

Default transports honor HTTP_PROXY, HTTPS_PROXY, ALL_PROXY, and NO_PROXY. Pass your own transport when you need custom routing.

max_retries=3 permits up to four transport attempts: the initial one plus three retries. Authentication and validation failures are not blindly retried. The policy belongs to the model adapter, not create_agent. For a total request deadline, see Integrate an application. A silent provider is bounded by the timeouts in the next section, and a timeout is not retried.

When the provider goes quiet

A provider that takes the request and never answers used to be waited on for ten minutes, and then asked three more times; the Google and Ollama adapters waited forever. Every adapter now gives up, and says which wait ran out. TimeoutOptions has one limit for each stretch where nothing may arrive:

Limit	Default	What it bounds
connect	5 s	Opening the connection. A reachable host answers in milliseconds.
first_byte	60 s	From sending the request to receiving the response headers. Streaming APIs answer the headers before the model has produced anything, so a long wait here means a gateway that took the request and did nothing.
idle	600 s	The longest gap between two chunks once the response has started. It is long because a reasoning model may think silently.

from zett_agent.model import TimeoutOptions

OpenAIProvider(
    model=os.environ["OPENAI_MODEL"],
    api_key=os.environ["OPENAI_API_KEY"],
    timeout=TimeoutOptions(first_byte=30, idle=900),
)

There is deliberately no limit on the total time: a long answer that keeps streaming is healthy. Set a limit to None to switch it off. A limit that runs out raises ProviderTimeoutError, whose phase is "connect", "first_byte" or "idle".

A timeout is not retried, except for a failed connection. Retrying a silent provider waits the whole limit a second time, and the first request may still be running, and billed, on the other side. If the response had already started when it went quiet, what arrived was delivered, so a UI has shown it; the request then fails rather than restarting.

Note

The Google and Ollama clients send their request when the first chunk is asked for, so for them first_byte covers connecting, the response headers and the first chunk, and the connection has no limit of its own. Ollama’s default allows ten minutes for that first chunk, because a local model may need to be loaded from disk first.

When the provider answers with a web page

Between you and a provider there is often a reverse proxy, a gateway, a firewall or a captive portal. When that fails it answers with an HTML page, a bare line of text, or nothing, and a failure used to be reduced to “Provider stream request failed”. The status, the provider’s own explanation, and the fact that it was a web page are now in the message:

openai returned HTTP 400: model [gpt-6-Astra] service info not found
openai answered HTTP 502 with a web page ("502 Bad Gateway") instead of an API
response. A proxy, gateway, firewall or login page in front of the API usually
does this; check the base_url and the network path to it.

Failures arrive as ProviderHTTPError, which has status_code, reason (the provider’s own words, when the body was a JSON error), preview and is_html. A rejected credential is still a ProviderAuthError, with the same detail added. The SDK’s own error stays on __cause__.

A streaming request that is answered with a web page or with JSON is an error too, whatever the status. A 200 with a login page, or with JSON because the gateway does not stream, used to be read as a successful reply with no text; it is now raised, naming the content type and what the body said. Any other content type is read as a stream, because some gateways send one as text/plain, and a stream that ends without a single event is raised as an error.

Streams that end early or carry broken tool calls

Providers keep a request alive through three failures that used to end it.

• Malformed or truncated tool arguments. If a tool call’s JSON is invalid, for example because the response hit its output limit or a gateway cut the stream, the call is not run. It becomes a failed tool result that says what happened and how to retry, so the model can resend a smaller call. Slightly malformed but complete JSON is still repaired silently. A cut-off payload is never completed by guesswork, because that could write truncated content. AnthropicProvider(max_tokens=...) sets the per-response output budget (default 16384); raise it for large writes, or lower it for a model with a smaller limit.

• Responses ``error`` events. A mid-stream error event (rate limit, overload) is raised with its own code and message instead of surfacing as a missing terminal response.

• Responses streams without a terminal event. If a gateway drops response.completed after every output item finished, the finished items are used and finish_reason is "incomplete". If an item never finished, or nothing arrived, the stream raised ProviderResponseError; an empty stream is retried because nothing had been shown yet.

Prompt caching

The runtime derives a prompt-cache key from the session, so every step of one conversation can use the same routing hint on endpoints that support it. Cache reuse and savings remain provider decisions, not runtime guarantees. Set cache_key on AgentRunConfig to route several conversations onto one prefix, or construct a provider with send_prompt_cache_key=False when an endpoint rejects the field.

OpenAI sends the key by default. DeepSeek omits it by default because its first-party service caches automatically; enable it only for a gateway that expects it. Other adapters may not expose cache routing.

When a provider reports cache hits, ModelUsage.cache_read_tokens and cache_hit_rate expose them. Requests through gateways that translate protocols often omit the breakdown; ModelUsage.cache_reported is then False, which distinguishes “unknown” from a measured 0% miss.

Always release the adapter

The client does not own a provider you passed in. Close it when the application is done, even if several clients shared it:

await model.aclose()

Changing models mid-conversation

Pass model=other_model to run or stream to override the default for one request. Ordinary messages and local tools remain provider-neutral, but vision support and hosted tools must still match the new endpoint. Signed reasoning/replay blocks are retained only for the provider/model identity that can consume them; do not treat them as portable instructions.

Next: Stream to a terminal or UI, or write your own adapter in Implement a model adapter.