Tutorial: move a LangGraph app onto Agent Adapter¶

Available · pip install "agent-adapter[langgraph,langchain]" · about 20 minutes

You keep your graph, your state and your checkpointer. You swap LangGraph's ToolNode and hand-written interrupt() approvals for registry tools run by an Executor. Approvals become signed, bound to one call, and usable once. Other starting points: Migration.

File	Does
`examples/tutorial_langgraph.py`	before, after (tool node, own node, compiled flow); prints each step
`tests/docs/test_tutorial_examples.py`	runs it offline (skipped when `langgraph` is not installed)

What changes¶

Before	After
`@tool` functions in a `ToolNode([...])`	the same tools in a `Registry` (`import_langchain`, or `@registry.tool`), run by `tool_node(registry, executor=...)`
`interrupt(...)` inside a tool, resumed with any value	`confirm=True` on the tool; `LangGraphApproval(key=...)` interrupts; only `port.resume_value(...)` answers
retries, timeouts, logging per tool, by hand	tool options (`timeout=`, `retries=`) and executor hooks (`LoggingHook`, `BudgetHook`)
cost tracking by hand	the executor's `Ledger`
your own node calling tools	the same node wrapped in `one_use(node)`
a fixed pipeline of nodes	optional: a `Flow` compiled with `compile_flow`

What stays: StateGraph, MessagesState, edges, the checkpointer, Command(resume=...).

Before¶

@tool
def shipping_rate(item: str, region: str = "us") -> float:
    """Shipping rate for an item, in percent of the order value (by region)."""
    return {"eu": 5.0, "us": 18.0}.get(region, 12.0)


@tool
def send_statement_unsafe(period: str) -> str:
    """Send the account statement for a period, after asking a human."""
    if interrupt({"period": period}) != "yes":  # any resume value "yes" approves
        return "cancelled"
    return f"sent {period}"

Any caller who can send Command(resume={id: "yes"}) approves. The answer is not bound to the arguments or the call, nobody signed it, and the same value works again on a replayed checkpoint. That is the gap this migration closes.

1. Tools into a registry¶

registry = Registry()
import_langchain(registry, [shipping_rate], parallel_safe=True)  # keep existing LangChain tools


@registry.tool(confirm=True, effects="destructive")
async def send_statement(period: str) -> str:
    """Send the account statement for a period (needs approval)."""
    return f"sent {period}"

import_langchain keeps your existing @tool objects (agent-adapter[langchain]). New tools use @registry.tool. Mark every tool with side effects confirm=True.

2. One port, one executor¶

KEY = b"load me from your secret store"  # >= 16 bytes; the same in every process
port = LangGraphApproval(key=KEY)
executor = Executor(registry, approval=port)  # hooks, budget, ledger as usual

LangGraphApproval pauses the graph with interrupt() and checks the answer. The key signs answers: at least 16 bytes, from your secret store, the same in every process. key=None makes a random per-process key, so a run then resumes only in that process.

3. `tool_node` replaces `ToolNode`¶

app = build(tool_node(registry, executor=executor), CALLS)  # replaces ToolNode([...])
config: Any = {"configurable": {"thread_id": "shop-1"}}
[pending] = (await app.ainvoke({"messages": []}, config, version="v2")).interrupts
request = pending.value["request"]  # show reviewers this, never the raw interrupt value
print("asks", request["tool"], request["arguments"])

shipping_rate runs; send_statement raises an interrupt whose value is {"type": "agent_adapter.approval", "binding", "request"}. Show reviewers request only.

answer = port.resume_value(pending, ApprovalDecision(approved=True, decided_by="ana"))
forged = {**answer, "decision": {"approved": True, "edited_args": {"period": "2020-01"}}}
for value in ("yes", forged):  # a bare answer; an edited decision the mac does not cover
    try:
        await app.ainvoke(Command(resume={pending.id: value}), config)
    except ValidationError:  # LangGraph's invalid resume: nothing saved, nothing runs
        print("refused", "bare" if value == "yes" else "forged")
