Skip to content

LangGraph Integration

This page covers InterruptApproval — the approval backend that pauses a LangGraph run using interrupt() and resumes it once a human (or your own application) supplies a decision from outside the graph entirely. This is the right approach for human-in-the-loop approval in a production LangGraph app, where you don't want to block a worker thread on a blocking terminal input() call the way CLIApproval does.

Requirements

InterruptApproval requires langgraph:

pip install -e ".[langgraph]"

Constructing InterruptApproval() without it installed raises a clear ImportError telling you to install it.

How it works

  1. A tool call triggers guard(), which sees the permission is in require_approval_for, and calls ctx.approval.request(req).
  2. InterruptApproval.request(...) calls LangGraph's interrupt() with a structured payload describing the request. This pauses the entire graph run — the compiled graph must have a checkpointer for this to work, since the run's state needs to be persisted while paused.
  3. app.invoke(...) returns immediately with "__interrupt__" in the result — nothing has been written/deleted/executed yet.
  4. Your own application (a UI, an API endpoint, a Slack bot) decides whether to approve, then resumes the exact same run with app.invoke(Command(resume={"approved": True/False}), config=config).
  5. The tool call that was waiting on approval receives that answer and either proceeds or raises ApprovalDeniedError.

The interrupt payload looks like this:

{
    "type": "mero_tools_approval",
    "action": "WRITE",
    "detail": "some/path.txt",
    "requester": "worker_1",
    "context": {"permission": Permission.WRITE, "path": "some/path.txt"},
}

InterruptApproval accepts the resume value either as a plain bool or as a dict with an "approved" key ({"approved": True}) — the latter is useful if your UI wants to pass back additional structured data alongside the decision.

Full working example

This is adapted directly from the package's own integration test (tests/test_langgraph_integration.py), which verifies this really pauses and resumes a graph via interrupt()/Command(resume=...), not just a mock.

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing_extensions import TypedDict

from langchain_mero_tools import (
    SecurityContext, PathEntry, Permission, InterruptApproval, make_file_tool,
)


class State(TypedDict):
    path: str
    content: str
    result: str


ctx = SecurityContext(
    name="worker",
    paths=[
        PathEntry(
            path="./workspace",
            slug="root",
            allowed_permission=Permission.WRITE,
            required_permission=Permission.WRITE,
        ),
    ],
    approval=InterruptApproval(),
)
tool = make_file_tool(ctx)


def write_node(state: State) -> State:
    result = tool.invoke({
        "operation": "write",
        "path": state["path"],
        "content": state["content"],
    })
    return {"result": result}


graph = StateGraph(State)
graph.add_node("write", write_node)
graph.add_edge(START, "write")
graph.add_edge("write", END)

# A checkpointer is required — the graph's state has to persist while paused.
app = graph.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "t1"}}

# First invoke pauses at interrupt() — nothing is written yet.
result = app.invoke(
    {"path": "./workspace/note.txt", "content": "hello", "result": ""},
    config=config,
)
assert "__interrupt__" in result
# ./workspace/note.txt does not exist yet.

# Some time later, from your own UI/API, once a human has decided:
final = app.invoke(Command(resume={"approved": True}), config=config)
# ./workspace/note.txt now contains "hello".
assert "Wrote" in final["result"]

If the human instead denies:

app.invoke(Command(resume={"approved": False}), config=config)
# final["result"] contains "Denied" / "not approved" — nothing was written.

Key details worth remembering

  • A checkpointer is mandatory. graph.compile(checkpointer=...) — the in-memory InMemorySaver works for examples and tests; use a persistent checkpointer (e.g. a database-backed one) in production so a paused run survives a process restart.
  • thread_id identifies the run. The same config["configurable"] ["thread_id"] must be used for both the initial invoke and the resuming invoke — that's how LangGraph knows which paused run to resume.
  • Nothing happens between pause and resume. The tool call is genuinely suspended mid-execution — no partial writes, no side effects — until the resume value comes back.
  • Combine with manager/worker delegation for the best of both. Wrap InterruptApproval in a ChainedApproval alongside ManagerDelegationApproval so routine, in-scope worker actions get auto-approved by a manager agent, and only genuinely novel requests pause for a human via interrupt():
from langchain_mero_tools import ChainedApproval, ManagerDelegationApproval

approval = ChainedApproval([
    ManagerDelegationApproval(registry, manager="manager_1"),
    InterruptApproval(),
])

See also