OpenAICompatibleProvider

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

class zett_agent.providers.openai_compatible.OpenAICompatibleProvider(*, model, api_key, base_url, transport=None, temperature=None, response=False, retry=RetryOptions(base_delay=0.25, max_delay=8.0, max_retries=3), send_prompt_cache_key=True, extra_headers=None, extra_query=None, extra_body=None, timeout=TimeoutOptions(connect=5.0, first_byte=60.0, idle=600.0))

Stream a service that speaks the OpenAI Chat Completions protocol.

This is the base class of OpenAIProvider and the providers for the services that bend the protocol. A provider for another one is a subclass that sets provider_name and overrides the hooks below where the service differs; it does not copy the stream loop.

The hooks are ordinary methods and never need super() to keep the protocol working: each default is the plain OpenAI behavior, and a subclass that wants to add to it calls super() itself.

• vendor_headers() -- headers sent on every request.

• request_fields() -- JSON fields of one request.

• replay_reasoning() -- the vendor extension of an earlier assistant message.

• read_delta() -- one streamed delta.

• merge_replay() -- how streamed opaque blocks are put together.

• read_usage() -- token counts.

• chat_stop() -- why the response ended.

• status_error() -- what an HTTP failure means.

参数:

• model (str)

• api_key (str)

• base_url (str)

• transport (httpx.AsyncBaseTransport | None)

• temperature (float | None)

• response (bool)

• retry (RetryOptions)

• send_prompt_cache_key (bool)

• extra_headers (Mapping[str, str] | None)

• extra_query (Mapping[str, str] | None)

• extra_body (Mapping[str, JsonValue] | None)

• timeout (TimeoutOptions)

provider_name: str = 'openai'

Name stored on messages this provider produced and compared before reasoning is replayed.

reasoning_keys: ClassVar[tuple[str, ...]] = ('reasoning_content', 'reasoning')

Fields of a streamed delta that carry reasoning text, in order of preference. Services differ on the name (reasoning_content and reasoning are the two in use), and reading both is harmless because only the first one that has text is used.

response = 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.

vendor_headers()

Return the headers this service wants on every request.

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

返回类型:

dict[str, str]

request_fields(request)

Return the JSON fields this service takes for request, beyond the standard ones.

The standard ones (model, messages, tools, temperature, tool_choice, prompt_cache_key) are set by the provider. This is where the reasoning level is spelled the way the service wants it. The default is OpenAI's top-level reasoning_effort. A field given in extra_body replaces one returned here.

参数:

request (ModelRequest)

返回类型:

dict[str, JsonValue]

replay_reasoning(message, payload)

Add what this service needs of an earlier assistant turn to its rendered payload.

The default returns the reasoning as reasoning_content, which reasoning services expect to see again on later requests. A message from another provider is skipped, because reasoning from one model is not context for another.

参数:

• message (AssistantMessage)

• payload (dict[str, JsonValue])

返回类型:

None

read_delta(delta)

Read one streamed delta.

The default reads content, refusal, tool_calls, and the first of reasoning_keys that has text. A service with fields of its own returns a ChatDelta that also carries them, usually by extending super().read_delta.

参数:

delta (Mapping[str, JsonValue])

返回类型:

ChatDelta

merge_replay(blocks, fragment)

Add one streamed opaque block (ChatDelta.replay) to the ones collected so far.

The default appends. A service that streams one block in pieces overrides this to join them, and blocks is what finally becomes replay_blocks.

参数:

• blocks (list[dict[str, JsonValue]])

• fragment (dict[str, JsonValue])

返回类型:

None

read_usage(usage)

Read the usage object of a chunk.

The default accepts every spelling of the cache and reasoning counters that the compatible services use.

参数:

usage (Mapping[str, JsonValue])

返回类型:

ModelUsage

chat_stop(finish_reason, *, has_tool_calls, refusal_text, stream_error, native_finish_reason)

Say why the response ended. Override it for a service with finish reasons of its own.

参数:

• finish_reason (str | None)

• has_tool_calls (bool)

• refusal_text (str)

• stream_error (Mapping[str, JsonValue])

• native_finish_reason (str | None)

返回类型:

StopInfo

status_error(error)

Turn an HTTP failure into the typed error the runtime reports.

The default reports a refusal the service sent as an HTTP error, a rejected credential, and everything else as ProviderHTTPError. A service that explains failures in a body of its own overrides this and falls back to super().

参数:

error (APIStatusError)

返回类型:

ProviderError

async aclose()

Close the SDK's owned HTTP connection pool.

返回类型:

None

native_deferred_tools = False

See zett_agent.model.NativeDeferredToolsModel.native_deferred_tools. A provider opts in by overriding this on its class; the default keeps Responses, Chat Completions, Google, and Ollama on the withhold-and-search path.

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.

示例

Use a model or tool from a plain function:

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

参数:

runtime (SyncRuntime | None)

返回类型:

SyncObject[Self]