Framework guides

How to add human approvals to OpenAI Agents SDK

Map OpenAI Agents SDK interruptions to Contro1 requests so humans can approve high-risk actions before execution continues.

OpenAI Agents SDK is well suited to tool-level approval flows where you need to gate sensitive actions but keep low-risk actions fast.

Use the integration skill

Copy this skill link into your code agent to add OpenAI Agents SDK and Contro1 to your system.

Copy skill link

Key takeaways

Mark risky tools with needs_approval=True so Runner.run() yields an interruption instead of executing.
Each interruption becomes one Contro1 request keyed by run_id + call_id for idempotency.
Wrap Runner.run() in a try/except to escalate unexpected errors to a human on call.
In the system prompt, tell the agent which tool names always require approval even if obvious.

When to reach for Contro1 with OpenAI Agents

The Agents SDK already exposes a clean seam - function_tool(needs_approval=True) - for pausing before a tool call executes. That seam is perfect for Contro1: we turn each interruption into a routable approval and return the operator's decision back into the run.

Use this when you want fine-grained tool-by-tool control rather than pausing the whole agent. Low-risk tools keep running at full speed; high-risk ones route through the right human every time.

Installation

pip install

pip install centcom flask python-dotenv

.env

CENTCOM_API_KEY=your_centcom_api_key
CENTCOM_BASE_URL=https://api.contro1.com/api/centcom/v1
CENTCOM_WEBHOOK_SECRET=whsec_your_signing_secret

Basic integration

bridge.py

from agents import Agent, Runner, function_tool
import centcom

@function_tool(needs_approval=True)
def issue_refund(customer_id: str, amount: int) -> str:
    return billing.refund(customer_id, amount)

result = Runner.run(agent, "Refund customer 8842 for $2,400")

while result.interruptions:
    for interruption in result.interruptions:
        req = centcom.create_request(
            type="approval",
            question=f"Approve tool call: {interruption.name}?",
            context=f"Arguments: {interruption.arguments}",
            required_role="manager",
            external_request_id=f"openai:{result.run_id}:{interruption.call_id}",
        )
        decision = centcom.wait_for_response(req["id"], interval=3, timeout=600)
        if decision["response"].get("approved"):
            result.state.approve(interruption)
        else:
            result.state.reject(
                interruption,
                rejection_message=decision["response"].get("comment", "Rejected in Contro1"),
            )
    result = Runner.resume(result)

bridge.ts

import { Runner, functionTool } from '@openai/agents';
import { centcom } from '@contro1/sdk';

const issueRefund = functionTool({
  name: 'issue_refund',
  needsApproval: true,
  handler: async ({ customerId, amount }) => billing.refund(customerId, amount),
});

let result = await Runner.run(agent, 'Refund customer 8842');

while (result.interruptions?.length) {
  for (const interruption of result.interruptions) {
    const req = await centcom.createRequest({
      type: 'approval',
      question: `Approve tool call: ${interruption.name}?`,
      context: JSON.stringify(interruption.arguments),
      required_role: 'manager',
      external_request_id: `openai:${result.runId}:${interruption.callId}`,
    });
    const decision = await centcom.waitForResponse(req.id, { timeoutSec: 600 });
    if (decision.response?.approved) {
      result.state.approve(interruption);
    } else {
      result.state.reject(interruption, decision.response?.comment ?? 'Rejected in Contro1');
    }
  }
  result = await Runner.resume(result);
}

Case continuity

Use one correlation_id per OpenAI Agents run - f"openai-{run_id}" works well. This groups every tool interruption, operator decision, and audit record for the run into one case timeline.

Keep call_id in external_request_id so duplicate interruptions return the original request instead of creating a new one.

openai_case.py

case_id = f"openai-{run_id}"

created = client.create_protocol_request({
    "title": f"Approve tool call: {tool_name}",
    "request_type": "decision",
    "source": {"integration": "openai-agents", "run_id": run_id},
    "continuation": {"mode": "decision", "callback_url": callback_url},
    "external_request_id": f"openai:{run_id}:{call_id}",
    "correlation_id": case_id,
})

Audit records and cases reference

Logging autonomous actions

Use log_action for tool outputs or model-side actions that were allowed to run without a human. This gives compliance and support teams evidence without slowing down low-risk work.

If the log describes what happened after an approved interruption, set in_reply_to to the Contro1 request id.

audit_action.py

client.log_action(
    action="openai_agents.email_sent",
    summary="Sent approved follow-up email to customer c-8821",
    source={"integration": "openai-agents", "run_id": run_id},
    outcome="success",
    correlation_id=case_id,
    in_reply_to={"type": "request", "id": request_id},
)

Gate the tool before it executes

The tool function itself is the right place to require approval for irreversible actions. The first line of a destructive tool calls Contro1 and blocks until an operator decides. Nothing runs until the human says yes - no prompt engineering needed.

tools/delete_file.py

import os
import centcom
from agents import function_tool

_centcom = centcom.Client(api_key=os.environ["CENTCOM_API_KEY"])

