Agents with Write Access by Default (Anti-Pattern) + Fixes + Code

from __future__ import annotations

from dataclasses import dataclass
import hashlib
import hmac
import json
from typing import Any


READ_TOOLS = {"search.read", "kb.read", "http.get"}
WRITE_TOOLS = {"ticket.close", "email.send", "db.write"}


def stable_json(obj: Any) -> bytes:
  return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")


def args_hash(args: dict[str, Any]) -> str:
  # IMPORTANT: hash ignores fields injected by the gateway.
  filtered = {k: v for k, v in args.items() if k not in {"idempotency_key", "approval_token"}}
  return hashlib.sha256(stable_json(filtered)).hexdigest()[:24]


@dataclass(frozen=True)
class Policy:
  allow: set[str]
  require_approval: set[str]  # usually write tools


class Denied(RuntimeError):
  pass


@dataclass(frozen=True)
class PendingApproval:
  approval_id: str
  checkpoint: str  # signed blob the server can resume


def evaluate(policy: Policy, tool: str) -> str:
  if tool not in policy.allow:
      raise Denied(f"not_allowed:{tool}")
  if tool in WRITE_TOOLS and tool in policy.require_approval:
      return "approve"
  return "allow"


def sign_checkpoint(payload: dict[str, Any], *, secret: bytes) -> str:
  raw = stable_json(payload)
  sig = hmac.new(secret, raw, hashlib.sha256).hexdigest()
  return sig + "." + raw.decode("utf-8")


def verify_checkpoint(blob: str, *, secret: bytes) -> dict[str, Any]:
  sig, raw = blob.split(".", 1)
  expected = hmac.new(secret, raw.encode("utf-8"), hashlib.sha256).hexdigest()
  if not hmac.compare_digest(sig, expected):
      raise Denied("bad_checkpoint_signature")
  return json.loads(raw)


def request_approval(*, tenant_id: str, tool: str, args_preview: dict[str, Any]) -> str:
  # Return an approval id/token from your approval system.
  # (pseudo)
  return "appr_31ac"


def call_tool(*, ctx: dict[str, Any], policy: Policy, tool: str, args: dict[str, Any], secret: bytes) -> Any:
  """
  Execute a tool with governance.

  CRITICAL:
  - tenant_id/env come from authenticated context (ctx), never from model output.
  - writes require approval and resume via checkpoint.
  """
  tenant_id = ctx["tenant_id"]
  env = ctx["env"]
  run_id = ctx["run_id"]
  step_id = ctx["step_id"]

  decision = evaluate(policy, tool)

  if decision == "approve":
      approval_id = request_approval(
          tenant_id=tenant_id,
          tool=tool,
          args_preview={"args_hash": args_hash(args), "args": {k: v for k, v in args.items() if k != "body"}},
      )
      checkpoint = sign_checkpoint(
          {
              "run_id": run_id,
              "step_id": step_id,
              "tenant_id": tenant_id,
              "env": env,
              "tool": tool,
              "args": args,
              "args_hash": args_hash(args),
              "kind": "tool_call",
          },
          secret=secret,
      )
      return PendingApproval(approval_id=approval_id, checkpoint=checkpoint)

  # Deterministic idempotency key injection for writes (gateway-owned, not model-owned).
  if tool in WRITE_TOOLS:
      base_hash = args_hash(args)
      args = {**args, "idempotency_key": f"{tenant_id}:{tool}:{base_hash}"}

  creds = load_scoped_credentials(tool=tool, tenant_id=tenant_id, env=env)  # (pseudo) NEVER from model
  return tool_impl(tool, args=args, creds=creds)  # (pseudo)


