OpenAI Agents SDK Integration Python

Add Heed as a tool in the OpenAI Agents SDK. Before your agent executes a sensitive action (like sending a refund), Heed intercepts it and waits for human approval.

Prerequisites

Quick Start

pip install openai-agents httpx

export HEED_API_KEY="hd_live_your_key_here"
export OPENAI_API_KEY="sk-your-openai-key"

curl -X POST https://heed.run/api/v1/hitl/policies \
  -H "Content-Type: application/json" \
  -H "X-API-Key: $HEED_API_KEY" \
  -d '{
    "name": "gate-refunds",
    "description": "Require approval before issuing refunds",
    "rules": [
      {
        "action": "issue_refund",
        "requires_approval": true
      }
    ],
    "is_active": true
  }'

Complete Example

This agent can issue refunds, but every issue_refund action is gated by Heed. The agent calls the Heed tool, which blocks until a human approves or rejects.

"""
OpenAI Agents SDK + Heed: Gate issue_refund behind human approval.

Before the agent issues a refund, it calls the Heed tool.
If a policy requires approval, the tool blocks (polls) until
a human decides. If approved, the agent proceeds. If rejected,
it informs the user.

Requires: pip install openai-agents httpx
"""

import os
import time
import json
import httpx
from agents import Agent, Runner, function_tool

HEED_BASE = "https://heed.run/api/v1"
HEED_API_KEY = os.environ["HEED_API_KEY"]
POLL_INTERVAL = 3   # seconds between polls
MAX_POLL_WAIT = 300  # 5 minutes max

_headers = {
    "Content-Type": "application/json",
    "X-API-Key": HEED_API_KEY,
}


@function_tool
def request_approval(action: str, summary: str, context: str) -> str:
    """Request human approval before executing a consequential action.
    Blocks until a decision is made (approved/rejected) or times out.

    Args:
        action: The action being gated, e.g. "issue_refund"
        summary: Human-readable description of what will happen
        context: JSON string with relevant details (amounts, customer, etc.)

    Returns:
        "approved" or "rejected"
    """
    payload = {
        "agent_id": "openai-refund-agent",
        "action": action,
        "summary": summary,
        "context": json.loads(context) if context.startswith("{") else {"detail": context},
        "on_expire": "reject",
    }

    resp = httpx.post(f"{HEED_BASE}/hitl/requests", json=payload, headers=_headers)
    resp.raise_for_status()
    data = resp.json()

    # Auto-approved (no matching policy)
    if data.get("status") == "approved":
        return "approved"

    # Pending — poll until decided
    poll_url = data.get("poll_url")
    if not poll_url:
        return "approved"

    elapsed = 0
    while elapsed < MAX_POLL_WAIT:
        time.sleep(POLL_INTERVAL)
        elapsed += POLL_INTERVAL

        poll_resp = httpx.get(poll_url, headers=_headers)
        poll_resp.raise_for_status()
        status = poll_resp.json().get("status")

        if status in ("approved", "rejected"):
            return status

    return "rejected"  # timeout fallback


@function_tool
def issue_refund(order_id: str, amount: float, reason: str) -> str:
    """Issue a refund for an order. Must be approved via Heed first."""
    # In a real app this would call your payment processor
    return f"Refund of ${amount:.2f} issued for order {order_id}: {reason}"


# Build the agent
refund_agent = Agent(
    name="Refund Agent",
    instructions="""You are a refund assistant. When you need to issue a refund,
you MUST first call request_approval with action="issue_refund", a summary of
the refund, and context as a JSON string with keys "order_id", "amount", "reason".

Only proceed with issue_refund if request_approval returns "approved".
If it returns "rejected", tell the user the refund was not approved.""",
    tools=[request_approval, issue_refund],
)


# Run it
if __name__ == "__main__":
    result = Runner.run_sync(
        refund_agent,
        messages=[{"role": "user", "content": "Issue a $250 refund for order ORD-9876. The customer was double-charged."}],
    )
    print(result.final_output)

How It Works

1. The agent decides it needs to issue a refund and calls request_approval with action="issue_refund"
2. Heed checks the gate-refunds policy → the action matches a rule that requires approval
3. Heed returns status: "pending" with a poll_url
4. The tool polls poll_url every 3 seconds until a human approves or rejects
5. The tool returns "approved" or "rejected"
6. The agent proceeds with issue_refund only if approved

Creating the Policy

The policy gates the issue_refund action. Without it, every request auto-approves. Create it with one curl call:

curl -X POST https://heed.run/api/v1/hitl/policies \
  -H "Content-Type: application/json" \
  -H "X-API-Key: hd_live_your_key" \
  -d '{
    "name": "gate-refunds",
    "description": "Require human approval before issuing refunds",
    "rules": [
      {
        "action": "issue_refund",
        "requires_approval": true
      }
    ],
    "is_active": true
  }'

Add conditions to gate only high-value refunds:

"rules": [
  {
    "action": "issue_refund",
    "conditions": { "amount": { ">": 500 } },
    "requires_approval": true
  }
]

Async Pattern

For async agents, use httpx.AsyncClient and asyncio.sleep:

import asyncio
import httpx
from agents import function_tool

@function_tool
async def request_approval(action: str, summary: str, context: str) -> str:
    """Request human approval (async)."""
    payload = {
        "agent_id": "openai-refund-agent",
        "action": action,
        "summary": summary,
        "context": json.loads(context) if context.startswith("{") else {"detail": context},
        "on_expire": "reject",
    }

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{HEED_BASE}/hitl/requests", json=payload, headers=_headers
        )
        resp.raise_for_status()
        data = resp.json()

        if data.get("status") == "approved":
            return "approved"

        poll_url = data.get("poll_url")
        if not poll_url:
            return "approved"

        for _ in range(100):  # 5 min with 3s interval
            await asyncio.sleep(3)
            poll_resp = await client.get(poll_url, headers=_headers)
            poll_resp.raise_for_status()
            status = poll_resp.json().get("status")
            if status in ("approved", "rejected"):
                return status

    return "rejected"

Then run with the async runner:

result = await Runner.run(
    refund_agent,
    messages=[{"role": "user", "content": "Issue a $250 refund for order ORD-9876"}],
)

Using Callbacks Instead of Polling

Provide a callback_url and Heed will POST the decision when a human decides:

payload = {
    "agent_id": "openai-refund-agent",
    "action": "issue_refund",
    "summary": "Issue $250 refund for ORD-9876",
    "context": {"order_id": "ORD-9876", "amount": 250, "reason": "Double charge"},
    "on_expire": "reject",
    "callback_url": "https://your-server.com/heed/callback"
}

See the main docs for webhook payload format and HMAC signature verification.

Multi-Agent Pattern

In a multi-agent setup, use a dedicated "supervisor" agent that holds the Heed tool, and delegate consequential actions through it:

from agents import Agent, Runner, function_tool

# Specialist agents (no Heed access)
refund_agent = Agent(
    name="Refund Specialist",
    instructions="You handle refund calculations and details.",
    tools=[issue_refund],
)

# Supervisor with Heed gate
supervisor = Agent(
    name="Supervisor",
    instructions="""You oversee agent actions. Before delegating a refund
or any destructive action, call request_approval first. Only delegate
to the specialist if approved.""",
    tools=[request_approval],
)

Next Steps

• Full API documentation — all endpoints, authentication, webhook signing
• Swagger UI — interactive API explorer
• More integration guides — LangChain, Claude Code / MCP
• Four Modes of Oversight — from strict gating to circuit breakers