butler/transport/anthropic_transport.py
Rama main · réplica de lectura del repositorio original.
"""Anthropic Messages transport — for Anthropic and MiniMax APIs.
Handles the Anthropic message format where system prompt is a
top-level parameter and tool results use content blocks.
"""
from __future__ import annotations
import json
import logging
from typing import Any, Dict, List, Optional
from butler.transport.base import ProviderTransport
from butler.transport.types import NormalizedResponse, ToolCall, Usage
logger = logging.getLogger(__name__)
_STOP_REASON_MAP = {
"end_turn": "stop",
"tool_use": "tool_calls",
"max_tokens": "length",
"model_context_window_exceeded": "length",
"stop_sequence": "stop",
"refusal": "content_filter",
}
def _convert_messages_to_anthropic(
messages: List[Dict[str, Any]],
) -> tuple[str, List[Dict[str, Any]]]:
"""Convert OpenAI-format messages to Anthropic format.
Returns (system_prompt, anthropic_messages).
"""
system_parts: list[str] = []
anthropic_msgs: list[dict[str, Any]] = []
for msg in messages:
role = msg.get("role", "")
content = msg.get("content", "")
if role == "system":
if content:
system_parts.append(content)
continue
if role == "assistant":
blocks: list[Any] = []
if content:
blocks.append({"type": "text", "text": content})
tool_calls = msg.get("tool_calls")
if tool_calls:
for tc in tool_calls:
fn = tc.get("function", tc)
args_str = fn.get("arguments", "{}")
try:
args = json.loads(args_str) if isinstance(args_str, str) else args_str
except json.JSONDecodeError:
args = {"raw": args_str}
blocks.append({
"type": "tool_use",
"id": tc.get("id", ""),
"name": fn.get("name", ""),
"input": args,
})
if blocks:
anthropic_msgs.append({"role": "assistant", "content": blocks})
continue
if role == "tool":
tool_call_id = msg.get("tool_call_id", "")
anthropic_msgs.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tool_call_id,
"content": str(content) if content else "",
}],
})
continue
if isinstance(content, list):
anthropic_msgs.append({"role": "user", "content": content})
else:
anthropic_msgs.append({"role": "user", "content": str(content) if content else ""})
return "\n\n".join(system_parts), anthropic_msgs
def _convert_tools_to_anthropic(
tools: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
"""Convert OpenAI function-calling tools to Anthropic tool format."""
from butler.transport.schema_sanitizer import sanitize_tool_schemas
tools = sanitize_tool_schemas(tools, keep_nullable_hint=False) or []
result = []
for tool in tools:
fn = tool.get("function", tool)
entry = {
"name": fn.get("name", ""),
"description": fn.get("description", ""),
"input_schema": fn.get("parameters", {"type": "object", "properties": {}}),
}
result.append(entry)
return result
class AnthropicTransport(ProviderTransport): # type: ignore[misc]
@property
def api_mode(self) -> str:
return "anthropic_messages"
def convert_messages(
self, messages: List[Dict[str, Any]], **kwargs: Any
) -> tuple[str, List[Dict[str, Any]]]:
# Sprint 29 P2-4.3: apply cache_control to last user msg (boundary 2)
# before OpenAI→Anthropic conversion. The list-form content carries
# through `_convert_messages_to_anthropic` (which now passes list
# content for role=user verbatim). Then add cache_control to the
# last tool_result block (boundary 4) on the converted side.
from butler.transport.cache_control import (
apply_cache_control_to_last_tool_result,
apply_cache_control_to_messages,
)
annotated = apply_cache_control_to_messages(
messages if isinstance(messages, list) else []
)
system_prompt, anthropic_messages = _convert_messages_to_anthropic(annotated)
anthropic_messages = apply_cache_control_to_last_tool_result(anthropic_messages)
return system_prompt, anthropic_messages
def convert_tools(self, tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
return _convert_tools_to_anthropic(tools)
def build_kwargs(
self,
model: str,
messages: Any,
tools: Optional[Any] = None,
**params: Any,
) -> Dict[str, Any]:
if isinstance(messages, tuple) and len(messages) == 2:
system_prompt, anthropic_messages = messages
else:
# Route through self.convert_messages so cache_control (boundaries
# 2 + 4) is applied consistently — Sprint 29 P2-4.3.
system_prompt, anthropic_messages = self.convert_messages(
messages if isinstance(messages, list) else []
)
kwargs: Dict[str, Any] = {
"model": model,
"messages": anthropic_messages,
"max_tokens": params.get("max_tokens", 4096),
}
if system_prompt:
# Sprint 29 P2-4.3: boundary 1 (system). apply_cache_control_to_system
# returns [] when disabled or system is empty → caller keeps the
# original str form (backward-compatible with HEAD~).
from butler.transport.cache_control import apply_cache_control_to_system
sys_blocks = apply_cache_control_to_system(system_prompt)
kwargs["system"] = sys_blocks if sys_blocks else system_prompt
if tools:
if isinstance(tools, list) and tools and "input_schema" not in tools[0]:
tools = _convert_tools_to_anthropic(tools)
# Sprint 29 P2-4.3: boundary 3 (tools). apply_cache_control_to_tools
# returns the input unchanged when disabled or empty.
from butler.transport.cache_control import apply_cache_control_to_tools
kwargs["tools"] = apply_cache_control_to_tools(tools)
temperature = params.get("temperature")
if temperature is not None:
kwargs["temperature"] = temperature
timeout = params.get("timeout")
if timeout is not None:
kwargs["timeout"] = timeout
stream = params.get("stream")
if stream is not None:
kwargs["stream"] = stream
return kwargs
def normalize_response(
self, response: Any, **kwargs: Any
) -> NormalizedResponse:
if isinstance(response, dict):
return self._normalize_dict(response)
return self._normalize_sdk(response)
def _normalize_dict(self, data: dict[str, Any]) -> NormalizedResponse:
content_blocks = data.get("content", [])
stop_reason = data.get("stop_reason", "end_turn")
text_parts: list[str] = []
reasoning_parts: list[str] = []
tool_calls: list[ToolCall] = []
for block in content_blocks:
btype = block.get("type", "")
if btype == "text":
text_parts.append(block.get("text", ""))
elif btype == "thinking":
reasoning_parts.append(block.get("thinking", ""))
elif btype == "tool_use":
args = block.get("input", {})
tool_calls.append(ToolCall(
id=block.get("id"),
name=block.get("name", ""),
arguments=json.dumps(args, ensure_ascii=False) if isinstance(args, dict) else str(args),
))
usage = None
raw_usage = data.get("usage")
if raw_usage:
usage = Usage(
prompt_tokens=raw_usage.get("input_tokens", 0),
completion_tokens=raw_usage.get("output_tokens", 0),
total_tokens=raw_usage.get("input_tokens", 0) + raw_usage.get("output_tokens", 0),
)
return NormalizedResponse(
content="\n".join(text_parts) if text_parts else None,
tool_calls=tool_calls or None,
finish_reason=_STOP_REASON_MAP.get(stop_reason, "stop"),
reasoning="\n".join(reasoning_parts) if reasoning_parts else None,
usage=usage,
)
def _normalize_sdk(self, response: Any) -> NormalizedResponse:
content_blocks = getattr(response, "content", []) or []
stop_reason = getattr(response, "stop_reason", "end_turn")
text_parts: list[str] = []
reasoning_parts: list[str] = []
tool_calls: list[ToolCall] = []
for block in content_blocks:
btype = getattr(block, "type", "")
if btype == "text":
text_parts.append(getattr(block, "text", ""))
elif btype == "thinking":
thinking = getattr(block, "thinking", "")
reasoning_parts.append(thinking)
elif btype == "tool_use":
args = getattr(block, "input", {})
tool_calls.append(ToolCall(
id=getattr(block, "id", None),
name=getattr(block, "name", ""),
arguments=json.dumps(args, ensure_ascii=False) if isinstance(args, dict) else str(args),
))
usage = None
raw_usage = getattr(response, "usage", None)
if raw_usage:
usage = Usage(
prompt_tokens=getattr(raw_usage, "input_tokens", 0),
completion_tokens=getattr(raw_usage, "output_tokens", 0),
total_tokens=getattr(raw_usage, "input_tokens", 0) + getattr(raw_usage, "output_tokens", 0),
)
return NormalizedResponse(
content="\n".join(text_parts) if text_parts else None,
tool_calls=tool_calls or None,
finish_reason=_STOP_REASON_MAP.get(stop_reason, "stop"),
reasoning="\n".join(reasoning_parts) if reasoning_parts else None,
usage=usage,
)
def validate_response(self, response: Any) -> bool:
if isinstance(response, dict):
return bool(response.get("content")) or response.get("stop_reason") == "end_turn"
content = getattr(response, "content", None)
return bool(content) or getattr(response, "stop_reason", None) == "end_turn"
from butler.transport import register_transport # noqa: E402
register_transport("anthropic_messages", AnthropicTransport)