""" title: LiteLLM Dashboard author: Kagamine Rin & VAI author_url: https://rin.captcouch.xyz description: Hardcore LiteLLM observability — health, cost breakdown, and request history with Dynamic FinOps Audit. required_open_webui_version: 0.4.0 requirements: requests version: 1.4.0 licence: MIT """ import requests import json from datetime import datetime, timedelta from zoneinfo import ZoneInfo from pydantic import BaseModel, Field from typing import Optional, List, Dict, Any class Tools: def __init__(self): self.valves = self.Valves() self._price_cache = {} class Valves(BaseModel): """LiteLLM proxy connection settings.""" litellm_url: str = Field( default="http://localhost:4000", description="LiteLLM proxy base URL (e.g. http://localhost:4000)", ) litellm_master_key: str = Field( default="", description="LiteLLM master API key for authentication", json_schema_extra={"input": {"type": "password"}}, ) default_hide_zero_spend: bool = Field( default=True, description="Default behavior for zero-spend model filtering.", ) timezone: str = Field( default="UTC", description="IANA Timezone name (e.g. America/New_York).", ) enable_dynamic_pricing: bool = Field( default=True, description="When True, attempts to fetch live pricing from /v1/model/info for the Audit Engine.", ) # Fallback registry for 2026 standard tier pricing MODEL_PRICING_FALLBACK = { "gemini/gemini-3.5-flash": { "input": 1.50 / 1_000_000, "input_cached": 0.15 / 1_000_000, "output": 9.00 / 1_000_000, }, "gemini/gemini-3-flash-preview": { "input": 0.50 / 1_000_000, "input_cached": 0.05 / 1_000_000, "output": 3.00 / 1_000_000, } } def _headers(self) -> dict: if not self.valves.litellm_master_key: return {} return {"Authorization": f"Bearer {self.valves.litellm_master_key}"} def _base_url(self) -> str: return self.valves.litellm_url.rstrip("/") def _fetch_live_prices(self): """Sync internal price cache with LiteLLM's live model registry.""" if not self.valves.enable_dynamic_pricing or self._price_cache: return try: url = f"{self._base_url()}/v1/model/info" resp = requests.get(url, headers=self._headers(), timeout=10) if resp.status_code == 200: data = resp.json().get("data", []) for m in data: name = m.get("model_name") info = m.get("model_info", {}) if name and info: self._price_cache[name] = { "input": info.get("input_cost_per_token", 0), "input_cached": info.get("cache_read_input_token_cost", 0), "output": info.get("output_cost_per_token", 0), "reasoning": info.get("output_cost_per_reasoning_token", info.get("output_cost_per_token", 0)) } except Exception: pass def _audit_cost(self, model: str, prompt: int, cached: int, completion: int, reported_cost: float, reasoning: int = 0) -> dict: """Audit reported cost with dynamic or fallback pricing.""" self._fetch_live_prices() pricing = self._price_cache.get(model) or self.MODEL_PRICING_FALLBACK.get(model) if not pricing: return {"status": "unsupported", "display": ""} uncached = max(0, prompt - cached) # Handle reasoning tokens separately if pricing allows actual_output = max(0, completion - reasoning) calc = (cached * pricing.get("input_cached", 0)) + \ (uncached * pricing.get("input", 0)) + \ (actual_output * pricing.get("output", 0)) + \ (reasoning * pricing.get("reasoning", pricing.get("output", 0))) delta = reported_cost - calc status = "mismatch" if abs(delta) > 0.005 else "ok" symbol = "⚠️" if status == "mismatch" else "✓" return { "audited": calc, "delta": delta, "status": status, "display": f" [calc: ${calc:.4f} {symbol}]" } def _resolve_date(self, date_str: str) -> str: date_str = date_str.strip().lower() try: today = datetime.now(ZoneInfo(self.valves.timezone)).date() except Exception: today = datetime.utcnow().date() if date_str == "today": return today.isoformat() if date_str == "yesterday": return (today - timedelta(days=1)).isoformat() if "days ago" in date_str: try: n = int(date_str.split()[0]) return (today - timedelta(days=n)).isoformat() except Exception: pass try: datetime.strptime(date_str, "%Y-%m-%d") return date_str except ValueError: pass return date_str def litellm_status(self) -> str: """ Get an ultra-lean snapshot of proxy health and total spend. Uses /health/readiness and /global/spend/models for maximum token efficiency. """ lines = [f"LiteLLM Status Snapshot — {self._base_url()}"] lines.append("=" * 40) # 1. Readiness (Infrastructure) try: r = requests.get(f"{self._base_url()}/health/readiness", timeout=5) if r.status_code == 200: data = r.json() lines.append(f"Proxy: ✅ {data.get('status')} | DB: {data.get('db')} | v{data.get('litellm_version')}") else: lines.append(f"Proxy Readiness: ⚠️ Status {r.status_code}") except Exception as e: lines.append(f"Proxy Readiness: ❌ Connection Error") # 2. Spend Snapshot (Token Efficient) if self.valves.litellm_master_key: try: s = requests.get(f"{self._base_url()}/global/spend/models", headers=self._headers(), timeout=5) if s.status_code == 200: models = s.json() total = sum(m.get("total_spend", 0) for m in models) top = sorted(models, key=lambda x: x.get("total_spend", 0), reverse=True)[:3] lines.append(f"Lifetime Spend: ${total:.2f}") if top: lines.append("Top Models: " + ", ".join([f"{m['model']}: ${m['total_spend']:.2f}" for m in top])) except Exception: pass return "\n".join(lines) def litellm_health(self) -> str: """Check LiteLLM proxy health and authentication status.""" url = f"{self._base_url()}/health" try: resp = requests.get(url, headers=self._headers(), timeout=15) except Exception as e: return f"❌ Connection failed: {str(e)}" if resp.status_code != 200: return f"❌ Error {resp.status_code}: {resp.text[:200]}" data = resp.json() healthy = data.get("healthy_endpoints", []) unhealthy = data.get("unhealthy_endpoints", []) lines = [f"LiteLLM Health Check — {self._base_url()}"] lines.append("=" * 50) lines.append(f"Healthy: {len(healthy)} | Unhealthy: {len(unhealthy)}") for ep in healthy: m = ep.get("model", "unknown") lines.append(f" ✅ {m:25s} | rpm: {ep.get('rpm','?'):>3} | timeout: {ep.get('timeout','?')}s") for ep in unhealthy: lines.append(f" ❌ {ep.get('model', 'unknown')}") return "\n".join(lines) def litellm_cost(self, start_date: str, end_date: str, hide_zero_spend: bool = False) -> str: """Get cost breakdown with Sparkline trend visualization.""" if not self.valves.litellm_master_key: return "❌ Master key required." effective_hide = hide_zero_spend or self.valves.default_hide_zero_spend r_start = self._resolve_date(start_date) r_end = self._resolve_date(end_date) url = f"{self._base_url()}/user/daily/activity" try: resp = requests.get(url, headers=self._headers(), params={"start_date": r_start, "end_date": r_end}, timeout=30) if resp.status_code != 200: return f"❌ Error {resp.status_code}" data = resp.json() except Exception as e: return f"❌ Failed: {str(e)}" results = data.get("results", []) if not results: return "No data found." # Sparkline Generation spends = [d.get("metrics", {}).get("spend", 0) for d in results] def get_spark(vals): if not vals: return "" chars = " ▂▃▄▅▆▇█" low, high = min(vals), max(vals) rng = (high - low) or 1 return "".join(chars[int((v - low) / rng * 7)] for v in vals) meta = data.get("metadata", {}) lines = [f"LiteLLM Cost Report: {r_start} to {r_end}"] lines.append("=" * 60) lines.append(f"Total Spend: ${meta.get('total_spend', 0):.2f} [{get_spark(spends)}]") lines.append(f"Requests: {meta.get('total_successful_requests', 0)} ok, {meta.get('total_failed_requests', 0)} fail") # Endpoint breakdown lines.append("\nBy Endpoint:") ends = results[0].get("breakdown", {}).get("endpoints", {}) if results else {} for path, edata in ends.items(): m = edata.get("metrics", {}) lines.append(f" {path:20s} | ${m.get('spend',0):.4f} | {m.get('successful_requests',0)} reqs") return "\n".join(lines) def litellm_history(self, limit: int = 20, show_failed: bool = True) -> str: """Request history with Thought Token detection and Dynamic Audit.""" if not self.valves.litellm_master_key: return "❌ Master key required." limit = max(1, min(50, limit)) try: resp = requests.get(f"{self._base_url()}/spend/logs", headers=self._headers(), timeout=30) if resp.status_code != 200: return f"❌ Error {resp.status_code}" logs = sorted(resp.json(), key=lambda x: x.get("startTime", ""), reverse=True) except Exception as e: return f"❌ Failed: {str(e)}" if not show_failed: logs = [l for l in logs if not l.get("metadata", {}).get("exception")] recent = logs[:limit] lines = [f"LiteLLM History (Most Recent {len(recent)})", "=" * 90] for i, entry in enumerate(recent, 1): ts = entry.get("startTime", "?")[5:19] model = entry.get("model", "unknown")[:25] spend = entry.get("spend", 0) prompt = entry.get("prompt_tokens", 0) or 0 completion = entry.get("completion_tokens", 0) or 0 # Thought / Reasoning tokens usage = entry.get("usage", {}) or {} thought = usage.get("completion_tokens_details", {}).get("reasoning_tokens", 0) thought_str = f" [💭{thought}]" if thought else "" # Cache cached = entry.get("cache_read_input_tokens", 0) or 0 cache_str = f" 💾{int(cached/prompt*100)}%" if prompt and cached else "" audit = self._audit_cost(entry.get("model"), prompt, cached, completion, spend, reasoning=thought) line = f"{i:2d}. {ts} | {model:25s} | ${spend:.4f}{audit['display']} | {prompt}→{completion}{thought_str}{cache_str}" lines.append(line) return "\n".join(lines)