Framework adapters

This page is for teams building an AI agent who want to pause it and ask a real human, delivered to that person's phone. It builds on Embed human approval and runs on the Partner plan.

Start here: ask or gate

There are two ways to involve a human, and picking the wrong one is the mistake that costs people a weekend.

Ask gives the model a tool it can call. Good for "check with someone before you guess". The catch is that it is the model's choice, and a model in a hurry can decide not to call it.

Gate hooks the framework's own approval mechanism. The runtime asks the human before the tool function runs. There is no path around it, because the model never gets a say. Use this for anything that spends money, deletes data, or leaves your system.

Most products want both: a general ask_human tool, plus a gate on the two or three tools that can actually hurt.

What you are not building

Both paths are a few lines because the hard parts already exist. If you wire a human into an agent yourself, this is the list you end up owning:

The honest version of the pitch: the ask is easy, and everything after it is not.

The contract underneath

Whatever framework you use, and whichever mode you pick, it is the same two calls:

1. Connect a phone once. enroll(externalId) returns a link. Show it to your end-user. One tap opens the Pushary app, which is what puts Approve and Deny on their lock screen. No account and no charge to them. Show the link as a QR they can re-scan rather than a tap-through: iOS does not re-deliver a universal link after an App Store install, so someone installing the app for the first time finishes by scanning it again from inside the app.
2. Ask a human. ask({ question, externalId }) creates a decision, delivers it, and blocks until they answer. You get a fail-closed approved flag back.

Any framework: @pushary/server

The core SDK is the floor under every adapter. If your framework can call a function, this works.

npm i @pushary/server

import { createPusharyServer } from '@pushary/server'

const pushary = createPusharyServer({ apiKey: process.env.PUSHARY_API_KEY! })

// once per end-user: show the link, they tap to connect their phone
const { universalLink } = await pushary.enroll(user.id)

// whenever your agent needs a human:
const { approved } = await pushary.decisions.ask({
  externalId: user.id,
  question: 'Issue a $50 refund?',
  type: 'confirm', // confirm | select | input
})
if (approved) await issueRefund()

Any framework, in Python: pushary

The Python SDK carries the same two calls with snake_case arguments and dict returns.

pip install pushary

import os
from pushary import PusharyServer

px = PusharyServer(api_key=os.environ["PUSHARY_API_KEY"])

# once per end-user: show the link, they tap to connect their phone
link = px.enroll("user_123")["universalLink"]

# whenever your agent needs a human:
d = px.decisions.ask(
    question="Issue a $50 refund?",
    external_id="user_123",
    type="confirm",  # confirm | select | input
)
if d["approved"]:
    issue_refund()

Vercel AI SDK

npm i @pushary/ai-sdk ai zod

Ask. The agent gets an askHuman tool it can call:

import { generateText, stepCountIs } from 'ai'
import { createPusharyTools, enroll } from '@pushary/ai-sdk'

await enroll({ apiKey: process.env.PUSHARY_API_KEY! }, user.id) // show the link once

const { text } = await generateText({
  model: 'openai/gpt-4o',
  tools: createPusharyTools({ apiKey: process.env.PUSHARY_API_KEY!, externalId: user.id }),
  stopWhen: stepCountIs(10),
  prompt: 'Issue the refund only if a human approves it.',
})

Gate. pusharyApproval plugs into the AI SDK's own toolApproval, which runs before a tool executes:

import { pusharyApproval } from '@pushary/ai-sdk'

const { text } = await generateText({
  model: 'openai/gpt-4o',
  tools: { issueRefund, lookupOrder },
  // only issueRefund asks a human; lookupOrder runs untouched
  toolApproval: pusharyApproval({ externalId: user.id, tools: ['issueRefund'] }),
  prompt: 'Refund order 1234.',
})

Drop tools to gate every call. For a single tool, pusharyToolApproval takes the per-tool form:

toolApproval: {
  issueRefund: pusharyToolApproval({ toolName: 'issueRefund', externalId: user.id }),
}

toolApproval does not exist in ai@5, so the gate needs a newer ai. The ask path above works from ai@5 on.

Eve

npm i @pushary/eve

Ask. Eve discovers tools by file. Drop in two one-line files:

// agent/tools/ask-human.ts
import { pusharyAskHuman } from '@pushary/eve'
export default pusharyAskHuman()

// agent/tools/connect-phone.ts
import { pusharyConnectPhone } from '@pushary/eve'
export default pusharyConnectPhone()

By default each tool asks the session principal, so run user-scoped auth and each end-user is their own principal. To bind a fixed end-user, pass one: pusharyAskHuman({ externalId: user.id }).

Gate. Eve has a per-tool approval field it evaluates before execute. Pass pusharyApproval() and the tool cannot run without a yes:

