Human-in-the-Loop Basics

8 examples to get you started with human-in-the-loop agent control - 5 basic and 3 intermediate.

You will classify tool risk, pause before an external send, collect an approve/reject decision, resume the run, and sketch a tiny kill flag.

Prerequisites

Python 3.11+ recommended
Comfort with a simple agent loop (model proposes a tool, host executes it)
No paid APIs required for these sketches

python -m venv .venv && source .venv/bin/activate
# stdlib only for the examples below

Basic Examples

1. Label Tools as Auto or Gated

Start by listing what the agent may call and whether a human must approve first.

from dataclasses import dataclass
 
 
@dataclass(frozen=True)
class ToolSpec:
    name: str
    requires_approval: bool
 
 
TOOLS = {
    "search": ToolSpec("search", requires_approval=False),
    "draft_email": ToolSpec("draft_email", requires_approval=False),
    "send_email": ToolSpec("send_email", requires_approval=True),
}
 
for t in TOOLS.values():
    mode = "GATE" if t.requires_approval else "AUTO"
    print(f"{t.name:12} {mode}")

Read and draft tools stay fast.
Send is the first gate; that is the HITL baseline for messaging agents.
Keep this table in code or config, not only in a prompt.

Related: Privilege Rings: Scoping What an Agent Can Do Without Asking

2. Pause the Loop Before a Gated Tool

When the model requests a gated tool, do not execute it. Emit a pending action instead.

from dataclasses import dataclass
 
 
@dataclass
class PendingAction:
    tool: str
    args: dict
    status: str = "awaiting_approval"
 
 
def dispatch(tool: str, args: dict) -> object:
    spec = TOOLS[tool]
    if spec.requires_approval:
        return PendingAction(tool=tool, args=args)
    return {"ok": True, "tool": tool, "args": args, "result": "done"}
 
 
print(dispatch("search", {"q": "refund policy"}))
print(dispatch("send_email", {"to": "user@example.com", "body": "Hello"}))

The host owns the pause, not the model.
PendingAction is a first-class run state, not an error.
Real systems persist this state so a crash does not lose the decision context.

3. Build a Decision Packet for the Human

Approvers need a short, honest summary - not a raw transcript dump.

def decision_packet(pending: PendingAction) -> dict:
    return {
        "title": f"Approve {pending.tool}?",
        "tool": pending.tool,
        "args": pending.args,
        "risk": "external_message",
        "choices": ["approve", "reject", "edit"],
    }
 
 
pending = PendingAction("send_email", {"to": "user@example.com", "body": "Hello"})
print(decision_packet(pending))

Show tool name, destination, and content that will leave the system.
Offer edit so humans can fix a bad draft without restarting the whole agent.
Log the packet id so audit trails can point at the same object later.

4. Apply Approve / Reject From the Human

Once a decision arrives, either run the tool or mark the run blocked.

def apply_decision(pending: PendingAction, decision: str, edited_args: dict | None = None) -> dict:
    if decision == "reject":
        return {"status": "rejected", "tool": pending.tool}
    if decision == "edit":
        args = edited_args or pending.args
    elif decision == "approve":
        args = pending.args
    else:
        raise ValueError(f"unknown decision: {decision}")
    # Real code: call the email provider here
    return {"status": "executed", "tool": pending.tool, "args": args}
 
 
print(apply_decision(pending, "approve"))
print(apply_decision(pending, "reject"))

Reject should stop or replan, not silently retry the same send.
Edit should re-validate args (email format, allowlisted domains) before execute.
Never treat "no response yet" as approve.

5. Wire a Minimal Agent Step With a Gate

Combine model-chosen tool selection (mocked here) with the gate.

def model_choose_tool(user_goal: str) -> tuple[str, dict]:
    # Stand-in for an LLM tool call
    if "email" in user_goal.lower():
        return "send_email", {"to": "user@example.com", "body": "We processed your request."}
    return "search", {"q": user_goal}
 
 
def agent_step(user_goal: str, auto_approve: bool = False) -> dict:
    tool, args = model_choose_tool(user_goal)
    outcome = dispatch(tool, args)
    if isinstance(outcome, PendingAction):
        if auto_approve:
            return apply_decision(outcome, "approve")
        return {"status": "awaiting_approval", "packet": decision_packet(outcome)}
    return {"status": "ok", "result": outcome}
 
 
