owi-litellm-tool/litellm_dashboard.py

"""
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.2.0
licence: MIT
"""

import requests
from datetime import datetime, timedelta
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 _resolve_date(self, date_str: str) -> str:
        """Resolve natural language date shortcuts to YYYY-MM-DD.

        Supports: today, yesterday, tomorrow, N days ago,
        monday-sunday (this week), last monday-last sunday,
        this week, last week, this month, last month.
        Falls back to passthrough if already YYYY-MM-DD or unrecognized.
        """
        date_str = date_str.strip().lower()
        today = datetime.utcnow().date()

        # Direct shortcuts
        if date_str == "today":
            return today.isoformat()
        if date_str == "yesterday":
            return (today - timedelta(days=1)).isoformat()
        if date_str == "tomorrow":
            return (today + timedelta(days=1)).isoformat()

        # "N days ago"
        if "days ago" in date_str:
            try:
                n = int(date_str.split()[0])
                return (today - timedelta(days=n)).isoformat()
            except (ValueError, IndexError):
                pass

        # Day names — most recent occurrence (including today if it matches)
        days = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"]
        if date_str in days:
            target_day = days.index(date_str)
            days_back = (today.weekday() - target_day) % 7
            return (today - timedelta(days=days_back)).isoformat()

        # "last monday", "last tuesday", etc.
        if date_str.startswith("last "):
            day_part = date_str[5:].strip()
            if day_part in days:
                target_day = days.index(day_part)
                days_back = (today.weekday() - target_day) % 7 + 7
                return (today - timedelta(days=days_back)).isoformat()

        # "this week" → Monday of this week
        if date_str == "this week":
            return (today - timedelta(days=today.weekday())).isoformat()

        # "last week" → Monday of last week
        if date_str == "last week":
            return (today - timedelta(days=today.weekday() + 7)).isoformat()

        # "this month" → 1st of this month
        if date_str == "this month":
            return today.replace(day=1).isoformat()

        # "last month" → 1st of last month
        if date_str == "last month":
            if today.month == 1:
                return today.replace(year=today.year - 1, month=12, day=1).isoformat()
            return today.replace(month=today.month - 1, day=1).isoformat()

        # Already YYYY-MM-DD — passthrough
        try:
            datetime.strptime(date_str, "%Y-%m-%d")
            return date_str
        except ValueError:
            pass

        # Unknown — return as-is (API will reject with clear error)
        return date_str

    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,
        show_zero_spend: bool = False,
    ) -> str:
        """
        Get LiteLLM cost breakdown for a date range.
        :param start_date: Start date — YYYY-MM-DD or natural language shortcut (e.g. "2026-06-01", "today", "yesterday", "monday", "last monday", "two days ago", "this week", "last month")
        :param end_date: End date — YYYY-MM-DD or natural language shortcut
        :param show_zero_spend: Show models with sub-$0.001 spend (default False). When False, these models are hidden and summarized as "... (N model(s) with sub-$0.001 spend hidden)". Useful when free-tier or subscription models (e.g. Ollama Cloud) dominate the breakdown with high token counts but negligible cost.
        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."

        resolved_start = self._resolve_date(start_date)
        resolved_end = self._resolve_date(end_date)

        url = f"{self._base_url()}/user/daily/activity"
        headers = self._headers()
        params = {"start_date": resolved_start, "end_date": resolved_end}

        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 = []
        hidden_count = 0
        for model_name in sorted(model_totals, key=lambda k: model_totals[k]["spend"], reverse=True):
            mt = model_totals[model_name]
            if not show_zero_spend and mt["spend"] < 0.001:
                hidden_count += 1
                continue
            # Strip common provider prefixes for readability
            display = model_name
            for prefix in ("gemini/", "openai/", "anthropic/", "mistral/", "cohere/"):
                if display.startswith(prefix):
                    display = display[len(prefix):]
                    break
            # Truncate long model names to fit table, appending ellipsis
            max_name = 24
            if len(display) > max_name:
                display = display[:max_name - 1] + "…"
            model_lines.append(
                f"  {display:25s} ${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: {resolved_start} to {resolved_end}")
        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":25s} {"Spend":>7s} | {"Requests":>14s} | {"Tokens":>10s}')
        lines.extend(model_lines)
        if hidden_count > 0:
            lines.append(f"  ... ({hidden_count} model(s) with sub-$0.001 spend hidden)")
        lines.append("")
        lines.append("By Date:")
        lines.extend(date_lines)

        return "\n".join(lines)

    def litellm_history(
        self,
        limit: int = 20,
        show_cost: bool = True,
        show_failed: bool = True,
    ) -> str:
        """
        Get recent LiteLLM proxy request history.
        :param limit: Number of recent requests to show (1-50, default 20)
        :param show_cost: Include per-request cost (default True)
        :param show_failed: Include failed requests in output (default True). Set False to hide auth failures, routing errors, etc.
        Returns a formatted table of recent LLM requests with model, status, tokens, duration, and optional cost.
        """
        if not self.valves.litellm_master_key:
            return "❌ Master key required for request history. Set the litellm_master_key valve."

        limit = max(1, min(50, limit))

        url = f"{self._base_url()}/spend/logs"
        headers = self._headers()

        try:
            resp = requests.get(url, headers=headers, 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 — request history may be large"

        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 request logs"
        if resp.status_code != 200:
            return f"❌ Unexpected status {resp.status_code}: {resp.text[:500]}"

        data = resp.json()
        if not isinstance(data, list) or len(data) == 0:
            return "No request history found."

        # Sort by startTime descending and take the most recent `limit` entries
        sorted_logs = sorted(data, key=lambda e: e.get("startTime", ""), reverse=True)
        
        # Filter out failed requests if show_failed is False
        if not show_failed:
            sorted_logs = [
                e for e in sorted_logs
                if not (
                    (e.get("metadata", {}) or {}).get("exception", "")
                    or (e.get("metadata", {}) or {}).get("status", "") == "failure"
                )
            ]
        
        recent = sorted_logs[:limit]
        total_count = len(data)
        filtered_note = " (failed requests hidden)" if not show_failed else ""

        lines = []
        lines.append(f"LiteLLM Request History — {self._base_url()}")
        lines.append(f"Showing {len(recent)} of {total_count} total requests{filtered_note} (most recent first)")
        lines.append("=" * 90)

        for i, entry in enumerate(recent, 1):
            # Timestamp
            start_time = entry.get("startTime", "?")[:19]  # YYYY-MM-DDTHH:MM:SS
            start_short = start_time[5:] if len(start_time) > 5 else start_time  # MM-DDTHH:MM:SS

            # Model — strip provider prefix for readability
            model = entry.get("model", "unknown")
            for prefix in ("gemini/", "openai/", "anthropic/", "mistral/", "cohere/"):
                if model.startswith(prefix):
                    model = model[len(prefix):]
                    break

            # Call type
            call_type = entry.get("call_type", "?")
            # Shorten common call types
            call_type = call_type.replace("acompletion", "chat").replace("aembedding", "embed").replace("aimage_generation", "image").replace("_", " ")

            # Status — check for errors
            metadata = entry.get("metadata", {}) or {}
            exception = metadata.get("exception", "")
            status_flag = metadata.get("status", "")
            is_failed = bool(exception) or status_flag == "failure"
            status = "❌ FAIL" if is_failed else "✅ OK"

            # Duration
            duration_ms = entry.get("request_duration_ms", 0) or 0
            if duration_ms >= 1000:
                duration_str = f"{duration_ms / 1000:.1f}s"
            else:
                duration_str = f"{duration_ms}ms"

            # Tokens
            prompt_tokens = entry.get("prompt_tokens", 0) or 0
            completion_tokens = entry.get("completion_tokens", 0) or 0
            total_tokens = entry.get("total_tokens", 0) or 0

            # Cache
            cache_read = entry.get("cache_read_input_tokens", 0) or 0

            # Spend
            spend = entry.get("spend", 0) or 0

            # Build line
            base = f"  {i:2d}. {start_short} | {model[:28]:28s} | {call_type[:10]:10s} | {status} | {duration_str:>7s}"

            # Token info
            if total_tokens > 0:
                # Compact token display
                if prompt_tokens >= 1000:
                    p_str = f"{prompt_tokens / 1000:.1f}k"
                else:
                    p_str = str(prompt_tokens)
                if completion_tokens >= 1000:
                    c_str = f"{completion_tokens / 1000:.1f}k"
                else:
                    c_str = str(completion_tokens)
                token_str = f"{p_str}→{c_str}"

                # Cache indicator
                if cache_read > 0:
                    cache_pct = (cache_read / prompt_tokens * 100) if prompt_tokens > 0 else 0
                    token_str += f" 💾{cache_pct:.0f}%"

                base += f" | {token_str}"

            # Cost
            if show_cost and spend > 0:
                base += f" | ${spend:.4f}"

            # Error info
            if is_failed:
                # Priority: exception string > error_information.error_message > generic status
                if exception:
                    err_short = exception[:80] + "…" if len(exception) > 80 else exception
                else:
                    err_info = metadata.get("error_information", {}) or {}
                    err_msg = err_info.get("error_message", "")
                    if err_msg:
                        err_short = err_msg[:80] + "…" if len(err_msg) > 80 else err_msg
                    else:
                        err_short = "status: failure"
                base += f"\n      └─ {err_short}"

            lines.append(base)

        # Summary
        if len(recent) > 0:
            recent_spend = sum(e.get("spend", 0) or 0 for e in recent)
            recent_prompt = sum(e.get("prompt_tokens", 0) or 0 for e in recent)
            recent_completion = sum(e.get("completion_tokens", 0) or 0 for e in recent)
            recent_cache = sum(e.get("cache_read_input_tokens", 0) or 0 for e in recent)
            recent_failed = sum(1 for e in recent if ((e.get("metadata", {}) or {}).get("exception")) or ((e.get("metadata", {}) or {}).get("status", "")) == "failure")

            lines.append("")
            lines.append(f"  Summary ({len(recent)} shown): ${recent_spend:.4f} spent | "
                        f"{recent_prompt:,} prompt + {recent_completion:,} completion tokens | "
                        f"{recent_failed} failed | "
                        f"cache hit {recent_cache / recent_prompt * 100:.0f}%" if recent_prompt > 0 and recent_cache > 0
                        else f"  Summary ({len(recent)} shown): ${recent_spend:.4f} spent | "
                             f"{recent_prompt:,} prompt + {recent_completion:,} completion tokens | "
                             f"{recent_failed} failed")

        return "\n".join(lines)