// agent/tools/issue-refund.ts
import { defineTool } from 'eve/tools'
import { z } from 'zod'
import { pusharyApproval } from '@pushary/eve'

export default defineTool({
  description: 'Refund an order',
  inputSchema: z.object({ amount: z.number(), customer: z.string() }),
  approval: pusharyApproval({ externalId: (ctx) => ctx.toolInput?.customer }),
  execute: async ({ amount }) => refund(amount),
})

It sits alongside Eve's built-in always(), never() and once(). Those decide statically; this one asks a person and waits.

The channel. pusharyChannel() is the third option, and the one to reach for in a real product. It puts everything Eve already pauses on onto a phone: any tool gated with approval, plus the built-in ask_question. Eve parks the turn durably while it waits, so an approval can sit for hours at zero idle compute.

// agent/channels/pushary.ts
import { pusharyChannel } from '@pushary/eve'
export default pusharyChannel()

PUSHARY_API_KEY=pk_...sk_...
PUSHARY_WEBHOOK_SECRET=whsec_...          # decisions.getWebhookSecret()
PUSHARY_CALLBACK_ORIGIN=https://your-agent.vercel.app

The channel also exposes POST /pushary/message, /pushary/stop and /pushary/reset, so the rest of the session is reachable from the phone too.

LangGraph and LangChain

Install pushary-langgraph (@pushary/langgraph in TypeScript).

pip install pushary-langgraph

Ask. A blocking approval inside a node:

from pushary_langgraph import connect, ask_human

connect("user_123")  # show the link once

def approve_transfer(state):
    d = ask_human("Approve this transfer?", external_id=state["user_id"], node="approval")
    return {"approved": d["approved"]}

Gate, durable. For waits longer than a request can hold, pushary_interrupt(..., callback_url=...) opens the decision and parks the graph with LangGraph's native interrupt(). Resume with Command(resume=answer) from your webhook route.

It handles the two footguns for you. It derives a deterministic idempotency key from external_id + node + question, so the node's re-run on resume hits the same decision instead of paging twice, and resolve_pushary_callback(raw, signature, secret) verifies and parses the callback in one call. Key your resume off correlationId, or pass context: thread_id at create time to carry your own resume key through.

In TypeScript the same pair is createAskHumanTool(config, { externalId }) and pusharyInterrupt(config, input). The blocking ask_human also works inside any LangChain @tool.

CrewAI

Install pushary-crewai and give the agent a ready-made tool instead of the console human_input=True prompt.

pip install pushary-crewai

from crewai import Agent
from pushary_crewai import connect, make_ask_human_tool

connect("user_123")  # show the link once

agent = Agent(role="Ops", goal="Ship safely", tools=[make_ask_human_tool("user_123")])

The external_id is bound to the tool, so a prompt-injected agent cannot ask the wrong person. The tool blocks fail-closed until they answer.

OpenAI Agents SDK

Install pushary-openai-agents (@pushary/openai-agents in TypeScript).

pip install pushary-openai-agents

Ask. A function tool bound to one end-user:

from agents import Agent
from pushary_openai_agents import connect, pushary_tool

connect("user_123")  # show the link once

agent = Agent(name="Support", instructions="Ask a human before risky steps.",
              tools=[pushary_tool("user_123")])

In TypeScript the same tool is pusharyTool(config, { externalId }), added to new Agent({ tools: [...] }). The externalId is bound in code, never taken from model input, so a prompt-injected model cannot ask the wrong person.

Gate. The OpenAI Agents SDK splits approval 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)

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, and the run continues or stops on the answer. TypeScript is the same shape:

import { pusharyNeedsApproval, resolvePusharyInterruptions } from '@pushary/openai-agents'

let result = await run(agent, 'Refund order 1234')
while (result.interruptions?.length) {
  const outcome = await resolvePusharyInterruptions(
    { externalId: user.id },
    { interruptions: result.interruptions, state: result.state },
  )
  if (!outcome.allApproved) break
  result = await run(agent, result.state)
}

A denial is handed back to the model as the rejection message, so it knows why it was stopped rather than retrying blindly.

Mastra

Install @pushary/mastra.

npm i @pushary/mastra

Ask. A blocking tool for any agent:

import { createPusharyAskTool, connect } from '@pushary/mastra'

await connect({ apiKey: process.env.PUSHARY_API_KEY! }, 'user_123') // show the link once

export const askHuman = createPusharyAskTool(
  { apiKey: process.env.PUSHARY_API_KEY! },
  { externalId: 'user_123' },
) // add to any Agent({ tools: { askHuman } })

Gate. Like the OpenAI Agents SDK, Mastra's requireApproval only decides that a human is needed; the run then suspends. resolvePusharyApprovals asks the person and resumes or declines:

import { pusharyRequireApproval, resolvePusharyApprovals } from '@pushary/mastra'