print(agent_step("Find the refund policy"))
print(agent_step("Email the customer a confirmation"))

Default production path waits; auto_approve is only for tests or dry-runs.
Search completes in one step; email returns a pending packet.
This is the core HITL contract: plan free, side effects gated.

Related: Building an Approval Gate Before Irreversible Actions

Intermediate Examples

6. Resume a Paused Run After Approval

Production agents must round-trip a run_id so the human UI can continue the same session.

from uuid import uuid4
 
 
RUNS: dict[str, dict] = {}
 
 
def start_run(goal: str) -> dict:
    run_id = str(uuid4())
    tool, args = model_choose_tool(goal)
    outcome = dispatch(tool, args)
    if isinstance(outcome, PendingAction):
        RUNS[run_id] = {"goal": goal, "pending": outcome, "status": "awaiting_approval"}
        return {"run_id": run_id, "status": "awaiting_approval", "packet": decision_packet(outcome)}
    RUNS[run_id] = {"goal": goal, "status": "done", "result": outcome}
    return {"run_id": run_id, "status": "done", "result": outcome}
 
 
def resume_run(run_id: str, decision: str, edited_args: dict | None = None) -> dict:
    run = RUNS[run_id]
    if run["status"] != "awaiting_approval":
        return {"error": "not_waiting"}
    result = apply_decision(run["pending"], decision, edited_args)
    run["status"] = result["status"]
    run["result"] = result
    return {"run_id": run_id, **result}
 
 
paused = start_run("Email the customer a confirmation")
print(paused)
print(resume_run(paused["run_id"], "approve"))

Persist RUNS in Redis/DB in real systems.
Resume must be idempotent: double-click approve should not send twice.
Store the decision actor and timestamp for audit.

7. Soft Notify vs Hard Gate

Not every medium-risk action needs a hard block. Some only need a visible notification with a short undo window.

from enum import Enum
 
 
class GateMode(Enum):
    AUTO = "auto"
    NOTIFY = "notify"
    HARD = "hard"
 
 
POLICY = {
    "search": GateMode.AUTO,
    "draft_email": GateMode.AUTO,
    "post_internal_note": GateMode.NOTIFY,
    "send_email": GateMode.HARD,
}
 
 
def plan_execution(tool: str) -> str:
    mode = POLICY[tool]
    if mode is GateMode.AUTO:
        return "execute_now"
    if mode is GateMode.NOTIFY:
        return "execute_and_notify"
    return "wait_for_approval"
 
 
print(plan_execution("post_internal_note"))
print(plan_execution("send_email"))

NOTIFY is for reversible, low-blast internal actions.
HARD is for external or irreversible effects.
Do not use NOTIFY for refunds, deletes, or customer email.

8. Add a Process-Level Kill Flag

Even with gates, a runaway loop of auto tools needs an emergency stop.

class KillSwitch:
    def __init__(self) -> None:
        self.tripped = False
 
    def trip(self, reason: str) -> None:
        self.tripped = True
        self.reason = reason
 
    def check(self) -> None:
        if self.tripped:
            raise RuntimeError(f"killed: {self.reason}")
 
 
def bounded_loop(steps: list[str], kill: KillSwitch) -> list[str]:
    out: list[str] = []
    for i, step in enumerate(steps):
        kill.check()
        if i == 3:
            kill.trip("too many steps in demo")
        out.append(step)
    return out
 
 
ks = KillSwitch()
try:
    print(bounded_loop(["a", "b", "c", "d", "e"], ks))
except RuntimeError as e:
    print(e)

Kill switches halt the host loop, not only the next gated tool.
Wire trip() to on-call buttons, budget caps, and anomaly detectors.
Pair with Kill Switches: Stopping a Runaway Agent Mid-Execution.

Related: Designing Escalation Paths When an Agent Gets Stuck

Privilege Rings: Scoping What an Agent Can Do Without Asking - tiered autonomy
Building an Approval Gate Before Irreversible Actions - full gate recipe
Audit Trails: Logging Every Agent Action for Later Review - decision logging
Human-in-the-Loop Best Practices - production checklist
Common Guardrail Mistakes That Undermine Human Oversight - failure modes

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

Human-in-the-Loop Basics

