OpenRouterProvider#

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

class zett_agent.providers.openrouter.OpenRouterProvider(model, api_key, transport=None, *, base_url=None, temperature=None, retry=RetryOptions(base_delay=0.25, max_delay=8.0, max_retries=3), timeout=TimeoutOptions(connect=5.0, first_byte=60.0, idle=600.0), site_url=None, app_name=None, send_prompt_cache_key=True, extra_headers=None, extra_query=None, extra_body=None)[源代码]#

Stream any model OpenRouter serves.

参数:

model (str) -- OpenRouter model slug, such as anthropic/claude-sonnet-4.5.
api_key (str) -- OpenRouter API key.
transport (httpx.AsyncBaseTransport | None) -- Optional HTTPX transport for custom routing or tests.
base_url (str | None) -- API root; None uses https://openrouter.ai/api/v1.
temperature (float | None) -- Optional sampling temperature.
retry (RetryOptions) -- Model-owned retry/backoff policy; never restarts an emitted stream.
timeout (TimeoutOptions) -- How long the provider may stay silent before the request is given up on. See TimeoutOptions.
site_url (str | None) -- Your app's URL, sent as HTTP-Referer so OpenRouter can attribute traffic.
app_name (str | None) -- Your app's name, sent as X-OpenRouter-Title.
send_prompt_cache_key (bool) -- Declare the run's prompt cache key on every request. OpenRouter uses it to keep a conversation on one upstream provider.
extra_headers (Mapping[str, str] | None) -- HTTP headers sent on every request after the adapter's own.
extra_query (Mapping[str, str] | None) -- URL query parameters sent on every request.
extra_body (Mapping[str, JsonValue] | None) -- JSON fields merged into every request body last. OpenRouter-only fields such as provider (routing preferences) or reasoning options go here; a reasoning given here replaces the one built from the request's effort.

备注

A reasoning level is always sent as reasoning.effort; OpenRouter passes it on in the form the upstream model takes and ignores it for a model that cannot reason. Reasoning that cannot be shown as text (encrypted or summarized) is kept on the message and sent back unchanged.

示例

Route one request through OpenRouter:

model = OpenRouterProvider(
    model="anthropic/claude-sonnet-4.5",
    api_key=os.environ["OPENROUTER_API_KEY"],
    site_url="https://example.com",
    app_name="My agent",
)
try:
    client = await create_agent(model)
    print((await client.run("Explain this function")).content)
finally:
    await model.aclose()

provider_name = 'openrouter'#: Name stored on messages this provider produced and compared before reasoning is replayed.

reasoning_keys: ClassVar[tuple[str, ...]] = ('reasoning', 'reasoning_content')#: 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.

vendor_headers()[源代码]#

Identify the app to OpenRouter.

返回类型:: dict[str, str]

request_fields(request)[源代码]#

Send the reasoning level as a reasoning object, and the parallel-tools preference.

参数:: request (ModelRequest)
返回类型:: dict[str, JsonValue]

replay_reasoning(message, payload)[源代码]#

Send back reasoning_details unchanged, or the plain reasoning text if there are none.

参数:

message (AssistantMessage)
payload (dict[str, JsonValue])

返回类型:

None

read_delta(delta)[源代码]#

Read the reasoning text, and keep every reasoning_details entry for replay.

参数:: delta (Mapping[str, JsonValue])
返回类型:: ChatDelta

merge_replay(blocks, fragment)[源代码]#

Join the pieces of one reasoning detail into one entry.

A detail can arrive in several chunks that share a type and an integer index. Their text and summary are concatenated; any other field takes the latest value that is not null, which is how a signature that closes a text detail is kept. A detail without an index cannot be told apart from its neighbour, so it is never merged.

参数:

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

返回类型:

None

async aclose()#

Close the SDK's owned HTTP connection pool.

返回类型:: None

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

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.

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

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.

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

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]