Framework adapters

Pushary has a human in the loop adapter for each major agent framework. Each one hooks the framework's own approval step, so your user approves or denies a tool call on their phone before it runs. If nobody answers, the tool does not run. Every adapter is a thin layer over the same core SDK and runs on the Partner plan.

New to this? Start with the builder quickstart. It covers the two calls every adapter makes.

Choose your adapterCopy Anchor Link

Ask or gate: pick the right oneCopy Anchor Link

There are two ways to involve a person.

Ask gives the model a tool it can call. It suits "check with someone before you guess". The model decides whether to call it, and a model in a hurry can skip it.

Gate hooks the framework's own approval step. The runtime asks the person before the tool function runs, and the model never gets a say. Use it 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 do real harm.

What the adapters handle for youCopy Anchor Link

If you wire a person into an agent yourself, this is the list you end up owning:

The two calls under every adapterCopy Anchor Link

Whatever framework you use, it comes down to 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 puts Approve and Deny on their lock screen. It costs them nothing and needs no account. Show the link as a QR code they can scan again: iOS does not reopen a universal link after an App Store install, so a first-time user finishes by scanning it from inside the app.
2. Ask a person. ask({ question, externalId }) creates a decision, delivers it and waits up to 55 seconds for the answer. You get back a fail-closed approved flag.

Any framework: the core SDKCopy Anchor Link

If your framework can call a function, the core SDK works. The Python SDK has the same two calls with snake_case arguments and dict returns.

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 person:
const { approved } = await pushary.decisions.ask({
  externalId: user.id,
  question: 'Issue a $50 refund?',
  type: 'confirm', // confirm | select | input
})
if (approved) await issueRefund()

Vercel AI SDKCopy Anchor Link

Full walkthrough: Vercel AI SDK guide.

pusharyApproval() goes in the AI SDK's toolApproval, so the SDK asks before the tool runs. createPusharyTools adds an optional askHuman question tool. A prompt that asks the model to seek permission does not protect another tool. Set policy: false when a person must answer. Your server binds externalId to the signed-in customer, then shows their enrollment link.

@pushary/ai-sdk 0.3 requires @pushary/server 2.1. For late answers, the delayed-review recipe saves the AI SDK messages and application state in SQLite. It keeps generated output and refuses to re-run an uncertain call. It needs AI SDK 7 and Node.js 22.13 or later. Your application owns persistence and recovery.

LangGraph and LangChainCopy Anchor Link

Full walkthrough: LangGraph guide.

An approval node calls pusharyInterrupt (Python: pushary_interrupt), and the graph routes to the risky node only on a confirm answer of 'yes'. Select and text answers do not grant permission. createAskHumanTool is an optional question tool for prebuilt LangChain agents.

A blocking question suits a short wait. For delayed answers, use a persistent LangGraph checkpointer, save the decision-to-thread mapping, and resume only after you read and validate the decision. @pushary/langgraph 0.4 and pushary-langgraph 0.4 require the shared SDK 2.1 and bind each review to its recipient and action. Follow the native interrupt guide and its restart test. Do not mix examples from different package versions. Your worker still handles callbacks, duplicate resumes, output storage and uncertain business effects.

OpenAI Agents SDKCopy Anchor Link

Full walkthrough: OpenAI Agents SDK guide. For OpenAI-hosted sessions, see the hosted Agents API recipe.

pusharyTool (Python: pushary_tool) lets the model ask a question. Native needsApproval (Python: needs_approval) protects a tool before it runs, and resolvePusharyInterruptions answers the interruptions. Your server picks the customer, shows their enrollment link, and resolves only the interruptions of that customer's saved run. Set policy: false when a person must answer.

@pushary/openai-agents 0.4 and pushary-openai-agents 0.4 require the shared SDK 2.1. The TypeScript delayed-answer recipe saves native RunState, fixed call bindings, claims and generated output in SQLite. It needs Node.js 22.13 or later and handles answers after a process restart without re-running uncertain calls. The Python resolver blocks inside one request, so Python callers build durable scheduling themselves.

