"""
Anthropic provider implementation for Praval framework.
Provides integration with Anthropic's Claude models through the
Messages API with support for conversation history and system messages.
"""
import os
import re
from typing import Any, Dict, Iterator, List, Optional
import anthropic
from ..core.exceptions import (
HITLConfigurationError,
InterventionRequired,
ProviderError,
)
from ..hitl.runtime import HITLRuntime
from ..model_runtime import (
call_with_retries,
execute_legacy_tool_call,
max_provider_retries,
)
from ..models import (
ContentPart,
ModelEvent,
ModelRequest,
ModelResponse,
ProviderCapabilities,
ToolCall,
ToolResult,
ToolSpec,
Usage,
)
from .errors import map_provider_exception, sdk_max_retries
from .registry import reasoning_parameters
from .usage import anthropic_usage
def _redact_secrets(message: str) -> str:
if not message:
return message
secrets = [
os.getenv("OPENAI_API_KEY"),
os.getenv("ANTHROPIC_API_KEY"),
os.getenv("COHERE_API_KEY"),
]
redacted = message
for secret in secrets:
if secret and secret in redacted:
redacted = redacted.replace(secret, "***")
return redacted
[docs]
class AnthropicProvider:
"""Anthropic provider for LLM interactions."""
provider_name = "anthropic"
capabilities = ProviderCapabilities(
tools=True,
streaming=True,
native_streaming=True,
tool_streaming=True,
structured_outputs=True,
json_schema_mode="json_schema",
multimodal=True,
image_input=True,
reasoning=True,
reasoning_effort=True,
reasoning_budget=True,
)
[docs]
def __init__(self, config):
self.config = config
try:
api_key_env = getattr(config, "api_key_env", None) or "ANTHROPIC_API_KEY"
api_key = os.getenv(api_key_env)
if not api_key:
raise ProviderError(f"{api_key_env} environment variable not set")
client_kwargs: Dict[str, Any] = {"api_key": api_key}
if getattr(config, "base_url", None):
client_kwargs["base_url"] = config.base_url
if getattr(config, "timeout", None):
client_kwargs["timeout"] = config.timeout
client_kwargs["max_retries"] = sdk_max_retries(config)
self.client = anthropic.Anthropic(**client_kwargs)
except Exception as e:
raise ProviderError(
f"Failed to initialize Anthropic client: {self._redact(str(e))}"
) from e
[docs]
def generate(
self,
messages: List[Dict[str, str]],
tools: Optional[List[Dict[str, Any]]] = None,
hitl_context: Optional[Dict[str, Any]] = None,
) -> str:
"""Generate a response using Anthropic's Messages API."""
try:
system_message = self._extract_system_message(messages)
conversation_messages = self._filter_conversation_messages(messages)
call_params = {
"model": self._model_name(),
"messages": conversation_messages,
"temperature": self.config.temperature,
"max_tokens": self._max_output_tokens(),
}
if system_message:
call_params["system"] = system_message
if tools:
formatted_tools = self._format_tools_for_anthropic(tools)
if formatted_tools:
call_params["tools"] = formatted_tools
response = self.client.messages.create(**call_params)
if response.content and len(response.content) > 0:
content_blocks = response.content
tool_uses = []
for block in content_blocks:
block_type = getattr(block, "type", None) or (
block.get("type") if isinstance(block, dict) else None
)
if block_type == "tool_use":
tool_uses.append(block)
elif block_type == "text" and not tool_uses:
return getattr(block, "text", None) or (
block.get("text", "") if isinstance(block, dict) else ""
)
if tool_uses and tools:
return self._handle_tool_calls(
tool_uses,
tools,
conversation_messages,
system_message,
hitl_context=hitl_context,
)
return ""
except (InterventionRequired, HITLConfigurationError):
raise
except Exception as e:
raise self._mapped_error(
e, f"Anthropic API error: {self._redact(str(e))}"
) from e
[docs]
def invoke(
self,
request: ModelRequest,
tools: Optional[List[Dict[str, Any]]] = None,
) -> ModelResponse:
"""Invoke Anthropic through the provider-neutral adapter surface."""
return self._invoke_messages(request, tools=tools)
def _invoke_messages(
self,
request: ModelRequest,
tools: Optional[List[Dict[str, Any]]] = None,
) -> ModelResponse:
call_params = self._messages_params(request, tools=tools)
response = self.client.messages.create(**call_params)
return self._messages_model_response(response, call_params)
def _messages_model_response(
self,
response: Any,
call_params: Dict[str, Any],
) -> ModelResponse:
content_blocks = getattr(response, "content", None)
if not isinstance(content_blocks, (list, tuple)):
content_blocks = []
serialized_content = self._serialize_content_blocks(list(content_blocks))
tool_uses = [
block for block in serialized_content if block.get("type") == "tool_use"
]
tool_calls = [
ToolCall(
id=str(block.get("id") or ""),
name=str(block.get("name") or ""),
arguments=(
block.get("input") if isinstance(block.get("input"), dict) else {}
),
raw=block,
)
for block in tool_uses
]
finish_reason = self._event_value(response, "stop_reason", None)
metadata: Dict[str, Any] = {
"anthropic_assistant_content": serialized_content,
}
if tool_uses:
# Keep every earlier turn so later rounds resend the whole run.
# Assistant blocks, including signed thinking blocks, stay as
# received because Anthropic requires them unchanged.
metadata["anthropic_messages"] = [
*call_params["messages"],
{"role": "assistant", "content": serialized_content},
]
return ModelResponse(
content=self._extract_text(response),
provider=self.provider_name,
model=call_params["model"],
tool_calls=tool_calls,
raw=response,
usage=self._extract_usage(response),
finish_reason=finish_reason if isinstance(finish_reason, str) else None,
metadata=metadata,
)
[docs]
def stream(
self,
request: ModelRequest,
tools: Optional[List[Dict[str, Any]]] = None,
) -> Iterator[ModelEvent]:
"""Stream Anthropic messages as provider-neutral events."""
call_params = self._messages_params(request, tools=tools)
try:
stream_factory = getattr(self.client.messages, "stream", None)
if callable(stream_factory):
yield from self._stream_messages_context(stream_factory, call_params)
return
call_params["stream"] = True
yield from self._stream_messages_create(call_params)
except Exception as e:
message = self._redact(str(e))
yield ModelEvent(type="error", metadata={"message": message})
raise self._mapped_error(e, f"Anthropic streaming error: {message}") from e
def _redact(self, message: str) -> str:
"""Redact known provider keys and the key from ``config.api_key_env``."""
redacted = _redact_secrets(message)
api_key_env = getattr(self.config, "api_key_env", None)
secret = os.getenv(api_key_env) if api_key_env else None
if secret and redacted:
redacted = redacted.replace(secret, "***")
return redacted
[docs]
def map_provider_error(self, exc: BaseException) -> ProviderError:
"""Map an Anthropic SDK exception to a typed, redacted ``ProviderError``."""
return self._mapped_error(exc, None)
def _mapped_error(
self, exc: BaseException, message: Optional[str]
) -> ProviderError:
return map_provider_exception(
exc,
provider=self.provider_name,
model=self._model_name(),
message=message,
redact=self._redact,
)
[docs]
def close(self) -> None:
"""Close the underlying SDK client when supported."""
close = getattr(self.client, "close", None)
if callable(close):
close()
def _model_name(self) -> str:
return str(getattr(self.config, "model", None) or "claude-sonnet-5")
def _max_output_tokens(self) -> int:
return int(
getattr(self.config, "max_output_tokens", None)
or getattr(self.config, "max_tokens", 1000)
)
def _messages_params(
self,
request: ModelRequest,
*,
tools: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
messages = [
message.model_dump(exclude_none=True) for message in request.messages
]
system_message = self._extract_system_message(messages)
conversation_messages = self._filter_conversation_messages(messages)
call_params: Dict[str, Any] = {
"model": request.model or self._model_name(),
"messages": conversation_messages,
"temperature": request.temperature,
"max_tokens": request.max_output_tokens or self._max_output_tokens(),
}
if system_message:
call_params["system"] = system_message
if tools:
formatted_tools = self._format_tools_for_anthropic(tools)
if formatted_tools:
call_params["tools"] = formatted_tools
elif request.tools:
call_params["tools"] = self._format_tool_specs_for_anthropic(request.tools)
experimental_tools = self._experimental_tools(request)
if experimental_tools:
call_params.setdefault("tools", [])
call_params["tools"].extend(experimental_tools)
thinking = self._anthropic_thinking(request)
if thinking:
call_params["thinking"] = thinking
output_config = self._anthropic_output_config(request)
if output_config:
call_params["output_config"] = output_config
if request.timeout is not None:
call_params["timeout"] = request.timeout
reserved = {
"api",
"capabilities",
"endpoint",
"allow_experimental_tools",
"experimental_tools",
"max_retries",
"discover_model",
"prompt_caching",
}
for key, value in request.provider_options.items():
if key not in reserved:
call_params.setdefault(key, value)
model_name = str(call_params["model"])
if call_params.get("temperature") is None:
call_params.pop("temperature", None)
modern_sampling = any(
model_name == family
or re.fullmatch(re.escape(family) + r"-\d{4}-?\d{2}-?\d{2}", model_name)
for family in (
"claude-sonnet-5-5",
"claude-haiku-5-5",
"claude-sonnet-5",
"claude-opus-4-8",
"claude-opus-4-7",
)
)
if modern_sampling or (
request.reasoning is not None and request.reasoning.level is not None
):
# Thinking requires provider-default sampling. New adaptive-only
# families require it even when thinking is explicitly disabled.
for key in ("temperature", "top_p", "top_k"):
call_params.pop(key, None)
caching = request.provider_options.get("prompt_caching", False)
if caching is not False:
if caching is True:
control = {"type": "ephemeral"}
elif (
isinstance(caching, dict)
and caching.get("type") == "ephemeral"
and set(caching) <= {"type", "ttl"}
and caching.get("ttl", "5m") in {"5m", "1h"}
):
control = dict(caching)
else:
raise ProviderError(
"prompt_caching must be True, False or an ephemeral "
"cache_control with ttl 5m/1h"
)
# extra_body works with SDK versions predating automatic caching.
body = dict(call_params.get("extra_body") or {})
body.setdefault("cache_control", control)
call_params["extra_body"] = body
return call_params
def _experimental_tools(self, request: ModelRequest) -> List[Dict[str, Any]]:
value = request.provider_options.get("experimental_tools")
if value is None:
return []
if request.provider_options.get("allow_experimental_tools") is not True:
raise ProviderError(
"experimental_tools requires allow_experimental_tools=True"
)
if not isinstance(value, list) or not all(
isinstance(tool, dict) for tool in value
):
raise ProviderError("experimental_tools must be a list of tool mappings")
return [dict(tool) for tool in value]
def _serialize_content_blocks(
self, content_blocks: List[Any]
) -> List[Dict[str, Any]]:
serialized: List[Dict[str, Any]] = []
for block in content_blocks:
if isinstance(block, dict):
serialized.append(dict(block))
continue
if hasattr(block, "model_dump"):
dumped = block.model_dump(exclude_none=True)
if isinstance(dumped, dict):
serialized.append(dict(dumped))
continue
block_type = getattr(block, "type", None)
if block_type == "tool_use":
serialized.append(
{
"type": "tool_use",
"id": getattr(block, "id", None),
"name": getattr(block, "name", None),
"input": getattr(block, "input", None) or {},
}
)
elif block_type == "text":
serialized.append({"type": "text", "text": getattr(block, "text", "")})
elif block_type == "thinking":
serialized.append(
{
"type": "thinking",
"thinking": getattr(block, "thinking", ""),
"signature": getattr(block, "signature", ""),
}
)
elif block_type == "redacted_thinking":
serialized.append(
{"type": "redacted_thinking", "data": getattr(block, "data", "")}
)
return serialized
def _anthropic_tool_result(self, result: ToolResult) -> Dict[str, Any]:
block: Dict[str, Any] = {
"type": "tool_result",
"tool_use_id": result.tool_call_id,
"content": result.content,
}
if result.is_error:
block["is_error"] = True
return block
def _stream_messages_context(
self,
stream_factory: Any,
call_params: Dict[str, Any],
) -> Iterator[ModelEvent]:
content_parts: List[str] = []
final_message: Any = None
with stream_factory(**call_params) as stream:
for text in getattr(stream, "text_stream", []):
if text:
content_parts.append(str(text))
yield ModelEvent(type="delta", delta=str(text))
get_final = getattr(stream, "get_final_message", None)
if callable(get_final):
final_message = get_final()
usage = (
self._extract_usage(final_message) if final_message is not None else None
)
response = ModelResponse(
content=self._extract_text(final_message) or "".join(content_parts),
provider=self.provider_name,
model=call_params["model"],
raw=final_message,
usage=usage,
)
if usage:
yield ModelEvent(type="usage", usage=usage)
yield ModelEvent(type="final", response=response, usage=usage)
def _stream_messages_create(
self,
call_params: Dict[str, Any],
) -> Iterator[ModelEvent]:
content_parts: List[str] = []
usage: Optional[Usage] = None
stream = self.client.messages.create(**call_params)
for event in stream:
event_type = str(self._event_value(event, "type", ""))
if event_type in {"content_block_delta", "text_delta"}:
delta = self._event_value(event, "delta", event)
text = self._event_value(delta, "text", "")
if text:
content_parts.append(str(text))
yield ModelEvent(type="delta", delta=str(text))
elif event_type == "message_delta":
usage = self._extract_usage(event) or usage
elif event_type == "error":
yield ModelEvent(
type="error",
metadata={"message": self._redact(str(event))},
)
response = ModelResponse(
content="".join(content_parts),
provider=self.provider_name,
model=call_params["model"],
usage=usage,
)
if usage:
yield ModelEvent(type="usage", usage=usage)
yield ModelEvent(type="final", response=response, usage=usage)
def _extract_text(self, response: Any) -> str:
content = getattr(response, "content", None)
if content is None and isinstance(response, dict):
content = response.get("content")
text_parts: List[str] = []
for block in content or []:
block_type = getattr(block, "type", None) or (
block.get("type") if isinstance(block, dict) else None
)
if block_type == "text":
value = getattr(block, "text", None) or (
block.get("text", "") if isinstance(block, dict) else ""
)
text_parts.append(str(value))
return "".join(text_parts)
def _anthropic_thinking(self, request: ModelRequest) -> Dict[str, Any]:
if request.reasoning is None:
return {}
reasoning = request.reasoning
native = reasoning_parameters(request)
thinking: Dict[str, Any] = dict(native.get("thinking") or {})
if reasoning.budget_tokens is None and not reasoning.mode and not thinking:
return {}
if reasoning.mode:
thinking["type"] = reasoning.mode
elif reasoning.budget_tokens is not None:
thinking["type"] = "enabled"
if reasoning.budget_tokens is not None:
thinking["budget_tokens"] = reasoning.budget_tokens
if reasoning.display:
thinking["display"] = reasoning.display
return thinking
def _anthropic_output_config(self, request: ModelRequest) -> Dict[str, Any]:
output_config: Dict[str, Any] = {}
if request.response_schema is not None:
output_config["format"] = {
"type": "json_schema",
"schema": request.response_schema.json_schema or {},
}
output_config.update(reasoning_parameters(request).get("output_config", {}))
if request.reasoning is not None and request.reasoning.effort:
output_config["effort"] = request.reasoning.effort
return output_config
def _extract_usage(self, response: Any) -> Optional[Usage]:
return anthropic_usage(response)
def _extract_system_message(self, messages: List[Dict[str, Any]]) -> Optional[str]:
parts = [
self._content_to_text(message.get("content", ""))
for message in messages
if message.get("role") == "system"
]
return "\n\n".join(parts) if parts else None
def _filter_conversation_messages(
self, messages: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
conversation_messages = []
for message in messages:
role = message.get("role", "")
if role in ["user", "assistant"]:
conversation_messages.append(
{
"role": role,
"content": self._format_anthropic_content(
message.get("content", "")
),
}
)
return conversation_messages
def _format_anthropic_content(self, content: Any) -> Any:
if not isinstance(content, list):
return content
blocks: List[Dict[str, Any]] = []
for item in content:
part = item if isinstance(item, ContentPart) else ContentPart(**item)
if part.type == "text":
blocks.append({"type": "text", "text": part.text or ""})
elif part.type in {"image_url", "image"}:
blocks.append(
{
"type": "image",
"source": {"type": "url", "url": part.url or ""},
}
)
elif part.type == "image_base64":
blocks.append(
{
"type": "image",
"source": {
"type": "base64",
"media_type": part.mime_type or "image/png",
"data": part.data or "",
},
}
)
else:
raise ProviderError(
"Anthropic provider cannot serialize content part type: "
f"{part.type}"
)
return blocks
def _content_to_text(self, content: Any) -> str:
if isinstance(content, str):
return content
if isinstance(content, list):
text_parts = []
for item in content:
part = item if isinstance(item, ContentPart) else ContentPart(**item)
if part.type == "text":
text_parts.append(part.text or "")
return "".join(text_parts)
return str(content)
def _event_value(self, value: Any, key: str, default: Any = None) -> Any:
if isinstance(value, dict):
return value.get(key, default)
return getattr(value, key, default)
def _format_tools_for_anthropic(
self, tools: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
formatted_tools = []
for tool in tools or []:
if "function" not in tool or "description" not in tool:
continue
tool_name = tool["function"].__name__
tool_def = {
"name": tool_name,
"description": tool["description"],
"input_schema": {
"type": "object",
"properties": {},
"required": [],
},
}
parameters = tool.get("parameters") or {}
if parameters.get("type") == "object":
tool_def["input_schema"] = parameters
formatted_tools.append(tool_def)
continue
for param_name, param_info in parameters.items():
param_type = param_info.get("type", "str")
json_type = self._python_type_to_json_schema(param_type)
tool_def["input_schema"]["properties"][param_name] = {"type": json_type}
if param_info.get("required", False):
tool_def["input_schema"]["required"].append(param_name)
formatted_tools.append(tool_def)
return formatted_tools
def _format_tool_specs_for_anthropic(
self, tool_specs: List[ToolSpec]
) -> List[Dict[str, Any]]:
return [
{
"name": spec.name,
"description": spec.description,
"input_schema": spec.parameters,
}
for spec in tool_specs
]
def _python_type_to_json_schema(self, python_type: str) -> str:
type_mapping = {
"str": "string",
"int": "integer",
"float": "number",
"bool": "boolean",
"list": "array",
"dict": "object",
"List": "array",
"Dict": "object",
}
return type_mapping.get(python_type, "string")
def _build_runtime(
self, hitl_context: Optional[Dict[str, Any]]
) -> Optional[HITLRuntime]:
if not hitl_context:
return None
run_id = hitl_context.get("run_id")
agent_name = hitl_context.get("agent_name")
provider_name = hitl_context.get("provider_name")
if not run_id or not agent_name or not provider_name:
return None
return HITLRuntime(
run_id=run_id,
agent_name=agent_name,
provider_name=provider_name,
hitl_enabled=bool(hitl_context.get("enabled", False)),
db_path=hitl_context.get("db_path"),
trace_id=hitl_context.get("trace_id"),
)
def _serialize_tool_uses(self, tool_uses: List[Any]) -> List[Dict[str, Any]]:
serialized: List[Dict[str, Any]] = []
for tool_use in tool_uses:
if isinstance(tool_use, dict):
tool_type = tool_use.get("type")
tool_id = tool_use.get("id")
tool_name = tool_use.get("name")
tool_input = tool_use.get("input") or {}
else:
tool_type = getattr(tool_use, "type", None)
tool_id = getattr(tool_use, "id", None)
tool_name = getattr(tool_use, "name", None)
tool_input = getattr(tool_use, "input", None) or {}
if tool_type == "tool_use":
serialized.append(
{
"type": "tool_use",
"id": tool_id,
"name": tool_name,
"input": tool_input,
}
)
return serialized
def _execute_tool_uses(
self,
*,
serialized_tool_uses: List[Dict[str, Any]],
available_tools: List[Dict[str, Any]],
messages: List[Dict[str, str]],
system_message: Optional[str],
hitl_context: Optional[Dict[str, Any]],
start_index: int = 0,
existing_tool_results: Optional[List[Dict[str, Any]]] = None,
resume_intervention: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, Any]]:
tool_results = list(existing_tool_results or [])
for idx in range(start_index, len(serialized_tool_uses)):
tool_use = serialized_tool_uses[idx]
tool_name = tool_use.get("name")
tool_id = tool_use.get("id")
tool_input = tool_use.get("input") or {}
continuation_state = None
if not (idx == start_index and resume_intervention is not None):
continuation_state = {
"schema": "anthropic_tool_v1",
"messages": messages,
"system_message": system_message,
"tool_uses": serialized_tool_uses,
"current_index": idx,
"tool_results": list(tool_results),
}
result_content = execute_legacy_tool_call(
hitl_context=hitl_context,
tool_call_id=str(tool_id),
function_name=str(tool_name),
raw_args=tool_input,
available_tools=available_tools or [],
continuation_state=continuation_state,
resume_intervention=(
resume_intervention
if idx == start_index and resume_intervention is not None
else None
),
)
tool_results.append(
{
"type": "tool_result",
"tool_use_id": tool_id,
"content": result_content,
}
)
return tool_results
def _follow_up_response(
self,
*,
messages: List[Dict[str, str]],
serialized_tool_uses: List[Dict[str, Any]],
tool_results: List[Dict[str, Any]],
system_message: Optional[str],
) -> str:
followup_messages = list(messages)
followup_messages.append({"role": "assistant", "content": serialized_tool_uses})
followup_messages.append({"role": "user", "content": tool_results})
call_params = {
"model": self._model_name(),
"messages": followup_messages,
"temperature": self.config.temperature,
"max_tokens": self._max_output_tokens(),
}
if system_message:
call_params["system"] = system_message
try:
# The tools have already run: retry only this request.
response = call_with_retries(
"follow_up",
lambda: self.client.messages.create(**call_params),
retries=max_provider_retries(self.config),
map_error=self.map_provider_error,
)
if response.content and len(response.content) > 0:
for block in response.content:
block_type = getattr(block, "type", None) or (
block.get("type") if isinstance(block, dict) else None
)
if block_type == "text":
return getattr(block, "text", None) or (
block.get("text", "") if isinstance(block, dict) else ""
)
return "No response generated after tool execution"
except Exception:
return "\n".join([str(r["content"]) for r in tool_results])
def _handle_tool_calls(
self,
tool_uses: List[Any],
available_tools: List[Dict[str, Any]],
messages: List[Dict[str, str]],
system_message: Optional[str],
hitl_context: Optional[Dict[str, Any]] = None,
) -> str:
serialized_tool_uses = self._serialize_tool_uses(tool_uses)
tool_results = self._execute_tool_uses(
serialized_tool_uses=serialized_tool_uses,
available_tools=available_tools or [],
messages=messages,
system_message=system_message,
hitl_context=hitl_context,
)
return self._follow_up_response(
messages=messages,
serialized_tool_uses=serialized_tool_uses,
tool_results=tool_results,
system_message=system_message,
)