An agent that can call delete_customer, wire_funds, or kubectl apply is one hallucinated argument away from an incident you cannot undo. The fix is not a better prompt. It is a gate: catch the tool call before it runs, hand it to a human, and continue only on their word. Mechanically this is one pattern — interrupt and resume — and every current framework ships it. The part that actually matters is where you put the gate. That's the last section.

Why you gate (and what you don't)#

Gate the calls a mistake makes expensive:

Do not gate reads and cheap, reversible calls. If a human has to click "approve" on every search() and read_file(), they stop reading the dialog — and the one dangerous call sails through on autopilot. A gate is only as good as its scarcity. (The sibling discipline is not handing the agent standing credentials it can misuse between prompts — see secrets management for AI agents.)

The interrupt-and-resume pattern#

Three moving parts, identical everywhere:

  1. Intercept the tool call before execution.
  2. Persist + surface the pending call to a human (approve / deny / edit).
  3. Resume from the saved checkpoint with the decision applied.

The frameworks differ only in the API names.

LangGraph: interrupt() + Command(resume=...)#

Call interrupt() inside a node. It raises a resumable GraphInterrupt, and the value you pass is surfaced to the client. A checkpointer is mandatory — the pause is durable because state is persisted. You resume by calling the graph again with Command(resume=value) on the same thread_id. Note the node re-executes from the top, so keep pre-interrupt code idempotent.

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START
from langgraph.types import interrupt, Command

def approve_transfer(state):
    decision = interrupt({           # pauses here, surfaces this payload
        "action": "wire_funds",
        "amount": state["amount"],
        "to": state["account"],
    })
    if decision != "approve":
        return {"status": "denied"}
    # ... perform the real transfer only after approval ...
    return {"status": "sent"}

builder = StateGraph(dict)
builder.add_node("approve_transfer", approve_transfer)
builder.add_edge(START, "approve_transfer")

graph = builder.compile(checkpointer=InMemorySaver())  # required
config = {"configurable": {"thread_id": "txn-42"}}

result = graph.invoke({"amount": 5000, "account": "ACME"}, config)
# result["__interrupt__"] -> (Interrupt(value={...}, id=...),)  the graph is paused

# ...human reviews, later...
final = graph.invoke(Command(resume="approve"), config)  # same thread_id

Swap InMemorySaver for PostgresSaver/SqliteSaver in production so a pause survives a restart. (LangChain interrupts docs; interrupt() source)

OpenAI Agents SDK: needsApproval + interruptions#

Mark a tool needsApproval: true (or an async function for conditional gating — approve cheap cases in code, escalate the rest). When such a tool is about to run, the SDK records a RunToolApprovalItem instead of executing, and the run pauses and returns result.interruptions. Resolve each with state.approve() / state.reject(), then re-run.

import { Agent, run, tool, RunState } from '@openai/agents';
import { z } from 'zod';

const cancelOrder = tool({
  name: 'cancelOrder',
  description: 'Cancel an order',
  parameters: z.object({ orderId: z.number() }),
  needsApproval: true,                     // or: async (_ctx, {orderId}) => orderId < 1000
  execute: async ({ orderId }) => `cancelled ${orderId}`,
});

const agent = new Agent({ name: 'ops', tools: [cancelOrder] });

let result = await run(agent, 'Cancel order 992');
while (result.interruptions?.length) {
  // state is serializable: result.state.toString() -> DB, resume later
  for (const item of result.interruptions) {
    const ok = await confirm(`Approve ${item.name} ${item.arguments}?`);
    ok ? result.state.approve(item) : result.state.reject(item);
  }
  result = await run(agent, result.state);  // resume the original agent
}
console.log(result.finalOutput);

Because RunState serializes (toString() / RunState.fromString(agent, s)), you can shut the request down, notify a reviewer out-of-band, and resume days later. (OpenAI Agents SDK HITL guide; example)

Claude Agent SDK: the canUseTool callback#

Pass a canUseTool callback. It fires for any tool not already resolved by an allow/deny rule or permission mode, receiving the tool name and input, and pausing until you return a decision. Return allow (optionally with edited input) or deny with a message the model sees.

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import (
    PermissionResultAllow, PermissionResultDeny, ToolPermissionContext,
)

async def can_use_tool(tool_name: str, input_data: dict, ctx: ToolPermissionContext):
    if tool_name == "Bash" and "rm -rf" in input_data.get("command", ""):
        return PermissionResultDeny(message="Refuse destructive delete; archive instead.")
    if not await human_approves(tool_name, input_data):
        return PermissionResultDeny(message="User denied this action")
    # approve — or edit the input before it runs (e.g. scope a path to a sandbox)
    return PermissionResultAllow(updated_input=input_data)

options = ClaudeAgentOptions(can_use_tool=can_use_tool)

One sharp edge: auto-approved tools never reach canUseTool. A bare entry in allowed_tools, acceptEdits, or bypassPermissions resolves the call earlier in the permission flow and skips your callback. For a check that must run on every call regardless of mode, use a PreToolUse hook instead. (canUseTool docs)

Where to put the gate: the decision that matters#

The framework is a detail. Placement is the design. Pick per tool:

Default to tiering. A flat "always ask" trains reviewers to rubber-stamp; a flat "auto-approve" is how the incident happens. Gate on the argument, and put a human only where the blast radius earns one.