"""
Anthropic (Claude) provider implementation.
This module implements the BaseLLMProvider interface for Anthropic's Claude API.
Key Differences from OpenAI:
- Uses 'messages' API with system prompts separate
- Tool calling format is different
- Has thinking/reasoning tokens
- Different error codes and rate limits
"""
from __future__ import annotations
import json
from typing import Any, Dict, List, Optional
from anthropic import AsyncAnthropic, APIError, RateLimitError as AnthropicRateLimitError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from .base import (
BaseLLMProvider,
LLMResponse,
ToolCall,
ToolCallStatus,
LLMProviderError,
RateLimitError,
AuthenticationError,
)
class AnthropicProvider(BaseLLMProvider):
"""
Anthropic (Claude) implementation of the LLM provider interface.
Supports:
- Claude 3 family (Opus, Sonnet, Haiku)
- Tool use (Anthropic's function calling)
- Extended context windows
- Thinking/reasoning capabilities
"""
def __init__(self, config: Dict[str, Any]):
"""
Initialize Anthropic provider.
Args:
config: Must contain:
- api_key: Anthropic API key
- model: Model name (e.g., 'claude-3-5-sonnet-20241022')
- max_tokens: Max tokens to generate
"""
super().__init__(config)
self.client = AsyncAnthropic(api_key=config["api_key"])
self.model = config["model"]
self.default_max_tokens = config.get("max_tokens", 4096)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type(AnthropicRateLimitError),
)
async def generate(
self,
messages: List[Dict[str, Any]],
tools: Optional[List[Dict[str, Any]]] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
**kwargs
) -> LLMResponse:
"""
Generate a response using Anthropic's Messages API.
Note: Anthropic requires system messages to be passed separately,
so we extract them from the messages list.
"""
try:
# Separate system messages from conversation
system_messages = [m["content"] for m in messages if m["role"] == "system"]
system_prompt = "\n\n".join(system_messages) if system_messages else None
# Filter out system messages from conversation
conversation = [m for m in messages if m["role"] != "system"]
# Prepare API call parameters
params = {
"model": self.model,
"messages": conversation,
"max_tokens": max_tokens if max_tokens is not None else self.default_max_tokens,
}
if system_prompt:
params["system"] = system_prompt
if temperature is not None:
params["temperature"] = temperature
# Add tools if provided
if tools:
params["tools"] = self._convert_mcp_tools_to_anthropic(tools)
# Make API call
response = await self.client.messages.create(**params)
# Extract text content
content_blocks = response.content
text_content = ""
tool_calls = []
for block in content_blocks:
if block.type == "text":
text_content += block.text
elif block.type == "tool_use":
tool_calls.append(ToolCall(
id=block.id,
name=block.name,
arguments=block.input,
status=ToolCallStatus.PENDING,
))
# Extract usage stats
usage = {
"prompt_tokens": response.usage.input_tokens,
"completion_tokens": response.usage.output_tokens,
"total_tokens": response.usage.input_tokens + response.usage.output_tokens,
}
return LLMResponse(
content=text_content.strip(),
tool_calls=tool_calls,
finish_reason=response.stop_reason,
usage=usage,
raw_response=response,
)
except AnthropicRateLimitError as e:
raise RateLimitError(
"Anthropic rate limit exceeded",
provider="anthropic",
original_error=e,
)
except APIError as e:
if "invalid_api_key" in str(e).lower() or "authentication" in str(e).lower():
raise AuthenticationError(
"Invalid Anthropic API key",
provider="anthropic",
original_error=e,
)
raise LLMProviderError(
f"Anthropic API error: {str(e)}",
provider="anthropic",
original_error=e,
)
except Exception as e:
raise LLMProviderError(
f"Unexpected error: {str(e)}",
provider="anthropic",
original_error=e,
)
async def generate_with_tools(
self,
messages: List[Dict[str, Any]],
tools: List[Dict[str, Any]],
max_turns: int = 10,
**kwargs
) -> tuple[str, List[Dict[str, Any]]]:
"""
Generate response with automatic tool calling loop.
Anthropic's tool use works similarly to OpenAI but with different
message formats for tool results.
"""
conversation = list(messages)
for turn in range(max_turns):
response = await self.generate(
messages=conversation,
tools=tools,
**kwargs
)
if not response.has_tool_calls:
return response.content, conversation
# Add assistant's response with tool calls
assistant_message = {
"role": "assistant",
"content": []
}
# Add text content if present
if response.content:
assistant_message["content"].append({
"type": "text",
"text": response.content
})
# Add tool use blocks
for tc in response.tool_calls:
assistant_message["content"].append({
"type": "tool_use",
"id": tc.id,
"name": tc.name,
"input": tc.arguments,
})
conversation.append(assistant_message)
# Return early - orchestrator will execute tools and continue
return "", conversation
from .base import MaxTurnsExceededError
raise MaxTurnsExceededError(
f"Tool calling loop exceeded {max_turns} turns",
provider="anthropic",
)
def format_tool_result(self, tool_call: ToolCall) -> Dict[str, Any]:
"""
Format tool result for Anthropic's expected format.
Anthropic expects tool results in a user message with tool_result blocks.
"""
return {
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": tool_call.id,
"content": json.dumps(tool_call.result) if tool_call.result else "{}",
}
]
}
def _convert_mcp_tools_to_anthropic(self, mcp_tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Convert MCP tool format to Anthropic's tool format.
Anthropic's format is simpler - just name, description, and input_schema.
"""
anthropic_tools = []
for tool in mcp_tools:
anthropic_tools.append({
"name": tool["name"],
"description": tool.get("description", ""),
"input_schema": tool.get("parameters", {}),
})
return anthropic_tools
@property
def provider_name(self) -> str:
return "anthropic"
@property
def model_name(self) -> str:
return self.model
