diff --git a/mcp/leadtech_bmad_mcp/scripts/mcp_token_report.py b/mcp/leadtech_bmad_mcp/scripts/mcp_token_report.py new file mode 100755 index 0000000..364cc62 --- /dev/null +++ b/mcp/leadtech_bmad_mcp/scripts/mcp_token_report.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +"""Mesure la consommation tokens des appels MCP leadtech dans un transcript Claude Code. + +Usage: + python3 mcp_token_report.py + python3 mcp_token_report.py --project RL799_V2 # dernier transcript du projet + +Deux niveaux mesurés: + - taille des RESULTATS d'outils leadtech (ce que le MCP injecte dans le contexte) + - total tokens de session (input/output/cache) pour situer la part MCP +""" +import json, sys, glob, os +from pathlib import Path + +LEADTECH_TOOLS = {"get_guidance","validate_plan","validate_patch","emit_checklist", + "propose_capitalization","triage_capitalization","route_to_project_memory"} + +def is_leadtech(name: str) -> bool: + n = (name or "").lower() + return "leadtech" in n or any(t in n for t in LEADTECH_TOOLS) + +def approx_tokens(s: str) -> int: + return round(len(s) / 4) + +def resolve_file(arg: str) -> str: + if arg.startswith("--project"): + proj = sys.argv[sys.argv.index("--project")+1] + base = Path.home()/".claude"/"projects" + cands = [] + for d in base.glob("*"): + if proj.replace("_","-").replace("/","-").lower() in d.name.lower(): + cands += glob.glob(str(d/"*.jsonl")) + if not cands: sys.exit(f"Aucun transcript pour projet {proj}") + return max(cands, key=os.path.getmtime) + return arg + +def main(): + if len(sys.argv) < 2: sys.exit(__doc__) + f = resolve_file(sys.argv[1]) + calls = [] # (name, args_len) + results = {} # tool_use_id -> result_chars + id_to_name = {} + tot_in = tot_out = tot_cache_r = tot_cache_w = 0 + + for line in open(f): + try: d = json.loads(line) + except: continue + msg = d.get("message", {}) + if not isinstance(msg, dict): continue + u = msg.get("usage") + if u: + tot_in += u.get("input_tokens",0) + tot_out += u.get("output_tokens",0) + tot_cache_r += u.get("cache_read_input_tokens",0) + tot_cache_w += u.get("cache_creation_input_tokens",0) + for c in (msg.get("content") or []): + if not isinstance(c, dict): continue + if c.get("type") == "tool_use" and is_leadtech(c.get("name","")): + calls.append((c.get("name"), approx_tokens(json.dumps(c.get("input",{}),ensure_ascii=False)))) + id_to_name[c.get("id")] = c.get("name") + if c.get("type") == "tool_result": + rid = c.get("tool_use_id") + if rid in id_to_name: + content = c.get("content") + txt = content if isinstance(content,str) else json.dumps(content,ensure_ascii=False) + results[rid] = approx_tokens(txt) + + print(f"# Transcript: {Path(f).name}\n") + print(f"Total session : in={tot_in} out={tot_out} cache_read={tot_cache_r} cache_write={tot_cache_w}") + print(f"Appels MCP leadtech : {len(calls)}") + if not calls: + print(" (aucun appel MCP leadtech dans cette session)") + return + by_tool = {} + for name, _ in calls: + by_tool[name] = by_tool.get(name,0)+1 + res_tokens = sum(results.values()) + print(f" par tool: " + ", ".join(f"{k}×{v}" for k,v in by_tool.items())) + print(f" tokens des résultats MCP injectés : ~{res_tokens}") + if tot_in: + print(f" part estimée des résultats MCP / input non caché : ~{round(100*res_tokens/max(tot_in,1))}%") + print(f"\n détail résultats par appel:") + for rid, name in id_to_name.items(): + print(f" {name:<24} ~{results.get(rid,0)} tok") + +if __name__ == "__main__": + main()