Wire Pushary human-in-the-loop into your agent framework with one tool and one enroll call, and connect your end-users' phones in one tap

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

The whole thing is two calls

Whatever framework you use, the integration is the same shape:

Connect a phone once. enroll(externalId) returns a link. Show it to your end-user. One tap turns on approvals. No account, no app store detour required, no charge to them.
Ask a human. ask({ question, externalId }) creates a decision, delivers it to that person, and blocks until they answer. You get a fail-closed approved flag back.

Everything else, durable delivery across native app push, PWA push notifications and Slack, webhook retries, the audit trail, is already handled by Pushary.

externalId is your own stable id for the end-user (a user id, email hash, tenant key). Use the same value for enroll and every ask for that person.

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 (1.3.0+) 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

import { generateText, stepCountIs } from 'ai'
import { openai } from '@ai-sdk/openai'
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.',
})

The agent gets an askHuman tool. A declined, expired, or unanswered confirm is reported to the model as "not approved, do not proceed."

Eve

npm i @pushary/eve

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

Verified against eve@0.24.x. Eve is in public preview, so pin your eve version.

LangGraph and LangChain

A LangGraph node is a plain function, so the ask goes inside it. The same call works in any LangChain @tool.

def approve_transfer(state):
    d = px.decisions.ask(
        question="Approve this transfer?",
        external_id=state["user_id"],
        type="confirm",
    )
    return {"approved": d["approved"]}

For long waits, pause with LangGraph's interrupt() instead, create the decision with callback_url, and resume with Command(resume=answer) from your webhook route. Two footguns: nodes re-run from the top on resume (use a deterministic idempotency_key so the re-run returns the same decision), and the callback body is {correlationId, answer, answeredAt} with no context echo (keep your own correlationId -> thread_id map).

CrewAI

Replace the console human_input=True prompt with a tool.

from crewai.tools import BaseTool

class AskHuman(BaseTool):
    name: str = "ask_human"
    description: str = "Ask a person to approve before acting."

    def _run(self, question: str) -> str:
        d = px.decisions.ask(question=question, external_id="user_123", type="confirm")
        return "approved" if d["approved"] else f"not approved ({d['status']})"

OpenAI Agents SDK

from agents import function_tool

@function_tool
def ask_human(question: str) -> str:
    d = px.decisions.ask(question=question, external_id="user_123", type="confirm")
    return "approved" if d["approved"] else f"not approved ({d['status']})"

The TypeScript Agents SDK works the same way with a tool() whose execute calls pushary.decisions.ask. Route needsApproval interruptions to the same call so a person resolves them instead of your code guessing.

Mastra

For a blocking approval, a createTool from @mastra/core/tools:

import { createTool } from '@mastra/core/tools'
import { z } from 'zod'

export const askHuman = createTool({
  id: 'ask-human',
  description: 'Ask a person to approve before acting.',
  inputSchema: z.object({ question: z.string() }),
  outputSchema: z.object({ approved: z.boolean() }),
  execute: async ({ question }) => {
    const d = await pushary.decisions.ask({ question, externalId: 'user_123', type: 'confirm' })
    return { approved: d.approved }
  },
})

For long waits, use a workflow step with suspend/resume: create the decision with a callbackUrl inside the step, suspend, and drive run.resume(...) 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, create the decision with a callbackUrl and let your workflow park until the signed webhook arrives.

const { decisionId } = await pushary.decisions.create({
  externalId: user.id,
  question: 'Approve this transfer?',
  callbackUrl: 'https://yourapp.com/webhooks/pushary',
  idempotencyKey: `${runId}:approve-transfer`,
})
// store decisionId -> runId in your own state, then suspend the workflow.

The callback body is { correlationId, answer, answeredAt }. Verify it with verifyWebhookSignature, look up your run by correlationId, and resume. The callback does not echo your own context, so keep the correlationId -> run map yourself.

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

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 uses the same durable decision ledger, so answers are never lost even if your process restarts mid-wait.

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

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

