Make Claude Code ask before it posts, edits your CRM, or deploys

mkdir -p ~/.claude/hooks
curl -fsSL https://agent-control.net/gate/hooks/claude-code.sh -o ~/.claude/hooks/action-gate.sh
chmod +x ~/.claude/hooks/action-gate.sh

export AGENT_CONTROL_API_KEY="paste the agent API key"

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash|PowerShell",
        "hooks": [
          {
            "type": "command",
            "command": "\"$HOME/.claude/hooks/action-gate.sh\"",
            "timeout": 660,
            "onFailure": "block",
            "statusMessage": "Waiting for you in Approval Inbox"
          }
        ]
      },
      {
        "matcher": "mcp__.*(?:slack|crm|hubspot|salesforce|pipedrive)",
        "hooks": [
          {
            "type": "command",
            "command": "\"$HOME/.claude/hooks/action-gate.sh\"",
            "timeout": 660,
            "onFailure": "block",
            "statusMessage": "Waiting for you in Approval Inbox"
          }
        ]
      }
    ]
  }
}

#!/usr/bin/env bash
# Action Gate hook for Claude Code PreToolUse (v2.1.310 or newer).
# Asks before a Slack post, a CRM write, or a deploy (git push, vercel deploy).
# A no or a timeout prints one JSON deny and exits 0. No output and exit 0
# lets the tool call continue. Needs python3. Set AGENT_CONTROL_API_KEY.
# You keep the key. Pair this file with onFailure "block" in ~/.claude/settings.json.
set -u
tmp=$(mktemp)
trap 'rm -f "$tmp"' EXIT
cat >"$tmp"
python3 - "$tmp" <<'PY'
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request

BASE = os.environ.get("ACTION_GATE_BASE", "https://agent-control.net").strip().rstrip("/")
KEY = os.environ.get("AGENT_CONTROL_API_KEY", "").strip()
PAY = "https://agent-control.net/billing/pay?plan=action"

DEPLOY_SEG = re.compile(
    r"^(?:(?:npx|bunx)\s+)?(?:git\s+push|vercel\s+deploy)\b",
    re.I,
)
SUBSHELL_DEPLOY = re.compile(
    r"(?:\$\(|`)\s*(?:sudo\s+)?(?:(?:npx|bunx)\s+)?(?:git\s+push|vercel\s+deploy)\b",
    re.I,
)
SLACK_SEG = re.compile(r"^slack\s+(?:chat\s+)?post\b", re.I)
CRM_SEG = re.compile(
    r"^(?:hubspot|salesforce|pipedrive)\b.*\b(?:update|create|upsert|write|patch|edit|delete)\b"
    r"|^sf\s+(?:data\s+)?(?:update|create|upsert|delete)\b",
    re.I,
)
WRITE_NAME = re.compile(r"post|send|write|update|create|edit|upsert|patch|delete|message", re.I)
SLACK_NAME = re.compile(r"slack", re.I)
CRM_NAME = re.compile(r"crm|hubspot|salesforce|pipedrive", re.I)
SPLIT = re.compile(r"\s*(?:&&|\|\||[;&|\n])\s*")
QUOTED = re.compile(r"'[^']*'|\"(?:\\.|[^\"])*\"")


def wait_seconds():
    raw = os.environ.get("ACTION_GATE_WAIT_S", "").strip()
    if not raw:
        return 600
    try:
        return max(0, int(raw))
    except ValueError:
        return 600


def clip(text, limit):
    text = " ".join(str(text).split())
    if len(text) <= limit:
        return text
    if limit <= 3:
        return text[:limit]
    return text[: limit - 3] + "..."


def deny(reason):
    json.dump(
        {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": clip(reason, 500),
            }
        },
        sys.stdout,
    )
    sys.stdout.write("\n")
    sys.exit(0)


def allow():
    sys.exit(0)


def normalize_seg(seg):
    seg = seg.strip()
    prev = None
    while seg and seg != prev:
        prev = seg
        seg = re.sub(r"^[A-Za-z_][A-Za-z0-9_]*=\S*\s+", "", seg)
        seg = re.sub(r"^sudo\s+(?:-n\s+)?", "", seg)
        seg = re.sub(r"^command\s+", "", seg)
    return seg.strip()


def segments(command):
    bare = QUOTED.sub(" ", command)
    for part in SPLIT.split(bare):
        seg = normalize_seg(part)
        if seg:
            yield seg


def tool_kind(name):
    if not name or not WRITE_NAME.search(name):
        return None
    if SLACK_NAME.search(name):
        return "slack.post"
    if CRM_NAME.search(name):
        return "crm.write"
    return None


def classify(tool_name, tool_input):
    command = ""
    if isinstance(tool_input, dict) and isinstance(tool_input.get("command"), str):
        command = tool_input["command"]
    elif isinstance(tool_input, str) and tool_name in ("Bash", "PowerShell"):
        command = tool_input
    if command.strip():
        segs = list(segments(command))
        if any(DEPLOY_SEG.match(seg) for seg in segs) or SUBSHELL_DEPLOY.search(command):
            return "deploy"
        if "hooks.slack.com" in command.lower() or any(SLACK_SEG.match(seg) for seg in segs):
            return "slack.post"
        if any(CRM_SEG.match(seg) for seg in segs):
            return "crm.write"
    return tool_kind(tool_name or "")


