""" title: LiteLLM Dashboard author: Kagamine Rin author_url: https://rin.captcouch.xyz description: Check LiteLLM proxy health + auth status and get cost breakdowns with configurable timerange. Two functions, that's it. required_open_webui_version: 0.4.0 requirements: requests version: 1.0.0 licence: MIT """ import requests from pydantic import BaseModel, Field from typing import Optional class Tools: def __init__(self): self.valves = self.Valves() 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"}}, ) def _headers(self) -> dict: """Build auth headers from valves.""" if not self.valves.litellm_master_key: return {} return {"Authorization": f"Bearer {self.valves.litellm_master_key}"} def _base_url(self) -> str: """Get base URL, strip trailing slash.""" return self.valves.litellm_url.rstrip("/") def litellm_health(self) -> str: """ Check LiteLLM proxy health and authentication status. Returns which models are healthy/unhealthy, their rate limits, and any auth errors. Use this to verify your LiteLLM proxy is up and your master key works. """ url = f"{self._base_url()}/health" headers = self._headers() try: resp = requests.get(url, headers=headers, timeout=15) except requests.exceptions.ConnectionError: return f"❌ Connection failed — LiteLLM proxy not reachable at {self._base_url()}" except requests.exceptions.Timeout: return f"❌ Connection timed out after 15s — proxy at {self._base_url()} may be overloaded" if resp.status_code == 401: return "❌ Authentication failed — check your litellm_master_key valve. The proxy rejected the key." if resp.status_code == 403: return "❌ Forbidden — your key doesn't have permission to access /health" if resp.status_code != 200: return f"❌ Unexpected status {resp.status_code}: {resp.text[:500]}" data = resp.json() healthy = data.get("healthy_endpoints", []) unhealthy = data.get("unhealthy_endpoints", []) healthy_count = data.get("healthy_count", len(healthy)) unhealthy_count = data.get("unhealthy_count", len(unhealthy)) # Strip litellm_metadata — it's ~2KB per model of auth context we don't need # Also strip other internal fields that bloat the response STRIP_KEYS = { "litellm_metadata", "safety_settings", "cache_control_injection_points", "use_in_pass_through", "use_litellm_proxy", "merge_reasoning_content_in_choices", } lines = [] lines.append(f"LiteLLM Health Check — {self._base_url()}") lines.append("=" * 50) if not self.valves.litellm_master_key: lines.append("⚠️ No master key configured — showing public health only") else: lines.append("✅ Authenticated with master key") lines.append(f"\nHealthy: {healthy_count} | Unhealthy: {unhealthy_count}") if healthy: lines.append(f"\n--- Healthy Models ({healthy_count}) ---") for ep in healthy: model = ep.get("model", "unknown") # Strip gemini/ prefix for readability display = model.replace("gemini/", "") if model.startswith("gemini/") else model rpm = ep.get("rpm", "?") timeout = ep.get("timeout", "?") extras = [] if ep.get("reasoning_effort") and ep["reasoning_effort"] != "none": extras.append(f"reasoning: {ep['reasoning_effort']}") if ep.get("service_tier"): extras.append(f"tier: {ep['service_tier']}") extra_str = f" ({', '.join(extras)})" if extras else "" lines.append(f" ✅ {display} — rpm: {rpm}, timeout: {timeout}s{extra_str}") if unhealthy: lines.append(f"\n--- Unhealthy Models ({unhealthy_count}) ---") for ep in unhealthy: model = ep.get("model", "unknown") display = model.replace("gemini/", "") if model.startswith("gemini/") else model lines.append(f" ❌ {display}") if not healthy and not unhealthy: lines.append("\n(no endpoint data returned)") return "\n".join(lines) def litellm_cost( self, start_date: str, end_date: str, ) -> str: """ Get LiteLLM cost breakdown for a date range. :param start_date: Start date in YYYY-MM-DD format (e.g. "2026-06-01") :param end_date: End date in YYYY-MM-DD format (e.g. "2026-06-04") Returns formatted cost report with per-model and per-day breakdowns, token counts, cache hit rate, and request stats. """ if not self.valves.litellm_master_key: return "❌ Master key required for cost data. Set the litellm_master_key valve." url = f"{self._base_url()}/user/daily/activity" headers = self._headers() params = {"start_date": start_date, "end_date": end_date} try: resp = requests.get(url, headers=headers, params=params, timeout=30) except requests.exceptions.ConnectionError: return f"❌ Connection failed — LiteLLM proxy not reachable at {self._base_url()}" except requests.exceptions.Timeout: return f"❌ Connection timed out after 30s — large date ranges may be slow" if resp.status_code == 401: return "❌ Authentication failed — check your litellm_master_key valve." if resp.status_code == 403: return "❌ Forbidden — your key doesn't have permission to access cost data" if resp.status_code != 200: return f"❌ Unexpected status {resp.status_code}: {resp.text[:500]}" data = resp.json() results = data.get("results", []) metadata = data.get("metadata", {}) if not results: return f"No cost data found for {start_date} to {end_date}." # Aggregate totals from metadata total_spend = metadata.get("total_spend", 0) total_prompt = metadata.get("total_prompt_tokens", 0) total_completion = metadata.get("total_completion_tokens", 0) total_tokens = metadata.get("total_tokens", 0) total_successful = metadata.get("total_successful_requests", 0) total_failed = metadata.get("total_failed_requests", 0) total_api = metadata.get("total_api_requests", 0) cache_read = metadata.get("total_cache_read_input_tokens", 0) cache_created = metadata.get("total_cache_creation_input_tokens", 0) # Cache hit rate total_input_tokens = total_prompt # cache_read is a subset of prompt if total_input_tokens > 0 and cache_read > 0: cache_pct = (cache_read / total_input_tokens) * 100 cache_str = f"{cache_pct:.1f}% ({cache_read:,} / {total_input_tokens:,} input tokens)" else: cache_str = "N/A" # Build per-model breakdown across all dates model_totals = {} for day in results: breakdown = day.get("breakdown", {}) models = breakdown.get("models", {}) for model_name, model_data in models.items(): m_metrics = model_data.get("metrics", {}) if model_name not in model_totals: model_totals[model_name] = { "spend": 0, "requests": 0, "prompt": 0, "completion": 0, "total": 0, "failed": 0, } model_totals[model_name]["spend"] += m_metrics.get("spend", 0) model_totals[model_name]["requests"] += m_metrics.get("successful_requests", 0) model_totals[model_name]["failed"] += m_metrics.get("failed_requests", 0) model_totals[model_name]["prompt"] += m_metrics.get("prompt_tokens", 0) model_totals[model_name]["completion"] += m_metrics.get("completion_tokens", 0) model_totals[model_name]["total"] += m_metrics.get("total_tokens", 0) # Build per-date breakdown date_lines = [] for day in results: date = day.get("date", "?") m = day.get("metrics", {}) date_lines.append( f" {date}: ${m.get('spend', 0):.2f} | " f"{m.get('successful_requests', 0)} ok, {m.get('failed_requests', 0)} fail | " f"{m.get('total_tokens', 0):,} tokens" ) # Build model lines sorted by spend model_lines = [] for model_name in sorted(model_totals, key=lambda k: model_totals[k]["spend"], reverse=True): mt = model_totals[model_name] # Strip provider prefix and abbreviate long model names for table readability display = model_name.replace("gemini/", "") if model_name.startswith("gemini/") else model_name # Abbreviate known long names display = display.replace("3.1-pro-preview-customtools", "3.1-pro-ct") display = display.replace("3.1-flash-image-preview", "3.1-flash-img") display = display.replace("3.1-flash-lite-preview", "3.1-flash-lite") display = display.replace("3-flash-preview", "3-flash-prev") model_lines.append( f" {display:20s} ${mt['spend']:7.2f} | " f"{mt['requests']:3d} ok, {mt['failed']:d} fail | " f"{mt['total']:>10,} tokens" ) # Assemble report lines = [] lines.append(f"LiteLLM Cost Report: {start_date} to {end_date}") lines.append("=" * 60) lines.append(f"Total Spend: ${total_spend:.2f}") lines.append(f"Total Tokens: {total_tokens:,} ({total_prompt:,} prompt + {total_completion:,} completion)") lines.append(f"Cache Hit Rate: {cache_str}") lines.append(f"Requests: {total_successful} successful, {total_failed} failed ({total_api} total)") if cache_created > 0: lines.append(f"Cache Created: {cache_created:,} tokens") lines.append("") lines.append("By Model:") lines.append(f' {"Model":20s} {"Spend":>7s} | {"Requests":>14s} | {"Tokens":>10s}') lines.extend(model_lines) lines.append("") lines.append("By Date:") lines.extend(date_lines) return "\n".join(lines)