LangGraph

A refund agent built as a plain StateGraph: an agent node calls the model, a tools node runs its tool calls. lookup_order runs freely; issue_refund is gated, so the tools node raises a LangGraph interrupt() and the run pauses until a reviewer decides in Vigilator. Source: Vigilator/examples/langgraph.

What you'll see

• Inbox - an interrupt titled issue_refund needs a decision carrying the refund's arguments and the transcript so far, with Approve, Edit and Reject. The tool allows editing, so the interrupt ships a JSON schema for its arguments and Edit opens a typed form.
• Live View - a refund-agent session showing the lookup call and its result. It ends when the agent pauses and a new one opens when the run resumes - see why.

Run it

git clone https://github.com/Vigilator/examples
cd examples/langgraph
uv sync

cp .env.example .env

uv run main.py

uv run fastapi dev webhook.py                             # the agent and the receiver, port 8000
svix listen http://localhost:8000/webhooks/vigilator      # relay deliveries to localhost

How it works

The tools node collects the gated calls into one interrupt(), shaped like LangChain's human-in-the-loop payload - one action_requests entry per call and a review_configs entry naming the allowed decisions:

GATED = {"issue_refund": ["approve", "edit", "reject"]}

def tools(state):
    calls = state["messages"][-1].tool_calls
    gated = [c for c in calls if c["name"] in GATED]
    decisions = {}
    if gated:
        response = interrupt({
            "action_requests": [{"name": c["name"], "args": c["args"]} for c in gated],
            "review_configs": [{"action_name": c["name"], "allowed_decisions": GATED[c["name"]]} for c in gated],
        })
        decisions = {c["id"]: d for c, d in zip(gated, response["decisions"])}
    return {"messages": [_run_tool(call, decisions.get(call["id"], {"type": "approve"})) for call in calls]}

The bridge maps that payload one-to-one onto a Vigilator interrupt: each action request keeps its name, args and allowed decisions, the transcript is attached as messages, and externalId is set to langgraph:<thread_id> so the outcome can be routed back:

request = InterruptsPostRequest(
    title=f"{name} needs a decision",
    description=...,
    externalId=f"langgraph:{thread_id}",
    messages=to_vigilator_messages(transcript),
    actionRequests=[ActionRequest(name=name, args=args, argsSchema=schema, allowedDecisions=[...])],
)

When the interrupt is answered - by polling get_interrupt or from the webhook - the decisions become the Command that resumes the graph. interrupt() returns it inside the tools node, which runs approved calls as proposed, edited calls with the reviewer's editedArgs, and turns rejections into an error tool message the model can explain to the customer:

def to_langgraph_resume(decided):
    decisions = []
    for action in decided:
        if action.decision == "approve":
            decisions.append({"type": "approve"})
        elif action.decision == "edit":
            decisions.append({"type": "edit", "edited_action": {"name": action.name, "args": action.edited_args}})
        elif action.decision == "reject":
            decisions.append({"type": "reject", "message": action.response_text or "Rejected by the reviewer."})
        elif action.decision == "respond":
            decisions.append({"type": "respond", "message": action.response_text or ""})
    return Command(resume={"decisions": decisions})

Adapting it

• Add tools to TOOLS and list the ones needing review in GATED with the decisions they allow. A question for a human is a tool whose only allowed decision is respond - see Ask a human.
• Keep the action_requests / review_configs shape and the bridge works unchanged. If you use create_agent, its middleware already produces it - see LangChain.
• Swap the in-memory checkpointer for a database one to resume from another process - see Durable resume.