const issueRefund = createTool({
  id: 'issue-refund',
  requireApproval: pusharyRequireApproval(),
  // ...
})

const output = await agent.generate('Refund order 1234', { requireToolApproval: true })
if (output.finishReason === 'suspended') {
  await resolvePusharyApprovals({ externalId: user.id }, { agent })
}

With no runs passed it lists the agent's own suspended runs, so a background worker can drain approvals for a whole thread with the same call. Decisions are keyed on runId plus toolCallId, so running it twice against the same suspended run resolves to the same decision rather than asking again.

Long waits. pusharyApprovalStep(config, { callbackUrl }) is the workflow form: it suspends, and resolvePusharyCallback plus run.resume(...) drives it from the signed webhook. Mastra persists the snapshot, so the wait holds no compute.

Hermes

pip install pushary into Hermes' environment, then register a tool in a plugin:

def ask_human(params, **kwargs):
    d = px.decisions.ask(
        question=params["question"],
        external_id=params["external_id"],
        type="confirm",
    )
    return "approved" if d["approved"] else f"not approved ({d['status']})"

def register(ctx):
    ctx.register_tool(
        name="ask_human",
        toolset="approvals",
        schema={"type": "object", "properties": {
            "question": {"type": "string"},
            "external_id": {"type": "string"},
        }, "required": ["question", "external_id"]},
        handler=ask_human,
        description="Ask a person to approve before acting.",
    )

In gateway mode, pass each user's external_id from your own user model rather than a session id.

OpenClaw

Gate risky tool calls with a before_tool_call hook:

import { definePluginEntry } from 'openclaw/plugin-sdk/plugin-entry'

export default definePluginEntry({
  id: 'pushary-approvals',
  name: 'Pushary approvals',
  register(api) {
    api.on('before_tool_call', async (event) => {
      if (event.toolName !== 'deploy_service') return
      const d = await pushary.decisions.ask({
        question: `Allow ${event.toolName}?`,
        externalId: 'user_123',
        type: 'confirm',
      })
      if (!d.approved) return { block: true, blockReason: 'A human declined this action.' }
    })
  },
})

Claude Agent SDK

The SDK's MCP support is the integration. One mcpServers entry, no install:

import { query } from '@anthropic-ai/claude-agent-sdk'

for await (const message of query({
  prompt: 'Clean up stale feature flags. Ask before deleting anything.',
  options: {
    mcpServers: {
      pushary: {
        type: 'http',
        url: 'https://pushary.com/api/mcp/mcp',
        headers: { Authorization: `Bearer ${process.env.PUSHARY_API_KEY}` },
      },
    },
    allowedTools: ['mcp__pushary__ask_user', 'mcp__pushary__wait_for_answer', 'mcp__pushary__send_notification'],
  },
})) {
  // ...
}

These MCP tools reach the phones connected to your workspace. When the agent serves your own end-users, call decisions.ask from @pushary/server inside a custom tool with each user's externalId instead.

Durable workflows (Inngest, Temporal, Vercel Workflow)

For waits of minutes or hours that must not hold idle compute, install @pushary/durable. It has no framework dependency of its own, so it drops into whichever runner you use.

npm i @pushary/durable

import { createApproval, resolveApproval, deterministicKey } from '@pushary/durable'

// inside a step: open the decision and park the run
const { correlationId } = await createApproval(
  { apiKey: process.env.PUSHARY_API_KEY! },
  {
    externalId: user.id,
    question: 'Approve this transfer?',
    callbackUrl: 'https://yourapp.com/webhooks/pushary',
    idempotencyKey: deterministicKey([runId, 'approve-transfer']),
  },
)
// store correlationId -> runId in your own state, then suspend/wait.

// in your callback route: verify + parse in one call
const approval = resolveApproval(rawBody, signature, process.env.PUSHARY_WEBHOOK_SECRET!)
if (approval?.approved) resumeRun(approval.correlationId)

resolveApproval verifies the HMAC signature first (a spoofed request never resumes your workflow), then returns { correlationId, answer, value, approved, context? }. Look up your run by correlationId, or pass context: runId at create time and read it straight off the callback. The README has copy-paste recipes for Inngest, Temporal, and Vercel Workflow.

Your own framework: @pushary/server/adapters

Every adapter above is a thin binding over one shared kernel, and that kernel is public. If you run an in-house harness, or a framework we have not shipped for, this is the same surface our own adapters are built on.

import { createAdapterKernel, renderApprovalQuestion } from '@pushary/server/adapters'

const kernel = createAdapterKernel('the Acme helpers')

// blocking ask, idempotency and fail-closed already handled
export const askHuman = kernel.askExternalUser

// durable create for a framework that parks its own run
export const openDecision = kernel.createDurableDecision

// an enforced gate: build it once, call it per tool call
const gate = kernel.createGate({ apiKey: process.env.PUSHARY_API_KEY! })

