Integrations

#!/usr/bin/env python3
"""Claude Code PreToolUse hook: check each tool call with MCP Guard before it runs."""
import json, os, sys, urllib.request

event = json.load(sys.stdin)
key = os.environ.get("MCP_GUARD_API_KEY")
if not key:
    sys.exit(0)  # no key: leave Claude Code's normal permission rules in charge

def last_user_text(path):
    """The latest thing the user typed, from the session transcript (best effort)."""
    try:
        with open(path) as f:
            entries = [json.loads(line) for line in f if line.strip()]
    except (OSError, ValueError):
        return None
    for e in reversed(entries):
        m = e.get("message") or {}
        if e.get("type") != "user" or m.get("role") != "user":
            continue
        c = m.get("content")
        text = c if isinstance(c, str) else "\n".join(
            b.get("text", "") for b in c or [] if isinstance(b, dict) and b.get("type") == "text")
        if text.strip():
            return text[-2000:]
    return None

check = {
    "action": {"tool": event["tool_name"], "args": event.get("tool_input", {})},
    "context": f"Claude Code coding agent, working directory {event.get('cwd', '?')}",
}
intent = last_user_text(event.get("transcript_path", ""))
if intent:
    check["intent"] = intent

req = urllib.request.Request(
    "https://api.mcp-guard.ai/v1/guard",
    data=json.dumps(check).encode(),
    headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
)
try:
    with urllib.request.urlopen(req, timeout=5) as r:
        g = json.load(r)
except Exception:
    sys.exit(0)  # guard unreachable: fall back to the normal permission rules

decision = {"block": "deny", "ask": "ask"}.get(g["verdict"])  # allow: no opinion
if decision:
    reasons = ", ".join(g["reasons"]) or "no specific reason"
    print(json.dumps({"hookSpecificOutput": {
        "hookEventName": "PreToolUse",
        "permissionDecision": decision,
        "permissionDecisionReason": f"MCP Guard {g['verdict']} (p_unsafe {g['p_unsafe']}): {reasons}",
    }}))

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash|Write|Edit|MultiEdit|WebFetch|mcp__.*",
        "hooks": [
          { "type": "command", "command": "python3 \"$CLAUDE_PROJECT_DIR\"/.claude/hooks/mcp_guard.py", "timeout": 10 }
        ]
      }
    ]
  }
}

import { guard } from "./guard"; // the helper from the agent-loop recipe

type Tools = Record<string, (args: unknown) => Promise<unknown>>;

/** Wrap the tools a code-mode sandbox exposes: every call the generated code makes is checked first. */
export function guardedTools<T extends Tools>(
  tools: T,
  ctx: { intent: string; context?: string; confirm?: (tool: string, reasons: string[]) => Promise<boolean> },
): T {
  return new Proxy(tools, {
    get(target, name) {
      const fn = target[name as keyof T];
      if (typeof name !== "string" || typeof fn !== "function") return fn;
      return async (args: unknown) => {
        const g = await guard({ action: { tool: name, args }, intent: ctx.intent, context: ctx.context });
        const ok = g.verdict === "allow" || (g.verdict === "ask" && (await ctx.confirm?.(name, g.reasons)));
        if (!ok) throw new Error(`MCP Guard ${g.verdict}: ${name} not run (${g.reasons.join(", ") || "unsafe"})`);
        return fn(args);
      };
    },
  });
}

// The model wrote `code`; run it with checked tools. With no confirm(), "ask" fails closed.
await sandbox.run(code, { tools: guardedTools(tools, { intent: userRequest, context: "Environment: production." }) });