agent_hook.py
Rama main · réplica de lectura del repositorio original.
"""
One pre-execution hook for the coding agents that have hooks, besides Claude Code (claude_hook.py):
python agent_hook.py cursor Cursor beforeShellExecution, beforeReadFile, beforeMCPExecution,
preToolUse for Write|Delete (~/.cursor/hooks.json)
python agent_hook.py gemini-cli Gemini CLI BeforeTool (~/.gemini/settings.json)
python agent_hook.py codex Codex CLI PreToolUse (~/.codex/hooks.json)
python agent_hook.py vscode VS Code PreToolUse (Copilot agent mode) (~/.copilot/hooks/)
python agent_hook.py antigravity Antigravity PreToolUse (~/.gemini/config/hooks.json)
python agent_hook.py qwen-code Qwen Code PreToolUse (~/.qwen/settings.json)
python agent_hook.py factory Factory Droid PreToolUse (~/.factory/hooks.json or settings.json)
python agent_hook.py kiro Kiro CLI PreToolUse (~/.kiro/hooks/squidbrake.json)
Each reads the agent's JSON on stdin, sends the action to Squidbrake, waits while a person decides if it's held, and
answers in that agent's format. Shell commands are sent as "Bash", reads as "Read", writes and edits as "Write" / "Edit",
so the same rules apply whichever agent ran them. MCP tools: Cursor's and Codex's are checked here as <server>.<tool>
(unless the server is already Squidbrake's proxy); the other agents' go through `connect guard`, which routes their MCP
servers through Squidbrake, so no call is recorded twice.
Installed by `squidbrake connect agents`. Settings come as arguments: --url, --key (like claude_hook.py).
"""
from __future__ import annotations
import json
import os
import re
import shlex
import sys
import time
import httpx
try: # what a command will change, and an undo for what it destroys (both best effort, never fatal)
import effects
import runs
import undo
except Exception: # pragma: no cover
effects = runs = undo = None
try:
import hooklog # which agent ran the hook and when, for `squidbrake doctor`
except Exception: # pragma: no cover
hooklog = None
try:
import offline # what to do while the gateway is down (rules.yaml when_unreachable)
except Exception: # pragma: no cover
offline = None
AGENTS = ("cursor", "gemini-cli", "codex", "vscode", "antigravity", "qwen-code", "factory", "kiro")
def _arg(flag: str, env: str, default: str = "") -> str:
if flag in sys.argv[1:-1]:
return sys.argv[sys.argv.index(flag) + 1]
return os.getenv(env, default)
URL = _arg("--url", "GATEWAY_URL", "http://localhost:8080").rstrip("/")
KEY = _arg("--key", "GATEWAY_API_KEY") or (offline.load_key(URL) if offline else "")
FAIL_OPEN = os.getenv("GATEWAY_FAIL_OPEN", "").lower() in ("1", "true", "yes")
MAX_WAIT = float(os.getenv("GATEWAY_MAX_WAIT", "540")) # under the 600 s timeout `connect agents` sets
# --------------------------------------------------------------------------- what the agent wants to do
def normalize(name: str, args) -> tuple[str, dict] | None:
"""(name, input) the way Squidbrake's rules see it, or None for actions left to other paths (MCP tools)."""
args = args if isinstance(args, dict) else {"value": args}
low = name.lower()
if low.startswith("mcp"):
return None
lower = {k.lower(): v for k, v in args.items()}
cmd = next((lower[k] for k in ("command", "commandline", "command_line", "cmd") if k in lower), None)
if cmd is not None:
if isinstance(cmd, list): # e.g. Codex: ["bash", "-lc", "rm -rf ..."]; keep the quoting
cmd = shlex.join(str(c) for c in cmd)
out = {"command": str(cmd)}
cwd = next((lower[k] for k in ("cwd", "workdir", "directory") if k in lower), None)
if cwd:
out["cwd"] = cwd
return "Bash", out
path = next((v for k, v in lower.items() if k in ("file_path", "filepath", "path", "absolute_path", "absolutepath",
"targetfile", "target_file", "file")), None)
if any(w in low for w in ("read", "view", "open")) and path:
return "Read", {"file_path": path}
if any(w in low for w in ("edit", "replace", "patch", "str_replace", "multi_edit")):
return "Edit", {**args, **({"file_path": path} if path else {})}
if any(w in low for w in ("write", "create")):
return "Write", {**args, **({"file_path": path} if path else {})}
return name, args
OURS = ("gateway_proxy", "squidbrake proxy", "gateway_mcp")
def mcp_call(server, tool, args, launched_by) -> tuple[str, dict] | None:
"""An MCP tool call as `<server>.<tool>`, the way the proxy names them, so the same rules apply. None when the
server is already Squidbrake's proxy (`connect guard` / `wrap`): it checks the call itself, once."""
server, tool = str(server or "mcp"), str(tool or "unknown")
if any(o in str(launched_by or "") for o in OURS) or server.startswith(("gw-", "gw_")) or server == "gateway-db":
return None
if isinstance(args, str): # Cursor sends the arguments as a JSON string
try:
args = json.loads(args) if args.strip() else {}
except ValueError:
args = {"value": args}
app = re.sub(r"[^A-Za-z0-9_-]+", "-", server).strip("-").lower() or "mcp"
return f"{app}.{tool}", args if isinstance(args, dict) else {"value": args}
def cursor_tool(name: str, args, cwd) -> tuple[str, dict] | None:
"""Cursor's preToolUse, installed for Write (every file edit) and Delete. Shell and reads have their own hooks."""
args = args if isinstance(args, dict) else {}
path = next((args[k] for k in ("file_path", "path", "target_file", "filePath", "file") if args.get(k)), None)
if name == "Delete" and isinstance(path, str):
# deleting a file is the same as `rm` it: the command checks know which deletes are everyday work (a log, a
# temp file), and those only count inside the project, so a path in it is made relative
# compared as text: os.path.isabs disagrees across OSes and Python versions on paths like /w/x (Windows, 3.13)
root = str(cwd or "").replace("\\", "/").rstrip("/") + "/"
posix = path.replace("\\", "/")
if cwd and posix.lower().startswith(root.lower()):
path = posix[len(root):] or path
return "Bash", {"command": f"rm -- {shlex.quote(path)}", **({"cwd": cwd} if cwd else {})}
if name in ("Shell", "Read", "Grep") or name.startswith("MCP:"):
return None # covered by beforeShellExecution / beforeReadFile / beforeMCPExecution
return normalize(name, {**args, **({"file_path": path} if path else {})})
def parse(agent: str, ev: dict) -> tuple[tuple[str, dict] | None, str | None]:
"""-> ((name, input) or None, session id)"""
if agent == "cursor":
session = ev.get("conversation_id")
if ev.get("hook_event_name") == "beforeShellExecution":
# cwd can come empty: then the open project, so `rm -rf build` is measured where it would run, not in the hook's folder
cwd = ev.get("cwd") or next(iter(ev.get("workspace_roots") or []), None)
if isinstance(cwd, str) and re.match(r"^/[A-Za-z]:[/\\]", cwd): # Windows roots come as /e:/project
cwd = cwd[1:]
return ("Bash", {k: v for k, v in (("command", ev.get("command")), ("cwd", cwd)) if v}), session
if ev.get("hook_event_name") == "beforeReadFile":
return ("Read", {"file_path": ev.get("file_path")}), session
if ev.get("hook_event_name") == "beforeMCPExecution":
return mcp_call(ev.get("mcp_server_name"), ev.get("tool_name"), ev.get("tool_input"),
ev.get("command") or ev.get("url") or ev.get("mcp_server_url")), session
if ev.get("hook_event_name") == "preToolUse": # Cursor's own tools: here for edits and deletes
return cursor_tool(str(ev.get("tool_name") or ""), ev.get("tool_input"), ev.get("cwd")), session
return normalize(str(ev.get("tool_name", "unknown")), ev.get("tool_input")), session
if agent in ("codex", "factory") and str(ev.get("tool_name", "")).startswith("mcp__"):
# Codex's (config.toml) and Droid's MCP servers aren't rewritten by `connect guard`: the hook checks them
_, server, tool = (str(ev["tool_name"]).split("__", 2) + ["", ""])[:3]
return mcp_call(server, tool, ev.get("tool_input"), None), ev.get("session_id")
if agent == "kiro" and ev.get("tool_name") == "delete_file":
return cursor_tool("Delete", ev.get("tool_input"), ev.get("cwd")), ev.get("session_id")
if agent == "antigravity":
call = ev.get("toolCall") or {}
return normalize(str(call.get("name", "unknown")), call.get("args")), ev.get("conversationId")
return normalize(str(ev.get("tool_name", "unknown")), ev.get("tool_input")), ev.get("session_id")
# --------------------------------------------------------------------------- answers, in each agent's format
def answer(agent: str, allow: bool, message: str = "", stop: bool = False) -> None:
if agent == "cursor":
out = {"permission": "allow"} if allow else {"permission": "deny", "user_message": message, "agent_message": message}
elif agent in ("gemini-cli", "antigravity"):
out = {"decision": "allow"} if allow else {"decision": "deny", "reason": message}
if stop and agent == "gemini-cli":
out.update({"continue": False, "stopReason": message})
elif agent == "kiro": # Kiro: exit code 2 blocks the tool, and stderr goes to the agent
if allow:
sys.exit(0)
print(message, file=sys.stderr)
sys.exit(2)
elif agent in ("codex", "qwen-code", "factory"):
if allow: # no output means "carry on" (an "allow" would skip the agent's own prompt)
sys.exit(0)
out = {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny",
"permissionDecisionReason": message}}
else: # vscode
out = {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "allow" if allow else "deny",
**({} if allow else {"permissionDecisionReason": message})}}
if stop:
out.update({"continue": False, "stopReason": message})
print(json.dumps(out))
sys.exit(0)
def gateway_client(url: str, key: str, timeout: float) -> httpx.Client:
"""The gateway on this computer is reached directly: 127.0.0.1, not localhost (Windows tries IPv6 ::1 first, and
waits about 2 s for each call to fail over), and never through a proxy (on Windows the system proxy setting is
read without its "bypass for local addresses" list, so a company proxy would get every hook call)."""
host = httpx.URL(url).host
if host == "localhost":
url = url.replace("//localhost", "//127.0.0.1", 1)
return httpx.Client(base_url=url, headers={"X-Gateway-Key": key}, timeout=timeout,
trust_env=host not in ("localhost", "127.0.0.1", "::1"))
def unreachable(url: str, e: httpx.HTTPError, what: str = "action") -> str:
"""Why everything is blocked, in words a person can act on (the agent passes it on). Same text in claude_hook.py."""
status = getattr(getattr(e, "response", None), "status_code", None)
if isinstance(e, (ValueError, KeyError, TypeError)):
why = f"Something at {url} answered, but not like Squidbrake does (a proxy or another app on that port?)"
elif "rejected" in str(e):
why = f"Squidbrake's dashboard at {url} rejected this computer's key (it may have been removed)"
elif status in (502, 503, 504) or isinstance(e, (httpx.ConnectError, httpx.TimeoutException)):
why = f"Squidbrake's dashboard at {url} isn't answering: it may be switched off, restarting, or deleted"
else:
why = f"Squidbrake at {url} is unreachable or misconfigured ({e})"
return (f"{why}. Every {what} must go through it, so this was blocked. Tell the user: start it again (or ask "
f"whoever runs it). If it was removed on purpose, take Squidbrake out of this computer's agents with: "
f"squidbrake connect all --remove")
def notes_text(d: dict) -> str:
"""Notes people left for this agent (Decision.notes). Same words as claude_hook.py and gw_async.py."""
return "".join(f"NOTE FROM {n.get('by')} (a person on your team, sent through Squidbrake): {n.get('text')}\n\n"
for n in d.get("notes") or [])
def check(agent: str, name: str, inp: dict, session: str | None) -> None:
# These agents read only a refusal's message for sure, so a note comes as one: that action waits a turn.
body = {"name": name, "kind": "agent_hook", "input": inp, "source": agent, "session_id": session,
"metadata": {"accepts_notes": True}}
line = inp.get("command") if name == "Bash" and effects is not None else None
cwd = inp.get("cwd") or os.getcwd()
if line:
try:
if found := effects.predict(line, cwd):
body.setdefault("metadata", {})["effects"] = found
except Exception:
pass
try: # what `make clean` / `npm run x` / `bash x.sh` runs underneath, so the gateway checks that too
if found := runs.expand(line, cwd):
body.setdefault("metadata", {})["runs"] = found
except Exception:
pass
try:
with gateway_client(URL, KEY, 15) as http:
r = http.post("/v1/events", json=body)
if r.status_code == 401:
raise httpx.HTTPError("the gateway rejected the key")
r.raise_for_status()
d = r.json()
end = time.monotonic() + MAX_WAIT
while d["decision"] == "review" and time.monotonic() < end:
wait = min(25.0, end - time.monotonic())
r = http.get(f"/v1/events/{d['event_id']}/decision", params={"wait": wait}, timeout=wait + 10)
r.raise_for_status()
d = r.json()
if offline is not None:
offline.refresh(http, URL)
told = notes_text(d)
if told and d["decision"] == "allow":
http.post(f"/v1/events/{d['event_id']}/result", json={"error": "not run: a person's note came first"})
answer(agent, False, f"{told}This action did not run, so you could read the note first. If it still "
"fits, run it again.")
if told: # notes come only with a first answer, so a refusal's reason carries them
d["reason"] = f"{told}{d.get('reason') or ''}"
if d["decision"] == "allow":
kept = undo.snapshot(line, cwd) if line and undo is not None else None
if kept is None and d.get("snapshot_strict"):
kept = {"skipped": f"Squidbrake can't keep a copy of what {name} changes from {agent}"}
if d.get("snapshot") and kept is not None:
http.post(f"/v1/events/{d['event_id']}/snapshot", json={"skipped": kept["skipped"]}
if kept.get("skipped") else {"kind": "local", "ref": kept["id"], "summary": undo.describe(kept)})
if kept and kept.get("skipped") and d.get("snapshot") == "required":
http.post(f"/v1/events/{d['event_id']}/result", json={"error": f"not run: no undo copy ({kept['skipped']})"})
answer(agent, False, f"Squidbrake keeps a copy before changes like this one and couldn't "
f"({kept['skipped']}), so it didn't run. Tell the user.")
if kept and not kept.get("skipped"):
print(undo.describe(kept), file=sys.stderr)
# these agents don't report results back: mark the call done so it isn't left "pending"
http.post(f"/v1/events/{d['event_id']}/result", json={"output": {"result": f"not reported by {agent}"}})
except (httpx.HTTPError, ValueError, KeyError, TypeError) as e: # not JSON / not our answer: blocked too
if offline is not None and isinstance(e, httpx.HTTPError) and "rejected" not in str(e):
run, why = offline.decide(URL, name=name, kind="agent_hook", source=agent, session_id=session, input=inp,
fail_open=FAIL_OPEN)
if run:
print(f"Squidbrake is unreachable; {why}.", file=sys.stderr)
answer(agent, True)
answer(agent, False, unreachable(URL, e) + f" ({why[:1].upper() + why[1:]}.)")
if FAIL_OPEN:
answer(agent, True)
answer(agent, False, unreachable(URL, e))
if d["decision"] == "allow":
answer(agent, True)
if d["decision"] == "review":
answer(agent, False, "Squidbrake: nobody approved this in time. Ask the user to approve it in the dashboard, then try again.")
by, note = d.get("decided_by"), d.get("decision_note")
if d.get("rule_id") in ("emergency-stop", "session-stop") or str(d.get("rule_id") or "").startswith("budget:") \
or (note or "").startswith("The session was stopped"):
answer(agent, False, f"Squidbrake: {note or d.get('reason')}. Stop working and tell the user.", stop=True)
if by == "timeout":
answer(agent, False, "Squidbrake: nobody approved this in time, so it didn't run. Ask the user to approve it in "
f"the dashboard ({URL}/dashboard) when you try again.")
if by:
answer(agent, False, f"Squidbrake: rejected by {by}." + (f' Note: "{note}".' if note else "")
+ " Don't retry it; ask the user how to proceed.")
answer(agent, False, f"Squidbrake blocked this (rule '{d.get('rule_id')}'): {(d.get('reason') or '').rstrip('.')}. "
"Don't try to work around it.")
def read_event() -> dict | None:
"""The event on stdin as UTF-8 whatever the console's code page, with or without a byte-order mark (Cursor on
Windows sends one). None if nothing came; ValueError if what came isn't a JSON object."""
text = sys.stdin.buffer.read().decode("utf-8-sig", errors="replace").strip()
if not text:
return None
ev = json.loads(text)
if not isinstance(ev, dict):
raise ValueError("not a JSON object")
return ev
def main() -> None:
agent = sys.argv[1] if len(sys.argv) > 1 else ""
if agent not in AGENTS:
sys.exit(f"usage: agent_hook.py {{{'|'.join(AGENTS)}}} --url URL --key KEY")
try:
ev = read_event()
except ValueError:
# Something arrived but it isn't an event we can read: never let that through unchecked
answer(agent, FAIL_OPEN, f"Squidbrake couldn't read what {agent} sent to its hook, so this was blocked to be "
"safe. Tell the user to run: squidbrake doctor")
if ev is None:
answer(agent, True) # nothing to check
if hooklog is not None:
hooklog.record(agent, str(ev.get("hook_event_name") or (ev.get("toolCall") or {}).get("name") or ev.get("tool_name") or ""))
hooklog.daily(URL)
action, session = parse(agent, ev)
if action is None:
answer(agent, True)
check(agent, action[0], action[1], session)
if __name__ == "__main__":
main()