Pbeslin/parse mcp history

changelog.d/20260512_132545_paul.beslin.ext_nhi_1628.md

@@ -0,0 +1,42 @@
+<!--
+A new scriv changelog fragment.
+
+Uncomment the section that is right (remove the HTML comment wrapper).
+For top level release notes, leave all the headers commented out.
+-->
+
+<!--
+### Removed
+
+- A bullet item for the Removed category.
+
+-->
+
+### Added
+
+- `ggshield ai discover --history` backfills historical MCP tool calls to GitGuardian (parsed from `~/.claude/projects/*/*.jsonl`). The API deduplicates events via idempotency keys, so reruns are safe.
+
+<!--
+### Changed
+
+- A bullet item for the Changed category.
+
+-->
+<!--
+### Deprecated
+
+- A bullet item for the Deprecated category.
+
+-->
+<!--
+### Fixed
+
+- A bullet item for the Fixed category.
+
+-->
+<!--
+### Security
+
+- A bullet item for the Security category.
+
+-->

ggshield/cmd/ai/discover.py

@@ -21,6 +21,7 @@
     save_discovery_cache,
     submit_ai_discovery,
 )
+from ggshield.verticals.ai.history import BackfillReport, backfill_mcp_history
 from ggshield.verticals.ai.models import Scope
 
 
@@ -32,11 +33,19 @@
     default=False,
     help="Output as JSON",
 )
+@click.option(
+    "--history",
+    "scan_history",
+    is_flag=True,
+    default=False,
+    help="Also backfill historical MCP tool calls parsed from agent transcripts.",
+)
 @add_common_options()
 @click.pass_context
 def discover_cmd(
     ctx: click.Context,
     use_json: bool,
+    scan_history: bool,
     **kwargs: Any,
 ) -> None:
     """
@@ -47,6 +56,7 @@ def discover_cmd(
     Examples:
       ggshield ai discover
       ggshield ai discover --json
+      ggshield ai discover --history
     """
 
     config = discover_ai_configuration()
@@ -61,9 +71,12 @@ def discover_cmd(
         )
         return
 
+    backfill_report = BackfillReport()
     try:
         config = submit_ai_discovery(client, config)
         save_discovery_cache(config)
+        if scan_history:
+            backfill_report = backfill_mcp_history(client, config)
     except Exception as exc:
         if "missing the following scope:" in str(exc):
             scope = str(exc).split("missing the following scope:")[1].strip()
