#!/usr/bin/env python3 """ Claude Console / Anthropic API Usage Cost Extractor ===================================================== Extracts actual API usage costs from Anthropic. STRATEGY (Native-First Protocol): 1. Try Admin API (/v1/organizations/usage_report/messages + /v1/organizations/cost_report) - Requires an Admin API key (sk-ant-admin...) — provision at: https://console.anthropic.com/settings/admin-keys 2. Fallback: Playwright browser agent against console.anthropic.com/usage Output: /mnt/e/genesis-system/data/context_state/claude_api_usage.json Usage: python3 scripts/claude_console_usage.py python3 scripts/claude_console_usage.py --headless false # show browser python3 scripts/claude_console_usage.py --days 30 # last 30 days python3 scripts/claude_console_usage.py --admin-key sk-ant-admin-xxx """ import argparse import json import os import sys import time from datetime import datetime, timezone, timedelta from pathlib import Path # Add genesis core to path for secrets loader sys.path.insert(0, str(Path(__file__).parent.parent / "core")) OUTPUT_PATH = Path("/mnt/e/genesis-system/data/context_state/claude_api_usage.json") # Claude model pricing in USD per million tokens (as of Feb 2026) MODEL_PRICING = { "claude-opus-4-6": {"input": 15.00, "output": 75.00, "cache_write": 18.75, "cache_read": 1.50}, "claude-sonnet-4-6": {"input": 3.00, "output": 15.00, "cache_write": 3.75, "cache_read": 0.30}, "claude-haiku-4-5": {"input": 0.80, "output": 4.00, "cache_write": 1.00, "cache_read": 0.08}, "claude-3-5-sonnet-20241022": {"input": 3.00, "output": 15.00, "cache_write": 3.75, "cache_read": 0.30}, "claude-3-5-haiku-20241022": {"input": 0.80, "output": 4.00, "cache_write": 1.00, "cache_read": 0.08}, "claude-3-opus-20240229": {"input": 15.00, "output": 75.00, "cache_write": 18.75, "cache_read": 1.50}, # Catch-all patterns checked via startswith in estimate_cost() } def get_secrets(): """Load API keys from secrets_loader or environment.""" try: from secrets_loader import get_anthropic_api_key standard_key = get_anthropic_api_key() except ImportError: standard_key = os.environ.get("ANTHROPIC_API_KEY") admin_key = os.environ.get("ANTHROPIC_ADMIN_KEY") or os.environ.get("ANTHROPIC_ADMIN_API_KEY") console_email = os.environ.get("ANTHROPIC_CONSOLE_EMAIL") console_password = os.environ.get("ANTHROPIC_CONSOLE_PASSWORD") # Try to load from secrets.env if not in environment secrets_env = Path("/mnt/e/genesis-system/config/secrets.env") if secrets_env.exists(): with open(secrets_env) as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue if "=" in line: key, _, val = line.partition("=") key = key.strip() val = val.strip().strip('"').strip("'") if key == "ANTHROPIC_ADMIN_KEY" and not admin_key: admin_key = val elif key == "ANTHROPIC_CONSOLE_EMAIL" and not console_email: console_email = val elif key == "ANTHROPIC_CONSOLE_PASSWORD" and not console_password: console_password = val return { "standard_key": standard_key, "admin_key": admin_key, "console_email": console_email, "console_password": console_password, } def estimate_cost(model: str, input_tokens: int, output_tokens: int, cache_creation_tokens: int = 0, cache_read_tokens: int = 0) -> float: """Estimate USD cost from token counts using known pricing.""" # Find pricing: exact match first, then prefix match pricing = MODEL_PRICING.get(model) if not pricing: for prefix, p in MODEL_PRICING.items(): if model.startswith(prefix.split("-20")[0]): # match base model name pricing = p break if not pricing: # Default to Sonnet pricing if unknown pricing = MODEL_PRICING["claude-sonnet-4-6"] cost = ( (input_tokens / 1_000_000) * pricing["input"] + (output_tokens / 1_000_000) * pricing["output"] + (cache_creation_tokens / 1_000_000) * pricing["cache_write"] + (cache_read_tokens / 1_000_000) * pricing["cache_read"] ) return round(cost, 6) # ───────────────────────────────────────────────────────────────── # STRATEGY 1: Native Admin API # ───────────────────────────────────────────────────────────────── def try_admin_api(admin_key: str, days: int = 30) -> dict | None: """ Use Anthropic's Admin API to fetch usage + cost data. Requires an Admin API key (sk-ant-admin...). Docs: https://platform.claude.com/docs/en/api/usage-cost-api """ import urllib.request import urllib.error if not admin_key or not admin_key.startswith("sk-ant-admin"): return None print(f"[Strategy 1] Trying Admin API with key {admin_key[:20]}...") end_dt = datetime.now(timezone.utc) start_dt = end_dt - timedelta(days=days) starting_at = start_dt.strftime("%Y-%m-%dT00:00:00Z") ending_at = end_dt.strftime("%Y-%m-%dT23:59:59Z") headers = { "x-api-key": admin_key, "anthropic-version": "2023-06-01", "content-type": "application/json", } def api_get(url: str) -> dict | None: req = urllib.request.Request(url, headers=headers) try: with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") print(f" HTTP {e.code}: {body[:200]}") return None except Exception as e: print(f" Error: {e}") return None # Fetch usage report grouped by model usage_url = ( f"https://api.anthropic.com/v1/organizations/usage_report/messages" f"?starting_at={starting_at}&ending_at={ending_at}" f"&group_by[]=model&bucket_width=1d" ) usage_data = api_get(usage_url) if not usage_data: return None # Fetch cost report cost_url = ( f"https://api.anthropic.com/v1/organizations/cost_report" f"?starting_at={starting_at}&ending_at={ending_at}" f"&group_by[]=description" ) cost_data = api_get(cost_url) # Aggregate usage by model by_model = {} total_input = 0 total_output = 0 total_cache_creation = 0 total_cache_read = 0 for bucket in usage_data.get("data", []): model = bucket.get("model") or "unknown" inp = bucket.get("input_tokens", 0) out = bucket.get("output_tokens", 0) cc = bucket.get("cache_creation_input_tokens", 0) cr = bucket.get("cache_read_input_tokens", 0) if model not in by_model: by_model[model] = {"input_tokens": 0, "output_tokens": 0, "cache_creation_tokens": 0, "cache_read_tokens": 0, "estimated_cost_usd": 0.0} by_model[model]["input_tokens"] += inp by_model[model]["output_tokens"] += out by_model[model]["cache_creation_tokens"] += cc by_model[model]["cache_read_tokens"] += cr by_model[model]["estimated_cost_usd"] += estimate_cost(model, inp, out, cc, cr) total_input += inp total_output += out total_cache_creation += cc total_cache_read += cr # Total actual cost from cost API (if available) total_cost_usd = None if cost_data: cents_total = sum( float(item.get("cost", 0)) for item in cost_data.get("data", []) ) total_cost_usd = round(cents_total / 100, 4) # API returns cents as decimal strings estimated_total = round( sum(m["estimated_cost_usd"] for m in by_model.values()), 4 ) result = { "source": "anthropic_admin_api", "extracted_at": datetime.now(timezone.utc).isoformat(), "period": { "start": starting_at, "end": ending_at, "days": days, }, "totals": { "input_tokens": total_input, "output_tokens": total_output, "cache_creation_tokens": total_cache_creation, "cache_read_tokens": total_cache_read, "total_tokens": total_input + total_output + total_cache_creation + total_cache_read, "total_cost_usd_actual": total_cost_usd, "total_cost_usd_estimated": estimated_total, }, "by_model": by_model, "raw_cost_response": cost_data, } print(f" [OK] Admin API success. Total tokens: {result['totals']['total_tokens']:,}") return result # ───────────────────────────────────────────────────────────────── # STRATEGY 2: Playwright Browser Agent # ───────────────────────────────────────────────────────────────── def try_playwright_browser(email: str, password: str, headless: bool = True) -> dict | None: """ Log into console.anthropic.com and scrape usage/billing data. Falls back to this when no admin API key is available. """ if not email or not password: print("[Strategy 2] Skipping Playwright — no console credentials configured.") print(" Set ANTHROPIC_CONSOLE_EMAIL and ANTHROPIC_CONSOLE_PASSWORD in secrets.env") return None try: from playwright.sync_api import sync_playwright, TimeoutError as PWTimeout except ImportError: print("[Strategy 2] Playwright not installed. Run: pip install playwright && playwright install chromium") return None print(f"[Strategy 2] Launching Playwright browser (headless={headless})...") result = {} with sync_playwright() as p: browser = p.chromium.launch( headless=headless, args=["--no-sandbox", "--disable-dev-shm-usage"], ) context = browser.new_context( viewport={"width": 1280, "height": 900}, user_agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36", ) page = context.new_page() try: # ── Step 1: Navigate to login ────────────────────────────── print(" Navigating to console.anthropic.com...") page.goto("https://console.anthropic.com/login", wait_until="networkidle", timeout=30000) time.sleep(1) # ── Step 2: Fill login form ──────────────────────────────── print(f" Logging in as {email}...") # Email field email_sel = 'input[type="email"], input[name="email"], input[placeholder*="email" i]' page.wait_for_selector(email_sel, timeout=15000) page.fill(email_sel, email) # Check if there's a "Continue" button before password appears continue_btn = page.query_selector('button:has-text("Continue"), button[type="submit"]') if continue_btn: continue_btn.click() time.sleep(1) # Password field pass_sel = 'input[type="password"], input[name="password"]' try: page.wait_for_selector(pass_sel, timeout=10000) page.fill(pass_sel, password) except PWTimeout: pass # Password may appear after email submit # Submit login submit_sel = 'button[type="submit"], button:has-text("Log in"), button:has-text("Sign in")' page.wait_for_selector(submit_sel, timeout=10000) page.click(submit_sel) # Wait for navigation after login try: page.wait_for_url("**/console.anthropic.com/**", timeout=20000) except PWTimeout: # Check for 2FA or error current_url = page.url print(f" Current URL after login attempt: {current_url}") if "login" in current_url or "auth" in current_url: # Check for error messages error_el = page.query_selector('.error, [role="alert"], .alert') if error_el: print(f" Login error: {error_el.inner_text()[:200]}") print(" WARNING: May require 2FA or SSO. Trying to continue anyway...") print(f" Post-login URL: {page.url}") time.sleep(2) # ── Step 3: Navigate to Usage page ──────────────────────── print(" Navigating to usage page...") page.goto("https://console.anthropic.com/usage", wait_until="networkidle", timeout=30000) time.sleep(2) usage_data = _scrape_usage_page(page) result.update(usage_data) # ── Step 4: Navigate to Cost/Billing page ───────────────── print(" Navigating to billing/cost page...") page.goto("https://console.anthropic.com/settings/billing", wait_until="networkidle", timeout=30000) time.sleep(2) billing_data = _scrape_billing_page(page) result.update(billing_data) # ── Step 5: Try /cost page ───────────────────────────────── try: page.goto("https://console.anthropic.com/cost", wait_until="networkidle", timeout=15000) time.sleep(2) cost_data = _scrape_cost_page(page) result.update(cost_data) except Exception: pass except Exception as e: print(f" Playwright error: {e}") # Try to save a screenshot for debugging try: screenshot_path = "/mnt/e/genesis-system/data/context_state/console_debug.png" page.screenshot(path=screenshot_path) print(f" Screenshot saved to: {screenshot_path}") except Exception: pass finally: browser.close() if not result: return None result["source"] = "playwright_browser" result["extracted_at"] = datetime.now(timezone.utc).isoformat() return result def _scrape_usage_page(page) -> dict: """Extract token usage data from /usage page.""" data = {"usage_page": {}} try: # Wait for content to load page.wait_for_load_state("networkidle", timeout=10000) # Grab all visible text for broad extraction body_text = page.inner_text("body") or "" # Look for common usage table patterns # Try to find data tables with model breakdowns tables = page.query_selector_all("table") table_data = [] for table in tables: rows = table.query_selector_all("tr") for row in rows: cells = row.query_selector_all("td, th") row_data = [cell.inner_text().strip() for cell in cells] if row_data and any(row_data): table_data.append(row_data) # Look for specific usage metric elements usage_elements = {} selectors_to_try = [ ('[data-testid*="usage"]', "usage_widget"), ('[class*="usage"]', "usage_class"), ('[class*="metric"]', "metric_class"), ('[class*="cost"]', "cost_class"), ('[class*="token"]', "token_class"), ] for sel, label in selectors_to_try: elements = page.query_selector_all(sel) if elements: texts = [el.inner_text().strip() for el in elements[:10] if el.inner_text().strip()] if texts: usage_elements[label] = texts data["usage_page"] = { "url": page.url, "tables": table_data[:20], # first 20 rows "elements": usage_elements, "page_text_preview": body_text[:3000], } print(f" Scraped usage page: {len(table_data)} table rows, text length {len(body_text)}") except Exception as e: data["usage_page"]["error"] = str(e) return data def _scrape_billing_page(page) -> dict: """Extract billing/cost data from settings/billing page.""" data = {"billing_page": {}} try: page.wait_for_load_state("networkidle", timeout=10000) body_text = page.inner_text("body") or "" # Look for credit balance, spending amounts import re dollar_amounts = re.findall(r'\$[\d,]+\.?\d*', body_text) credit_mentions = re.findall(r'(?i)credit[s]?\s*[:\-]?\s*\$?[\d,]+\.?\d*', body_text) tables = page.query_selector_all("table") table_data = [] for table in tables: rows = table.query_selector_all("tr") for row in rows: cells = row.query_selector_all("td, th") row_text = [c.inner_text().strip() for c in cells] if row_text and any(row_text): table_data.append(row_text) data["billing_page"] = { "url": page.url, "dollar_amounts_found": dollar_amounts[:20], "credit_mentions": credit_mentions[:10], "tables": table_data[:20], "page_text_preview": body_text[:2000], } print(f" Scraped billing page: {len(dollar_amounts)} $ amounts found") except Exception as e: data["billing_page"]["error"] = str(e) return data def _scrape_cost_page(page) -> dict: """Extract data from /cost page.""" data = {"cost_page": {}} try: page.wait_for_load_state("networkidle", timeout=10000) body_text = page.inner_text("body") or "" tables = page.query_selector_all("table") table_data = [] for table in tables: rows = table.query_selector_all("tr") for row in rows: cells = row.query_selector_all("td, th") row_text = [c.inner_text().strip() for c in cells] if row_text and any(row_text): table_data.append(row_text) data["cost_page"] = { "url": page.url, "tables": table_data[:30], "page_text_preview": body_text[:2000], } print(f" Scraped cost page: {len(table_data)} table rows") except Exception as e: data["cost_page"]["error"] = str(e) return data # ───────────────────────────────────────────────────────────────── # STRATEGY 3: Local observability estimation from events.jsonl # ───────────────────────────────────────────────────────────────── def try_local_observability() -> dict | None: """ Estimate usage from the local observability events log. This data is less accurate (no token counts) but always available. The observability_logger.py tracks tool calls but NOT token usage directly. """ events_file = Path("/mnt/e/genesis-system/data/observability/events.jsonl") metrics_file = Path("/mnt/e/genesis-system/data/observability/metrics.json") if not events_file.exists() and not metrics_file.exists(): return None print("[Strategy 3] Reading local observability data...") data = { "source": "local_observability_estimate", "extracted_at": datetime.now(timezone.utc).isoformat(), "note": ( "Token/cost data NOT available from local observability. " "This shows tool call counts only. " "For actual costs, provision an Admin API key at " "https://console.anthropic.com/settings/admin-keys" ), } # Load metrics summary if metrics_file.exists(): try: with open(metrics_file) as f: content = f.read().strip() # metrics.json may contain concatenated JSON objects (JSONL-like) — take last valid one if content: # Try single JSON first try: metrics = json.loads(content) except json.JSONDecodeError: # Handle trailing extra brace (known corruption: "}}") or JSONL concat # Strip trailing extra chars until valid JSON for trim in range(1, 10): try: metrics = json.loads(content[:-trim]) break except json.JSONDecodeError: continue else: metrics = {} data["tool_call_metrics"] = metrics print(f" Total tool calls tracked: {metrics.get('total_tool_calls', 0)}") except Exception as e: print(f" Warning: Could not parse metrics file: {e}") data["tool_call_metrics"] = {} # Count events by type from JSONL if events_file.exists(): event_counts = {} session_ids = set() recent_events = [] with open(events_file) as f: for line in f: line = line.strip() if not line: continue try: event = json.loads(line) etype = event.get("event_type", "unknown") event_counts[etype] = event_counts.get(etype, 0) + 1 sess = event.get("agent", {}).get("session_id", "") if sess and sess != "unknown": session_ids.add(sess) recent_events.append(event) except json.JSONDecodeError: pass # Keep only last 5 events for preview recent_events = recent_events[-5:] data["event_summary"] = { "total_events": sum(event_counts.values()), "by_type": event_counts, "unique_sessions": len(session_ids), "recent_events": recent_events, } # Cost estimation note data["cost_estimation"] = { "method": "UNAVAILABLE — observability logger tracks tool calls, not token counts", "action_required": ( "To get real cost data:\n" "1. Go to https://console.anthropic.com/settings/admin-keys\n" "2. Create an Admin API key (starts with sk-ant-admin...)\n" "3. Add to secrets.env: ANTHROPIC_ADMIN_KEY=sk-ant-admin-xxx\n" "4. Re-run this script" ), "console_url": "https://console.anthropic.com/usage", "billing_url": "https://console.anthropic.com/settings/billing", } return data # ───────────────────────────────────────────────────────────────── # Main orchestrator # ───────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Extract Claude API usage costs from Anthropic" ) parser.add_argument("--days", type=int, default=30, help="Number of days to look back (default: 30)") parser.add_argument("--headless", type=str, default="true", help="Run browser headless (default: true)") parser.add_argument("--admin-key", type=str, default=None, help="Override Anthropic Admin API key (sk-ant-admin...)") parser.add_argument("--no-browser", action="store_true", help="Skip Playwright browser fallback") parser.add_argument("--output", type=str, default=str(OUTPUT_PATH), help=f"Output JSON path (default: {OUTPUT_PATH})") args = parser.parse_args() headless = args.headless.lower() not in ("false", "0", "no") output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) secrets = get_secrets() admin_key = args.admin_key or secrets["admin_key"] print("=" * 60) print("Claude Console Usage Cost Extractor") print("=" * 60) print(f"Period: last {args.days} days") print(f"Admin key: {'SET (' + admin_key[:20] + '...)' if admin_key else 'NOT SET'}") print(f"Console email: {secrets['console_email'] or 'NOT SET'}") print(f"Output: {output_path}") print() result = None # ── Strategy 1: Admin API (best, most accurate) ────────────── if admin_key: result = try_admin_api(admin_key, days=args.days) # ── Strategy 2: Playwright browser scrape ──────────────────── if result is None and not args.no_browser: result = try_playwright_browser( email=secrets["console_email"], password=secrets["console_password"], headless=headless, ) # ── Strategy 3: Local observability fallback ───────────────── if result is None: result = try_local_observability() if result is None: result = { "source": "none", "extracted_at": datetime.now(timezone.utc).isoformat(), "error": "All strategies failed. No admin key, no browser credentials, no local data.", "action_required": { "admin_api": ( "Provision Admin API key at " "https://console.anthropic.com/settings/admin-keys " "then add ANTHROPIC_ADMIN_KEY=sk-ant-admin-xxx to " "/mnt/e/genesis-system/config/secrets.env" ), "browser": ( "Add ANTHROPIC_CONSOLE_EMAIL and ANTHROPIC_CONSOLE_PASSWORD " "to /mnt/e/genesis-system/config/secrets.env" ), "manual": "View costs at https://console.anthropic.com/settings/billing", }, } # Save output with open(output_path, "w") as f: json.dump(result, f, indent=2, default=str) print() print("=" * 60) print(f"Result saved to: {output_path}") print(f"Source strategy: {result.get('source', 'unknown')}") # Print summary if "totals" in result: t = result["totals"] print() print("USAGE SUMMARY:") print(f" Total tokens: {t.get('total_tokens', 0):>15,}") print(f" Input tokens: {t.get('input_tokens', 0):>15,}") print(f" Output tokens: {t.get('output_tokens', 0):>15,}") print(f" Cache creation: {t.get('cache_creation_tokens', 0):>15,}") print(f" Cache read: {t.get('cache_read_tokens', 0):>15,}") print() actual = t.get("total_cost_usd_actual") estimated = t.get("total_cost_usd_estimated") if actual is not None: print(f" ACTUAL COST (USD): ${actual:>13.4f}") if estimated is not None: print(f" ESTIMATED COST: ${estimated:>13.4f}") print() if "by_model" in result: print(" BY MODEL:") for model, stats in sorted(result["by_model"].items()): print(f" {model}:") print(f" tokens: {stats.get('input_tokens',0) + stats.get('output_tokens',0):,}") print(f" est. cost: ${stats.get('estimated_cost_usd', 0):.4f}") elif "tool_call_metrics" in result: m = result["tool_call_metrics"] print() print("TOOL CALL SUMMARY (local observability — no token counts):") print(f" Total tool calls: {m.get('total_tool_calls', 0)}") print(f" Successful: {m.get('success_count', 0)}") print(f" Failed: {m.get('failure_count', 0)}") if "by_tool" in m: top = sorted(m["by_tool"].items(), key=lambda x: x[1], reverse=True)[:5] print(" Top tools:", ", ".join(f"{t}({c})" for t, c in top)) print() print(" NOTE: Actual API costs require an Admin API key.") print(" Action: https://console.anthropic.com/settings/admin-keys") elif "billing_page" in result or "usage_page" in result: print() print("BROWSER SCRAPE RESULTS:") if "billing_page" in result: bp = result["billing_page"] if "dollar_amounts_found" in bp: print(f" Dollar amounts found: {bp['dollar_amounts_found'][:10]}") if "usage_page" in result: up = result["usage_page"] tables = up.get("tables", []) if tables: print(f" Usage table rows: {len(tables)}") for row in tables[:5]: print(f" {row}") print("=" * 60) return result if __name__ == "__main__": main()