Claude Code token usage analyzer — track cache reads, subagents, and session costs locally. Adapted from @kieranklaassen's original script (https://gist.github.com/kieranklaassen).

claude_token_usage.py

#!/usr/bin/env python3
"""
claude-token-usage.py

Analyzes Claude Code session files (~/.claude/projects/) and reports
token consumption broken down by project, session, and subagent.

Usage:
    python3 claude_token_usage.py                    # all time, all projects
    SINCE_DAYS=7 python3 claude_token_usage.py       # last 7 days
    PROJECT=myequal-ai-app python3 claude_token_usage.py  # single project

Output:
    - Console summary (sessions, totals, subagent count)
    - Markdown report at ~/claude-token-report/token_report.md
"""

import json
import os
import sys
from pathlib import Path
from collections import defaultdict
from datetime import datetime, timedelta, timezone

# ── Config ────────────────────────────────────────────────────────────────────

PROJECTS_DIR = Path.home() / ".claude" / "projects"
OUTPUT_DIR   = Path(os.environ.get("OUTPUT_DIR", str(Path.home() / "claude-token-report")))

SINCE_DAYS     = int(os.environ.get("SINCE_DAYS", "0")) or None
SINCE_DATE     = os.environ.get("SINCE_DATE")       # e.g. "2026-03-30"
PROJECT_FILTER = os.environ.get("PROJECT")          # e.g. "myequal-ai-app"

# ── Helpers ───────────────────────────────────────────────────────────────────

def fmt(n: int) -> str:
    return f"{n:,}"


def get_cutoff() -> datetime | None:
    if SINCE_DATE:
        return datetime.fromisoformat(SINCE_DATE).replace(tzinfo=timezone.utc)
    if SINCE_DAYS:
        return datetime.now(timezone.utc) - timedelta(days=SINCE_DAYS)
    return None


def in_range(session: dict, cutoff: datetime | None) -> bool:
    if not cutoff or not session["timestamp_start"]:
        return True
    try:
        ts = datetime.fromisoformat(session["timestamp_start"].replace("Z", "+00:00"))
        return ts >= cutoff
    except ValueError:
        return True


def readable_project_name(dir_name: str) -> str:
    """
    Strip the OS path prefix that Claude Code uses for project directories.
    e.g. "-Users-adit-StudioProjects-myequal-ai-app" → "myequal-ai-app"
    """
    parts = dir_name.lstrip("-").split("-")
    # Find where the actual project name starts — after the home directory segments
    home_parts = [p for p in str(Path.home()).split("/") if p]
    # Skip segments that match home path components
    i = 0
    for part in home_parts:
        if i < len(parts) and parts[i].lower() == part.lower():
            i += 1
    return "-".join(parts[i:]) or dir_name


# ── Parsing ───────────────────────────────────────────────────────────────────

def extract_text(content) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for item in content:
            if isinstance(item, dict) and item.get("type") == "text":
                parts.append(item.get("text", ""))
            elif isinstance(item, str):
                parts.append(item)
        return "\n".join(parts).strip()
    return ""


def is_human_message(obj: dict) -> bool:
    content = obj.get("message", {}).get("content", "")
    if isinstance(content, list):
        types = [i.get("type") for i in content if isinstance(i, dict)]
        if types and all(t == "tool_result" for t in types):
            return False
    return True