8 examples to get you started with human-in-the-loop agent control - 5 basic and 3 intermediate.

You will classify tool risk, pause before an external send, collect an approve/reject decision, resume the run, and sketch a tiny kill flag.

Prerequisites

Python 3.11+ recommended
Comfort with a simple agent loop (model proposes a tool, host executes it)
No paid APIs required for these sketches

python -m venv .venv && source .venv/bin/activate
# stdlib only for the examples below

Basic Examples

1. Label Tools as Auto or Gated

Start by listing what the agent may call and whether a human must approve first.

from dataclasses import dataclass
 
 
@dataclass(frozen=True)
class ToolSpec:
    name: str
    requires_approval: bool
 
 
TOOLS = {
    "search": ToolSpec("search", requires_approval=False),
    "draft_email": ToolSpec("draft_email", requires_approval=False),
    "send_email": ToolSpec("send_email", requires_approval=True),
}
 
for t in TOOLS.values():
    mode = "GATE" if t.requires_approval else "AUTO"
    print(f"{t.name:12} {mode}")

Read and draft tools stay fast.
Send is the first gate; that is the HITL baseline for messaging agents.
Keep this table in code or config, not only in a prompt.

Related: Privilege Rings: Scoping What an Agent Can Do Without Asking

2. Pause the Loop Before a Gated Tool

When the model requests a gated tool, do not execute it. Emit a pending action instead.

from dataclasses import dataclass
 
 
@dataclass
class PendingAction:
    tool: str
    args: dict
    status: str = "awaiting_approval"
 
 
def dispatch(tool: str, args: dict) -> object:
    spec = TOOLS[tool]
    if spec.requires_approval:
        return PendingAction(tool=tool, args=args)
    return {"ok": True, "tool": tool, "args": args, "result": "done"}
 
 
print(dispatch("search", {"q": "refund policy"}))
print(dispatch("send_email", {"to": "user@example.com", "body": "Hello"}))

The host owns the pause, not the model.
PendingAction is a first-class run state, not an error.
Real systems persist this state so a crash does not lose the decision context.

3. Build a Decision Packet for the Human

Approvers need a short, honest summary - not a raw transcript dump.

def decision_packet(pending: PendingAction) -> dict:
    return {
        "title": f"Approve {pending.tool}?",
        "tool": pending.tool,
        "args": pending.args,
        "risk": "external_message",
        "choices": ["approve", "reject", "edit"],
    }
 
 
pending = PendingAction("send_email", {"to": "user@example.com", "body": "Hello"})
print(decision_packet(pending))

Show tool name, destination, and content that will leave the system.
Offer edit so humans can fix a bad draft without restarting the whole agent.
Log the packet id so audit trails can point at the same object later.

4. Apply Approve / Reject From the Human

Once a decision arrives, either run the tool or mark the run blocked.

def apply_decision(pending: PendingAction, decision: str, edited_args: dict | None = None) -> dict:
    if decision == "reject":
        return {"status": "rejected", "tool": pending.tool}
    if decision == "edit":
        args = edited_args or pending.args
    elif decision == "approve":
        args = pending.args
    else:
        raise ValueError(f"unknown decision: {decision}")
    # Real code: call the email provider here
    return {"status": "executed", "tool": pending.tool, "args": args}
 
 
print(apply_decision(pending, "approve"))
print(apply_decision(pending, "reject"))

Reject should stop or replan, not silently retry the same send.
Edit should re-validate args (email format, allowlisted domains) before execute.
Never treat "no response yet" as approve.

5. Wire a Minimal Agent Step With a Gate

Combine model-chosen tool selection (mocked here) with the gate.

def model_choose_tool(user_goal: str) -> tuple[str, dict]:
    # Stand-in for an LLM tool call
    if "email" in user_goal.lower():
        return "send_email", {"to": "user@example.com", "body": "We processed your request."}
    return "search", {"q": user_goal}
 
 
def agent_step(user_goal: str, auto_approve: bool = False) -> dict:
    tool, args = model_choose_tool(user_goal)
    outcome = dispatch(tool, args)
    if isinstance(outcome, PendingAction):
        if auto_approve:
            return apply_decision(outcome, "approve")
        return {"status": "awaiting_approval", "packet": decision_packet(outcome)}
    return {"status": "ok", "result": outcome}
 
 
