KimiAnthropicProvider#

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

class zett_agent.providers.kimi.KimiAnthropicProvider(model, api_key, transport=None, *, base_url=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), send_cache_key=True, max_tokens=16384, extra_headers=None, extra_query=None, extra_body=None)[源代码]#

Stream Kimi through its Anthropic Messages protocol.

Use it for a tool that speaks Messages, or to keep a conversation on the protocol the rest of the application uses. KimiProvider is the same service over Chat Completions.

参数:

model (str) -- Kimi model identifier, such as kimi-k3.
api_key (str) -- Moonshot API key, sent as a bearer token.
transport (httpx.AsyncBaseTransport | None) -- Optional HTTPX transport for custom routing or tests.
base_url (str | None) -- API root; None uses https://api.moonshot.ai/anthropic.
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.
send_cache_key (bool) -- Send the run's prompt cache key as metadata.user_id, which Kimi uses to improve its cache hit rate. Use an opaque id, not a personal one.
max_tokens (int) -- Output budget for one response, excluding thinking.
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 (dict[str, JsonValue]) -- JSON fields merged into every request body last.

备注

kimi-k3 always reasons, so an OFF effort sends nothing, and its level goes in output_config.effort. No reasoning field is sent to any other model; extra_body is the way to give it one. Kimi reads only a top-level cache_control and caches by default, so no cache marker is put in the messages. Thinking blocks are kept with their signature and sent back unchanged.

示例

Stream a Kimi model over Messages:

model = KimiAnthropicProvider(model="kimi-k3", api_key=os.environ["MOONSHOT_API_KEY"])
try:
    client = await create_agent(model)
    print((await client.run("Explain this function")).content)
finally:
    await model.aclose()

provider_name: str = 'kimi-anthropic'#: Name stored on messages this provider produced. Signed blocks are sent back only to the same name and model.

bearer_auth: ClassVar[bool] = True#: Authenticate with Authorization: Bearer as well as x-api-key. The services that take only the bearer token ignore the other header.

request_fields(request)[源代码]#

Give kimi-k3 its reasoning level, and the cache key as metadata.user_id.

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

start_usage(usage)[源代码]#

Read the usage counts, with the reasoning tokens Kimi reports under output_tokens_details.

参数:: usage (Mapping[str, JsonValue])
返回类型:: ModelUsage

finish_usage(usage, update)[源代码]#

Fold in the final counts, keeping the reasoning tokens already read if the update has none.

参数:

usage (ModelUsage)
update (Mapping[str, JsonValue])

返回类型:

ModelUsage

async aclose()#

Close the owned SDK connection pool.

返回类型:: None

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.

参数:

body (dict[str, JsonValue])
client_tools (list[dict[str, JsonValue]])

返回类型:

None

message_stop(finish_reason, details, *, has_tool_calls)#

Say why the response ended, from stop_reason and stop_details.

参数:

finish_reason (str | None)
details (Mapping[str, JsonValue])
has_tool_calls (bool)

返回类型:

StopInfo

native_deferred_tools: bool = False#: See zett_agent.model.NativeDeferredToolsModel.native_deferred_tools. Only Anthropic itself loads a defer_loading tool when a tool_reference names it.

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.

status_error(error)#

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

参数:: error (AnthropicError)
返回类型:: 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]

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]