ZerdaZerda

AnthropicInterface

On this page

llm.AnthropicInterface — sync & async Claude (Messages API) interface with native token streaming.

Constructor

Python · 5 lines
AnthropicInterface(
    model_name: str = "claude-sonnet-4-20250514",
    api_key: Optional[str] = None,
    **kwargs,
)

Creates both anthropic.Anthropic (sync) and anthropic.AsyncAnthropic (async) clients eagerly. Raises ImportError("install anthropic: pip install anthropic") if the anthropic package is missing.

ParameterTypeDefaultNotes
model_namestr"claude-sonnet-4-20250514"Any Claude model id accepted by the Messages API.
api_keyOptional[str]NonePassed to both clients; if None, the anthropic SDK falls back to its own environment variable (ANTHROPIC_API_KEY).
**kwargs——Stored via the base class; not otherwise consumed here.

generate(messages=None, prompt=None, max_tokens=2048, temperature=0.3, tools=None, tool_choice=None, **kwargs) -> Union[str, dict]

Python · 9 lines
response = self._sync_client.messages.create(
    model=self.model_name, max_tokens=max_tokens, temperature=temperature,
    messages=msgs, **kwargs,
)
calls = parse_anthropic_tool_calls(response.content)
if calls:
    text = "".join(b.text for b in response.content if getattr(b, "type", None) == "text")
    return {"content": text or None, "tool_calls": tool_calls_to_dicts(calls), "stop_reason": response.stop_reason}
return response.content[0].text

Parameters — same prompt/messages resolution pattern as OpenAIInterface: messages wins if both provided; _build_messages raises a ValueError if neither is given, but (see below) that's caught before it reaches you.

NameTypeNotes
messagesOptional[list[dict]]Standard Anthropic-style message list.
promptOptional[str]Wrapped into [{"role": "user", "content": prompt}].
max_tokensintDefault 2048. Anthropic's Messages API requires max_tokens, unlike some other providers — this default ensures the call never omits it.
temperaturefloatDefault 0.3.
toolsOptional[list[dict]]Tool definitions in the unified schema; converted via llm.tools.to_anthropic_tools into messages.create's native tools=[{"name", "description", "input_schema"}] shape.
tool_choiceOptional[Union[str, dict]]Forwarded as-is to messages.create(tool_choice=...) when given — accepts Anthropic's own values, e.g. {"type": "auto"}, {"type": "any"}, {"type": "tool", "name": "..."}.
**kwargs—Forwarded into messages.create(...) — e.g. system, top_p, stop_sequences.

Returns: If the model doesn't call a tool, response.content[0].text — note this only reads the first content block, so if a response happens to contain multiple non-tool-use content blocks, only the first block's .text is returned, which will raise an AttributeError if that first block isn't a text block. If the model calls one or more tools, parse_anthropic_tool_calls scans all of response.content for blocks with type == "tool_use" (each .input is already a dict, no JSON parsing needed) and this method returns {"content": ..., "tool_calls": [...], "stop_reason": response.stop_reason} instead — content here is the concatenation of any text-type blocks found alongside the tool-use block(s), or None if there were none. On any exception, returns {"error": str(e)}. See Tool Calling for the full contract.

Raises: nothing directly — same as OpenAI's interface, _build_messages's ValueError("Either 'prompt' or 'messages' must be provided.") is caught by this method's own try/except (not propagated uncaught) and converted into {"error": "..."}. All other exceptions are caught and logged the same way.

generate_async(...)

Same contract as generate, calling await self._async_client.messages.create(...). Same tools/tool_choice handling.

astream(messages=None, prompt=None, max_tokens=2048, temperature=0.3, **kwargs) -> AsyncIterator[str]

Uses Anthropic's dedicated streaming context manager:

Python · 6 lines
async with self._async_client.messages.stream(
    model=self.model_name, max_tokens=max_tokens, temperature=temperature,
    messages=msgs, **kwargs,
) as stream:
    async for text in stream.text_stream:
        yield text

stream.text_stream yields only text deltas (not tool-use or other event types), so this is genuine incremental streaming. On error, yields f"Error: {e}" as stream content rather than raising. astream does not accept tools/tool_choice and never reconstructs tool-use deltas — since stream.text_stream filters to text only, a tool call made mid-stream simply produces no text output at all rather than an error. Use generate/generate_async for tool calling; see Tool Calling § Streaming limitation.

Example

Python · 9 lines
import asyncio
from zerda.llm import AnthropicInterface

async def main():
    llm = AnthropicInterface(api_key="sk-ant-...")
    async for chunk in llm.astream(prompt="اكتب قصة قصيرة عن صحراء."):
        print(chunk, end="", flush=True)

asyncio.run(main())

cleanup() / acleanup()

Same pattern as OpenAIInterface — closes the sync/async client respectively, logging (not raising) on failure.

  • BaseLLMInterface
  • OpenAIInterface — near-identical structure, different SDK.
  • AWSInterface — can also serve Claude models, via Bedrock's converse API instead of Anthropic's own SDK; its tool-calling shape (toolConfig/toolUse) is Bedrock-native rather than Anthropic-native, even for the same underlying Claude model.
  • Tool Calling — the unified tools/tool_choice contract and per-provider conversion table.

Related pages