pushary-openai-agents

Human-in-the-loop for the OpenAI Agents SDK (Python). A function tool that asks a real human to approve, delivered to their phone, and blocks on a fail-closed answer.

Requires the Pushary Partner plan.

Install

pip install pushary-openai-agents

Set PUSHARY_API_KEY (get it in your dashboard).

Connect a phone once

from pushary_openai_agents import connect

link = connect("user_123")  # show this to your end-user; one tap connects their phone

The tool

from agents import Agent, Runner
from pushary_openai_agents import pushary_tool

agent = Agent(
    name="Support",
    instructions="Call ask_human before issuing any refund.",
    tools=[pushary_tool("user_123")],
)
result = await Runner.run(agent, "Refund order 5?")

When the model calls the tool, Pushary delivers the question to that user's phone and the call blocks until they answer. The tool returns a fail-closed instruction. The external_id is bound when you build the tool, never taken from the model, so a prompt-injected agent cannot ask the wrong person.

Gating a tool the model cannot skip

pushary_tool is a tool the model chooses to call. That is right for "go ask someone about this", and wrong for "this must not happen without a yes", because a model that does not want to be interrupted can decline to call it.

The SDK's own gate splits in two: needs_approval decides whether a human is needed, and the run then stops with result.interruptions. Nothing asks anyone. Resolving those interruptions is the caller's job, and resolve_pushary_interruptions is that job done:

from agents import Agent, Runner, function_tool
from pushary_openai_agents import pushary_needs_approval, resolve_pushary_interruptions

@function_tool(needs_approval=pushary_needs_approval())
def issue_refund(amount: float) -> str:
    return charge_back(amount)

agent = Agent(name="Support", instructions="Refund when asked.", tools=[issue_refund])

result = await Runner.run(agent, "Refund order 1234")
while result.interruptions:
    outcome = resolve_pushary_interruptions(result, external_id="user_123")
    if not outcome.all_approved:
        break
    result = await Runner.run(agent, outcome.state)

Resume with outcome.state, not result.to_input_list(). The second replays the conversation without the decisions on it, so the model asks for the same tool again and the person gets paged twice.

Each interruption becomes one decision on the phone, resolved in order so the person sees one question at a time. A denial is handed back to the model as the rejection message, so it knows why it was stopped rather than retrying blindly.

Fail-closed: a denial, an expiry, or nobody answering all reject. For a multi-tenant product, resolve the end-user per interruption:

resolve_pushary_interruptions(
    result, external_id=lambda item: owner_of(item.raw_item.call_id)
)

Pass run_id= when you replay a run under ids you mint yourself, so the replay resolves to the same decisions instead of paging twice.

Durable approvals

For a wait longer than a request can hold, drive your own flow off ask_human with a callback_url on decisions.create and resolve the signed callback:

from pushary_openai_agents import resolve_pushary_callback, SIGNATURE_HEADER

def callback(request):
    cb = resolve_pushary_callback(request.body, request.headers.get(SIGNATURE_HEADER), SECRET)
    if not cb:
        return ("bad signature", 401)
    # look up your parked run by cb["correlationId"], approve/reject, resume
    return ("ok", 200)

For TypeScript, use npm i @pushary/openai-agents.

API

• connect(external_id, *, api_key=None, base_url=None) — enroll an end-user's phone.
• pushary_tool(external_id, *, name="ask_human", ...) — an OpenAI Agents function tool bound to that user.
• ask_human(question, *, external_id, type="confirm", ...) — blocking, returns the decision dict.
• pushary_needs_approval() — a needs_approval predicate that routes every call to a human.
• resolve_pushary_interruptions(result, *, external_id, run_id="", ...) — ask about each interruption, then approve or reject it on the run's context.
• resolve_pushary_callback(raw_body, signature, secret) — verify + parse a callback for the durable path.
• create_pushary_gate(...) — the raw fail-closed gate, for anything the helpers above do not cover.
• describe_answer(type, result), is_affirmative(answer), render_approval_question(tool, input), deterministic_key(parts), SIGNATURE_HEADER.

License

MIT