|
| 1 | +# Human-in-the-loop |
| 2 | + |
| 3 | +Use the human-in-the-loop (HITL) flow to pause agent execution until a person approves or rejects sensitive tool calls. Tools declare when they need approval, run results surface pending approvals as interruptions, and `RunState` lets you serialize and resume runs after decisions are made. |
| 4 | + |
| 5 | +## Marking tools that need approval |
| 6 | + |
| 7 | +Set `needs_approval` to `True` to always require approval or provide an async function that decides per call. The callable receives the run context, parsed tool parameters, and the tool call ID. |
| 8 | + |
| 9 | +```python |
| 10 | +from agents import Agent, Runner, function_tool |
| 11 | + |
| 12 | + |
| 13 | +@function_tool(needs_approval=True) |
| 14 | +async def cancel_order(order_id: int) -> str: |
| 15 | + return f"Cancelled order {order_id}" |
| 16 | + |
| 17 | + |
| 18 | +async def requires_review(_ctx, params, _call_id) -> bool: |
| 19 | + return "refund" in params.get("subject", "").lower() |
| 20 | + |
| 21 | + |
| 22 | +@function_tool(needs_approval=requires_review) |
| 23 | +async def send_email(subject: str, body: str) -> str: |
| 24 | + return f"Sent '{subject}'" |
| 25 | + |
| 26 | + |
| 27 | +agent = Agent( |
| 28 | + name="Support agent", |
| 29 | + instructions="Handle tickets and ask for approval when needed.", |
| 30 | + tools=[cancel_order, send_email], |
| 31 | +) |
| 32 | +``` |
| 33 | + |
| 34 | +`needs_approval` is available on [`function_tool`][agents.tool.function_tool], [`Agent.as_tool`][agents.agent.Agent.as_tool], [`ShellTool`][agents.tool.ShellTool], and [`ApplyPatchTool`][agents.tool.ApplyPatchTool]. Local MCP servers also support approvals through `require_approval` on [`MCPServerStdio`][agents.mcp.server.MCPServerStdio], [`MCPServerSse`][agents.mcp.server.MCPServerSse], and [`MCPServerStreamableHttp`][agents.mcp.server.MCPServerStreamableHttp]. Hosted MCP servers support approvals via [`HostedMCPTool`][agents.tool.HostedMCPTool] with `tool_config={"require_approval": "always"}` and an optional `on_approval_request` callback. Shell and apply_patch tools accept an `on_approval` callback if you want to auto-approve or auto-reject without surfacing an interruption. |
| 35 | + |
| 36 | +## How the approval flow works |
| 37 | + |
| 38 | +1. When the model emits a tool call, the runner evaluates `needs_approval`. |
| 39 | +2. If an approval decision for that tool call is already stored in the [`RunContextWrapper`][agents.run_context.RunContextWrapper] (for example, from `always_approve=True`), the runner proceeds without prompting. Per-call approvals are scoped to the specific call ID; use `always_approve=True` to allow future calls automatically. |
| 40 | +3. Otherwise, execution pauses and `RunResult.interruptions` (or `RunResultStreaming.interruptions`) contains `ToolApprovalItem` entries with details such as `agent.name`, `name`, and `arguments`. |
| 41 | +4. Convert the result to a `RunState` with `result.to_state()`, call `state.approve(...)` or `state.reject(...)` (optionally passing `always_approve` or `always_reject`), and then resume with `Runner.run(agent, state)` or `Runner.run_streamed(agent, state)`. |
| 42 | +5. The resumed run continues where it left off and will re-enter this flow if new approvals are needed. |
| 43 | + |
| 44 | +## Example: pause, approve, resume |
| 45 | + |
| 46 | +The snippet below mirrors the JavaScript HITL guide: it pauses when a tool needs approval, persists state to disk, reloads it, and resumes after collecting a decision. |
| 47 | + |
| 48 | +```python |
| 49 | +import asyncio |
| 50 | +import json |
| 51 | +from pathlib import Path |
| 52 | + |
| 53 | +from agents import Agent, Runner, RunState, function_tool |
| 54 | + |
| 55 | + |
| 56 | +async def needs_oakland_approval(_ctx, params, _call_id) -> bool: |
| 57 | + return "Oakland" in params.get("city", "") |
| 58 | + |
| 59 | + |
| 60 | +@function_tool(needs_approval=needs_oakland_approval) |
| 61 | +async def get_temperature(city: str) -> str: |
| 62 | + return f"The temperature in {city} is 20° Celsius" |
| 63 | + |
| 64 | + |
| 65 | +agent = Agent( |
| 66 | + name="Weather assistant", |
| 67 | + instructions="Answer weather questions with the provided tools.", |
| 68 | + tools=[get_temperature], |
| 69 | +) |
| 70 | + |
| 71 | +STATE_PATH = Path(".cache/hitl_state.json") |
| 72 | + |
| 73 | + |
| 74 | +def prompt_approval(tool_name: str, arguments: str | None) -> bool: |
| 75 | + answer = input(f"Approve {tool_name} with {arguments}? [y/N]: ").strip().lower() |
| 76 | + return answer in {"y", "yes"} |
| 77 | + |
| 78 | + |
| 79 | +async def main() -> None: |
| 80 | + result = await Runner.run(agent, "What is the temperature in Oakland?") |
| 81 | + |
| 82 | + while result.interruptions: |
| 83 | + # Persist the paused state. |
| 84 | + state = result.to_state() |
| 85 | + STATE_PATH.parent.mkdir(parents=True, exist_ok=True) |
| 86 | + STATE_PATH.write_text(state.to_string()) |
| 87 | + |
| 88 | + # Load the state later (could be a different process). |
| 89 | + stored = json.loads(STATE_PATH.read_text()) |
| 90 | + state = await RunState.from_json(agent, stored) |
| 91 | + |
| 92 | + for interruption in result.interruptions: |
| 93 | + approved = await asyncio.get_running_loop().run_in_executor( |
| 94 | + None, prompt_approval, interruption.name or "unknown_tool", interruption.arguments |
| 95 | + ) |
| 96 | + if approved: |
| 97 | + state.approve(interruption, always_approve=False) |
| 98 | + else: |
| 99 | + state.reject(interruption) |
| 100 | + |
| 101 | + result = await Runner.run(agent, state) |
| 102 | + |
| 103 | + print(result.final_output) |
| 104 | + |
| 105 | + |
| 106 | +if __name__ == "__main__": |
| 107 | + asyncio.run(main()) |
| 108 | +``` |
| 109 | + |
| 110 | +In this example, `prompt_approval` is synchronous because it uses `input()` and is executed with `run_in_executor(...)`. If your approval source is already asynchronous (for example, an HTTP request or async database query), you can use an `async def` function and `await` it directly instead. |
| 111 | + |
| 112 | +To stream output while waiting for approvals, call `Runner.run_streamed`, consume `result.stream_events()` until it completes, and then follow the same `result.to_state()` and resume steps shown above. |
| 113 | + |
| 114 | +## Other patterns in this repository |
| 115 | + |
| 116 | +- **Streaming approvals**: `examples/agent_patterns/human_in_the_loop_stream.py` shows how to drain `stream_events()` and then approve pending tool calls before resuming with `Runner.run_streamed(agent, state)`. |
| 117 | +- **Agent as tool approvals**: `Agent.as_tool(..., needs_approval=...)` applies the same interruption flow when delegated agent tasks need review. |
| 118 | +- **Shell and apply_patch tools**: `ShellTool` and `ApplyPatchTool` also support `needs_approval`. Use `state.approve(interruption, always_approve=True)` or `state.reject(..., always_reject=True)` to cache the decision for future calls. For automatic decisions, provide `on_approval` (see `examples/tools/shell.py`); for manual decisions, handle interruptions (see `examples/tools/shell_human_in_the_loop.py`). |
| 119 | +- **Local MCP servers**: Use `require_approval` on `MCPServerStdio` / `MCPServerSse` / `MCPServerStreamableHttp` to gate MCP tool calls (see `examples/mcp/get_all_mcp_tools_example/main.py` and `examples/mcp/tool_filter_example/main.py`). |
| 120 | +- **Hosted MCP servers**: Set `require_approval` to `"always"` on `HostedMCPTool` to force HITL, optionally providing `on_approval_request` to auto-approve or reject (see `examples/hosted_mcp/human_in_the_loop.py` and `examples/hosted_mcp/on_approval.py`). Use `"never"` for trusted servers (`examples/hosted_mcp/simple.py`). |
| 121 | +- **Sessions and memory**: Pass a session to `Runner.run` so approvals and conversation history survive multiple turns. SQLite and OpenAI Conversations session variants are in `examples/memory/memory_session_hitl_example.py` and `examples/memory/openai_session_hitl_example.py`. |
| 122 | +- **Realtime agents**: The realtime demo exposes WebSocket messages that approve or reject tool calls via `approve_tool_call` / `reject_tool_call` on the `RealtimeSession` (see `examples/realtime/app/server.py` for the server-side handlers). |
| 123 | + |
| 124 | +## Long-running approvals |
| 125 | + |
| 126 | +`RunState` is designed to be durable. Use `state.to_json()` or `state.to_string()` to store pending work in a database or queue and recreate it later with `RunState.from_json(...)` or `RunState.from_string(...)`. Pass `context_override` if you do not want to persist sensitive context data in the serialized payload. |
| 127 | + |
| 128 | +## Versioning pending tasks |
| 129 | + |
| 130 | +If approvals may sit for a while, store a version marker for your agent definitions or SDK alongside the serialized state. You can then route deserialization to the matching code path to avoid incompatibilities when models, prompts, or tool definitions change. |
0 commit comments