# LangGraph

Source: https://pushary.com/docs/agents/build/langgraph
Last updated: 2026-10-04

> LangGraph human in the loop with phone approvals. An approval node asks your user to Approve or Deny before the risky node runs, in TypeScript or Python.

In LangGraph, the graph decides what runs next. So the way to add human in the loop to LangGraph is a node: put an approval node in front of the risky node, and draw the edge to it only when the answer is yes. `@pushary/langgraph` (TypeScript) and `pushary-langgraph` (Python) give you that node. It asks your user on their phone and returns their answer.

<Tabs items="[&#x22;TypeScript&#x22;, &#x22;Python&#x22;]">
  <Tab value="TypeScript">
    ```bash
    npm i @pushary/langgraph @langchain/langgraph @langchain/core zod
    export PUSHARY_API_KEY=pk_xxx.xxx
    ```
  </Tab>

  <Tab value="Python">
    ```bash
    pip install pushary-langgraph
    export PUSHARY_API_KEY=pk_xxx.xxx
    ```
  </Tab>
</Tabs>

Get the key at [Partner onboarding](https://pushary.com/onboarding/partner). The Python package installs `langgraph` for you.

## Set up LangGraph human in the loop [#set-up-langgraph-human-in-the-loop]

<Steps>
  <Step>
    ### Connect a user once [#connect-a-user-once]

    <Tabs items="[&#x22;TypeScript&#x22;, &#x22;Python&#x22;]">
      <Tab value="TypeScript">
        ```ts
        import { connect } from '@pushary/langgraph'

        const { universalLink } = await connect({ apiKey: process.env.PUSHARY_API_KEY }, user.id)
        ```
      </Tab>

      <Tab value="Python">
        ```python
        from pushary_langgraph import connect

        link = connect(user.id)
        ```
      </Tab>
    </Tabs>

    Show the link to that user. They open it on their phone once. They install the free Pushary app and never sign up or pay.
  </Step>

  <Step>
    ### Put an approval node before the risky step [#put-an-approval-node-before-the-risky-step]

    <Tabs items="[&#x22;TypeScript&#x22;, &#x22;Python&#x22;]">
      <Tab value="TypeScript">
        ```ts
        import { Annotation, END, START, StateGraph } from '@langchain/langgraph'
        import { pusharyInterrupt } from '@pushary/langgraph'

        const config = { apiKey: process.env.PUSHARY_API_KEY }

        const Refund = Annotation.Root({
          orderId: Annotation<string>,
          customerId: Annotation<string>,
          approved: Annotation<boolean>,
        })

        const graph = new StateGraph(Refund)
          .addNode('approval', async (state) => {
            const answer = await pusharyInterrupt(config, {
              externalId: state.customerId,
              question: `Refund order ${state.orderId} in full?`,
              type: 'confirm',
            })
            return { approved: answer === 'yes' }
          })
          .addNode('refund', async (state) => {
            await refundOrder(state.orderId)
            return {}
          })
          .addEdge(START, 'approval')
          .addConditionalEdges('approval', (state) => (state.approved ? 'refund' : END))
          .addEdge('refund', END)
          .compile()

        await graph.invoke({ orderId: '4471', customerId: user.id, approved: false })
        ```
      </Tab>

      <Tab value="Python">
        ```python
        from typing import TypedDict

        from langgraph.graph import END, START, StateGraph
        from pushary_langgraph import pushary_interrupt

        class Refund(TypedDict):
            order_id: str
            customer_id: str
            approved: bool

        def approval(state: Refund) -> dict:
            answer = pushary_interrupt(
                f"Refund order {state['order_id']} in full?",
                external_id=state["customer_id"],
            )
            return {"approved": answer == "yes"}

        def refund(state: Refund) -> dict:
            refund_order(state["order_id"])
            return {}

        graph = (
            StateGraph(Refund)
            .add_node("approval", approval)
            .add_node("refund", refund)
            .add_edge(START, "approval")
            .add_conditional_edges("approval", lambda state: "refund" if state["approved"] else END)
            .add_edge("refund", END)
            .compile()
        )

        graph.invoke({"order_id": "4471", "customer_id": user.id, "approved": False})
        ```
      </Tab>
    </Tabs>

    `refundOrder` and `refund_order` are your own functions.
  </Step>

  <Step>
    ### Run the graph and answer on the phone [#run-the-graph-and-answer-on-the-phone]

    Your user's phone buzzes with "Refund order 4471 in full?". They tap Approve, the node returns `approved: true`, and the graph takes the edge to `refund`.
  </Step>
</Steps>

The model has no way around this node. The graph's edges decide what runs, not the model.

## What happens when nobody answers [#what-happens-when-nobody-answers]

The refund does not run. The approval node gets one of three answers:

| Answer                    | What it means                                                               | Edge taken |
| ------------------------- | --------------------------------------------------------------------------- | ---------- |
| `'yes'`                   | Your user tapped Approve                                                    | `refund`   |
| `'no'`                    | Your user tapped Deny                                                       | `END`      |
| `null` (`None` in Python) | Nobody answered within 55 seconds, or the question expired or was cancelled | `END`      |

A timeout does not cancel the question on the phone. It stays open for 1 hour by default, so the user can still answer. To act on that late answer, use the durable path below.

## Let the model ask a question [#let-the-model-ask-a-question]

<Tabs items="[&#x22;TypeScript&#x22;, &#x22;Python&#x22;]">
  <Tab value="TypeScript">
    ```ts
    import { ChatOpenAI } from '@langchain/openai'
    import { createReactAgent } from '@langchain/langgraph/prebuilt'
    import { createAskHumanTool } from '@pushary/langgraph'

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

    const agent = createReactAgent({ llm: new ChatOpenAI({ model: 'gpt-4o' }), tools: [askHuman] })

    await agent.invoke({
      messages: [{ role: 'user', content: 'Find out which order the customer means.' }],
    })
    ```
  </Tab>

  <Tab value="Python">
    ```python
    from pushary_langgraph import ask_human

    reply = ask_human("Which order do you need help with?", external_id=user.id, type="input")
    order_id = reply["value"]
    ```
  </Tab>
</Tabs>

The TypeScript example also needs `npm i @langchain/openai`. The tool is called `ask_human` by default.

A tool is the model's choice to call, so use it for questions like "which order". Use the approval node above for anything that must not happen without a yes.

## Wait longer than 55 seconds [#wait-longer-than-55-seconds]

Pass a `callbackUrl` and an `idempotencyKey` (Python: `callback_url` and `idempotency_key`). The node then pauses the graph with LangGraph's own `interrupt()` instead of waiting, and you resume it when the answer arrives.

```ts
const answer = await pusharyInterrupt(config, {
  externalId: state.customerId,
  idempotencyKey: state.reviewOperationId,
  question: `Refund order ${state.orderId} in full?`,
  type: 'confirm',
  callbackUrl: process.env.PUSHARY_CALLBACK_URL,
})
```

`reviewOperationId` is a stable id you keep in graph state for this one action. For this path you need:

1. A persistent checkpointer (`MemorySaver` does not survive a restart) and a stable, customer-scoped `thread_id`.
2. A callback route that verifies the raw body with `resolvePusharyCallback(raw, signature, webhookSecret)` before it trusts anything.
3. A saved mapping from customer, thread id and interrupt id to the Pushary `correlationId`.

Then resume the thread by interrupt id with the correlated answer:

```ts
import { Command } from '@langchain/langgraph'

await graph.invoke(new Command({
  resume: {
    [pendingInterrupt.id]: {
      correlationId: verifiedCallback.correlationId,
      answer: verifiedCallback.answer,
    },
  },
}), { configurable: { thread_id: trustedThreadId } })
```

The resume value is checked. A wrong `correlationId`, a confirm answer that is not `yes` or `no`, or a choice that was not offered throws instead of running the node. To release a graph whose question expired, resume with `{ correlationId, status: 'expired', answer: null }` (or `'cancelled'`). Never make up an approval to free a waiting graph.

The interrupt payload carries `decisionId`, `correlationId`, `operationKey`, `type`, `question`, `options` and `externalId`. The [restart test](https://github.com/Pushary/pushary-langgraph/blob/main/src/interrupt.test.ts) runs this path against a real SQLite checkpointer.

## Production notes [#production-notes]

### Retries and idempotency [#retries-and-idempotency]

LangGraph runs the whole node again on resume, so code before the interrupt must be safe to repeat. The adapter fingerprints your `idempotencyKey` with the recipient and the request. The same request reuses one question. A changed recipient, argument or question gets a new one. Use a fresh key for a new action.

### Approve means yes, nothing else [#approve-means-yes-nothing-else]

Only a confirm answer of `'yes'` is permission. A `select` or `input` answer is data, even when its text is `yes`. Route to the risky node only from a confirm.

### Pick the right person [#pick-the-right-person]

Take the user id from your signed-in user, never from the model's input. A prompt injection could otherwise send the approval to someone else.

### Versions [#versions]

`@pushary/langgraph` 0.4 and `pushary-langgraph` 0.4 require the shared SDK 2.1 (`@pushary/server` 2.1 or `pushary` 2.1). A durable resume must send the correlated object above. Let runs paused under 0.3 finish on 0.3.

Source and more examples: [pushary-langgraph on GitHub](https://github.com/Pushary/pushary-langgraph). LangGraph reference: [interrupts in TypeScript](https://docs.langchain.com/oss/javascript/langgraph/interrupts) and [in Python](https://docs.langchain.com/oss/python/langgraph/interrupts).

## Next steps [#next-steps]

<Cards>
  <Card title="Try a simulated decision" description="Approve or deny a refund in your browser before you connect a phone." href="/try" />

  <Card title="See the LangGraph approval workflow" description="The whole flow for LangGraph agents, from connect to resume." href="/human-in-the-loop-langgraph" />

  <Card title="Test your integration" description="Use a sandbox key to check approve, deny and no answer without a phone." href="/docs/agents/embed-testing" />
</Cards>