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.

Connect your OpenAI Agents SDK agent to Contro1

A OpenAI Agents SDK agent runs in your own code, so it connects with an Agent Credential: a key bound to one agent, so every call is attributed to it and nothing in a request can change which agent it is.

  • 1. Register the agent: contro1 init --name "<name>" --framework openai-agents, and finish the setup link it prints (purpose and owner).
  • 2. Connect the application account under Apps, if it is not connected yet.
  • 3. Give the agent the Actions it needs under Access. It starts with none.
  • 4. Create an Agent Credential for it (Settings, API keys) and store it as CONTRO1_API_KEY in your secret manager.
  • 5. Call Actions from your tools as below. The same credential creates approval requests for work your own code does.

Connect an agent: every path, in full

Run an application Action from a OpenAI Agents SDK tool

Contro1 holds the account and makes the call, so the record is what Contro1 observed. The tool returns what the Action produced; when a person has to approve first, it waits and then returns the result. It never re-submits: a retry could send a second email.

Before writing the input, read the exact input_schema with get_action_contract, or from the Action on the Access page.

agent.py
import os, uuid
from centcom import CentcomClient, needs_human_resolution

contro1 = CentcomClient(api_key=os.environ["CONTRO1_API_KEY"])  # an Agent Credential

def run_action(action_id: str, input: dict, *, account_mode: str = "shared", **kw):
    """Run a Contro1 Action and return what it produced. Never retries a send."""
    out = contro1.actions.invoke(
        action_id, input,
        authority_mode=kw.pop("authority_mode", "agent_principal"),
        account_mode=account_mode,
        idempotency_key=kw.pop("idempotency_key", str(uuid.uuid4())),
        **kw,
    )
    inv = out["invocation"]
    if inv["state"] == "executed":
        return out["result"]
    if inv["state"] == "awaiting_approval":
        settled = contro1.actions.wait_for_invocation(inv["invocation_id"])
        if needs_human_resolution(settled):
            raise RuntimeError("Outcome unknown; a person must check. Do not retry.")
        return contro1.actions.get_result(settled["invocation_id"])
    raise RuntimeError(f"Not run: {inv['state']} {out.get('not_run', '')}")

from agents import Agent, function_tool

@function_tool
def list_recent_emails(max_results: int = 5) -> list:
    """List the most recent emails in the team mailbox."""
    return run_action("gmail.message.list", {"max_results": max_results})

agent = Agent(name="Inbox assistant", tools=[list_recent_emails])

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 './contro1Client';

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, 3000, 600_000);
    if (decision.response?.approved) {
      result.state.approve(interruption);
    } else {
      result.state.reject(interruption, decision.response?.comment ?? 'Rejected in Contro1');
    }
  }
  result = await Runner.resume(result);
}

Send context the reviewer can trust

A request that only says "Approve this tool call?" forces the reviewer to rubber-stamp. Send three things with every gated call: the exact tool input your gate intercepted (machine-observed fact), the agent's own justification (make reason a required parameter of the risky tool, so the model produces it at decision time), and the trigger - the user message or event that started the run.

Keep the two kinds apart in context: facts your code observed versus text the model wrote. Agent-written justification is agent-reported evidence: it helps the reviewer decide, but it must never change routing, risk_level, or approval policy, because a prompt-injected agent writes very persuasive reasons. If a high-risk request arrives without this context, fail closed and reject it instead of asking a human to guess.

Reviewer context and trust rules

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 './contro1Client';

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, 3000, 600_000);
      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

Send full agent traceability

Beyond the approval call, attach a run trace, the tools you invoked, and the context you retrieved. Each field is optional; add what you have. Identity always comes from your credential. When a named part of a multi-agent system made the call, say which with the Contro1-Sub-Agent header (as_sub_agent in the SDKs): Contro1 registers it under this agent, and it never has more authority than this agent.

  • trace_id / parent_trace_id - link one run (and sub-agent runs) into a single trace.
  • tool_calls[] - what the agent tried to do, so reviewers see the actions. A hook can report every tool call for you.
  • retrieved_context[] - the data the decision was based on (RAG provenance).
  • Then export a signed evidence packet from GET /requests/:id/evidence.
Send full traceability
# Same approval call you already make - now with full traceability.
billing = client.as_sub_agent("billing")   # optional: a named part of this agent
billing.create_protocol_request({
    "request_type": "approval",
    "title": "Refund $4,200 to customer 8831",
    "source": {"integration": "openai-agents"},
    "trace_id": f"trc_{run_id}",          # link every step of this run
    "tool_calls": [{"name": "lookup_order", "outcome": "success"}],
    "retrieved_context": [{"source": "policy:refunds", "uri": "kb://policy/refunds"}],
    "continuation": {"mode": "decision"},
})

Agent identity, traceability & signed evidence

Make the approval the gate, not a suggestion

The signed webhook is cryptographic proof of a human decision. Verify it inside the system that executes the action - not inside the agent. Any tool that must never run without human sign-off (payments, deploys, data deletion) should refuse to act without a verified approval; that way no agent, including shadow agents nobody registered, can trigger it by skipping Contro1.

  • Verify the signature and reject timestamps older than 5 minutes (replay protection; the timestamp marks callback delivery, not request creation, so long SLAs are unaffected).
  • Bind the approval to the exact action parameters via metadata / correlation_id - never treat "an approval arrived" as permission for a different action.
  • Execute each request_id exactly once (idempotency on your side).
  • When in doubt, confirm state directly with GET /v1/requests/:id using a read-only API key.

Full guardrail pattern with code: Webhooks

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.

Related resources