Integration

Human in the Loop with the OpenAI Agents SDK

Using needs_approval, interruptions and RunState to pause OpenAI Agents SDK tools for human approval — and what to do when the human is not sitting at the keyboard.

5 July 2026 · 8 min read · by the Yoonet team

The OpenAI Agents SDK ships human in the loop as a first class feature: mark a tool with needs_approval, and instead of executing it the run stops and hands you a list of interruptions to approve or reject. The machinery is clean. What the docs leave open — deliberately — is the question that decides whether it works in production: who answers, and how does the answer get back?

This guide covers the SDK mechanics in Python, then the two production moves: persisting a paused run with RunState, and routing approvals to a reviewer who is not sitting at your keyboard.

Step 1 — Mark the risky tool

needs_approval takes either True or an async predicate. Prefer the predicate: approvals cost human attention, and attention is the scarcest resource in the whole loop. Gate by amount, by customer tier, by confidence — by whatever separates the routine from the regrettable.

from agents import Agent, Runner, function_tool

async def refunds_over_100(_ctx, params, _call_id) -> bool:
    # Approval is conditional: only pause when the stakes are real.
    return params.get("amount", 0) > 100

@function_tool(needs_approval=refunds_over_100)
async def issue_refund(order_id: str, amount: float) -> str:
    do_refund(order_id, amount)              # your implementation
    return f"Refunded {amount} on {order_id}"

agent = Agent(
    name="Support agent",
    instructions="Resolve support cases. Use tools when needed.",
    tools=[issue_refund],
)

Step 2 — Handle the interruptions

When the model decides to call a gated tool, the run returns early with result.interruptions populated. Convert the result to a RunState, record a verdict on each interruption, and run the agent again with the state — it continues from exactly where it paused.

result = await Runner.run(agent, "Customer 1042's order arrived broken, refund $180")

while result.interruptions:
    state = result.to_state()
    for interruption in result.interruptions:
        if ask_someone(interruption.name, interruption.arguments):   # the hard part
            state.approve(interruption)
        else:
            state.reject(interruption)
    result = await Runner.run(agent, state)

print(result.final_output)

The loop matters: an approved tool call can lead the model to attempt another gated call, so treat approvals as a cycle, not a single toll gate.

Step 3 — Persist the pause

Real approvers do not answer within a Python process's lifetime. RunState serialises to JSON, which turns "waiting for a human" from a blocked coroutine into a stored row — the same durability trick LangGraph gets from checkpointers.

from pathlib import Path
import json
from agents import RunState

# Pause: serialise the run and stop. No process waits for the human.
state = result.to_state()
Path(f"runs/{run_id}.json").write_text(state.to_string())

# Resume (any process, any time later): rebuild state and continue.
stored = json.loads(Path(f"runs/{run_id}.json").read_text())
state = await RunState.from_json(agent, stored)
state.approve(interruption)                  # or state.reject(...)
result = await Runner.run(agent, state)

Store it keyed by a run id, park a deadline next to it, and your agent can wait an hour or a week for its answer at no cost.

Step 4 — The reviewer problem

ask_someone() in the loop above is doing a lot of quiet work. In every SDK example it is input() — the developer approving their own agent. The production versions are a Slack bot (someone must watch the channel), an internal review UI (someone must staff it), or a managed human endpoint.

The third option treats the reviewer like you treat the model: capacity behind an API. Post the interruption to hitl.ph as a decision.own task and a named specialist on Yoonet's Balanga floor makes the call and returns it — with reasoning — to your webhook, which loads the stored RunState, approves or rejects, and reruns the agent.

import requests

def route_to_reviewer(interruption, run_id: str) -> str:
    """Send a pending approval to a managed human reviewer."""
    r = requests.post(
        "https://api.hitl.ph/v1/tasks",
        headers={"Authorization": f"Bearer {HITL_API_KEY}"},
        json={
            "type": "decision.own",
            "brief": (
                f"An agent wants to call {interruption.name} with "
                f"{interruption.arguments}. Approve or decline, one line of reasoning."
            ),
            "context": {"run_id": run_id},
            "return": {"mode": "webhook", "url": "https://your-app.com/hooks/hitl"},
        },
    )
    return r.json()["id"]

The same pattern covers more than approvals. A tool can itself be a human task: a phone call to confirm a delivery (voice.call), a judgement call on a messy record (data.validate) — the five task types are effectively tools whose implementation is a person. The API docs have the full request shape and the CLI equivalent.

Production notes

Predicates over blanket gates. If every call needs approval, the approver stops reading. Gate the tail, not the distribution.
Use always_approve sparingly. It exists for "stop asking me about this tool" — convenient in development, an audit hole in production.
Rejections need reasons. The model sees the rejection and tries another route; a reason steers it somewhere useful instead of into retry loops.
Decide expiry up front. A stored state with no deadline is an agent that silently never finishes. Auto reject on timeout and say so in the case log.

Step 1 — Mark the risky tool

Step 2 — Handle the interruptions

Step 3 — Persist the pause

Step 4 — The reviewer problem

Production notes

Read next