def parse_session(jsonl_path: Path, is_subagent: bool = False) -> dict | None:
    usage        = defaultdict(int)
    models       = defaultdict(int)
    prompts      = []
    session_id   = None
    timestamp    = None
    subagents    = []

    try:
        lines = jsonl_path.read_text().splitlines()
    except Exception:
        return None

    for line in lines:
        try:
            obj = json.loads(line)
        except json.JSONDecodeError:
            continue

        if not timestamp and obj.get("timestamp"):
            timestamp = obj["timestamp"]
        if not session_id and obj.get("sessionId"):
            session_id = obj["sessionId"]

        if obj.get("type") == "assistant":
            u = obj.get("message", {}).get("usage", {})
            usage["input"]          += u.get("input_tokens", 0)
            usage["cache_create"]   += u.get("cache_creation_input_tokens", 0)
            usage["cache_read"]     += u.get("cache_read_input_tokens", 0)
            usage["output"]         += u.get("output_tokens", 0)
            model = obj.get("message", {}).get("model", "unknown")
            if model:
                models[model] += 1

        elif obj.get("type") == "user":
            text = extract_text(obj.get("message", {}).get("content", ""))
            if text and not obj.get("isSidechain") and is_human_message(obj):
                prompts.append({
                    "text":      text,
                    "timestamp": obj.get("timestamp"),
                })

    # Discover subagent session files
    session_dir  = jsonl_path.parent / jsonl_path.stem
    subagents_dir = session_dir / "subagents"
    if subagents_dir.is_dir():
        for sub_file in subagents_dir.glob("*.jsonl"):
            sub = parse_session(sub_file, is_subagent=True)
            if sub:
                sub["filename"] = sub_file.name
                subagents.append(sub)

    total = sum(usage.values())
    return {
        "session_id":      session_id or jsonl_path.stem,
        "timestamp_start": timestamp,
        "is_subagent":     is_subagent,
        "usage":           dict(usage),
        "models":          dict(models),
        "total":           total,
        "prompts":         prompts,
        "subagents":       subagents,
    }


# ── Aggregation ───────────────────────────────────────────────────────────────

def load_projects() -> dict[str, list]:
    projects = defaultdict(list)
    cutoff   = get_cutoff()

    for project_dir in sorted(PROJECTS_DIR.iterdir()):
        if not project_dir.is_dir():
            continue
        name = readable_project_name(project_dir.name)
        if PROJECT_FILTER and PROJECT_FILTER not in project_dir.name and PROJECT_FILTER not in name:
            continue
        for jsonl_file in sorted(project_dir.glob("*.jsonl")):
            session = parse_session(jsonl_file)
            if session and session["total"] > 0 and in_range(session, cutoff):
                projects[name].append(session)

    return dict(projects)


def project_summary(sessions: list) -> dict:
    totals         = defaultdict(int)
    subagent_total = 0
    subagent_count = 0
    for s in sessions:
        for k, v in s["usage"].items():
            totals[k] += v
        for sub in s["subagents"]:
            subagent_total += sub["total"]
            subagent_count += 1
    return {
        "sessions":        len(sessions),
        "usage":           dict(totals),
        "total":           sum(totals.values()),
        "subagent_total":  subagent_total,
        "subagent_count":  subagent_count,
    }


def top_sessions(projects: dict, n: int = 20) -> list:
    rows = [(proj, s) for proj, sessions in projects.items() for s in sessions]
    return sorted(rows, key=lambda x: x[1]["total"], reverse=True)[:n]


def top_subagents(projects: dict, n: int = 20) -> list:
    rows = []
    for proj, sessions in projects.items():
        for s in sessions:
            for sub in s["subagents"]:
                rows.append((proj, s["session_id"], sub))
    return sorted(rows, key=lambda x: x[2]["total"], reverse=True)[:n]


# ── Output ────────────────────────────────────────────────────────────────────

def print_summary(projects: dict) -> None:
    summaries = {p: project_summary(s) for p, s in projects.items()}
    grand_total    = sum(v["total"] for v in summaries.values())
    total_sessions = sum(v["sessions"] for v in summaries.values())

    print(f"\nTotal: {fmt(grand_total)} tokens across {total_sessions} sessions in {len(summaries)} projects\n")
    print(f"{'Project':<50} {'Sessions':>8} {'Total Tokens':>14} {'Subagents':>10}")
    print("-" * 86)
    for proj, s in sorted(summaries.items(), key=lambda x: x[1]["total"], reverse=True)[:30]:
        print(f"{proj:<50} {s['sessions']:>8,} {fmt(s['total']):>14} {s['subagent_count']:>10,}")

    print("\nTop 10 costliest sessions:")
    for proj, session in top_sessions(projects, n=10):
        ts    = session["timestamp_start"][:10] if session["timestamp_start"] else "?"
        first = session["prompts"][0]["text"][:80].replace("\n", " ") if session["prompts"] else ""
        print(f"  [{ts}] {proj}: {fmt(session['total'])} — {first}")