MastraCopy Anchor Link

Full walkthrough: Mastra guide.

Set Mastra's requireApproval: true on the tool, then call resolvePusharyApprovals for the trusted suspended runs of this customer and action. Set policy: false when a person must answer. Save every returned generation and check pending and error results. A model-callable ask tool protects nothing else. Confirm yes or no is permission to execute. A choice or typed answer is data, even the word yes.

Before you open a review for a product customer, connect them with their enrollment link, wait for consent, and check reachability. An owner demo on your own phone is separate from connecting a customer.

For delayed answers, openPusharyAgentReview saves the exact call and customer binding, and resumePusharyAgentReview reads the authoritative decision before it resumes the saved run. @pushary/mastra 0.4 requires @pushary/server 2.1, @mastra/core 1.59 or later below 2, and Node.js 22.13 or later. The restart and live-phone recipe needs no model key and moves no money. Simulated HTTP checks alone do not prove phone delivery.

Claude Agent SDKCopy Anchor Link

Full walkthrough: Claude Agent SDK guide.

@pushary/claude-agent-sdk is a canUseTool. The SDK calls it before a tool that needs permission runs, and it returns your user's answer from their phone.

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

for await (const message of query({
  prompt: 'Refund order 1234',
  options: { canUseTool: pusharyCanUseTool({ externalId: user.id }) },
})) {
  if (message.type === 'result') console.log(message.subtype)
}

For your own agents, the SDK's MCP support works with no install. One mcpServers entry puts the hosted server's tools in front of the model:

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'],
  },
})) {
  if (message.type === 'result') console.log(message.subtype)
}

These MCP tools reach the phones connected to your workspace. When the agent serves your own end users, use canUseTool with each user's externalId.

EveCopy Anchor Link

npm i @pushary/eve

@pushary/eve needs eve@0.31 or newer and Node.js 24 or later.

Ask. Eve finds tools by file. Add 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 evaluates a per-tool approval field 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() }),
  approval: pusharyApproval({ externalId: (ctx) => ctx.session.auth.current?.principalId }),
  execute: async ({ amount }) => refund(amount),
})

The person who answers is the signed-in user of the session. Never take it from ctx.toolInput: the model writes the tool input, so a prompt injection could send the approval to a different customer. It sits next to Eve's built-in always(), never() and once(). Those decide on their own. This one asks a person and waits.

Channel. pusharyChannel() is the option to use 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 with no idle compute.

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

PUSHARY_API_KEY=pk_xxx.xxx
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.

Pydantic AICopy Anchor Link

pushary-pydantic-ai adds PusharyApprovals, a Pydantic AI capability. When the model calls a tool marked requires_approval=True, it asks your user on their phone and waits. Approve runs the tool in the same run. Deny, or no answer in time, returns a denial the model can read, and the tool does not run. The decision page shows every argument. A call too long to show in full is denied without asking anyone.

pip install pushary-pydantic-ai

from pydantic_ai import Agent
from pushary_pydantic_ai import PusharyApprovals

agent = Agent(
    'openai:gpt-4.1-mini',
    capabilities=[PusharyApprovals(external_id=customer.id)],
)

@agent.tool_plain(requires_approval=True)
def issue_refund(order_id: str) -> str:
    return refund_order_once(order_id)

For a multi-tenant agent, pass a function that reads the person from trusted deps, such as external_id=lambda ctx: ctx.deps.user_id, never from the tool input. The run waits timeout_seconds (55 by default) and the question expires with it. Set policy=False when a person must always answer.

Use run, iter or run_stream_events. run_stream ends before deferred calls are handled. Under Temporal, DBOS or Prefect the wait runs as a durable operation. To wait minutes or hours without holding a worker, let the run pause with DeferredToolRequests and resume it with create_reviews and resolve_reviews. The package README covers that flow, the ask_human question tool and the storage it needs.

