@pushary/langgraph

Human-in-the-loop for LangGraph and LangChain. Ask a real human to approve, and get the answer on their phone. Two seams:

• A blocking ask_human tool for a straightforward approval inside a request.
• A durable interrupt() wrapper that parks the graph in your checkpointer and resumes on a signed webhook, so an hour-long wait holds no compute and survives a restart.

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

Install

npm i @pushary/langgraph @langchain/langgraph @langchain/core

Set PUSHARY_API_KEY (get it in your dashboard).

Connect a phone once

import { connect } from '@pushary/langgraph'

const { universalLink } = await connect({ apiKey: process.env.PUSHARY_API_KEY! }, user.id)
// show universalLink to the user; one tap connects their phone

Blocking tool

import { createAskHumanTool } from '@pushary/langgraph'
import { createReactAgent } from '@langchain/langgraph/prebuilt'

const askHuman = createAskHumanTool({ apiKey: process.env.PUSHARY_API_KEY! }, { externalId: user.id })
const agent = createReactAgent({ llm, tools: [askHuman] })

The tool blocks until the person answers and returns a fail-closed instruction to the model ("The human declined. Do not proceed."). externalId is bound in code, never taken from model input, so a prompt-injected model cannot ask the wrong person.

Durable interrupt

Wrap LangGraph's native interrupt(). Pass a callbackUrl to park the graph instead of blocking.

import { pusharyInterrupt } from '@pushary/langgraph'
import { StateGraph, MemorySaver, Command } from '@langchain/langgraph'

async function approvalNode(state) {
  const answer = await pusharyInterrupt(
    { apiKey: process.env.PUSHARY_API_KEY! },
    {
      externalId: state.userId,
      question: 'Approve this transfer?',
      node: 'approval',
      callbackUrl: process.env.PUSHARY_CALLBACK_URL, // omit to block instead of park
    },
  )
  return { approved: answer === 'yes' }
}

const graph = builder.compile({ checkpointer: new MemorySaver() }) // a checkpointer is required

The whole node re-runs on resume, so keep any code before pusharyInterrupt idempotent. The decision's idempotency key is derived from externalId + node + question, so the re-run lands on the same decision instead of paging the human twice.

Resume from the webhook

import { resolvePusharyCallback } from '@pushary/langgraph'
import { Command } from '@langchain/langgraph'

// POST /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 threadId = await lookupThread(cb.correlationId) // your own correlationId -> thread_id map
  await graph.invoke(new Command({ resume: cb.answer }), { configurable: { thread_id: threadId } })
  return new Response('ok')
}

API

• connect(config, externalId) — enroll an end-user's phone.
• createAskHumanTool(config, { externalId }) — a LangChain tool() that blocks on a human.
• pusharyInterrupt(config, input) — ask from a node: blocking, or durable when callbackUrl is set.
• resolvePusharyCallback(raw, signature, secret) — verify + parse a callback into { correlationId, answer, approved, ... }.
• askExternalUser, createDurableDecision, describeAnswer, isAffirmative, deterministicKey, SIGNATURE_HEADER.

Python

The same two seams ship for Python (LangGraph's Python interrupt() plus a blocking ask_human). The package lives in python/ and on PyPI:

pip install pushary-langgraph

See python/README.md for the Python API.

Other frameworks

The same two calls work in CrewAI, the Vercel AI SDK, Mastra, the OpenAI Agents SDK, and more.

Example

A runnable example is in examples/.

License

MIT © Pushary