def write_report(projects: dict) -> Path:
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    report_path = OUTPUT_DIR / "token_report.md"
    summaries   = {p: project_summary(s) for p, s in projects.items()}
    cutoff      = get_cutoff()
    date_range  = f"Since {cutoff.strftime('%Y-%m-%d')}" if cutoff else "All time"

    lines = [
        "# Claude Code Token Usage",
        f"\nGenerated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | Range: {date_range}\n",
    ]

    # Grand totals
    grand = defaultdict(int)
    for s in summaries.values():
        for k, v in s["usage"].items():
            grand[k] += v
    grand_total = sum(grand.values())
    total_subs  = sum(s["subagent_count"] for s in summaries.values())
    sub_tokens  = sum(s["subagent_total"] for s in summaries.values())

    lines += [
        "## Totals\n",
        f"- **Projects**: {len(summaries)}",
        f"- **Sessions**: {sum(s['sessions'] for s in summaries.values()):,}",
        f"- **Total tokens**: {fmt(grand_total)}",
        f"  - Input:          {fmt(grand['input'])}",
        f"  - Cache creation: {fmt(grand['cache_create'])}",
        f"  - Cache read:     {fmt(grand['cache_read'])}  ({grand['cache_read']*100//grand_total if grand_total else 0}%)",
        f"  - Output:         {fmt(grand['output'])}",
        f"- **Subagents**: {total_subs} sessions ({fmt(sub_tokens)} tokens)",
        "",
    ]

    # Per-project
    lines += ["## By Project\n",
              "| Project | Sessions | Total | Cache Read % | Subagents |",
              "|---------|----------|-------|--------------|-----------|"]
    for proj, s in sorted(summaries.items(), key=lambda x: x[1]["total"], reverse=True):
        cache_pct = s["usage"].get("cache_read", 0) * 100 // s["total"] if s["total"] else 0
        lines.append(
            f"| {proj} | {s['sessions']} | {fmt(s['total'])} | {cache_pct}% | {s['subagent_count']} ({fmt(s['subagent_total'])}) |"
        )
    lines.append("")

    # Costliest sessions
    lines.append("## Costliest Sessions\n")
    for i, (proj, session) in enumerate(top_sessions(projects, n=20), 1):
        u  = session["usage"]
        ts = session["timestamp_start"][:19].replace("T", " ") if session["timestamp_start"] else "?"
        cache_pct = u.get("cache_read", 0) * 100 // session["total"] if session["total"] else 0
        lines += [
            f"### {i}. {proj} — {fmt(session['total'])} tokens",
            f"- **Session**: `{session['session_id']}`",
            f"- **Started**: {ts}",
            f"- **Cache read**: {fmt(u.get('cache_read', 0))} ({cache_pct}%)",
            f"- **Output**: {fmt(u.get('output', 0))}",
            f"- **Subagents**: {len(session['subagents'])}",
        ]
        if session.get("models"):
            model_str = ", ".join(f"{m} ({c}x)" for m, c in sorted(session["models"].items(), key=lambda x: -x[1]))
            lines.append(f"- **Models**: {model_str}")
        if session["prompts"]:
            first = session["prompts"][0]["text"][:300].replace("\n", " ")
            lines.append(f"- **First prompt**: {first}")
        lines.append("")

    # Costliest subagents
    lines += ["## Costliest Subagents\n",
              "| # | Project | Parent Session | File | Total | Output |",
              "|---|---------|----------------|------|-------|--------|"]
    for i, (proj, parent_id, sub) in enumerate(top_subagents(projects, n=20), 1):
        lines.append(
            f"| {i} | {proj} | `{parent_id[:8]}` | `{sub.get('filename','?')}` "
            f"| {fmt(sub['total'])} | {fmt(sub['usage'].get('output', 0))} |"
        )
    lines.append("")

    report_path.write_text("\n".join(lines))
    print(f"Report: {report_path}")
    return report_path


# ── Main ──────────────────────────────────────────────────────────────────────

def main():
    print("Scanning ~/.claude/projects/ ...")
    projects = load_projects()
    if not projects:
        print("No sessions found. Check PROJECTS_DIR or SINCE_DAYS filter.")
        sys.exit(0)

    print(f"Found {len(projects)} project(s)\n")
    print_summary(projects)
    write_report(projects)


if __name__ == "__main__":
    main()