LangGraph
LangGraph is the best-matched framework for Confirm, because it can pause without polling: checkpoint the graph, end the process, and resume when our webhook says a human decided. Hours or days can pass in between and nothing is blocked. This guide builds that flow. Python, about fifteen minutes.
Before you start
- A Confirm account and API key (sign in → Dashboard → API Keys).
- A webhook endpoint registered in your dashboard, pointing at the handler you'll write in step 3. For local development,
ngrok http 8000gives you a public URL. - Python 3.10+.
pip install langgraph requests fastapi uvicornexport CONFIRM_API_KEY=cfm_live_...Step 1: a graph that pauses on approval
The key move: the approval node creates the Confirm request, stashes the graph's thread_id in the request's metadata, then calls interrupt(). LangGraph checkpoints the run and stops. No polling loop, no blocked worker.
import os
from typing import TypedDict
import requests
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt
BASE = "https://confirm.dev/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['CONFIRM_API_KEY']}"}
class State(TypedDict):
customer_id: str
amount: float
outcome: str
def request_approval(state: State):
"""Create the Confirm request, then pause the graph until the webhook resumes it."""
thread_id = state["customer_id"] # use your real thread id scheme
requests.post(
f"{BASE}/requests",
headers=HEADERS,
json={
"summary": f"Refund ${state['amount']:,.2f} to customer {state['customer_id']}",
"payload": {"customerId": state["customer_id"], "amount": state["amount"]},
"notify": "finance@company.com", # or "group:finance" on Pro
"reasoning": "Customer requested a refund; amount exceeds the $1,000 policy threshold.",
"recentActions": ["Verified the order", "Checked the refund policy threshold"],
"ttlMinutes": 4320, # 3 days: nobody is polling, so be generous
"metadata": {"threadId": thread_id}, # how the webhook finds this run
# LangGraph's checkpointer holds the state, so agentState isn't needed here.
# Without a checkpointer, park it: "agentState": serialized_state,
},
timeout=30,
).raise_for_status()
# Pause here. The value passed to interrupt() is surfaced to the caller;
# the value the webhook resumes with becomes interrupt()'s return value.
verdict = interrupt({"waiting_on": "confirm.dev"})
if verdict["status"] != "APPROVED":
return {"outcome": f"blocked: {verdict.get('resolutionNote') or verdict['status']}"}
approved = verdict["effectivePayload"] # includes any human edits
issue_refund(approved["customerId"], approved["amount"]) # your real logic
return {"outcome": f"refunded ${approved['amount']:,.2f}"}
builder = StateGraph(State)
builder.add_node("request_approval", request_approval)
builder.add_edge(START, "request_approval")
builder.add_edge("request_approval", END)
# Use a durable checkpointer in production (Postgres/SQLite); memory works for the demo.
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)Step 2: kick off a run
from graph import graph
config = {"configurable": {"thread_id": "7133"}}
result = graph.invoke({"customer_id": "7133", "amount": 1850.0}, config)
# The graph is now paused at the interrupt. Inspect it:
print(result["__interrupt__"]) # [Interrupt(value={'waiting_on': 'confirm.dev'}, ...)]
# The process can exit here. State lives in the checkpointer.At this moment the approver's inbox has the email, your process has nothing to do, and the run is frozen in the checkpointer under thread_id 7133.
Step 3: resume from the webhook
When the human decides, Confirm POSTs to your endpoint. The handler verifies the signature, pulls the threadId out of metadata, and resumes the graph with the verdict as the value interrupt() returns.
import hmac
import hashlib
import os
import time
from fastapi import FastAPI, Request, HTTPException
from langgraph.types import Command
from graph import graph
app = FastAPI()
SECRET = os.environ["CONFIRM_WEBHOOK_SECRET"] # whsec_..., from Dashboard -> Webhooks
def verify(header: str, raw_body: bytes) -> bool:
parts = dict(p.split("=", 1) for p in header.split(","))
if abs(time.time() - int(parts["t"])) > 300:
return False
expected = hmac.new(
SECRET.encode(), f"{parts['t']}.".encode() + raw_body, hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected, parts["v1"])
@app.post("/webhooks/confirm")
async def confirm_webhook(request: Request):
raw = await request.body()
if not verify(request.headers.get("x-confirm-signature", ""), raw):
raise HTTPException(401, "bad signature")
event = await request.json()
if event["event"] not in ("request.approved", "request.rejected", "request.expired"):
return {"ok": True}
data = event["data"]
thread_id = (data.get("metadata") or {}).get("threadId")
if not thread_id:
return {"ok": True} # not a graph-managed request
config = {"configurable": {"thread_id": thread_id}}
# Command(resume=...) feeds the verdict into interrupt() and runs the graph onward.
result = graph.invoke(Command(resume=data), config)
print(f"thread {thread_id} finished: {result['outcome']}")
return {"ok": True}uvicorn webhook:app --port 8000Step 4: watch the full loop
- Run
python run.py. It prints the interrupt and exits. Nothing is blocked. - Open the approval email. Reject it with a note, or edit the amount and approve.
- Watch the webhook process log: the graph resumed on your decision and printed the outcome. If you rejected,
outcomecarries your note; if you edited, the refund used your numbers.
WHY THIS BEATS POLLING
ttlMinutes: 4320) and your infrastructure doesn't care. Expired requests also arrive as webhooks, so abandoned runs resolve themselves through the same handler.Next steps
- Swap
MemorySaverfor a durable checkpointer (Postgres) so resumes survive restarts. - Use
X-Confirm-Delivery-Idfor idempotency: webhook retries can deliver twice, andgraph.invokeon a finished thread should be a no-op you enforce. - Webhooks reference · Approval groups · API reference