export const approve = async (toolName: string, callId: string, input: unknown) => {
  const decision = await gate({
    toolName,
    callId,
    sessionId: currentRunId,
    question: renderApprovalQuestion(toolName, input),
    externalId: kernel.requireExternalId(currentUserId),
    // optional: lets a rule decide on the arguments, not only the action name
    input,
  })
  return decision // { approved: true } | { approved: false, reason }
}

The gate asks your policy before it asks a person. A rule that names the action resolves it with nobody paged: allow returns approved and opens no decision, deny returns a reason the model reads, and anything no rule names still asks. A site with no rules behaves exactly as before, because a rule has to name the action and * is never selected. Pass policy: false to restore the always-ask gate.

The label you pass to createAdapterKernel is what appears in the error when a key or an end-user is missing, so it points at your helpers rather than ours.

One protected action: protect()

createGate answers a yes/no. protect() is that plus running the thing, so the authorize, escalate and execute steps stop being three pieces of plumbing in your business code.

const protect = kernel.protect({ apiKey: process.env.PUSHARY_API_KEY! })

const outcome = await protect({
  action: 'refund.create',
  target: 'order_4471',
  externalId: customer.id,
  facts: { amount: 4800, currency: 'EUR' },   // what a rule may decide on
  callId: attemptId,
  runId: sessionId,
  run: () => stripe.refunds.create({ charge, amount: 4800 }),
})

outcome.ok ? outcome.result : outcome.reason

A rule that names refund.create resolves it with nobody paged. Anything no rule names asks a person. run is called only after the action is authorized, and an error it throws is not caught: reporting it as ok: false would be indistinguishable from a refusal.

protect() is at most once. The decision is idempotent — a replay of the same runId + callId + action lands on the same approval instead of asking twice — and the approval is then spent against a durable permit before run is called. A retry, a concurrent worker and a resumed run all reach the same permit and exactly one of them proceeds; the rest come back ok: false with a reason worded to stop the model rather than invite another attempt.

The permit is bound to the exact subject that was authorized, so changing the amount between the approval and the execution leaves you with no permit for the action you now want to run. protect() then records succeeded or failed against it. A process that dies mid-action leaves the permit unresolved, which is visible and reconcilable, rather than an action that ran twice, which is not.

See Run it once, and prove it ran for the two calls underneath, and the refusals you can branch on.

Python is the same two-step surface: protect = kernel.create_protect(), then protect(action, run, external_id=..., call_id=..., run_id=...). Building it once is what makes a missing key raise where the protector is defined rather than on the first action.

A runnable end-to-end version, against the sandbox and with no phone involved, is in examples/protected-action.ts in @pushary/server.

Also on the subpath: describeAnswer turns an outcome into an instruction the model cannot misread, resolvePusharyCallback verifies and parses a webhook in one call (SIGNATURE_HEADER is the header to read it from), and idempotencyKeyFor builds the re-run-safe key from externalId + node + question. isApproved is the fail-closed check itself, if you only want the rule.

Python has the same surface as pushary.adapters:

from pushary.adapters import AdapterKernel, render_approval_question

kernel = AdapterKernel("the Acme helpers")
ask_human = kernel.ask_human
gate = kernel.create_gate()

create_gate(policy=False) is the same escape hatch. Neither language evaluates a rule locally; the verdict is the server's, so the two cannot disagree about what a policy means.

Both give you the same guarantees the shipped adapters have: idempotency keyed on the call so a replay never asks twice, a denial reason the model can read, and silence treated as a no.

No code at all: MCP

If your agent speaks the Model Context Protocol, point it at the Pushary MCP server with your API key and skip the SDK entirely. See Connect any agent.

On the Partner plan, the same server exposes the end-user decision tools, so an MCP agent can run the two-call flow with no SDK:

• enroll_end_user({ externalId }) mints the keyless one-tap connect link for one of your users.
• create_decision({ question, externalId, ... }) writes a durable decision and delivers it to that user's phone.
• get_decision({ decisionId, waitSeconds }) is the durable wait: loop it (waitSeconds caps at 20) until the status leaves pending.
• cancel_decision({ decisionId }) retracts a stale request.

Same fail-closed rule as the SDKs: only an answered, approving decision is a yes. The API key holder controls which externalId a decision targets, so treat the key like the production credential it is.

What to reach for

• The model should decide when to ask: the ask tool.
• This must not happen without a yes: the gate.
• One synchronous approval in a request that finishes in under a minute: ask().
• Long waits or crash-safe resume: create() + callbackUrl under a durable runner.
• Zero code: MCP.

Every path writes to the same durable decision ledger, so an answer is never lost even if your process restarts mid-wait, and every one of them shows up in the same audit log.

Each framework also has a condensed integration page with the end-user story: the human-in-the-loop hub links to all of them.