1  #!/usr/bin/env python3
  2  """Extract token counts and costs from Claude Code session JSONL.
  3  
  4  Usage: python3 scripts/session-metrics.py [session_jsonl_path]
  5  
  6  If no path given, looks for the most recent session in ~/.claude/sessions/.
  7  Outputs a summary suitable for pasting into ops/metrics.md.
  8  """
  9  
 10  import json
 11  import sys
 12  import os
 13  from pathlib import Path
 14  from datetime import datetime
 15  
 16  
 17  def find_latest_session():
 18      """Find the most recently modified session JSONL file."""
 19      # Claude Code stores session JSONL in project dirs
 20      claude_dir = Path.home() / ".claude" / "projects"
 21      if not claude_dir.exists():
 22          # Fallback to sessions dir
 23          claude_dir = Path.home() / ".claude" / "sessions"
 24      if not claude_dir.exists():
 25          print(f"Claude directory not found: {claude_dir}", file=sys.stderr)
 26          sys.exit(1)
 27      jsonl_files = list(claude_dir.rglob("*.jsonl"))
 28      # Exclude subagent files
 29      jsonl_files = [f for f in jsonl_files if "subagents" not in str(f)]
 30      if not jsonl_files:
 31          print("No session files found", file=sys.stderr)
 32          sys.exit(1)
 33      return max(jsonl_files, key=lambda f: f.stat().st_mtime)
 34  
 35  
 36  def parse_session(path):
 37      """Parse a session JSONL file and extract metrics."""
 38      total_input = 0
 39      total_output = 0
 40      total_cache_read = 0
 41      total_cache_write = 0
 42      messages = 0
 43      tool_uses = 0
 44      first_ts = None
 45      last_ts = None
 46  
 47      with open(path) as f:
 48          for line in f:
 49              line = line.strip()
 50              if not line:
 51                  continue
 52              try:
 53                  entry = json.loads(line)
 54              except json.JSONDecodeError:
 55                  continue
 56  
 57              # Extract timestamp
 58              ts = entry.get("timestamp") or entry.get("ts")
 59              if ts:
 60                  if first_ts is None:
 61                      first_ts = ts
 62                  last_ts = ts
 63  
 64              # Claude Code JSONL: usage lives in entry.message.usage
 65              msg = entry.get("message", {})
 66              if isinstance(msg, dict):
 67                  usage = msg.get("usage", {})
 68              else:
 69                  usage = entry.get("usage", {})
 70              if usage:
 71                  total_input += usage.get("input_tokens", 0)
 72                  total_output += usage.get("output_tokens", 0)
 73                  total_cache_read += usage.get("cache_read_input_tokens", 0)
 74                  total_cache_write += usage.get("cache_creation_input_tokens", 0)
 75  
 76              # Count messages and tool uses
 77              entry_type = entry.get("type")
 78              if entry_type == "assistant":
 79                  messages += 1
 80                  content = msg.get("content", []) if isinstance(msg, dict) else []
 81                  if isinstance(content, list):
 82                      tool_uses += sum(1 for c in content if isinstance(c, dict) and c.get("type") == "tool_use")
 83  
 84      return {
 85          "input_tokens": total_input,
 86          "output_tokens": total_output,
 87          "cache_read": total_cache_read,
 88          "cache_write": total_cache_write,
 89          "total_tokens": total_input + total_output,
 90          "messages": messages,
 91          "tool_uses": tool_uses,
 92          "first_ts": first_ts,
 93          "last_ts": last_ts,
 94      }
 95  
 96  
 97  def estimate_cost(metrics):
 98      """Estimate cost using Claude Opus pricing (approximate)."""
 99      # Opus pricing: $15/M input, $75/M output, $1.50/M cache read, $3.75/M cache write
100      input_cost = (metrics["input_tokens"] / 1_000_000) * 15.0
101      output_cost = (metrics["output_tokens"] / 1_000_000) * 75.0
102      cache_read_cost = (metrics["cache_read"] / 1_000_000) * 1.50
103      cache_write_cost = (metrics["cache_write"] / 1_000_000) * 3.75
104      return input_cost + output_cost + cache_read_cost + cache_write_cost
105  
106  
107  def main():
108      if len(sys.argv) > 1:
109          path = Path(sys.argv[1])
110      else:
111          path = find_latest_session()
112          print(f"Using session: {path}", file=sys.stderr)
113  
114      if not path.exists():
115          print(f"File not found: {path}", file=sys.stderr)
116          sys.exit(1)
117  
118      m = parse_session(path)
119      cost = estimate_cost(m)
120  
121      print(f"### Session Token Summary")
122      print(f"")
123      print(f"| Metric | Value |")
124      print(f"|--------|-------|")
125      print(f"| Input tokens | {m['input_tokens']:,} |")
126      print(f"| Output tokens | {m['output_tokens']:,} |")
127      print(f"| Cache read | {m['cache_read']:,} |")
128      print(f"| Cache write | {m['cache_write']:,} |")
129      print(f"| Total tokens | {m['total_tokens']:,} |")
130      print(f"| Assistant messages | {m['messages']} |")
131      print(f"| Tool uses | {m['tool_uses']} |")
132      print(f"| Estimated cost | ${cost:.2f} |")
133      if m["first_ts"] and m["last_ts"]:
134          print(f"| First timestamp | {m['first_ts']} |")
135          print(f"| Last timestamp | {m['last_ts']} |")
136  
137  
138  if __name__ == "__main__":
139      main()