print(agent_step("Find the refund policy"))
print(agent_step("Email the customer a confirmation"))

Default production path waits; auto_approve is only for tests or dry-runs.
Search completes in one step; email returns a pending packet.
This is the core HITL contract: plan free, side effects gated.

Related: Building an Approval Gate Before Irreversible Actions

Intermediate Examples

6. Resume a Paused Run After Approval

Production agents must round-trip a run_id so the human UI can continue the same session.

from uuid import uuid4
 
 
RUNS: dict[str, dict] = {}
 
 
def start_run(goal: str) -> dict:
    run_id = str(uuid4())
    tool, args = model_choose_tool(goal)
    outcome = dispatch(tool, args)
    if isinstance(outcome, PendingAction):
        RUNS[run_id] = {"goal": goal, "pending": outcome, "status": "awaiting_approval"}
        return {"run_id": run_id, "status": "awaiting_approval", "packet": decision_packet(outcome)}
    RUNS[run_id] = {"goal": goal, "status": "done", "result": outcome}
    return {"run_id": run_id, "status": "done", "result": outcome}
 
 
def resume_run(run_id: str, decision: str, edited_args: dict | None = None) -> dict:
    run = RUNS[run_id]
    if run["status"] != "awaiting_approval":
        return {"error": "not_waiting"}
    result = apply_decision(run["pending"], decision, edited_args)
    run["status"] = result["status"]
    run["result"] = result
    return {"run_id": run_id, **result}
 
 
paused = start_run("Email the customer a confirmation")
print(paused)
print(resume_run(paused["run_id"], "approve"))

Persist RUNS in Redis/DB in real systems.
Resume must be idempotent: double-click approve should not send twice.
Store the decision actor and timestamp for audit.

7. Soft Notify vs Hard Gate

Not every medium-risk action needs a hard block. Some only need a visible notification with a short undo window.

from enum import Enum
 
 
class GateMode(Enum):
    AUTO = "auto"
    NOTIFY = "notify"
    HARD = "hard"
 
 
POLICY = {
    "search": GateMode.AUTO,
    "draft_email": GateMode.AUTO,
    "post_internal_note": GateMode.NOTIFY,
    "send_email": GateMode.HARD,
}
 
 
def plan_execution(tool: str) -> str:
    mode = POLICY[tool]
    if mode is GateMode.AUTO:
        return "execute_now"
    if mode is GateMode.NOTIFY:
        return "execute_and_notify"
    return "wait_for_approval"
 
 
print(plan_execution("post_internal_note"))
print(plan_execution("send_email"))

NOTIFY is for reversible, low-blast internal actions.
HARD is for external or irreversible effects.
Do not use NOTIFY for refunds, deletes, or customer email.

8. Add a Process-Level Kill Flag

Even with gates, a runaway loop of auto tools needs an emergency stop.

class KillSwitch:
    def __init__(self) -> None:
        self.tripped = False
 
    def trip(self, reason: str) -> None:
        self.tripped = True
        self.reason = reason
 
    def check(self) -> None:
        if self.tripped:
            raise RuntimeError(f"killed: {self.reason}")
 
 
def bounded_loop(steps: list[str], kill: KillSwitch) -> list[str]:
    out: list[str] = []
    for i, step in enumerate(steps):
        kill.check()
        if i == 3:
            kill.trip("too many steps in demo")
        out.append(step)
    return out
 
 
ks = KillSwitch()
try:
    print(bounded_loop(["a", "b", "c", "d", "e"], ks))
except RuntimeError as e:
    print(e)

Kill switches halt the host loop, not only the next gated tool.
Wire trip() to on-call buttons, budget caps, and anomaly detectors.
Pair with Kill Switches: Stopping a Runaway Agent Mid-Execution.

Related: Designing Escalation Paths When an Agent Gets Stuck

Privilege Rings: Scoping What an Agent Can Do Without Asking - tiered autonomy
Building an Approval Gate Before Irreversible Actions - full gate recipe
Audit Trails: Logging Every Agent Action for Later Review - decision logging
Human-in-the-Loop Best Practices - production checklist
Common Guardrail Mistakes That Undermine Human Oversight - failure modes

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

Human-in-the-Loop Basics

Buscar en la documentación

Human-in-the-Loop Basics