""" Anthropic • Priced Pipe for OpenWebUI Author: dIROX & fatalv.nu version: 1.00 Description: Anthropic (Claude) pipe that lists live models, calls the Messages API, prints a full token & cost breakdown (Base Input, Cache Writes [5m/1h], Cache Hits & Refreshes, Output), and shows per-model rate limits. Supports USD + optional SEK. """ from pydantic import BaseModel, Field from typing import Dict, Any, List, Union, Iterator, Optional import requests # ---------------- Pricing (USD per 1M tokens) ---------------- # Keys are pricing "families" we map Anthropic model IDs to. ANTHROPIC_PRICING: Dict[str, Dict[str, float]] = { # base_in write_5m write_1h cache_hit output "opus-4.1": {"in": 15.00, "w5": 18.75, "w1": 30.00, "hit": 1.50, "out": 75.00}, "opus-4": {"in": 15.00, "w5": 18.75, "w1": 30.00, "hit": 1.50, "out": 75.00}, "opus-3": { "in": 15.00, "w5": 18.75, "w1": 30.00, "hit": 1.50, "out": 75.00, }, # deprecated "sonnet-4": {"in": 3.00, "w5": 3.75, "w1": 6.00, "hit": 0.30, "out": 15.00}, "sonnet-3.7": {"in": 3.00, "w5": 3.75, "w1": 6.00, "hit": 0.30, "out": 15.00}, "sonnet-3.5": { "in": 3.00, "w5": 3.75, "w1": 6.00, "hit": 0.30, "out": 15.00, }, # deprecated family label, still priced "haiku-3.5": {"in": 0.80, "w5": 1.00, "w1": 1.60, "hit": 0.08, "out": 4.00}, "haiku-3": {"in": 0.25, "w5": 0.30, "w1": 0.50, "hit": 0.03, "out": 1.25}, } # ---------------- Rate limits per family ---------------- # Values you provided: RPM, Input TPM, Output TPM, with ≤200k context notes. FAMILY_LIMITS: Dict[str, Dict[str, Any]] = { "opus-4.1": { "rpm": 50, "tpm_in": 30000, "tpm_out": 8000, "ctx_note": "≤200k context (input excludes cache reads)", }, "opus-4": { "rpm": 50, "tpm_in": 30000, "tpm_out": 8000, "ctx_note": "≤200k context (input excludes cache reads)", }, "sonnet-4": { "rpm": 50, "tpm_in": 30000, "tpm_out": 8000, "ctx_note": "≤200k context (input excludes cache reads)", }, "sonnet-3.7": { "rpm": 50, "tpm_in": 20000, "tpm_out": 8000, "ctx_note": "≤200k context (input excludes cache reads)", }, "sonnet-3.5": { "rpm": 50, "tpm_in": 40000, "tpm_out": 8000, "ctx_note": "≤200k context", }, "haiku-3.5": { "rpm": 50, "tpm_in": 50000, "tpm_out": 10000, "ctx_note": "≤200k context", }, "haiku-3": { "rpm": 50, "tpm_in": 50000, "tpm_out": 10000, "ctx_note": "≤200k context", }, "opus-3": { "rpm": 50, "tpm_in": 20000, "tpm_out": 4000, "ctx_note": "≤200k context", }, } # Map known model IDs → pricing/limits family ID_TO_FAMILY: Dict[str, str] = { "claude-opus-4-20250514": "opus-4", "claude-sonnet-4-20250514": "sonnet-4", "claude-3-7-sonnet-20250219": "sonnet-3.7", "claude-3-5-sonnet-20241022": "sonnet-3.5", "claude-3-5-sonnet-20240620": "sonnet-3.5", "claude-3-opus-20240229": "opus-3", "claude-3-5-haiku-20241022": "haiku-3.5", "claude-3-haiku-20240307": "haiku-3", # If Anthropic exposes a dedicated Opus 4.1 model id, add it here → "opus-4.1" } # Fallback visible list (your current available models) FALLBACK_MODEL_IDS: List[str] = [ "claude-3-5-sonnet-20240620", "claude-3-opus-20240229", "claude-3-haiku-20240307", "claude-3-5-haiku-20241022", "claude-3-5-sonnet-20241022", "claude-3-7-sonnet-20250219", "claude-opus-4-20250514", "claude-sonnet-4-20250514", ] # ---------------- Helpers ---------------- def _family_for(model_id: str) -> Optional[str]: if model_id in ID_TO_FAMILY: return ID_TO_FAMILY[model_id] low = model_id.lower() if "opus-4.1" in low: return "opus-4.1" if "opus-4" in low: return "opus-4" if "opus-3" in low: return "opus-3" if "sonnet-4" in low: return "sonnet-4" if "3-7-sonnet" in low or "sonnet-3.7" in low: return "sonnet-3.7" if "3-5-sonnet" in low or "sonnet-3.5" in low: return "sonnet-3.5" if "haiku-3.5" in low: return "haiku-3.5" if "haiku-3" in low: return "haiku-3" return None def _normalize_model_from_ui(model: str, name_prefix: str, func_id: str) -> str: """ Strip UI prefix (e.g. 'ANTHROPIC/') and function-id prefix (e.g. 'anthropic_priced.'). """ if not isinstance(model, str): return model m = model.strip() if name_prefix and m.startswith(name_prefix): m = m[len(name_prefix) :] if func_id and m.startswith(func_id + "."): m = m[len(func_id) + 1 :] return m def _extract_usage(u: Dict[str, Any]) -> Dict[str, int]: """ Normalize Anthropic usage: - base_input = input_tokens (non-cached) - cache_write = cache_creation_input_tokens - cache_hit = cache_read_input_tokens - output = output_tokens """ if not isinstance(u, dict): return {"base_input": 0, "cache_write": 0, "cache_hit": 0, "output": 0} return { "base_input": int(u.get("input_tokens") or 0), "cache_write": int(u.get("cache_creation_input_tokens") or 0), "cache_hit": int(u.get("cache_read_input_tokens") or 0), "output": int(u.get("output_tokens") or 0), } def _fmt_money(x: Optional[float], code: str = "USD") -> str: return f"{code} " + ("-" if x is None else f"{x:.6f}".rstrip("0").rstrip(".")) def _calc_cost_usd( row: Dict[str, float], usage: Dict[str, int], write_tier: str ) -> Dict[str, float]: """ row keys: in, w5, w1, hit, out usage keys: base_input, cache_write, cache_hit, output write_tier: '5m' or '1h' """ per_m = 1_000_000.0 wkey = "w5" if str(write_tier).lower().startswith("5") else "w1" c_in = (usage["base_input"] / per_m) * row["in"] c_w = (usage["cache_write"] / per_m) * row[wkey] c_hit = (usage["cache_hit"] / per_m) * row["hit"] c_out = (usage["output"] / per_m) * row["out"] total = c_in + c_w + c_hit + c_out return { "base_input_usd": round(c_in, 6), "cache_write_usd": round(c_w, 6), "cache_hit_usd": round(c_hit, 6), "output_usd": round(c_out, 6), "total_usd": round(total, 6), } def _usd_to_sek(costs_usd: Dict[str, float], rate: float) -> Dict[str, float]: def mult(x: float) -> float: return round((x or 0.0) * rate, 6) return { "base_input_sek": mult(costs_usd["base_input_usd"]), "cache_write_sek": mult(costs_usd["cache_write_usd"]), "cache_hit_sek": mult(costs_usd["cache_hit_usd"]), "output_sek": mult(costs_usd["output_usd"]), "total_sek": mult(costs_usd["total_usd"]), } def _get_models(base_url: str, api_key: str, api_version: str) -> List[str]: headers = {"x-api-key": api_key, "anthropic-version": api_version} r = requests.get(f"{base_url}/models", headers=headers, timeout=20) r.raise_for_status() data = r.json() raw = data.get("data") or [] ids = [m.get("id") for m in raw if isinstance(m, dict) and m.get("id")] return sorted(ids, key=str.lower) def _coerce_text(content: Any) -> str: """Join different OpenWebUI content shapes into plain text for Anthropic.""" if isinstance(content, str): return content if isinstance(content, list): texts = [] for part in content: if isinstance(part, dict) and part.get("type") == "text": texts.append(str(part.get("text") or "")) return " ".join([t for t in texts if t]).strip() return str(content or "") class Pipe: class Valves(BaseModel): NAME_PREFIX: str = Field( default="ANTHROPIC/", description="UI prefix (removed before sending)." ) ANTHROPIC_API_BASE_URL: str = Field( default="https://api.anthropic.com/v1", description="Anthropic base URL." ) ANTHROPIC_API_KEY: str = Field(default="", description="Anthropic API key.") ANTHROPIC_API_VERSION: str = Field( default="2023-06-01", description="Anthropic API version header." ) DEFAULT_MODEL: str = Field( default="claude-3-5-sonnet-20241022", description="Default model if none provided.", ) MAX_TOKENS: int = Field( default=1024, description="Max output tokens for the generation." ) DEFAULT_PARAMS: dict = Field( default={"temperature": 0.7}, description="Default generation params." ) ENFORCE_KNOWN_MODEL: bool = Field( default=True, description="Validate model exists via /models." ) CACHE_WRITE_TIER: str = Field( default="5m", description="Cache write tier used for pricing: '5m' or '1h'." ) # Currency display SHOW_USD: bool = Field(default=True, description="Show USD prices.") SHOW_SEK: bool = Field(default=False, description="Show SEK prices.") USD_TO_SEK: float = Field(default=10.95, description="Manual FX rate USD→SEK.") def __init__(self): self.type = "anthropic_priced" self.id = "anthropic_priced" self.name = "Anthropic • Priced" self.valves = self.Valves() # ---- Model dropdown ---- def pipes(self) -> List[dict]: pref = self.valves.NAME_PREFIX or "" key = self.valves.ANTHROPIC_API_KEY ver = self.valves.ANTHROPIC_API_VERSION ids: List[str] = [] if key: try: ids = _get_models(self.valves.ANTHROPIC_API_BASE_URL, key, ver) except Exception: ids = [] if not ids: ids = FALLBACK_MODEL_IDS[:] return [{"id": mid, "name": f"{pref}{mid}"} for mid in ids] # ---- Core call (non-stream to capture usage) ---- def pipe(self, body: dict) -> Union[str, Iterator[str]]: if not self.valves.ANTHROPIC_API_KEY: return "Error: Please set ANTHROPIC_API_KEY." model_in = body.get("model") or self.valves.DEFAULT_MODEL model_id = _normalize_model_from_ui(model_in, self.valves.NAME_PREFIX, self.id) # Build Anthropic messages + system sys_msgs: List[str] = [] msg_list: List[Dict[str, Any]] = [] incoming = body.get("messages") if isinstance(incoming, list) and incoming: for m in incoming: role = m.get("role") content = _coerce_text(m.get("content")) if not content: continue if role == "system": sys_msgs.append(content) elif role in ("user", "assistant"): msg_list.append({"role": role, "content": content}) else: prompt = str(body.get("prompt") or "") msg_list = [{"role": "user", "content": prompt}] if prompt else [] system_text = "\n".join(sys_msgs) if sys_msgs else None # Optional validation: ensure model exists if self.valves.ENFORCE_KNOWN_MODEL and self.valves.ANTHROPIC_API_KEY: try: live_ids = set( _get_models( self.valves.ANTHROPIC_API_BASE_URL, self.valves.ANTHROPIC_API_KEY, self.valves.ANTHROPIC_API_VERSION, ) ) except Exception: live_ids = set() if live_ids and model_id not in live_ids: sample = ", ".join(sorted(list(live_ids))[:15]) return { "error": f"Model '{model_id}' is not available to your Anthropic account.", "hint": "Pick one from the dropdown (it reflects your live account models).", "available": sample, } # Build request allowed = { "temperature", "top_p", "top_k", "stop_sequences", "metadata", "system", "tools", "tool_choice", } payload = {k: v for k, v in body.items() if k in allowed} for k, v in (self.valves.DEFAULT_PARAMS or {}).items(): payload.setdefault(k, v) payload["model"] = model_id payload["max_tokens"] = int(body.get("max_tokens") or self.valves.MAX_TOKENS) payload["messages"] = msg_list if system_text: payload["system"] = system_text headers = { "x-api-key": self.valves.ANTHROPIC_API_KEY, "anthropic-version": self.valves.ANTHROPIC_API_VERSION, "content-type": "application/json", } try: r = requests.post( f"{self.valves.ANTHROPIC_API_BASE_URL}/messages", json=payload, headers=headers, timeout=300, ) except Exception as e: return f"Error calling Anthropic: {e}" if r.status_code >= 400: try: return r.json() except Exception: return {"error": r.text, "status": r.status_code} data = r.json() # Extract text text = "" try: blocks = data.get("content") or [] parts = [] for b in blocks: if isinstance(b, dict) and b.get("type") == "text": parts.append(str(b.get("text") or "")) text = "\n".join([p for p in parts if p]).strip() except Exception: pass # Usage usage_raw = data.get("usage") or {} usage = _extract_usage(usage_raw) # Pricing + limits fam = _family_for(model_id) row = ANTHROPIC_PRICING.get(fam) if fam else None lim = FAMILY_LIMITS.get(fam) if fam else None # Output out: List[str] = [] if text: out.append(text) out.append("") out.append("---") out.append( f"**Model:** `{model_id}`" + (f" | **Family:** {fam}" if fam else "") ) out.append( f"**Tokens:** base_in {usage['base_input']}, " f"cache_write {usage['cache_write']}, " f"cache_hit {usage['cache_hit']}, " f"out {usage['output']}, " f"total {usage['base_input'] + usage['cache_write'] + usage['cache_hit'] + usage['output']}" ) if lim: lim_line = f"**Limits:** {lim['rpm']} rpm | {lim['tpm_in']:,} in tpm | {lim['tpm_out']:,} out tpm" if lim.get("ctx_note"): lim_line += f" | {lim['ctx_note']}" out.append(lim_line) if row: costs_usd = _calc_cost_usd(row, usage, self.valves.CACHE_WRITE_TIER) if self.valves.SHOW_USD: out.append( f"**Cost (USD):** {_fmt_money(costs_usd['base_input_usd'])} base_in " f"+ {_fmt_money(costs_usd['cache_write_usd'])} cache_write({self.valves.CACHE_WRITE_TIER}) " f"+ {_fmt_money(costs_usd['cache_hit_usd'])} cache_hit " f"+ {_fmt_money(costs_usd['output_usd'])} out " f"= **{_fmt_money(costs_usd['total_usd'])}**" ) out.append( f"_Rates per 1M (USD): base_in {_fmt_money(row['in'])}, " f"write(5m) {_fmt_money(row['w5'])}, write(1h) {_fmt_money(row['w1'])}, " f"hit {_fmt_money(row['hit'])}, out {_fmt_money(row['out'])}_" ) if self.valves.SHOW_SEK: rate = float(self.valves.USD_TO_SEK or 0.0) sek = _usd_to_sek(costs_usd, rate) out.append( f"**Cost (SEK @ {rate}):** {_fmt_money(sek['base_input_sek'],'SEK')} base_in " f"+ {_fmt_money(sek['cache_write_sek'],'SEK')} cache_write({self.valves.CACHE_WRITE_TIER}) " f"+ {_fmt_money(sek['cache_hit_sek'],'SEK')} cache_hit " f"+ {_fmt_money(sek['output_sek'],'SEK')} out " f"= **{_fmt_money(sek['total_sek'],'SEK')}**" ) out.append( f"_Rates per 1M (SEK): base_in {_fmt_money(row['in']*rate,'SEK')}, " f"write(5m) {_fmt_money(row['w5']*rate,'SEK')}, write(1h) {_fmt_money(row['w1']*rate,'SEK')}, " f"hit {_fmt_money(row['hit']*rate,'SEK')}, out {_fmt_money(row['out']*rate,'SEK')}_" ) else: out.append("_No pricing row found for this model; showing tokens only._") return "\n".join(out)