from __future__ import annotations
import base64
import json
import ssl
from collections.abc import AsyncIterator, Mapping, Sequence
from dataclasses import replace
from typing import TYPE_CHECKING, Any, cast
from ..messages import (
AgentMessage,
AnyMessage,
AssistantMessage,
ImageBytesSource,
ImageContent,
ImageUrlSource,
SystemMessage,
TextContent,
ToolMessage,
UserMessage,
)
from ..model import (
DEFAULT_RETRY_OPTIONS,
DEFAULT_TIMEOUT_OPTIONS,
ModelEvent,
ModelRequest,
ModelResponse,
ModelUsage,
ReasoningEffort,
RetryOptions,
ServerToolCall,
ServerToolDefinition,
ServerToolInputDelta,
ServerToolResult,
TimeoutOptions,
ToolCallDelta,
ToolDefinition,
validate_response,
validate_retry,
validate_timeout,
)
from ..stop import StopInfo, StopReason, classify_stop
from .base import (
ProviderAuthError,
ProviderResponseError,
RetryingProvider,
_empty_stream_message,
_parse_tool_calls,
_reasoning_effort_to_budget,
_to_model_data,
_ToolCallAccumulator,
describe_status_error,
guard_idle,
http_error,
httpx_timeout,
require_event_stream,
retry_model_stream,
validate_extra_body,
validate_extra_headers,
validate_extra_query,
)
from .responses import local_tool_search_name
from .stop_reasons import withhold_replay_blocks
from .tool_arguments import ToolArgumentsError, decode_tool_arguments
from .tool_images import expand_tool_images
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable
import httpx
from anthropic import AsyncStream
#: Output budget per response. The original 4096 truncated any tool call that wrote a
#: sizeable file, leaving its JSON arguments incomplete.
DEFAULT_MAX_TOKENS = 16_384
def _tools_to_anthropic_payload(
tools: Sequence[ToolDefinition], server_tools: Sequence[ServerToolDefinition] = ()
) -> list[dict[str, Any]]:
"""Render local tools, deferring the ones tool search is meant to load.
A tool marked ``deferred`` is declared with ``defer_loading``: Anthropic
keeps it out of the model's context until a ``tool_reference`` names it.
That only works when something can produce the reference, so the flag is
honored only when the request carries a search: the local search tool, or
Anthropic's hosted ``tool_search_tool_*``. Without one the tool is declared
normally, because a deferred tool nothing can load would be unusable.
The local search tool is an ordinary function here, since Anthropic has no
client-side search endpoint; the model calls it and its result carries the
references (see :func:`_search_result_blocks`).
"""
if sum(tool.local_tool_search for tool in tools) > 1:
raise ValueError("Only one local tool search tool can be registered")
can_load = any(tool.local_tool_search for tool in tools) or any(
tool.type.startswith("tool_search_tool_") for tool in server_tools
)
rendered: list[dict[str, Any]] = []
for tool in tools:
payload: dict[str, Any] = {
"name": tool.name,
"description": tool.description,
"input_schema": dict(tool.parameters),
}
if tool.deferred and can_load:
payload["defer_loading"] = True
rendered.append(payload)
return rendered
def _search_result_blocks(message: Any) -> list[dict[str, Any]]:
"""Turn a local tool search result into ``tool_reference`` blocks.
The search tool returns the definitions it found; Anthropic needs only their
names, because each deferred tool is already declared in the request. A
search that finds nothing is reported as text, since an empty reference list
is not a valid result.
"""
try:
found = json.loads(message.content)
except (TypeError, json.JSONDecodeError) as error:
raise ProviderResponseError("A local tool search tool must return JSON tool definitions") from error
if not isinstance(found, list):
raise ProviderResponseError("A local tool search tool must return a list of definitions")
names = [item["name"] for item in found if isinstance(item, Mapping) and isinstance(item.get("name"), str)]
if len(names) != len(found):
raise ProviderResponseError("Every definition a local tool search tool returns needs a name")
if not names:
return [{"type": "text", "text": "No matching tools found."}]
return [{"type": "tool_reference", "tool_name": name} for name in names]
def _server_tools_to_anthropic_payload(tools: Sequence[ServerToolDefinition]) -> list[dict[str, Any]]:
"""Render Anthropic server tools using their versioned native schemas."""
return [{"type": tool.type, **dict(tool.configuration)} for tool in tools]
def _apply_prompt_cache(body: dict[str, Any], client_tools: Sequence[dict[str, Any]]) -> None:
"""Mark Anthropic cache breakpoints on the largest stable request prefixes.
Anthropic caches only what a ``cache_control`` marker asks for, then reads
each cached prefix at a fraction of the input price. The three markers below
cover the tool definitions, the whole system prompt, and the conversation so
far, so a repeated prefix is read instead of re-billed as fresh input. A
prefix shorter than the model's minimum cacheable length is ignored
silently.
"""
breakpoint = {"type": "ephemeral"}
# Server tools are provider-owned definitions and must not carry the marker.
# A deferred tool is not part of the cached prefix, so the marker goes on the
# last tool that is loaded up front.
loaded = [tool for tool in client_tools if not tool.get("defer_loading")]
if loaded:
loaded[-1]["cache_control"] = breakpoint
system = body.get("system")
if isinstance(system, str) and system:
body["system"] = [{"type": "text", "text": system, "cache_control": breakpoint}]
messages = body.get("messages")
if messages:
content = messages[-1].get("content")
if isinstance(content, list) and content and isinstance(content[-1], dict):
content[-1]["cache_control"] = breakpoint
[文档]
def _to_anthropic_content_blocks(
message: AnyMessage, *, model: str | None = None, search_tool: str | None = None
) -> list[dict[str, Any]]:
"""Convert a domain message into Anthropic content blocks."""
blocks: list[dict[str, Any]] = []
match message:
case UserMessage() | AgentMessage():
content = message.content
if isinstance(content, str):
return [{"type": "text", "text": content}] if content else []
for part in content:
match part:
case TextContent(text=text):
blocks.append({"type": "text", "text": text})
case ImageContent(source=source):
blocks.append(_to_anthropic_image_block(source))
return blocks
case AssistantMessage():
if message.provider == "anthropic" and message.model == model and message.replay_blocks:
return [dict(block) for block in message.replay_blocks]
if message.content:
blocks.append({"type": "text", "text": message.content})
for call in message.tool_calls:
blocks.append(
{
"type": "tool_use",
"id": call.id,
"name": call.name,
"input": dict(call.arguments),
}
)
return blocks
case ToolMessage():
result = (
_search_result_blocks(message)
if search_tool is not None and message.name == search_tool and message.success
else message.content
)
return [
{
"type": "tool_result",
"tool_use_id": message.tool_call_id,
"content": result,
"is_error": not message.success,
}
]
case _:
return []
def _to_anthropic_image_block(source: Any) -> dict[str, Any]:
"""Convert a typed image source into an Anthropic-compatible content block."""
match source:
case ImageUrlSource(url=url):
if url.startswith("data:"):
header, separator, data = url.partition(",")
if not separator or not header.endswith(";base64"):
raise ProviderResponseError("Anthropic image data URLs must contain base64 data")
return {"type": "image", "source": {"type": "base64", "media_type": header[5:-7], "data": data}}
return {"type": "image", "source": {"type": "url", "url": url}}
case ImageBytesSource(data=data, media_type=media_type):
encoded = base64.b64encode(data).decode("ascii")
return {
"type": "image",
"source": {"type": "base64", "media_type": media_type, "data": encoded},
}
case _:
raise ProviderResponseError(f"Unsupported image content source: {source!r}")
[文档]
class AnthropicProvider(RetryingProvider):
"""Map Anthropic SDK streams, including signed thinking and tool blocks.
Args:
model: Model identifier accepted by the Anthropic-compatible endpoint.
api_key: API credential supplied by the application.
base_url: API root, defaulting to the official Anthropic endpoint.
transport: Optional HTTPX transport for isolated tests or custom routing.
response: Must remain false; this adapter uses Anthropic Messages API.
retry: Exponential backoff for transient failures before output starts.
timeout: 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 :class:`~zett_agent.model.TimeoutOptions`; a timeout raises
:class:`~zett_agent.providers.base.ProviderTimeoutError` and is not retried.
prompt_cache: 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: 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: 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: URL query parameters sent on every request.
extra_body: 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.
.. seealso::
:class:`~zett_agent.messages.AssistantMessage` documents ``replay_blocks``;
:doc:`/concepts/context` explains replay during tool round trips.
"""
#: See :attr:`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.
native_deferred_tools: bool = True
def __init__(
self,
model: str,
api_key: str,
base_url: str = "https://api.anthropic.com",
transport: httpx.AsyncBaseTransport | None = None,
*,
response: bool = False,
retry: RetryOptions = DEFAULT_RETRY_OPTIONS,
prompt_cache: bool = True,
max_tokens: int = DEFAULT_MAX_TOKENS,
extra_headers: Mapping[str, str] | None = None,
extra_query: Mapping[str, str] | None = None,
extra_body: Mapping[str, Any] | None = None,
timeout: TimeoutOptions = DEFAULT_TIMEOUT_OPTIONS,
) -> None:
import httpx
import truststore
validate_response(response)
if response:
raise ValueError("AnthropicProvider does not support the Responses API")
if not api_key:
raise ValueError("api_key is required")
validate_retry(retry)
validate_timeout(timeout)
if not isinstance(prompt_cache, bool):
raise ValueError("prompt_cache must be a boolean")
if isinstance(max_tokens, bool) or not isinstance(max_tokens, int) or max_tokens < 1:
raise ValueError("max_tokens must be a positive integer")
self.max_tokens = max_tokens
self.extra_headers = validate_extra_headers(extra_headers)
self.extra_query = validate_extra_query(extra_query)
self.extra_body = validate_extra_body(extra_body)
self.retry = retry
self.timeout = timeout
self.response = response
self.prompt_cache = prompt_cache
self.model = model
self._http_client = httpx.AsyncClient(
transport=transport,
verify=truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT),
trust_env=True,
)
from anthropic import AsyncAnthropic
self._client = AsyncAnthropic(
api_key=api_key,
base_url=base_url,
http_client=self._http_client,
max_retries=0,
timeout=httpx_timeout(timeout),
default_headers=self.extra_headers or None,
default_query=self.extra_query or None,
)
[文档]
async def aclose(self) -> None:
"""Close the owned SDK connection pool."""
await self._client.close()
@retry_model_stream
async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
messages: list[dict[str, Any]] = []
system: list[str] = []
search_tool = local_tool_search_name(request.tools)
for message in expand_tool_images(request.messages):
match message:
case SystemMessage():
system.append(message.content)
case UserMessage() | AgentMessage():
content = _to_anthropic_content_blocks(message, model=self.model)
if content:
messages.append({"role": "user", "content": content})
case AssistantMessage():
content = _to_anthropic_content_blocks(message, model=self.model)
if content:
messages.append({"role": "assistant", "content": content})
case ToolMessage():
content = _to_anthropic_content_blocks(message, model=self.model, search_tool=search_tool)
if content:
messages.append({"role": "user", "content": content})
client_tools = _tools_to_anthropic_payload(request.tools, request.server_tools)
body: dict[str, Any] = {
"model": self.model,
"system": "\n\n".join(system),
"messages": messages,
"tools": [
*client_tools,
*_server_tools_to_anthropic_payload(request.server_tools),
],
"max_tokens": self.max_tokens,
"stream": True,
}
if not system:
body.pop("system")
if self.prompt_cache:
_apply_prompt_cache(body, client_tools)
if request.tool_choice:
body["tool_choice"] = {
"type": "tool",
"name": request.tool_choice,
"disable_parallel_tool_use": not request.parallel_tool_call,
}
elif not request.parallel_tool_call and (request.tools or request.server_tools):
body["tool_choice"] = {"type": "auto", "disable_parallel_tool_use": True}
if request.reasoning_effort != ReasoningEffort.OFF and not request.tool_choice:
body["max_tokens"] = _reasoning_effort_to_budget(request.reasoning_effort) + self.max_tokens
body["thinking"] = {
"type": "enabled",
"budget_tokens": _reasoning_effort_to_budget(request.reasoning_effort),
}
else:
body["thinking"] = {"type": "disabled"}
text = ""
reasoning = ""
finish_reason = None
stop_details: Mapping[str, Any] = {}
streams: dict[int, _ToolCallAccumulator] = {}
server_streams: dict[int, _ToolCallAccumulator] = {}
server_initial_inputs: dict[int, dict[str, Any]] = {}
usage = ModelUsage()
replay: dict[int, dict[str, Any]] = {}
from anthropic import AnthropicError
try:
# The request body is assembled as a dict, which the SDK's many typed overloads cannot match
# (they select on literal ``stream=True``), so the call is typed by what it returns here.
create = cast("Callable[..., Awaitable[AsyncStream[Any]]]", self._client.messages.create)
response = await self._first_byte(
create(**body, **({"extra_body": self.extra_body} if self.extra_body else {}))
)
except AnthropicError as error:
if getattr(error, "status_code", None) is None:
# No HTTP response at all: the connection failed or timed out. The SDK's own
# text ("Connection error.") is what says why; a timeout is recognized from the
# chain by the retry layer, which turns it into a typed error.
raise ProviderResponseError(f"Anthropic request failed: {error}") from error
status, failure, description = describe_status_error("anthropic", error)
if status == 401:
raise ProviderAuthError(f"Provider rejected API credential. {description}") from error
raise http_error(status, failure, description) from error
await require_event_stream(response, "anthropic")
events_seen = 0
try:
async for event in guard_idle(response, self.timeout):
events_seen += 1
data = _to_model_data(event)
event_type = data.get("type")
match event_type:
case "message_start":
message_usage = data.get("message", {}).get("usage", {})
usage = ModelUsage(
input_tokens=sum(
int(message_usage.get(key, 0) or 0)
for key in ("input_tokens", "cache_read_input_tokens", "cache_creation_input_tokens")
),
cache_read_tokens=int(message_usage.get("cache_read_input_tokens", 0) or 0),
cache_write_tokens=int(message_usage.get("cache_creation_input_tokens", 0) or 0),
output_tokens=int(message_usage.get("output_tokens", 0) or 0),
)
case "content_block_delta":
index = int(data.get("index", 0))
delta = data.get("delta", {})
match delta.get("type"):
case "text_delta":
chunk = delta.get("text", "")
text += chunk
replay[index]["text"] = replay[index].get("text", "") + chunk
yield ModelEvent.text(chunk)
case "thinking_delta":
chunk = delta.get("thinking", "")
reasoning += chunk
replay[index]["thinking"] = replay[index].get("thinking", "") + chunk
yield ModelEvent.reasoning(chunk)
case "signature_delta":
replay[index]["signature"] = replay[index].get("signature", "") + delta.get(
"signature", ""
)
case "input_json_delta":
partial = delta.get("partial_json", "")
if not partial:
continue
stream = server_streams.get(index)
if stream is not None:
stream.argument_buffer += partial
yield ModelEvent.server_tool_input(
ServerToolInputDelta(stream.call_id, partial),
)
continue
stream = streams.setdefault(index, _ToolCallAccumulator(index=index))
stream.append(ToolCallDelta(index=index, arguments_delta=partial))
yield ModelEvent.tool_call(ToolCallDelta(index=index, arguments_delta=partial))
case _:
pass
case "content_block_start":
index = int(data.get("index", 0))
block = data.get("content_block", {})
replay[index] = dict(block)
match block.get("type"):
case "tool_use":
stream = streams.setdefault(index, _ToolCallAccumulator(index=index))
stream.call_id = block.get("id", stream.call_id)
stream.name = block.get("name", stream.name)
yield ModelEvent.tool_call(
ToolCallDelta(index=index, id_delta=stream.call_id, name_delta=stream.name)
)
case "server_tool_use":
call_id = str(block.get("id", ""))
name = str(block.get("name", ""))
initial_input = block.get("input")
input_data = initial_input if isinstance(initial_input, dict) else {}
stream = _ToolCallAccumulator(index=index, call_id=call_id, name=name)
server_streams[index] = stream
server_initial_inputs[index] = input_data
yield ModelEvent.server_tool_started(ServerToolCall(call_id, name, input_data or None))
case str(block_type) if block_type.endswith("_tool_result"):
call_id = str(block.get("tool_use_id", ""))
name = block_type.removesuffix("_tool_result")
output = block.get("content")
error_code = _anthropic_server_tool_error(output)
result = ServerToolResult(call_id, name, output, error_code)
yield (
ModelEvent.server_tool_failed(result)
if error_code is not None
else ModelEvent.server_tool_completed(result)
)
case _:
pass
case "content_block_stop":
index = int(data.get("index", 0))
# A malformed block is not fatal here: the replay input
# falls back to an empty object, and the completed
# response reports the call as failed to the model.
if index in streams:
replay[index]["input"] = _lenient_arguments(streams[index].argument_buffer, index)
if index in server_streams:
stream = server_streams[index]
replay[index]["input"] = (
_lenient_arguments(stream.argument_buffer, index)
if stream.argument_buffer
else server_initial_inputs[index]
)
case "message_delta":
delta = data.get("delta", {})
finish_reason = delta.get("stop_reason")
if isinstance(details := delta.get("stop_details"), Mapping):
stop_details = details
token_info = data.get("usage", {})
usage = replace(
usage,
output_tokens=int(token_info.get("output_tokens", 0) or 0),
)
case _:
pass
finally:
await response.close()
if not events_seen:
raise ProviderResponseError(_empty_stream_message("anthropic", response))
tool_calls = _parse_tool_calls(streams, finish_reason)
for index, stream in streams.items():
replay[index]["input"] = _lenient_arguments(stream.argument_buffer, index)
stop = _anthropic_stop(finish_reason, stop_details, has_tool_calls=bool(tool_calls))
blocks: tuple[Mapping[str, Any], ...] = tuple(replay[index] for index in sorted(replay))
if stop.reason in (StopReason.REFUSAL, StopReason.INTERRUPTED):
# A refusal can arrive after a complete ``tool_use`` was streamed. The call is
# withheld from the message, and a ``tool_use`` block replayed without a result
# would be rejected, so the block goes as well.
blocks = withhold_replay_blocks(blocks, lambda block: block.get("type") == "tool_use")
yield ModelEvent.completed(
ModelResponse(
message=AssistantMessage(
content=text,
reasoning=reasoning or None,
tool_calls=tool_calls,
provider="anthropic",
model=self.model,
replay_blocks=blocks,
),
finish_reason=finish_reason,
usage=usage,
stop=stop,
)
)
def _anthropic_stop(finish_reason: str | None, details: Mapping[str, Any], *, has_tool_calls: bool) -> StopInfo:
"""Build the stop of a Messages response from ``stop_reason`` and ``stop_details``.
``stop_details`` is null for every reason but ``refusal``. On a refusal its
``category`` and ``explanation`` can both be null, which is a normal value and is not
treated as missing data; the category then falls back to the reason itself.
"""
kept = {
key: details[key]
for key in ("fallback_credit_token", "fallback_has_prefill_claim")
if details.get(key) is not None
}
category, explanation, recommended = (details.get(key) for key in ("category", "explanation", "recommended_model"))
return classify_stop(
finish_reason,
has_tool_calls=has_tool_calls,
category=str(category) if category else None,
message=str(explanation) if explanation else None,
recommended_model=str(recommended) if recommended else None,
detail=kept,
)
def _lenient_arguments(payload: str, index: int) -> dict[str, Any]:
"""Decode a replay block's input, using an empty object when it is unusable."""
try:
return decode_tool_arguments(payload, index=index)
except ToolArgumentsError:
return {}
def _anthropic_server_tool_error(output: Any) -> str | None:
"""Return Anthropic's structured hosted-tool error code, when present."""
if not isinstance(output, dict):
return None
output_type = output.get("type")
if not isinstance(output_type, str) or not output_type.endswith("_error"):
return None
code = output.get("error_code")
return str(code) if code else output_type
