92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
"""Anthropic-native /v1/messages backend.
|
|
|
|
For endpoints that speak the Anthropic Messages API (api.anthropic.com or any
|
|
proxy compatible with it). Uses x-api-key + anthropic-version headers.
|
|
"""
|
|
from __future__ import annotations
|
|
import json
|
|
import urllib.error
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
|
|
|
|
DEFAULT_ANTHROPIC_VERSION = "2023-06-01"
|
|
|
|
|
|
@dataclass
|
|
class AnthropicNativeBackend:
|
|
cfg: dict
|
|
|
|
@property
|
|
def endpoint(self) -> str:
|
|
return self.cfg["endpoint"].rstrip("/")
|
|
|
|
@property
|
|
def model(self) -> str:
|
|
return self.cfg["model"]
|
|
|
|
@property
|
|
def api_key(self) -> str:
|
|
return self.cfg.get("api_key") or "missing-key"
|
|
|
|
@property
|
|
def max_output_tokens(self) -> int:
|
|
return int(self.cfg.get("max_output_tokens", 4000))
|
|
|
|
@property
|
|
def timeout(self) -> int:
|
|
return int(self.cfg.get("timeout_seconds", 60))
|
|
|
|
@property
|
|
def anthropic_version(self) -> str:
|
|
return self.cfg.get("anthropic_version", DEFAULT_ANTHROPIC_VERSION)
|
|
|
|
def summarise(self, system_prompt: str, conversation_text: str) -> str:
|
|
url = f"{self.endpoint}/v1/messages"
|
|
body = {
|
|
"model": self.model,
|
|
"max_tokens": self.max_output_tokens,
|
|
"temperature": 0.2,
|
|
"system": system_prompt,
|
|
"messages": [
|
|
{"role": "user", "content": conversation_text},
|
|
],
|
|
}
|
|
data = json.dumps(body).encode("utf-8")
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"x-api-key": self.api_key,
|
|
"anthropic-version": self.anthropic_version,
|
|
}
|
|
# Some proxies (e.g. wolfai.top) accept Bearer tokens too — send both
|
|
# so we work whether the upstream wants x-api-key or Authorization.
|
|
if self.api_key.startswith("sk-"):
|
|
headers["Authorization"] = f"Bearer {self.api_key}"
|
|
|
|
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
|
|
payload = json.loads(resp.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as e:
|
|
detail = e.read().decode("utf-8", errors="replace")[:500]
|
|
raise RuntimeError(f"LLM HTTP {e.code}: {detail}") from e
|
|
except urllib.error.URLError as e:
|
|
raise RuntimeError(f"LLM connection failed: {e.reason}") from e
|
|
|
|
# Anthropic format: {"content": [{"type":"text","text":"..."}], ...}
|
|
content = payload.get("content")
|
|
if isinstance(content, list):
|
|
for block in content:
|
|
if isinstance(block, dict) and block.get("type") == "text":
|
|
text = block.get("text", "")
|
|
if text.strip():
|
|
return text.strip()
|
|
# Some proxies pass through OpenAI shape — fall back gracefully.
|
|
choices = payload.get("choices") or []
|
|
if choices:
|
|
msg = choices[0].get("message") or {}
|
|
content = msg.get("content")
|
|
if isinstance(content, str) and content.strip():
|
|
return content.strip()
|
|
raise RuntimeError(f"LLM returned unexpected payload shape: {payload}")
|