@function_tool
def delete_file(path: str) -> str:
    req = _centcom.create_request(
        type="approval",
        question=f"Permanently delete: {path}?",
        context={"path": path},
        required_role="developer",
        external_request_id=f"delete:{path}",
    )
    decision = _centcom.wait_for_response(req["id"], timeout=600)
    if not (decision.get("response") or {}).get("approved"):
        raise PermissionError("Deletion rejected by operator")
    return fs.delete(path)

Pause the agent on system error - orchestrator level

Wrap Runner.run() in an outer try/except so an unhandled agent-loop failure becomes a Contro1 request instead of a stack trace in logs. The operator chooses whether to retry the run or mark it as terminally failed.

run_with_error_gate.py

from agents import Runner
import centcom, traceback

def run_with_error_gate(agent, prompt, *, session_id):
    try:
        return Runner.run(agent, prompt)
    except Exception as exc:
        req = centcom.create_request(
            type="approval",
            question=f"Agent run {session_id} failed - retry?",
            context=f"{type(exc).__name__}: {exc}\n\n{traceback.format_exc()[:2000]}",
            required_role="oncall",
            priority="urgent",
            external_request_id=f"openai-run-error:{session_id}",
        )
        decision = centcom.wait_for_response(req["id"], timeout=900)
        if decision["response"].get("approved"):
            return Runner.run(agent, prompt)
        return {"status": "failed_by_human_decision", "error": str(exc)}

Escalate tool errors to a human

Inside each risky tool, catch domain errors (provider outage, validation failure, rate limit) and escalate via Contro1 before the exception propagates back into the agent loop. The operator picks retry / skip / cancel.

tools/refund.py

import centcom
from agents import function_tool

@function_tool(needs_approval=True)
def issue_refund(customer_id: str, amount: int) -> str:
    try:
        return billing.refund(customer_id, amount)
    except billing.ProviderError as exc:
        req = centcom.create_request(
            type="approval",
            question=f"Refund for {customer_id} failed. Retry, skip, or cancel?",
            context=f"{type(exc).__name__}: {exc}",
            required_role="oncall",
            external_request_id=f"refund-err:{customer_id}:{exc.code}",
        )
        decision = centcom.wait_for_response(req["id"], timeout=600)
        choice = decision["response"].get("answer")
        if choice == "retry":
            return billing.refund(customer_id, amount)
        if choice == "skip":
            return "skipped by operator"
        raise

tools/refund.ts

import { functionTool } from '@openai/agents';
import { centcom } from '@contro1/sdk';

export const issueRefund = functionTool({
  name: 'issue_refund',
  needsApproval: true,
  handler: async ({ customerId, amount }) => {
    try {
      return await billing.refund(customerId, amount);
    } catch (err) {
      const req = await centcom.createRequest({
        type: 'approval',
        question: `Refund ${customerId} failed. Retry, skip, or cancel?`,
        context: (err as Error).message,
        required_role: 'oncall',
      });
      const decision = await centcom.waitForResponse(req.id, { timeoutSec: 600 });
      if (decision.response?.answer === 'retry') return billing.refund(customerId, amount);
      if (decision.response?.answer === 'skip') return 'skipped by operator';
      throw err;
    }
  },
});

Prompt engineering: force the agent to pause

needs_approval=True gates a tool at the SDK level, but the model still decides whether to call the tool. Use the system prompt to teach it when to call - and when NOT to look for workarounds.

agent_instructions.md

BEFORE calling any tool marked "approval required" (issue_refund, modify_crm_record, send_customer_email, run_sql_write, deploy_release), stop reasoning about alternatives and just call it with your best-judgment arguments. The human reviewer sees your arguments and decides.

If the reviewer rejects, the SDK will surface a rejection_message. Treat that message as final:
  - Do NOT rephrase the request and retry.
  - Do NOT try a "lighter" version of the same action.
  - Tell the user exactly what was blocked and why, quoting the reviewer.

If the tool raises because the approval service is unreachable, stop. Report the outage to the user. Never fabricate a successful outcome.

See our GitHub integration repo

Our open-source OpenAI Agents connector shows the full bridge in production style, including signature verification on the callback path.

centcom-openai-agents on GitHub · openai_agents_bridge.py - interruption → approval mapping

Frequently asked questions

Should every tool require approval?

No. Gate only the high-risk, irreversible, or policy-sensitive tools so the agent remains useful without becoming unsafe. A good starting set is anything that touches money, customers, or production data.

How do I persist run state across the human wait?

The SDK lets you serialize result.state. Store it keyed by run_id before calling wait_for_response, and rehydrate it in the webhook handler if the process restarts.

Can the operator pass arguments back to the tool?

Yes. The approve call takes an overridden arguments payload - Contro1's response.comment is a natural place to put that override if you want the operator to adjust before approving.

Does this work with the Assistants API?

The pattern is the same: map required_action → Contro1 request → submit_tool_outputs with the operator decision. See our managed-agents example for Claude and adapt it.

Why the external_request_id with run_id + call_id?

It guarantees idempotency. If your wrapper crashes and the loop retries the same interruption, Contro1 returns the original request instead of creating a duplicate.