#!/usr/bin/env python3 """Tool snapshot classifier and drift check for the marketing-mcp-setup skill. Usage (from the skill folder): python3 -I scripts/mcp_tools_diff.py CURRENT.json [--baseline BASELINE.json] [--mode ro|rw] [--json] A snapshot is the tool names a client lists for one server, as JSON (format: assets/example-acme-tools-current.json). Never put tool results in it. Mark a tool "blocked": true when the client blocks it, and "denied_by": "role", "iam", "grant", "scope" or "server" only after test T4 proved that the source denies it. "reviewed_as": "read", "write" or "hard" overrides the name-based class. Rules: D01 a new write, hard-to-reverse or unknown tool on a read-only connection (FAIL; INFO when already blocked or denied; WARN on a read-write one) D02 a new read tool (WARN) D03 a removed tool (INFO) D04 a write, hard-to-reverse or unknown tool with no blocked or denied_by marker on a read-only connection, with or without a baseline (FAIL) D05 readOnlyHint true on a tool whose name reads as a write (WARN) D06 a write, hard-to-reverse or unknown tool that was blocked or denied in the baseline and is neither now (FAIL) Standard library only. No network calls, no subprocesses, no file writes. Exit codes: 0 no FAIL, 1 at least one FAIL, 2 bad input or unreadable file. """ import argparse import json import re import sys ORDER = {"FAIL": 0, "WARN": 1, "INFO": 2} DENIED_BY = ("role", "iam", "grant", "scope", "server") HARD = {"delete", "remove", "destroy", "send", "launch", "publish", "activate", "boost", "suppress", "unsubscribe", "subscribe", "merge", "apply", "execute", "deletion", "cancel", "stop", "revoke", "trigger"} WRITE = {"create", "update", "edit", "manage", "add", "set", "rename", "duplicate", "archive", "unarchive", "submit", "import", "upload", "sync", "enable", "disable", "toggle", "assign", "move", "copy", "replace", "revert", "restore", "dismiss", "bulk", "write", "upsert"} READ = {"get", "list", "search", "fetch", "read", "query", "describe", "discover", "report", "reports", "insight", "insights", "explain", "compare", "audit", "inspect", "display", "render", "status", "context", "schema", "guidance", "prime", "readonly"} STRONG_READ = {"get", "list", "search", "fetch", "read", "query", "describe"} STRONG_HARD = {"delete", "remove", "destroy", "send", "launch", "publish", "activate", "boost", "suppress", "unsubscribe", "merge", "deletion", "revoke"} STRONG_WRITE = {"create", "update", "edit", "write", "upsert", "rename", "duplicate", "submit", "replace", "manage"} UNTRUSTED = {"conversation", "conversations", "note", "notes", "review", "reviews", "ticket", "tickets", "comment", "comments", "chat", "chats", "transcript", "transcripts", "reply", "replies", "submission", "submissions", "inbox"} class InputError(Exception): """Bad arguments or unreadable input: exit code 2.""" def plural(n, word): return f"{n} {word}" if n == 1 else f"{n} {word}s" def load_json_file(path, what): try: with open(path, encoding="utf-8") as handle: return json.load(handle) except FileNotFoundError: raise InputError(f"{what} file not found: {path}") except (OSError, ValueError) as exc: raise InputError(f"cannot read {what} file {path}: {exc}") def name_tokens(name): if name.startswith("mcp__"): name = name.split("__", 2)[-1] spaced = re.sub(r"([a-z0-9])([A-Z])", r"\1 \2", name) spaced = re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1 \2", spaced) return [t.lower() for t in re.split(r"[^A-Za-z0-9]+", spaced) if t] def classify_tool(name, annotations): tokens = name_tokens(name) readonly = "readonly" in tokens or ("read" in tokens and "only" in tokens) verbs = [t for t in tokens if t in READ or t in WRITE or t in HARD] lead = verbs[0] if verbs else None if lead in STRONG_READ and not any(t in STRONG_HARD or t in STRONG_WRITE for t in tokens): cls = "read" elif any(t in HARD and not (t == "execute" and readonly) for t in tokens): cls = "hard" elif any(t in WRITE for t in tokens): cls = "write" elif any(t in READ for t in tokens): cls = "read" else: cls = "unknown" if annotations.get("destructiveHint") is True: cls = "hard" conflict = annotations.get("readOnlyHint") is True and cls in ("write", "hard") untrusted = any(t in UNTRUSTED for t in tokens) return cls, conflict, untrusted def load_snapshot(path): data = load_json_file(path, "snapshot") if not isinstance(data, dict) or not isinstance(data.get("tools"), list): raise InputError(f"{path}: expected an object with a tools list") if not data["tools"]: raise InputError(f"{path}: the snapshot lists no tools, so nothing was checked. Take it from the client's " f"server tool view, or record the server as listing none.") tools = {} for item in data["tools"]: entry = {"name": item} if isinstance(item, str) else item if not isinstance(entry, dict) or not entry.get("name"): raise InputError(f"{path}: every tool needs a name") denied = entry.get("denied_by") if denied is not None and denied not in DENIED_BY: raise InputError(f"{path}: denied_by on {entry['name']} must be one of {', '.join(DENIED_BY)}") annotations = entry.get("annotations") if isinstance(entry.get("annotations"), dict) else {} cls, conflict, untrusted = classify_tool(entry["name"], annotations) reviewed = entry.get("reviewed_as") if reviewed in ("read", "write", "hard"): cls = reviewed blocked = entry.get("blocked") is True tools[entry["name"]] = {"class": cls, "blocked": blocked, "denied_by": denied, "held": blocked or bool(denied), "conflict": conflict, "untrusted": untrusted} data["_tools"] = tools return data def how_held(tool): parts = (["blocked in the client"] if tool["blocked"] else []) + ( [f"denied by {tool['denied_by']}"] if tool["denied_by"] else []) return " and ".join(parts) or "neither blocked nor denied" def label(tool): return "unknown (treated as write)" if tool["class"] == "unknown" else tool["class"] def diff(args): current = load_snapshot(args.current) baseline = load_snapshot(args.baseline) if args.baseline else None mode = args.mode or current.get("mode") or "ro" if mode not in ("ro", "rw"): raise InputError("mode must be ro or rw") if baseline and (baseline.get("server"), baseline.get("client")) != (current.get("server"), current.get("client")): raise InputError("the baseline and current snapshots name different servers or clients") tools = current["_tools"] findings, reported = [], set() def add(severity, rule, name, message): findings.append({"severity": severity, "rule": rule, "tool": name, "message": message}) reported.add(name) u_note = " It may return text written by outsiders (U): keep send-out tools out of this profile." if baseline: old = baseline["_tools"] for name in sorted(set(tools) - set(old)): tool = tools[name] if tool["class"] == "read": add("WARN", "D02", name, "new read tool; review what it returns before the next run." + (u_note if tool["untrusted"] else "")) elif mode == "rw": add("WARN", "D01", name, f"new {tool['class']} tool; review it before use (agent-write-safety).") elif tool["held"]: add("INFO", "D01", name, f"new {tool['class']} tool, already {how_held(tool)}; record it in the inventory.") else: add("FAIL", "D01", name, f"new {label(tool)} tool on a read-only connection. Block it in the client or " f"prove it absent, then save this snapshot as the new baseline.") for name in sorted(set(old) - set(tools)): add("INFO", "D03", name, "removed; update the inventory and any allowlist that names it.") if mode == "ro": for name in sorted(set(tools) & set(old)): tool, before = tools[name], old[name] if tool["class"] != "read" and before["held"] and not tool["held"]: add("FAIL", "D06", name, f"{label(tool)} tool was {how_held(before)} in the baseline and is neither " f"now. Restore the block or the source denial, rerun T4, then save a new baseline.") if mode == "ro": for name in sorted(tools): tool = tools[name] if name not in reported and tool["class"] != "read" and not tool["held"]: add("FAIL", "D04", name, f"{label(tool)} tool with no blocked or denied_by marker on a read-only " f"connection. Block it in the client, or mark it once the client shows it " f"blocked or T4 proves the source denies it.") for name in sorted(tools): if tools[name]["conflict"]: findings.append({"severity": "WARN", "rule": "D05", "tool": name, "message": ( "readOnlyHint is true but the name reads as a write; annotations are hints, so treat it as a write " "until a denied test proves otherwise.")}) findings.sort(key=lambda f: (ORDER[f["severity"]], f["rule"], f["tool"])) counts = {k: sum(1 for f in findings if f["severity"] == k) for k in ORDER} by_class = {} for name in sorted(tools): tool = tools[name] key = tool["class"] if tool["class"] != "read" and tool["held"]: key += f" ({how_held(tool)})" by_class.setdefault(key, []).append(name + (" [U]" if tool["untrusted"] and tool["class"] == "read" else "")) class_counts = {c: sum(1 for t in tools.values() if t["class"] == c) for c in ("read", "write", "hard", "unknown")} if args.json: print(json.dumps({"server": current.get("server"), "client": current.get("client"), "captured": current.get("captured"), "mode": mode, "baseline_captured": baseline.get("captured") if baseline else None, "classes": class_counts, "tools_by_class": by_class, "findings": findings, "summary": counts}, indent=2)) else: base = f" Baseline captured {baseline.get('captured')}." if baseline else " No baseline." print(f"Tool snapshot {current.get('server')} ({current.get('client')}), captured {current.get('captured')}, " f"mode {mode}.{base} Nothing was changed.") blocked = sum(1 for t in tools.values() if t["blocked"]) denied = sum(1 for t in tools.values() if t["denied_by"]) print(f"{plural(len(tools), 'tool')}: {class_counts['read']} read, {class_counts['write']} write, {class_counts['hard']} " f"hard to reverse, {class_counts['unknown']} unknown; {blocked} marked blocked, {denied} denied at the source.") print("\nFindings") for f in findings: print(f"{f['severity']} {f['rule']} {f['tool']}: {f['message']}") if not findings: print("- none") print("\nTools by class") for key in sorted(by_class): print(f"- {key}: {', '.join(by_class[key])}") print(f"\nSummary: {counts['FAIL']} FAIL, {counts['WARN']} WARN, {counts['INFO']} INFO") return 1 if counts["FAIL"] else 0 def main(argv=None): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("current", help="tool snapshot JSON for the connection as it is now") parser.add_argument("--baseline", help="the last approved snapshot for the same server and client") parser.add_argument("--mode", choices=["ro", "rw"], help="read-only (default) or read-write connection") parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) try: return diff(args) except InputError as exc: print(f"error: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main())