def resume_after_approval(*, checkpoint: str, approval_token: str, secret: bytes) -> dict[str, Any]:
  """
  Continuation pattern:
  - verify signed checkpoint
  - attach approval token
  - execute exactly once (idempotent)
  """
  payload = verify_checkpoint(checkpoint, secret=secret)
  tool = payload["tool"]
  args = payload["args"]

  # Keep hash stable by putting approval_token outside args hashing.
  args = {**args, "approval_token": approval_token}

  if tool in WRITE_TOOLS:
      base_hash = payload["args_hash"]
      args = {**args, "idempotency_key": f"{payload['tenant_id']}:{tool}:{base_hash}"}

  creds = load_scoped_credentials(tool=tool, tenant_id=payload["tenant_id"], env=payload["env"])  # (pseudo)
  out = tool_impl(tool, args=args, creds=creds)  # (pseudo)
  return {"status": "ok", "run_id": payload["run_id"], "step_id": payload["step_id"], "tool": tool, "result": out}

import crypto from "node:crypto";

const READ_TOOLS = new Set(["search.read", "kb.read", "http.get"]);
const WRITE_TOOLS = new Set(["ticket.close", "email.send", "db.write"]);

function stableJson(obj) {
return JSON.stringify(obj, Object.keys(obj).sort());
}

export function argsHash(args) {
// IMPORTANT: hash ignores fields injected by the gateway.
const filtered = {};
for (const [k, v] of Object.entries(args || {})) {
  if (k === "idempotency_key" || k === "approval_token") continue;
  filtered[k] = v;
}
return crypto.createHash("sha256").update(stableJson(filtered), "utf8").digest("hex").slice(0, 24);
}

export class Denied extends Error {}

export function evaluate(policy, tool) {
// policy.allow / policy.requireApproval are Sets (use .has)
if (!policy.allow.has(tool)) throw new Denied("not_allowed:" + tool);
if (WRITE_TOOLS.has(tool) && policy.requireApproval.has(tool)) return "approve";
return "allow";
}

export function signCheckpoint(payload, { secret }) {
const raw = JSON.stringify(payload);
const sig = crypto.createHmac("sha256", secret).update(raw, "utf8").digest("hex");
return sig + "." + raw;
}

export function verifyCheckpoint(blob, { secret }) {
const [sig, raw] = blob.split(".", 2);
const expected = crypto.createHmac("sha256", secret).update(raw, "utf8").digest("hex");
if (!crypto.timingSafeEqual(Buffer.from(sig), Buffer.from(expected))) throw new Denied("bad_checkpoint_signature");
return JSON.parse(raw);
}

export function requestApproval({ tenantId, tool, argsPreview }) {
// (pseudo) send to your approval system, return approval id
return "appr_31ac";
}

export function callTool({ ctx, policy, tool, args, secret }) {
const { tenant_id: tenantId, env, run_id: runId, step_id: stepId } = ctx;
const decision = evaluate(policy, tool);

if (decision === "approve") {
  const approvalId = requestApproval({
    tenantId,
    tool,
    argsPreview: { args_hash: argsHash(args), args: args },
  });

  const checkpoint = signCheckpoint(
    {
      run_id: runId,
      step_id: stepId,
      tenant_id: tenantId,
      env,
      tool,
      args,
      args_hash: argsHash(args),
      kind: "tool_call",
    },
    { secret },
  );

  return { status: "needs_approval", approval_id: approvalId, checkpoint };
}

if (WRITE_TOOLS.has(tool)) {
  const baseHash = argsHash(args);
  args = { ...args, idempotency_key: tenantId + ":" + tool + ":" + baseHash };
}

const creds = loadScopedCredentials({ tool, tenantId, env }); // (pseudo) NEVER from model
return toolImpl(tool, { args, creds }); // (pseudo)
}

export function resumeAfterApproval({ checkpoint, approvalToken, secret }) {
const payload = verifyCheckpoint(checkpoint, { secret });
const tool = payload.tool;

// Keep hash stable by putting approval_token outside args hashing.
let args = { ...payload.args, approval_token: approvalToken };

if (WRITE_TOOLS.has(tool)) {
  const baseHash = payload.args_hash;
  args = { ...args, idempotency_key: payload.tenant_id + ":" + tool + ":" + baseHash };
}

const creds = loadScopedCredentials({ tool, tenantId: payload.tenant_id, env: payload.env }); // (pseudo)
const out = toolImpl(tool, { args, creds }); // (pseudo)
return { status: "ok", run_id: payload.run_id, step_id: payload.step_id, tool, result: out };
}