windlass.providers.llm.anthropic

anthropic

Anthropic Claude adapter.

Claude's Messages API differs from OpenAI's in three ways that matter, and this adapter absorbs all three so callers never have to think about them:

• the system prompt is a top-level parameter, not a message;
• content is a list of typed blocks (text, tool_use, tool_result) rather than a string;
• consecutive same-role messages must be merged.

Install with::

pip install "windlass[anthropic]"

Example

from windlass import Windlass # doctest: +SKIP Windlass.llm("anthropic").complete("Say hi").content # doctest: +SKIP 'Hi!'

AnthropicLLM

AnthropicLLM(
    model: str = "",
    *,
    api_key: str | None = None,
    base_url: str | None = None,
    max_retries: int = 0,
    thinking_budget: int | None = None,
    **config: Any
)

Bases: LLM

Chat completions via the official anthropic SDK.

Parameters:

Name	Type	Description	Default
model	str	Model id, e.g. claude-sonnet-4-5.	''
api_key	str | None	Credential. Falls back to ANTHROPIC_API_KEY.	None
base_url	str | None	Endpoint override.	None
max_retries	int	SDK-level retries; 0 because Windlass retries itself.	0
thinking_budget	int | None	Enables extended thinking with this token budget. None leaves it off.	None
**config	Any	Forwarded to :class:~windlass.interfaces.llm.LLM.	{}

Raises:

Type	Description
MissingDependencyError	When anthropic is not installed.
AuthenticationError	When no API key can be found.

Source code in src\windlass\providers\llm\anthropic.py

def __init__(
    self,
    model: str = "",
    *,
    api_key: str | None = None,
    base_url: str | None = None,
    max_retries: int = 0,
    thinking_budget: int | None = None,
    **config: Any,
) -> None:
    super().__init__(model=model, **config)
    self._sdk = require("anthropic", extra="anthropic", feature="The Anthropic provider")
    key = api_key or settings().require_secret(
        "anthropic_api_key", provider="anthropic", env_var="ANTHROPIC_API_KEY"
    )
    self.thinking_budget = thinking_budget
    self._client = self._sdk.AsyncAnthropic(
        api_key=key,
        base_url=base_url,
        timeout=self.timeout,
        max_retries=max_retries,
    )

default_model classmethod

default_model() -> str

Return "claude-sonnet-4-5".

Source code in src\windlass\providers\llm\anthropic.py

@classmethod
def default_model(cls) -> str:
    """Return ``"claude-sonnet-4-5"``."""
    return "claude-sonnet-4-5"

native

native() -> Any

Return the underlying anthropic.AsyncAnthropic client.

Source code in src\windlass\providers\llm\anthropic.py

def native(self) -> Any:
    """Return the underlying ``anthropic.AsyncAnthropic`` client."""
    return self._client

agenerate async

agenerate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> Completion

Request one message completion.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation. A leading system message is lifted into the top-level system parameter automatically.	required
tools	list[dict[str, Any]] | None	Anthropic-format tool definitions (input_schema).	None
**kwargs	Any	Request overrides.	{}

Returns:

Type	Description
Completion	The completion, with the SDK response on raw.

Raises:

Type	Description
AuthenticationError	Invalid credentials.
RateLimitError	Quota or rate limit hit.
ProviderError	Any other API failure.

Source code in src\windlass\providers\llm\anthropic.py

async def agenerate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> Completion:
    """Request one message completion.

    Args:
        messages: The conversation. A leading system message is lifted into
            the top-level ``system`` parameter automatically.
        tools: Anthropic-format tool definitions (``input_schema``).
        **kwargs: Request overrides.

    Returns:
        The completion, with the SDK response on ``raw``.

    Raises:
        AuthenticationError: Invalid credentials.
        RateLimitError: Quota or rate limit hit.
        ProviderError: Any other API failure.
    """
    payload = self._payload(messages, tools, kwargs)
    try:
        response = await self._client.messages.create(**payload)
    except Exception as exc:
        raise self._translate(exc) from exc

    text_parts: list[str] = []
    calls: list[ToolCall] = []
    for block in response.content:
        if block.type == "text":
            text_parts.append(block.text)
        elif block.type == "tool_use":
            calls.append(
                ToolCall(
                    id=block.id,
                    name=block.name,
                    arguments=dict(block.input or {}),
                )
            )

    return Completion(
        content="".join(text_parts),
        tool_calls=calls,
        finish_reason=self._finish_reason(response.stop_reason),
        model=response.model or self.model,
        usage=Usage(
            prompt_tokens=response.usage.input_tokens or 0,
            completion_tokens=response.usage.output_tokens or 0,
            cached_tokens=getattr(response.usage, "cache_read_input_tokens", 0) or 0,
        ),
        raw=response,
    )

astream_generate async

astream_generate(
    messages: list[Message], *, tools: list[dict[str, Any]] | None = None, **kwargs: Any
) -> AsyncIterator[StreamEvent]

