@pushary/mastra

Full walkthrough: Human-in-the-loop for Mastra. Reaching your own end-users on their phones is the Pushary Partner plan.

Human-in-the-loop for Mastra. Ask a real human to approve, and get the answer on their phone. Two seams, pick by wait length:

• A blocking createTool for an approval that resolves in well under a minute.
• A durable workflow step (suspend/resume) for long waits. Mastra persists the snapshot, so the wait holds no idle compute and survives a restart.

Requires the Pushary Partner plan and @mastra/core v1.

Install

npm i @pushary/mastra @mastra/core zod

Set PUSHARY_API_KEY (get it in your dashboard).

Connect a phone once

import { connect } from '@pushary/mastra'
const { universalLink } = await connect({ apiKey: process.env.PUSHARY_API_KEY! }, user.id)

Blocking tool

import { createPusharyAskTool } from '@pushary/mastra'
import { Agent } from '@mastra/core/agent'

const askHuman = createPusharyAskTool({ apiKey: process.env.PUSHARY_API_KEY! }, { externalId: user.id })

const agent = new Agent({
  name: 'Support',
  instructions: 'Call ask-human before any refund.',
  model,
  tools: { askHuman },
})

The tool blocks until the person answers and returns { approved, value, status }, fail-closed. externalId is bound in code, never taken from model input.

Gating a tool the model cannot skip

createPusharyAskTool 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.

Mastra's own gate splits in two: requireApproval decides whether a human is needed, and the run then suspends. Nothing asks anyone. resolvePusharyApprovals asks the person and resumes or declines:

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

const issueRefund = createTool({
  id: 'issue-refund',
  description: 'Refund an order',
  inputSchema: z.object({ amount: z.number() }),
  outputSchema: z.object({ ok: z.boolean() }),
  requireApproval: pusharyRequireApproval(),
  execute: async ({ amount }) => ({ ok: await chargeBack(amount) }),
})

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. Scope it with threadId or resourceId, or hand in runs you already have.

Each decision is keyed on runId plus toolCallId, so running this twice against the same suspended run resolves to the same decision rather than paging twice. A tool that suspended for its own resume data (rather than for approval) is left alone.

Fail-closed: a denial, an expiry, or nobody answering all decline, with the reason handed to the model. For a multi-tenant product, resolve the end-user per call:

resolvePusharyApprovals({ externalId: (pending) => ownerOf(pending.runId) }, { agent })

Durable step

import { pusharyApprovalStep } from '@pushary/mastra'
import { createWorkflow } from '@mastra/core/workflows'
import { z } from 'zod'

const approval = pusharyApprovalStep(
  { apiKey: process.env.PUSHARY_API_KEY! },
  { callbackUrl: `${process.env.PUBLIC_URL}/api/pushary/callback` },
)

export const refund = createWorkflow({
  id: 'refund',
  inputSchema: z.object({ question: z.string(), externalId: z.string() }),
  outputSchema: z.object({ approved: z.boolean(), value: z.string() }),
})
  .then(approval)
  .commit()

Run it, and it suspends at the approval step. Resume from the callback route:

import { resolvePusharyCallback } from '@pushary/mastra'

// POST /api/pushary/callback
export async function POST(req: Request) {
  const raw = await req.text()
  const cb = resolvePusharyCallback(raw, req.headers.get('x-pushary-signature'), process.env.PUSHARY_WEBHOOK_SECRET!)
  if (!cb) return new Response('bad signature', { status: 401 })
  const runId = await lookupRun(cb.correlationId) // your own correlationId -> runId map
  const run = await refund.createRun({ runId })
  await run.resume({ step: approval, resumeData: { answer: cb.answer } })
  return new Response('ok')
}

API

• connect(config, externalId) — enroll an end-user's phone.
• createPusharyAskTool(config, { externalId }) — a Mastra createTool that blocks on a human.
• pusharyRequireApproval() — a requireApproval predicate that routes every call to a human.
• resolvePusharyApprovals(config, { agent, runs?, threadId?, resourceId? }) — ask about each suspended tool call, then approve or decline it.
• pusharyApprovalStep(config, { callbackUrl }) — a durable createStep with suspend/resume.
• createPusharyGate(config) — the raw fail-closed gate, for anything the helpers above do not cover.
• resolvePusharyCallback(raw, signature, secret) — verify + parse a callback into { correlationId, answer, approved, ... }.
• askExternalUser, createDurableDecision, describeAnswer, isAffirmative, deterministicKey, SIGNATURE_HEADER.

Example

A runnable example is in examples/.

License

MIT