Skip to content

How do I pause a tool for human approval?

When you'd want this

Any tool that mutates the world — publishing content, spending money, sending a message, filing a ticket, running a shell command — should not run without a human saying yes when the run is being watched. The run should suspend to a checkpoint and hand control back to your driver code, which decides whether to resume with "approve", "reject", or a modified argument list.

Assumes ANTHROPIC_API_KEY in the environment

The snippet wires providers.claude(...) through an Invoker so the reader sees the real shape they'll ship. Swap providers.claude for providers.openai (and set OPENAI_API_KEY) if that's what you have — nothing else changes.

Working code

"""Requires ANTHROPIC_API_KEY in the environment."""

import asyncio
import os

from agentkit import Agent, Scope, Suspended, tool
from agentkit.adapters.checkpoint import InMemoryCheckpointStore
from agentkit.adapters.llm import providers
from agentkit.agents.cognition import ReActCognition
from agentkit.capabilities import Checkpointer
from agentkit.runtime import Invoker, RunContext, Services


@tool(side_effecting=True)
async def send_email(to: str, subject: str) -> str:
    """Send an email. Side-effecting — always gated under Autonomy.GATED."""
    return f"sent to {to!r}: {subject!r}"


async def main() -> None:
    llm = providers.claude(
        api_key=os.environ["ANTHROPIC_API_KEY"],
        model="claude-sonnet-4-6",
    )
    services = Services(
        invoker=Invoker(llm=llm),
        checkpointer=Checkpointer(port=InMemoryCheckpointStore()),
    )
    ctx = RunContext(
        correlation_id="run-1",
        scope=Scope(),
        services=services,
        autonomy="gated",  # gate every side_effecting tool
    )
    agent = Agent(
        name="notifier",
        model="claude-sonnet-4-6",
        prompt="Draft a one-line brief for the team, then call send_email exactly once.",
        cognition=ReActCognition(tools=[send_email]),
    )

    # First run — the loop suspends before send_email fires.
    result = await agent.run("Send the team a brief about octopus cognition.", ctx)
    susp = result.evals.get("suspended")
    assert isinstance(susp, Suspended)
    print(f"awaiting approval for: {[tc.name for tc in susp.pending]}")

    # Driver decides. Decisions are keyed by ToolCall.id:
    #   "approve"          — run with the model's args
    #   "reject" / "deny"  — inject a DENIED tool result, keep the loop going
    #   any other string   — parsed as a JSON args override, then run
    decisions = {tc.id: "approve" for tc in susp.pending}
    final = await agent.resume(susp.run_id, decisions, ctx)
    print(f"final: {final.output!r}")


if __name__ == "__main__":
    asyncio.run(main())

How it works

Autonomy is a run-wide tier carried on RunContext.autonomy. should_gate(autonomy, requires_approval, key_step) is the shared policy every gating surface consults:

Autonomy Gates
"auto" Only tools explicitly marked requires_approval=True
"gated" Also every side_effecting=True tool (key step)
"manual" Every tool call

When the ReActCognition sees a gated tool call, it snapshots its state through the Checkpointer (status "suspended"), emits an interrupt event per pending tool call, and returns an AgentResult with partial=True and a Suspended value in evals["suspended"]. Suspended.pending is a frozen tuple of ToolCalls — the operator UI reads it and the resume path threads it back verbatim, so a mutable list can't desync the two ends of the handshake.

agent.resume(run_id, decisions, ctx) loads the snapshot, applies your per-call decisions, appends the tool results to the transcript, and drives the loop to a final answer. You can call it from an entirely fresh process — that's the point of the checkpointer.

Gotchas

  • side_effecting= is required on every @tool. The framework fails at decoration time (not at call time) if you forget — because the gating and idempotency primitives cannot guess.
  • You must wire a checkpointer for resume(...) to work. Without one, the loop still emits Suspended, but the state isn't persisted, so resume raises ValueError("no suspended run … to resume"). InMemoryCheckpointStore is fine for tests; production wants PostgresCheckpointStore (extra: arc-agentkit[postgres]).
  • agent.resume(...) only works on a ReActCognition. Calling it on a SingleCallCognition or a coordinator is a contract violation and raises RuntimeError explicitly.
  • A Suspended result has partial=True and an empty output. The final text comes from resume(...), not the original run(...).