CrewAICopy Anchor Link

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

link = connect("user_123")  # show this link to the user 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 until they answer and fails closed.

HermesCopy Anchor Link

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

import os
from pushary import PusharyServer

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

def ask_human(params, **kwargs):
    d = px.decisions.ask(
        question=params["question"],
        external_id=APPROVER_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"},
        }, "required": ["question"]},
        handler=ask_human,
        description="Ask a person to approve before acting.",
    )

The approver comes from your code, not from the tool arguments, so a prompt-injected agent cannot send the question to the wrong person. In gateway mode, look up each user's external_id in your own user model for the session that is running. Never let the model fill it in.

OpenClawCopy Anchor Link

Gate risky tool calls with a before_tool_call hook:

import { definePluginEntry } from 'openclaw/plugin-sdk/plugin-entry'
import { createPusharyServer } from '@pushary/server'

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

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.' }
    })
  },
})

Durable workflows (Inngest, Temporal, Vercel Workflow)Copy Anchor Link

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 fits 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 or wait.

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

idempotencyKey is required and must stay the same across retries and replays. A durable step runs again on resume, so a per-attempt key would ask your user twice.

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

Build your own adapterCopy Anchor Link

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 without an adapter, build on the same surface:

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

const kernel = createAdapterKernel('the Acme helpers')

// Blocking ask, with idempotency and fail-closed handling built in.
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, authorization? } | { approved: false, reason }
}

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

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

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 names the header to read.
• idempotencyKeyFor keeps an explicit idempotencyKey, or makes a fresh key for an independent ask.
• isApproved is the fail-closed check on its own.

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 always-ask option. Neither language evaluates a rule locally. The verdict comes from the server, so the two cannot disagree about what a policy means.

Both give you the guarantees the shipped adapters have: explicit operation keys deduplicate retries, independent blocking calls get fresh identities, durable creates require a key, and unanswered decisions count as a no.

Protect one action with protect()Copy Anchor Link

createGate answers yes or no. protect() does that and then runs the action, so authorize, escalate and execute are one call in your business code.

protect() uses the authorization and execution-permit endpoints. Configure it with a separately authorized server credential. The onboarding runtime key cannot call those endpoints. Keep that credential away from the model and the customer's browser.

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 settles it with nobody paged. Anything no rule names asks a person. run is called only after the action is authorized. An error it throws is not caught, because reporting it as ok: false would look the same as a refusal.

protect() runs the action at most once. The decision is idempotent: a replay of the same runId, callId and action lands on the same approval instead of asking twice. 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, not to invite another try.

The permit is bound to the exact subject that was authorized. Change the amount between approval and execution and you have no permit for the new action. protect() then records succeeded or failed against the permit. A process that dies mid-action leaves the permit unresolved, which you can see and reconcile, instead of an action that ran twice.

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

In Python, build it once with protect = kernel.create_protect(), then call protect(action, run, external_id=..., call_id=..., run_id=...). Building it once makes a missing key fail where the protector is defined, not on the first action.

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

No code: MCPCopy Anchor Link

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

On the Partner plan, an MCP connection with the right authorization exposes end-user decision tools. Use a recipient-bound credential issued by your trusted backend. The onboarding runtime key's REST access does not grant these MCP tools. An MCP agent can then use:

• enroll_end_user({ externalId }) makes the one-tap connect link for one of your users. The user needs no account or key.
• 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 until the status leaves pending. waitSeconds caps at 20.
• cancel_decision({ decisionId }) withdraws a stale request.

The same fail-closed rule applies: only an answered, approving decision is a yes. Whoever holds the API key controls which externalId a decision targets, so treat the key like the production credential it is. See the MCP reference.

Which one to useCopy Anchor Link

Every path writes to the same decision store, so an answer survives a restart of your process. Every decision shows up in the same audit log.

Next stepsCopy Anchor Link