AnthropicProvider#

Connect a provider and configure retries, credentials, and transport. See the guide.

class zett_agent.providers.anthropic.AnthropicProvider(model, api_key, base_url='https://api.anthropic.com', transport=None, *, response=False, retry=RetryOptions(base_delay=0.25, max_delay=8.0, max_retries=3), prompt_cache=True, max_tokens=16384, extra_headers=None, extra_query=None, extra_body=None, timeout=TimeoutOptions(connect=5.0, first_byte=60.0, idle=600.0))[source]#

Map Anthropic SDK streams, including signed thinking and tool blocks.

Parameters:

  • model (str) – Model identifier accepted by the Anthropic-compatible endpoint.

  • api_key (str) – API credential supplied by the application.

  • base_url (str) – API root, defaulting to the official Anthropic endpoint.

  • transport (httpx.AsyncBaseTransport | None) – Optional HTTPX transport for isolated tests or custom routing.

  • response (bool) – Must remain false; this adapter uses Anthropic Messages API.

  • retry (RetryOptions) – Exponential backoff for transient failures before output starts.

  • timeout (TimeoutOptions) – How long the provider may stay silent before the request is given up on: connecting, waiting for the response headers, and between chunks of a response that has started. See TimeoutOptions; a timeout raises ProviderTimeoutError and is not retried.

  • prompt_cache (bool) – Mark cache breakpoints on the tool definitions, the system prompt, and the conversation so far. Anthropic caches only what a marker asks for, so disabling this trades prompt-cache reads for a payload without cache_control.

  • max_tokens (int) – Output budget for one response, excluding thinking. A tool call that writes a large file must fit in it: a response cut off by this limit leaves the call’s JSON incomplete, and the model is then told to retry with a smaller call. Lower it for models with a smaller output limit.

  • extra_headers (Mapping[str, str] | None) – HTTP headers sent on every request, for gateways that route or bill on a header or for anthropic-beta flags. A header given here replaces the SDK default of the same name.

  • extra_query (Mapping[str, str] | None) – URL query parameters sent on every request.

  • extra_body (dict[str, JsonValue]) – JSON fields merged into every request body last, so they add an option the adapter does not model and may override one it does.

Note

Client-side tool search works here too: ToolSearchExtension ranks the deferred tools locally, and this adapter declares them with defer_loading and answers the search with tool_reference blocks.

Note

Signed replay blocks belong to the originating provider and model. Persist them unchanged for tool round trips; do not render them as user-facing reasoning. Await aclose() to release client resources.

Examples

Stream an Anthropic model through the provider-neutral client:

model = AnthropicProvider(
    model=os.environ["ANTHROPIC_MODEL"],
    api_key=os.environ["ANTHROPIC_API_KEY"],
)
try:
    client = await create_agent(model)
    reply = await client.run("Review the proposed API")
finally:
    await model.aclose()

Warning

Signed thinking and tool replay blocks must be returned unchanged only to the same provider and model. They are not interchangeable with plain reasoning text.

See also

AssistantMessage documents replay_blocks; What the model sees explains replay during tool round trips.

native_deferred_tools: bool = True#

See zett_agent.model.NativeDeferredToolsModel.native_deferred_tools. The Messages API accepts defer_loading tools and loads one when a tool_reference names it, so every deferred definition travels with the request.

async aclose()#

Close the owned SDK connection pool.

Return type:

None

bearer_auth: ClassVar[bool] = False#

Authenticate with Authorization: Bearer as well as x-api-key. The services that take only the bearer token ignore the other header.

finish_usage(usage, update)#

Fold the usage object of the message_delta event into usage.

The default takes the final output_tokens, which is the only count Anthropic updates.

Parameters:
Return type:

ModelUsage

mark_cache(body, client_tools)#

Put cache markers into body (and the tool definitions in client_tools).

The default marks Anthropic’s breakpoints on the largest stable request prefixes: the tool definitions, the whole system prompt, and the conversation so far. Anthropic caches only what a cache_control marker asks for, then reads each cached prefix at a fraction of the input price. A prefix shorter than the model’s minimum cacheable length is ignored silently. Nothing is marked unless prompt_cache is on.

Parameters:
Return type:

None

message_stop(finish_reason, details, *, has_tool_calls)#

Say why the response ended, from stop_reason and stop_details.

Parameters:

  • finish_reason (str | None)

  • details (Mapping[str, JsonValue])

  • has_tool_calls (bool)

Return type:

StopInfo

provider_name: str = 'anthropic'#

Name stored on messages this provider produced. Signed blocks are sent back only to the same name and model.

request_fields(request)#

Return the reasoning and output-budget fields of request.

These are merged over the standard body. The default is Anthropic’s: extended thinking with a token budget taken from the reasoning level, and max_tokens raised by that budget because thinking is paid out of it. Thinking cannot be combined with a forced tool choice, and is explicitly disabled when the level is OFF. A field given in extra_body replaces one returned here.

Parameters:

request (ModelRequest)

Return type:

dict[str, JsonValue]

response: bool = False#

Whether this adapter uses a Responses-style endpoint instead of its provider’s traditional message or chat-completions endpoint. Only the providers read it; the runtime does not.

start_usage(usage)#

Read the usage object of the message_start event.

Cache reads and writes are reported next to input_tokens, not inside it, so the total input is their sum.

Parameters:

usage (Mapping[str, JsonValue])

Return type:

ModelUsage

status_error(error)#

Turn an SDK failure from opening the request into the typed error the runtime reports.

Parameters:

error (AnthropicError)

Return type:

ProviderError

sync(*, runtime=None)#

Return a context-managed synchronous view; optionally share a runtime.

Omitting runtime creates a new background thread for this view. Passing an existing runtime reuses its loop and leaves its shutdown to the caller. This API is for service entry points, not Extension lifecycle hooks.

Examples

Use a model or tool from a plain function:

with read_file.sync() as read:
    result = read({"path": "README.md"})
Parameters:

runtime (SyncRuntime | None)

Return type:

SyncObject[Self]

vendor_headers()#

Return the headers this service wants on every request.

They are sent before extra_headers, so the application can still replace one.

Return type:

dict[str, str]