Stream a message completion.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation.	required
tools	list[dict[str, Any]] | None	Anthropic-format tool definitions.	None
**kwargs	Any	Request overrides.	{}

Yields:

Type	Description
AsyncIterator[StreamEvent]	Text deltas, then completed tool calls, then done.

Source code in src\windlass\providers\llm\anthropic.py

async def astream_generate(
    self,
    messages: list[Message],
    *,
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]:
    """Stream a message completion.

    Args:
        messages: The conversation.
        tools: Anthropic-format tool definitions.
        **kwargs: Request overrides.

    Yields:
        Text deltas, then completed tool calls, then ``done``.
    """
    import json as _json

    payload = self._payload(messages, tools, kwargs)
    partials: dict[int, dict[str, Any]] = {}
    usage = Usage()
    stop_reason: str | None = None

    try:
        async with self._client.messages.stream(**payload) as stream:
            async for event in stream:
                kind = getattr(event, "type", "")
                if kind == "content_block_start":
                    block = event.content_block
                    if block.type == "tool_use":
                        partials[event.index] = {
                            "id": block.id,
                            "name": block.name,
                            "arguments": "",
                        }
                elif kind == "content_block_delta":
                    delta = event.delta
                    if delta.type == "text_delta":
                        yield StreamEvent(type="text", delta=delta.text, raw=event)
                    elif delta.type == "input_json_delta" and event.index in partials:
                        partials[event.index]["arguments"] += delta.partial_json
                elif kind == "message_delta":
                    stop_reason = getattr(event.delta, "stop_reason", None) or stop_reason
                    if getattr(event, "usage", None):
                        usage = Usage(
                            prompt_tokens=usage.prompt_tokens,
                            completion_tokens=event.usage.output_tokens or 0,
                        )
                elif kind == "message_start":
                    usage = Usage(
                        prompt_tokens=event.message.usage.input_tokens or 0,
                        completion_tokens=0,
                    )
    except Exception as exc:
        raise self._translate(exc) from exc

    for slot in partials.values():
        try:
            arguments = _json.loads(slot["arguments"]) if slot["arguments"] else {}
        except ValueError:
            arguments = {}
        yield StreamEvent(
            type="tool_call",
            tool_call=ToolCall(
                id=slot["id"],
                name=slot["name"],
                arguments=arguments,
                raw_arguments=slot["arguments"],
            ),
        )
    yield StreamEvent(type="done", finish_reason=self._finish_reason(stop_reason), usage=usage)

aclose async

aclose() -> None

Close the SDK's HTTP connection pool.

Source code in src\windlass\providers\llm\anthropic.py

async def aclose(self) -> None:
    """Close the SDK's HTTP connection pool."""
    close = getattr(self._client, "close", None)
    if close is not None:
        await close()

to_anthropic_messages

to_anthropic_messages(messages: list[Message]) -> tuple[str, list[dict[str, Any]]]

Translate Windlass messages into Anthropic's format.

Performs the three transformations Claude requires: system extraction, typed content blocks, and merging of consecutive same-role turns (the API rejects two user messages in a row, which happens naturally when several tools return at once).

Parameters:

Name	Type	Description	Default
messages	list[Message]	The conversation.	required

Returns:

Type	Description
tuple[str, list[dict[str, Any]]]	A (system_prompt, messages) pair.

Example

from windlass.core.types import Message system, msgs = to_anthropic_messages( ... [Message.system("be nice"), Message.user("hi")] ... ) system, msgs[0]["role"] ('be nice', 'user')

Source code in src\windlass\providers\llm\anthropic.py

def to_anthropic_messages(messages: list[Message]) -> tuple[str, list[dict[str, Any]]]:
    """Translate Windlass messages into Anthropic's format.

    Performs the three transformations Claude requires: system extraction,
    typed content blocks, and merging of consecutive same-role turns (the API
    rejects two ``user`` messages in a row, which happens naturally when several
    tools return at once).

    Args:
        messages: The conversation.

    Returns:
        A ``(system_prompt, messages)`` pair.

    Example:
        >>> from windlass.core.types import Message
        >>> system, msgs = to_anthropic_messages(
        ...     [Message.system("be nice"), Message.user("hi")]
        ... )
        >>> system, msgs[0]["role"]
        ('be nice', 'user')
    """
    system_parts: list[str] = []
    converted: list[dict[str, Any]] = []

    for message in messages:
        role = message.role.value
        if role == "system":
            system_parts.append(message.content)
            continue

        if role == "tool":
            block: dict[str, Any] = {
                "type": "tool_result",
                "tool_use_id": message.tool_call_id or "",
                "content": message.content,
            }
            if message.metadata.get("is_error"):
                block["is_error"] = True
            _append(converted, "user", block)
            continue

        blocks: list[dict[str, Any]] = []
        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": call.arguments,
                }
            )
        if not blocks:
            continue
        for block in blocks:
            _append(converted, role, block)

    return "\n\n".join(p for p in system_parts if p), converted