def one_line(tool_name, tool_input):
    if isinstance(tool_input, dict) and isinstance(tool_input.get("command"), str):
        command = tool_input["command"].strip()
        if command:
            return command
    if isinstance(tool_input, str) and tool_input.strip():
        return tool_input.strip()
    if isinstance(tool_input, dict):
        return json.dumps(tool_input, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
    return tool_name or "tool call"


def exact_input(tool_name, tool_input):
    if isinstance(tool_input, dict):
        rendered = json.dumps(tool_input, ensure_ascii=False, indent=2, sort_keys=True)
    else:
        rendered = str(tool_input)
    return f"{tool_name}\n{rendered}"


def read_event():
    path = sys.argv[1] if len(sys.argv) > 1 else ""
    raw = ""
    if path:
        with open(path, encoding="utf-8") as handle:
            raw = handle.read()
    if not raw.strip():
        deny("Action Gate could not read the tool call. Stop.")
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        deny("Action Gate could not read the tool call. Stop.")
    if not isinstance(data, dict):
        deny("Action Gate could not read the tool call. Stop.")
    return data


def request_json(url, method, payload, timeout):
    data = None if payload is None else json.dumps(payload).encode("utf-8")
    headers = {"Authorization": "Bearer " + KEY, "Accept": "application/json"}
    if data is not None:
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as res:
            raw = res.read().decode("utf-8", "replace")
    except urllib.error.HTTPError as err:
        raw = err.read().decode("utf-8", "replace")
    except Exception:
        deny("Action Gate could not be reached. Stop.")
    try:
        body = json.loads(raw) if raw.strip() else {}
    except json.JSONDecodeError:
        deny("Action Gate could not be reached. Stop.")
    if not isinstance(body, dict):
        deny("Action Gate could not be reached. Stop.")
    return body


def reason_text(payload, fallback):
    reasons = payload.get("reasons")
    if isinstance(reasons, list):
        text = " ".join(str(item).strip() for item in reasons if str(item).strip())
        if text:
            return text
    err = payload.get("error")
    if err:
        return str(err)
    return fallback


def poll_target(payload):
    approval = str(payload.get("approval_id") or "").strip()
    raw = str(payload.get("poll_url") or "").strip()
    prefix = "/api/v1/approvals/"
    if raw.startswith(BASE + prefix):
        return raw
    if raw.startswith(prefix):
        return BASE + raw
    if approval:
        return BASE + prefix + approval
    return ""


def verdict(payload):
    decision = str(payload.get("decision") or "").strip().lower()
    if decision in ("go", "allow"):
        allow()
    if decision in ("wait", "hold"):
        return
    if decision:
        deny(reason_text(payload, "Action Gate said stop."))
    deny(reason_text(payload, "Action Gate could not be reached. Stop."))


def ask(action_type, tool_name, tool_input):
    if not KEY:
        deny("Set AGENT_CONTROL_API_KEY. Get Action Gate, $49 for 30 days: " + PAY)
    line = one_line(tool_name, tool_input)
    summary = clip(f"Ask before {action_type}: {line}", 280)
    preview = exact_input(tool_name, tool_input)[:4000]
    if len(preview) == 4000:
        preview = preview[:3997] + "..."
    target = clip(line or tool_name, 200)
    risk = "medium" if action_type == "slack.post" else "high"
    deadline = time.monotonic() + wait_seconds()
    body = request_json(
        BASE + "/api/v1/check_action",
        "POST",
        {
            "action_type": action_type,
            "summary": summary,
            "preview": preview,
            "target": target,
            "risk": risk,
        },
        20,
    )
    while True:
        verdict(body)
        if time.monotonic() >= deadline:
            deny("No decision before the timeout. Stop.")
        url = poll_target(body)
        if not url:
            deny("Action Gate did not return a request to check. Stop.")
        try:
            delay = float(body.get("poll_after_ms") or 2000) / 1000
        except (TypeError, ValueError):
            delay = 2
        delay = min(max(delay, 0.2), 5)
        remaining = deadline - time.monotonic()
        if remaining < 1:
            deny("No decision before the timeout. Stop.")
        time.sleep(min(delay, remaining))
        if time.monotonic() >= deadline:
            deny("No decision before the timeout. Stop.")
        body = request_json(url, "GET", None, min(20, deadline - time.monotonic()))


def run():
    event = read_event()
    tool_name = str(event.get("tool_name") or "")
    tool_input = event.get("tool_input")
    kind = classify(tool_name, tool_input)
    if not kind:
        allow()
    ask(kind, tool_name, tool_input if tool_input is not None else {})


try:
    run()
except SystemExit:
    raise
except Exception:
    deny("Action Gate hook failed. Stop.")
PY
exit $?

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "^Bash$",
        "hooks": [
          {
            "type": "command",
            "command": "\"$HOME/.claude/hooks/action-gate.sh\"",
            "timeout": 660,
            "statusMessage": "Waiting for you in Approval Inbox"
          }
        ]
      },
      {
        "matcher": "mcp__.*(?:slack|crm|hubspot|salesforce|pipedrive)",
        "hooks": [
          {
            "type": "command",
            "command": "\"$HOME/.claude/hooks/action-gate.sh\"",
            "timeout": 660,
            "statusMessage": "Waiting for you in Approval Inbox"
          }
        ]
      }
    ]
  }
}