out = await app.ainvoke(Command(resume={pending.id: answer}), config, version="v2")
print("after:", [m.content for m in out.value["messages"][1:]])

sequenceDiagram
  participant G as graph (tool_node)
  participant P as LangGraphApproval
  participant R as Reviewer UI
  G->>P: send_statement needs approval
  P-->>G: interrupt(binding, request)
  G-->>R: request (tool, arguments)
  R->>P: resume_value(interrupt, decision)
  P-->>R: binding + decision + mac (HMAC-SHA256, port key)
  R->>G: Command(resume={id: value})
  G->>P: check binding (run, tool, args, call) and mac
  P-->>G: decision, used once
  G->>G: run send_statement

Resume value	Result
`"yes"`, or any bare decision	invalid resume (`ValidationError`): nothing saved, ask again
a signed value with an edited decision	invalid: the `mac` covers the decision, edits included
the signed value, used again (a replay, a concurrent resume)	`ApprovalAlreadyDecided`; the tool ran once

4. Your own nodes: `one_use`¶

async def send_node(state: MessagesState, config: RunnableConfig) -> dict[str, Any]:
    """Your own node calling a confirm tool: wrap it in one_use."""
    thread = str(config["configurable"]["thread_id"])
    ctx = executor.context(run_id=thread)
    receipt = await executor.run("send_statement", {"period": "2026-10"}, ctx, call_id="send-1")
    return {"messages": [AIMessage(str(receipt))]}

Wrap it: graph.add_node("send", one_use(send_node)). LangGraph re-runs an interrupted node from the top; one_use uses each answer once and replays calls the node finished before the pause (give each a stable call_id). Without the wrapper LangGraphApproval raises ToolUsageError before it asks.

5. Optional: a compiled flow with a store¶

class Statement(BaseModel):
    rate: float = 0.0
    receipt: str = ""


flow = (
    Flow("statement", state=Statement)
    .node("rate", Call("shipping_rate", args={"item": "laptop"}, into="rate"))
    .node("send", Call("send_statement", args={"period": "2026-11"}, into="receipt"))
    .edge("rate", "send")
    .end("send")
)


async def flow_demo() -> None:
    store = MemoryApprovalStore()  # a database store: one use across processes
    flows = Executor(registry, approval=DeferredApproval(store))
    app = compile_flow(flow, executor=flows).compile(checkpointer=InMemorySaver())
    config: Any = {"configurable": {"thread_id": "statement-1"}}

    paused = await run_compiled(app, {}, config)
    assert paused.pending is not None
    print(paused.status, paused.pending.tool, paused.pending.arguments)
    # the reviewer's process decides the record; any worker then continues with None
    await store.decide(paused.pending.approval_id, ApprovalDecision(approved=True))
    done = await run_compiled(app, None, config)
    print(done.status, done.state.rate, done.state.receipt)

A store-backed port (DeferredApproval(store)) decides only from its store, and each record is used once across every process that shares the store. The reviewer decides the record; any worker continues with run_compiled(app, None, config).

Safety rules¶

Rule	Why
framework approval never counts alone	a resume value approves only when signed with the port's key and bound to the run, tool, arguments digest and call id
one key, from a secret store, in every process	a value signed elsewhere is invalid; `key=None` pins a run to one process
one use across processes: a store-backed port	with a shared key, a resume value can be replayed in another process after the run went on; a store consumes each record once
wrap your own tool-calling nodes in `one_use`	each answer used once; finished calls replay instead of running twice
no raw `interrupt()` in a `one_use` node	it shifts resume positions; ask through `LangGraphApproval` only
a new process re-runs a node's finished calls (`tool_node`, `one_use`)	the replay journal is in-process; use `compile_flow` (its progress is sealed into the interrupt) or one confirm call per node
encrypt the checkpointer (`EncryptedSerializer`), restrict its access	interrupt values carry arguments and finished results; signed, not encrypted
never forward raw interrupt values to a UI	show `value["request"]` only

More detail: LangGraph integration and Approval.