The whole thing is two calls

Whatever framework you use, the integration is the same shape:

Connect a phone once. enroll(externalId) returns a link. Show it to your end-user. One tap turns on approvals. No account, no app store detour required, no charge to them.
Ask a human. ask({ question, externalId }) creates a decision, delivers it to that person, and blocks until they answer. You get a fail-closed approved flag back.

Everything else, durable delivery across native app push, PWA push notifications and Slack, webhook retries, the audit trail, is already handled by Pushary.

externalId is your own stable id for the end-user (a user id, email hash, tenant key). Use the same value for enroll and every ask for that person.

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 (1.3.0+) 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

import { generateText, stepCountIs } from 'ai'
import { openai } from '@ai-sdk/openai'
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.',
})

The agent gets an askHuman tool. A declined, expired, or unanswered confirm is reported to the model as "not approved, do not proceed."

Eve

npm i @pushary/eve

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

Verified against eve@0.24.x. Eve is in public preview, so pin your eve version.

LangGraph and LangChain

A LangGraph node is a plain function, so the ask goes inside it. The same call works in any LangChain @tool.

def approve_transfer(state):
    d = px.decisions.ask(
        question="Approve this transfer?",
        external_id=state["user_id"],
        type="confirm",
    )
    return {"approved": d["approved"]}

CrewAI

Replace the console human_input=True prompt with a tool.

from crewai.tools import BaseTool

class AskHuman(BaseTool):
    name: str = "ask_human"
    description: str = "Ask a person to approve before acting."

    def _run(self, question: str) -> str:
        d = px.decisions.ask(question=question, external_id="user_123", type="confirm")
        return "approved" if d["approved"] else f"not approved ({d['status']})"

OpenAI Agents SDK

from agents import function_tool

@function_tool
def ask_human(question: str) -> str:
    d = px.decisions.ask(question=question, external_id="user_123", type="confirm")
    return "approved" if d["approved"] else f"not approved ({d['status']})"

Mastra

For a blocking approval, a createTool from @mastra/core/tools:

import { createTool } from '@mastra/core/tools'
import { z } from 'zod'

export const askHuman = createTool({
  id: 'ask-human',
  description: 'Ask a person to approve before acting.',
  inputSchema: z.object({ question: z.string() }),
  outputSchema: z.object({ approved: z.boolean() }),
  execute: async ({ question }) => {
    const d = await pushary.decisions.ask({ question, externalId: 'user_123', type: 'confirm' })
    return { approved: d.approved }
  },
})

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

Durable workflows (Inngest, Temporal, Vercel Workflow)

For waits of minutes or hours that must not hold idle compute, create the decision with a callbackUrl and let your workflow park until the signed webhook arrives.

const { decisionId } = await pushary.decisions.create({
  externalId: user.id,
  question: 'Approve this transfer?',
  callbackUrl: 'https://yourapp.com/webhooks/pushary',
  idempotencyKey: `${runId}:approve-transfer`,
})
// store decisionId -> runId in your own state, then suspend the workflow.

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.

What to reach for

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 uses the same durable decision ledger, so answers are never lost even if your process restarts mid-wait.

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

Framework adapters

The whole thing is two calls

Any framework: `@pushary/server`

Any framework, in Python: `pushary`

Vercel AI SDK

Eve

LangGraph and LangChain

CrewAI

OpenAI Agents SDK

Mastra

Hermes

OpenClaw

Claude Agent SDK

Durable workflows (Inngest, Temporal, Vercel Workflow)

No code at all: MCP

What to reach for

On this page

Framework adapters

The whole thing is two calls

Any framework: `@pushary/server`

Any framework, in Python: `pushary`

Vercel AI SDK

Eve

LangGraph and LangChain

CrewAI

OpenAI Agents SDK

Mastra

Hermes

OpenClaw

Claude Agent SDK

Durable workflows (Inngest, Temporal, Vercel Workflow)

No code at all: MCP

What to reach for

On this page