Implement a model adapter#

Use a custom adapter for a new protocol or a deterministic test double. For an OpenAI-compatible HTTP service, first consider OpenAIProvider with a custom base_url; a new adapter may be unnecessary.

The protocol#

AgentModel accepts one fully assembled ModelRequest and returns an async iterator of ModelEvent. It has a model-owned RetryOptions policy. The protocol method is declared as a normal def returning AsyncIterator; an implementation may use async def with yield to satisfy that contract.

retry and stream are all an adapter needs. One more attribute is optional: native_deferred_tools: bool, declared by NativeDeferredToolsModel. Set it to True only when your endpoint keeps a deferred tool out of the model’s context yet can load it when a search result names it, as Anthropic’s defer_loading does: the adapter then receives every deferred definition and marks it deferred itself. Leave it out when the model only learns a deferred tool from a search result, as with the Responses API; ToolSearchExtension then withholds those definitions from the request.

from collections.abc import AsyncIterator
from zett_agent.messages import AssistantMessage
from zett_agent.model import (
    ModelEvent,
    ModelRequest,
    ModelResponse,
    RetryOptions,
)

class FixedModel:
    retry = RetryOptions(max_retries=0)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        yield ModelEvent.text("Hello")
        yield ModelEvent.completed(
            ModelResponse(AssistantMessage(content="Hello"))
        )

Required stream invariants#

  • Yield zero or more text, reasoning, or tool-call fragments.

  • End a successful stream with exactly one completed ModelResponse.

  • Include the complete answer and complete ToolCalls in that response, even when they were also streamed as deltas. The runtime does not reconstruct the final message from display text.

  • Do not emit anything after the final response.

  • Raise on transport/protocol failure; do not disguise failure as a successful empty answer.

ModelRequest contains the complete context for this step. Support the applicable message roles and typed image parts explicitly. AgentMessage is internal input, not assistant output; built-in adapters map it to user input for vendor APIs. Declare cache_key to the endpoint when its protocol caches prompt prefixes; the OpenAI adapter sends it as prompt_cache_key on both of its protocols.

Tool calls and usage#

Use ToolCallDelta for partial ID/name/argument fragments and ToolCall for final parsed arguments. Correlate ToolMessage results by call ID, not list position. Report ModelUsage on the final response; input includes cache reads/writes and output includes reasoning when reported. Missing counters remain zero, but a protocol that reports no cache counters at all must pass cache_reported=False so callers can tell unknown cache usage from a measured miss.

Replay signed vendor blocks only for the same provider/model identity. Preserve them in AssistantMessage.replay_blocks instead of concatenating them into text.

Retry policy belongs here#

Retry transient failures only before an event escapes to the caller. Restarting after partial text or tool deltas would duplicate visible output and could confuse tool-call assembly. Validation/authentication failures should not be blindly retried. Close SDK streams in finally/context managers when the consumer stops early.

Run a test adapter: Your first agent. Inspect built-in implementations under Providers for concrete vendor mappings.