@@ -73,15 +86,15 @@ def discover_cmd(
         ui.display_warning(f"Could not upload AI discovery to GitGuardian: {reason}")
 
     # Summarize after sending to GIM, so we can benefit from its fixes.
-    summary = _summarize_discovery(config)
+    summary = _summarize_discovery(config, backfill_report)
 
     if use_json:
         click.echo(json.dumps(summary, indent=2))
     else:
         print_summary(summary)
 
 
-def _summarize_discovery(config: AIDiscovery) -> Dict[str, Any]:
+def _summarize_discovery(config: AIDiscovery, report: BackfillReport) -> Dict[str, Any]:
     """Summarize what we want to show of the discovery."""
     agent_names = set()
     servers = []
@@ -110,6 +123,12 @@ def _summarize_discovery(config: AIDiscovery) -> Dict[str, Any]:
     return {
         "agents": [AGENTS[name].display_name for name in agent_names],
         "servers": servers,
+        "history": {
+            "parsed": report.parsed,
+            "ingested": report.ingested,
+            "duplicates": report.duplicates,
+            "skipped": report.skipped,
+        },
     }
 
 
@@ -158,3 +177,13 @@ def print_summary(summary: Dict[str, Any]) -> None:
                 click.echo(f"{indent}{connector} {project}")
 
     click.echo()
+
+    history = summary.get("history")
+    if history and history.get("parsed"):
+        click.echo(f"{format_text('Backfilling MCP usage history…', STYLE['key'])}")
+        click.echo(f"  • Parsed {history['parsed']:,} events")
+        click.echo(
+            f"  • Recorded {history['ingested']:,} events "
+            f"({history['duplicates']:,} already known, "
+            f"{history.get('skipped', 0):,} skipped)"
+        )

ggshield/verticals/ai/agents/claude_code.py

@@ -1,7 +1,8 @@
 import json
 import re
+from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, Iterator, Literal
+from typing import Any, Dict, Iterator, Literal, Optional
 
 import click
 from pygitguardian.models import AIDiscovery, MCPActivityRequest
@@ -229,14 +230,7 @@ def parse_mcp_activity(
         # Remove the optional "claude_ai_" prefix
         server_cfg_name = server_cfg_name.removeprefix("claude_ai_")
 
-        # Lookup the server name based on its configuration name
-        # Fallback to the server name if not found
-        server_name = server_cfg_name
-        for server in ai_config.servers:
-            for configuration in server.configurations:
-                if _mangle_server_name(configuration.name) == server_cfg_name:
-                    server_name = server.name
-                    break
+        server_name = self._resolve_server_name(server_cfg_name, ai_config)
 
         return MCPActivityRequest(
             user=ai_config.user,
@@ -246,8 +240,82 @@ def parse_mcp_activity(
             model="",
             cwd=payload.raw.get("cwd", ""),
             input=payload.raw.get("tool_input", {}),
+            timestamp=payload.timestamp,
         )
 
+    def iter_history_events(
+        self, ai_config: Optional[AIDiscovery]
+    ) -> Iterator[MCPActivityRequest]:
+        """Walk every Claude session transcript and yield its MCP tool_use events."""
+        for path in self._history_files():
+            for entry in self._load_jsonl_file(path):
+                yield from self._parse_history_entry(entry, ai_config)
+
+    def _history_files(self) -> Iterator[Path]:
+        """Yield every Claude Code session transcript file we know about."""
+        yield from sorted(self.config_folder.glob("projects/*/*.jsonl"))
+
+    def _parse_history_entry(
+        self,
+        entry: Dict[str, Any],
+        ai_config: Optional[AIDiscovery],
+    ) -> Iterator[MCPActivityRequest]:
+        """Turn one parsed transcript entry into zero-or-more MCPActivityRequest events.
+
+        Returns nothing for non-MCP tool uses or sidechain entries.
+        Server names are resolved against ai_config when available.
+        """
+        if not isinstance(entry, dict) or entry.get("isSidechain"):
+            return
+
+        message = entry.get("message") or {}
+        content = message.get("content") or []
+        if not isinstance(content, list):
+            return
+
+        try:
+            ts = datetime.fromisoformat(entry["timestamp"].replace("Z", "+00:00"))
+        except (KeyError, AttributeError, ValueError):
+            return
+
+        cwd = entry.get("cwd", "")
+        model = message.get("model", "")
+
+        for block in content:
+            if not isinstance(block, dict):
+                continue
+            if block.get("type") != "tool_use":
+                continue
+            raw_name = block.get("name", "")
+            if not raw_name.startswith("mcp__"):
+                continue
+            parts = raw_name.split("__")
+            tool = parts[-1]
+            server_cfg_name = "__".join(parts[1:-1]).removeprefix("claude_ai_")
+            server_name = self._resolve_server_name(server_cfg_name, ai_config)
+            yield MCPActivityRequest(
+                user=self._user_or_default(ai_config),
+                tool=tool,
+                server=server_name,
+                agent=self.name,
+                model=model,
+                cwd=cwd,
+                input=block.get("input") or {},
+                timestamp=ts,
+            )
+
+    def _resolve_server_name(
+        self, cfg_name: str, ai_config: Optional[AIDiscovery]
+    ) -> str:
+        """Look up the canonical server name; fall back to the configuration name."""
+        if ai_config is None:
+            return cfg_name
+        for server in ai_config.servers:
+            for configuration in server.configurations:
+                if _mangle_server_name(configuration.name) == cfg_name:
+                    return server.name
+        return cfg_name
+
 
 MANGLING_PATTERN = re.compile(r"[^A-Za-z0-9-]")
 

ggshield/verticals/ai/agents/codex.py

@@ -1,6 +1,7 @@
 import json
+from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, Iterator, Literal
+from typing import Any, Dict, Iterator, Literal, Optional
 
 import click
 from pygitguardian.models import AIDiscovery, MCPActivityRequest
@@ -115,19 +116,106 @@ def parse_mcp_activity(
         tool = parts[-1]
         server_cfg_name = "__".join(parts[1:-1])
 
-        server_name = server_cfg_name
-        for server in ai_config.servers:
-            for configuration in server.configurations:
-                if _mangle_server_name(configuration.name) == server_cfg_name:
-                    server_name = server.name
-                    break
-
         return MCPActivityRequest(
             user=ai_config.user,
             tool=tool,
-            server=server_name,
+            server=self._resolve_server_name(server_cfg_name, ai_config),
             agent=self.name,
             model=payload.raw.get("model", ""),
             cwd=payload.raw.get("cwd", ""),
             input=payload.raw.get("tool_input", {}),
+            timestamp=payload.timestamp,
         )
+
+    def iter_history_events(
+        self, ai_config: Optional[AIDiscovery]
+    ) -> Iterator[MCPActivityRequest]:
+        """Walk every Codex session rollout and yield its MCP tool_use events."""
+        for path in self._history_files():
+            yield from self._parse_session_file(path, ai_config)
+
+    def _history_files(self) -> Iterator[Path]:
+        """Yield every Codex session rollout file we know about."""
+        yield from sorted(self.config_folder.glob("sessions/*/*/*/rollout-*.jsonl"))
+
+    def _parse_session_file(
+        self, path: Path, ai_config: Optional[AIDiscovery]
+    ) -> Iterator[MCPActivityRequest]:
+        """Yield MCPActivityRequest events from a single Codex session rollout."""
+        cwd = ""
+        model = ""
+        for entry in self._load_jsonl_file(path):
+            if not isinstance(entry, dict):
+                continue
+            payload = entry.get("payload")
+            if not isinstance(payload, dict):
+                continue
+            entry_type = entry.get("type")
+            if entry_type == "session_meta":
+                cwd = payload.get("cwd") or cwd
+                continue
+            if entry_type == "turn_context":
+                cwd = payload.get("cwd") or cwd
+                model = payload.get("model") or model
+                continue
+            if entry_type != "response_item":
+                continue
+            if payload.get("type") != "function_call":
+                continue
+            namespace = payload.get("namespace") or ""
+            if not namespace.startswith("mcp__"):
+                continue
+            event = self._build_activity_from_function_call(
+                entry, payload, namespace, cwd, model, ai_config
+            )
+            if event is not None:
+                yield event
+
+    def _build_activity_from_function_call(
+        self,
+        entry: Dict[str, Any],
+        payload: Dict[str, Any],
+        namespace: str,
+        cwd: str,
+        model: str,
+        ai_config: Optional[AIDiscovery],
+    ) -> Optional[MCPActivityRequest]:
+        """Turn one function_call response_item into an MCPActivityRequest."""
+        tool = payload.get("name") or ""
+        if not tool:
+            return None
+        server_cfg_name = namespace.removeprefix("mcp__").removesuffix("__")
+        try:
+            tool_input = json.loads(payload.get("arguments") or "{}")
+        except (json.JSONDecodeError, TypeError):
+            tool_input = {}
+        if not isinstance(tool_input, dict):
+            tool_input = {}
+        try:
+            ts = datetime.fromisoformat(
+                str(entry.get("timestamp", "")).replace("Z", "+00:00")
+            )
+        except ValueError:
+            return None
+        return MCPActivityRequest(
+            user=self._user_or_default(ai_config),
+            tool=tool,
+            server=self._resolve_server_name(server_cfg_name, ai_config),
+            agent=self.name,
+            model=model,
+            cwd=cwd,
+            input=tool_input,
+            timestamp=ts,
+        )
+
+    def _resolve_server_name(
+        self, cfg_name: str, ai_config: Optional[AIDiscovery]
+    ) -> str:
+        """Look up the canonical server name; fall back to the configuration name."""
+        if ai_config is None:
+            return cfg_name
+        for server in ai_config.servers:
+            for configuration in server.configurations:
+                if _mangle_server_name(configuration.name) == cfg_name:
+                    return server.name
+        return cfg_name