A LangGraph Support Agent with Tool Permissions and Human Handoff

A support agent that only answers questions is a search box with manners. The moment it can look up an order or issue a refund, it is an API exposed to untrusted natural language, and it needs the same controls as any other API: authorisation, limits, audit and a way to stop.

This is how I build that agent in LangGraph. Every code block comes from a small reference implementation that I ran in this environment: 37 pytest tests, a FastAPI layer, a Postgres checkpointer and a scripted evaluation set. The model in every test is a fake that replays scripted messages behind the same BaseChatModel interface as the production client. No real LLM was called and no number here describes a real model's quality.

It follows the same principles as Semai AI Support (status: Built): answers from approved sources, limited tool use, human handoff and operator review. The code below is a fresh reference implementation written for this article, not Semai's code, and it makes no claim about Semai's results.

The shape of the graph

Seven nodes, and every exit from the happy path goes to one place: handoff.

Each node owns one decision:

• classify picks an intent with structured output. Code, not the model, decides what that intent allows.
• retrieve searches only the sources the charter allows for that intent, filtered by tenant.
• agent is the only node that lets the model choose anything. It sees only the tools the charter grants.
• gate is pure. It decides which tool calls may run, and it is the only node that calls interrupt().
• act runs the permitted calls with identity injected from server state.
• answer refuses to send any reply that cites no approved source.
• handoff writes a structured record for a person.

Nodes route with Command(goto=..., update=...) and a Literal return type, so LangGraph can draw the graph from the type hints and a node can route and write state in one return value. Identity travels as runtime context: StateGraph(SupportState, context_schema=SupportContext) and graph.invoke(..., context=ctx). For when a state machine like this is worth the overhead compared with a plain chain, see LangChain vs LangGraph.

The charter is data, and code enforces it

The first control is a written statement of what the agent is for. If that statement only lives in a system prompt, every prompt edit becomes an unreviewed policy change. So I keep it in a file with one entry per intent: allowed sources, allowed tools and an escalation rule.

[policy]
min_confidence = 0.7
max_tool_rounds = 3

[tools.lookup_order]
kind = "read"

[tools.issue_refund]
kind = "write"
approval_above_pence = 5000

[intents.order_status]
sources = ["delivery-policy"]
tools = ["lookup_order"]
escalate = "when_ungrounded"

[intents.refund_request]
sources = ["refund-policy"]
tools = ["lookup_order", "issue_refund"]
escalate = "when_ungrounded"

[intents.account_takeover]
sources = []
tools = []
escalate = "always"
# ... password_reset and legal_complaint follow the same pattern

At startup, a Pydantic model with extra="forbid" loads the file. A validator rejects any intent that grants a tool the charter does not define. The loader then checks every charter tool against the registered implementations and refuses to boot if one is missing. A typo such as approval_above_penc fails the tests rather than silently disabling the approval limit, and there is a test for exactly that.

Two small methods on the charter answer every policy question the graph asks: allows(intent, tool) and needs_approval(tool, args). The graph never reads the TOML directly, so there is one place to change and one place to test. The file also gives review a clean diff: when someone grants issue_refund to a new intent, it shows up as a one-line change in a pull request, not a paragraph buried in a prompt.

Unknown intents go to a person, not to the nearest match. If the classifier returns crypto_advice, decide_after_classify returns ("handoff", "unknown_intent") before retrieval runs. account_takeover and legal_complaint never reach the agent node at all. The test for that scripts only one model message, so a second model call would fail the test.

Identity never comes from the model

The rule I care about most: the model supplies what the customer wants, and the server supplies who the customer is. Tenant and customer come from the authenticated session. The API layer puts them in a frozen SupportContext, and tools receive it as an injected argument that never appears in the tool schema the model sees.

@dataclass(frozen=True)
class SupportContext:
    """Set by the API layer from the authenticated session. Never from the model."""

tenant_id: str
    customer_id: str
    thread_id: str
    trace_id: str

Ctx = Annotated[SupportContext, InjectedToolArg]

def build_tools(orders: OrderStore, ledger: RefundLedger, outbox: list) -> dict[str, BaseTool]:
    def owned_order(ctx: SupportContext, order_id: str):
        order = orders.get(ctx.tenant_id, order_id)
        # Someone else's order and a missing order look identical to the model.
        return order if order and order.customer_id == ctx.customer_id else None

@tool
    def lookup_order(order_id: str, ctx: Ctx) -> dict:
        """Look up the status of one of the signed-in customer's orders."""
        order = owned_order(ctx, order_id)
        if order is None:
            return {"error": "order_not_found", "order_id": order_id}
        return {"order_id": order.order_id, "status": order.status,
                "total_pence": order.total_pence}
    # ... issue_refund and send_password_reset use owned_order and ctx the same way

# graph.py, inside the act node (dedented):
# Keep only the arguments the model is allowed to supply, then inject identity.
args = {k: v for k, v in call["args"].items() if k in tool.tool_call_schema.model_fields}
msg = tool.invoke({**call, "args": {**args, "ctx": runtime.context}})

Four decisions in that block each close a specific hole:

• Injected, not described. InjectedToolArg removes ctx from tool_call_schema. I bound the tool to a ChatOpenAI instance (constructed with a dummy key, no request sent) and the function parameters it would send contain only order_id. The model has no field to fill with someone else's identity.
• Filter, then inject. A model can still emit arguments that are not in the schema, including one called ctx. The executor drops everything outside tool_call_schema.model_fields, and only then adds the server's context. If the order were reversed, a smuggled ctx could overwrite the real one.
• Tenant is part of the key. Orders are fetched by (tenant_id, order_id), so an order number that exists in two tenants cannot cross between them.
• Not found, not forbidden. Another customer's order returns the same order_not_found as a missing one. A different error would let a script enumerate valid order numbers through the chat window.

send_password_reset takes no model arguments at all. The link goes to the address on file for the signed-in customer. A model that asks to send it to attacker@example.com gets its argument filtered out, and a test asserts the outbox holds exactly one message, addressed to the right customer.

The permission check lives in the tool, not the prompt. I use the same split between conversation layer and tool layer in the MedReclaim AI design (implementation-ready, not shipped), where a FastAPI tool server does its own tenant and identity checks behind the LangGraph gateway.

Proving the model cannot read someone else's order

A claim like "tools are tenant-safe" needs a test that plays the attacker. The scripted model does the worst thing a confused or prompt-injected model could do here: it asks for Bob's order while signed in as Alice, and it tries to smuggle Bob's identity in as extra arguments.

def test_model_cannot_read_another_customers_order(make_graph, alice):
    llm = scripted(
        classify_as("order_status"),
        # Prompt-injected or confused model asks for Bob's order, and even
        # tries to smuggle Bob's identity in as extra arguments.
        call_tool("lookup_order", order_id="ORD-2001", customer_id="cus_bob",
                  ctx={"tenant_id": "acme", "customer_id": "cus_bob"}),
        say("I can't find that order on your account [delivery-policy#1]."),
    )
    graph = make_graph(llm)
    out = graph.invoke({"messages": [("user", "Where is my order ORD-2001? It has shipped")]},
                       run_config(alice), context=alice, version="v2")
    results = tool_results(out.value)
    assert results == [{"error": "order_not_found", "order_id": "ORD-2001"}]
    assert "8000" not in json.dumps(results)  # nothing about Bob's order leaks
    assert out.value["actions"][0]["outcome"] == "executed"

The last assertion matters. The tool did run: the charter allows lookup_order for order_status. The refusal came from the tool's ownership check, which is the layer that still holds when the charter is wrong. Sibling tests cover the other layers. The same ORD-1001 in two tenants returns only Alice's tenant's order. issue_refund requested under the order_status intent is refused by the gate with tool_not_permitted, and the ledger stays empty.

The scripted model's bind_tools is a no-op, so the script can call tools the agent was never offered. That is deliberate. Real models hallucinate tool names and follow injected instructions, so the gate re-checks the charter even though the agent node only binds the allowed tools. For a broader set of attacks against this kind of graph, see red-teaming LangGraph agents.

Refunds above the threshold wait for a person

Large refunds pause for approval. The pause happens in gate, and that placement is the main design decision in this section. When a graph resumes after interrupt(), LangGraph re-runs the interrupted node from the top. Any side effect before the interrupt() call runs twice. So the node that interrupts must be pure: it computes a plan from state, asks the operator, and writes the plan. The refund itself runs later, in act.

class OperatorDecision(BaseModel):
    approved: bool
    operator_id: str
    note: str = ""

# ... inside build_graph
    def gate(state: SupportState, runtime: Runtime[SupportContext]) -> dict:
        log = node_logger("gate", runtime)
        plan = plan_tool_calls(charter, state["intent"], state["messages"][-1])
        for item in plan:
            if item["decision"] != "needs_approval":
                continue
            call = item["call"]
            decision = interrupt(
                {"kind": "refund_approval", "tool_call_id": call["id"], "args": call["args"],
                 "trace_id": runtime.context.trace_id, "customer_id": runtime.context.customer_id},
                response_schema=OperatorDecision,
            )
            item["decision"] = "approved" if decision.approved else "rejected"
            item["operator"] = decision.model_dump()
        # ... log the plan, including who approved it
        return {"plan": plan}

# tests/test_interrupt_resume.py
    first = graph.invoke({"messages": [("user", ASK)]}, config, context=alice, version="v2")
    [pending] = first.interrupts
    assert pending.value["args"]["amount_pence"] == 12_000
    assert graph.get_state(config).next == ("gate",)
    assert world.ledger.refunds == {}  # nothing paid while waiting

decision = {"approved": True, "operator_id": "op_sam", "note": "photos checked"}
    done = graph.invoke(Command(resume=decision), config, context=alice, version="v2")

Details that were worth verifying against langgraph 1.2.12 rather than recalling:

• `response_schema` on interrupt() accepts a Pydantic model. With one, the interrupt carries a JSON Schema the operator UI can render as a form, and the resume value is validated before interrupt() returns it. A test resumes with {"approved": "maybe"}: it raises ValidationError, nothing is paid, and the thread is still waiting.
• `version="v2"` makes invoke return a GraphOutput with .value and .interrupts, instead of an __interrupt__ key mixed into the state dict.
• Context is not checkpointed. Every resume must pass context= again. A resume without it hits the PermissionError that every node raises through node_logger before doing anything else, so a forgotten context fails closed.

Rejection is not a failure path. A rejected refund reaches the model as a ToolMessage with not_approved_by_operator, and the model tells the customer a colleague will follow up. The ledger stays empty.

The FastAPI layer rebuilds identity for the resume from its own thread registry (thread_id to tenant and customer), not from the request body, and it takes operator_id from the operator's session. An operator from another tenant gets a 404. So does a customer who tries to approve their own refund. A second click after the decision gets a 409, because get_state(...).interrupts is empty. The refund tool is also idempotent on thread_id:tool_call_id, so a retried act step cannot pay twice.

Approval waits last minutes or days, across deploys. InMemorySaver is only for tests. One test interrupts with a PostgresSaver (langgraph-checkpoint-postgres 3.1.2), closes the connection, builds a new graph on a new connection and resumes. The refund goes through exactly once.

The handoff record is a contract

When the agent stops, the person picking up the thread should not have to read the transcript and guess why. The handoff is a typed record with a closed set of reasons, each mapped to the question the operator actually needs to answer.

HandoffReason = Literal[
    "charter_always_escalates",
    "unknown_intent",
    "low_confidence",
    "no_grounded_sources",
    "uncited_answer",
    "tool_round_limit",
]
# ... QUESTIONS maps each reason to the question the operator must answer

class HandoffRecord(BaseModel):
    thread_id: str
    trace_id: str
    tenant_id: str
    customer_id: str
    intent: str | None
    reason: HandoffReason
    question_for_operator: str
    summary: str
    actions: list[dict] = Field(default_factory=list)
    sources: list[CitedSource] = Field(default_factory=list)
    draft_reply: str | None = None
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))

def extractive_summary(messages: list[AnyMessage], actions: list[dict]) -> str:
    """The customer's own words plus what the agent did. No model in this path."""
    said = [m.text for m in messages if isinstance(m, HumanMessage)]
    lines = [f"Customer: {s}" for s in said]
    lines += [f"Agent {a['outcome']}: {a['tool']}({a['args']})" for a in actions]
    return "\n".join(lines)

The summary is extractive on purpose. An LLM summary reads better. But the handoff path is where the model has already failed or been ruled out, and a generated summary can drop the one sentence the operator needs ("I've already been charged twice"). The record keeps the customer's own words and every action with its outcome (executed, refused, rejected). A human-readable summary can be added on top, but never in place of this.

Two reasons deserve a note. uncited_answer fires when the model writes a reply that cites none of the retrieved sources. The reply is not sent; it goes into draft_reply so the operator can fix it rather than start from nothing. no_grounded_sources fires when retrieval finds nothing in the sources the charter allows. That is a knowledge-base gap, and it should end up as a new help article, not a guessed answer. Retrieval here is deliberately a keyword stand-in. The pgvector version, with tenant and permission filters inside the SQL, is in keeping a knowledge base fresh.

One trace id from request to tool call

When an operator asks why the agent refunded something, I want to answer from logs, not from memory. The API layer mints a trace id per request. The thread id lives as long as the conversation. Every node binds both before doing any work.

def node_logger(node: str, runtime: Runtime[SupportContext]) -> structlog.typing.FilteringBoundLogger:
    ctx = runtime.context
    if not isinstance(ctx, SupportContext):
        # Every node starts here, so a run without authenticated context fails closed.
        raise PermissionError("graph invoked without an authenticated SupportContext")
    return structlog.get_logger().bind(
        node=node, trace_id=ctx.trace_id, thread_id=ctx.thread_id, tenant_id=ctx.tenant_id
    )

def run_config(ctx: SupportContext) -> RunnableConfig:
    """Config for graph.invoke: the checkpoint thread plus a root run id equal to
    our trace id, so LangSmith (if enabled) and our logs share one identifier."""
    return {
        "configurable": {"thread_id": ctx.thread_id},
        "run_id": uuid.UUID(ctx.trace_id),
        "metadata": {"trace_id": ctx.trace_id, "tenant_id": ctx.tenant_id},
        "tags": ["support-agent"],
    }

# The gate's log line after an approved resume (trimmed):
# {"node": "gate", "trace_id": "aeb1196b-...", "thread_id": "f2d138d0...", "tenant_id": "acme",
#  "decisions": {"issue_refund": "approved"}, "approved_by": ["op_eval"], "event": "tool_plan", ...}

The tests pin this down in three ways:

• structlog.testing.capture_logs shows that one order-status run emits exactly route, retrieved, model_step, tool_plan, tool_call, model_step, answer, all with the same trace and thread ids.
• No log line carries customer_id. The handoff record has it; the log aggregator does not need it.
• A callback handler records the root run id. It equals the trace id, so a LangSmith trace and a log search use the same key. I did not send traces to LangSmith for this article; the test proves only that the id is propagated through LangChain's callback system.

A resumed run gets a new trace id and keeps the thread id. A resume is a separate request, often from a different person hours later, so it gets its own trace. The thread id joins the two.

Evaluating decisions with scripted conversations

Model quality needs evaluation with a real model: answer correctness, tone, classification accuracy. That belongs in shadow mode and eval gates, covered in shipping LLM features behind eval gates. But much of what makes this agent safe does not depend on the model. Given these model outputs, did the system route correctly, run the right tools, refuse the wrong ones and pause where it should? Those questions have exact answers, so they run on every commit.

Each case is a conversation: customer turns, operator decisions, the model's scripted outputs, and the expected outcome.

REFUND_ASK = "My delivered order ORD-1002 arrived damaged, please refund it"
BIG_REFUND = call_tool("issue_refund", order_id="ORD-1002", amount_pence=12_000, reason="damaged")

CASES = [
    # ...
    Case("big_refund_waits", [REFUND_ASK], [classify_as("refund_request"), BIG_REFUND],
         "awaiting_approval"),
    Case("big_refund_rejected", [REFUND_ASK, REJECT],
         [classify_as("refund_request"), BIG_REFUND,
          say("A colleague will be in touch [refund-policy#1].")],
         "answered", [("issue_refund", "rejected")]),
    Case("refund_tool_under_status_intent", ["Where is my order ORD-1001? Shipped yet?"],
         [classify_as("order_status"),
          call_tool("issue_refund", order_id="ORD-1001", amount_pence=3_500, reason="sorry"),
          say("It was delivered [delivery-policy#1].")],
         "answered", [("issue_refund", "refused")]),
    # ...
]

# evals/runner.py, run_case (trimmed)
    for turn in case.turns:
        ctx = SupportContext("acme", "cus_alice", thread, str(uuid.uuid4()))
        payload = (Command(resume=turn) if isinstance(turn, dict)
                   else {"messages": [HumanMessage(turn)]})
        if isinstance(turn, dict) and not graph.get_state(run_config(ctx)).interrupts:
            return {"name": case.name, "passed": False,
                    "failures": ["operator decision scripted but nothing awaited approval"]}
        # ... invoke, then compare outcome, (tool, outcome) pairs and handoff reason

There are 11 cases, covering answered questions, own-order lookups, small and large refunds (waiting, approved, rejected), an out-of-charter tool, forced escalation, low confidence, an uncited answer, and a two-turn conversation that moves from order status to a refund. The runner fails a case in two situations beyond a wrong outcome:

• The graph wants a model step the script did not provide. The fake raises ScriptExhausted instead of a bare StopIteration. That usually means a policy stopped pausing or escalating.
• Scripted steps are left over. That means the graph took a shorter path than expected.

A test suite for guardrails only earns trust if it fails when the guardrails go. So I stubbed Charter.allows and Charter.needs_approval to permit everything and re-ran the set. Four of the 11 cases failed, each with a specific message: the big refund no longer paused, the approve and reject cases had nothing awaiting approval, and the out-of-charter refund executed. The ownership check in the tool held regardless, which is the point of having it in two places.

Checklist

• Keep the charter (intent to allowed sources, allowed tools and escalation rule) in a reviewed file, validate it with extra="forbid", and refuse to boot if it names a tool with no implementation.
• Send unknown intents, low-confidence classifications and always-escalate topics to a person before retrieval or tools run.
• Put tenant and customer identity in server-built context (context_schema plus InjectedToolArg), never in a tool argument the model can fill.
• Drop model arguments outside tool_call_schema before injecting identity, so a smuggled ctx cannot overwrite the real one.
• Return the same error for "not yours" and "does not exist".
• Re-check the charter at execution time even though the model only saw the allowed tools.
• Call interrupt() only from a pure node. Run side effects in a later node, with an idempotency key.
• Pass a Pydantic response_schema to interrupt(), take operator_id from the operator's session, and return 409 when nothing is waiting.
• Use a durable checkpointer (PostgresSaver) for approval waits, and test a resume on a fresh connection.
• Make the handoff a typed record: a closed set of reasons, the operator's question, an extractive summary, actions with outcomes, cited sources and the draft reply.
• Refuse to send answers that cite no approved source.
• Bind trace id, thread id and tenant into every log line, make the root run_id equal the trace id, and keep customer identifiers out of logs.
• Run scripted conversations in CI that grade routing and tool decisions, and check that they fail when the policy is stubbed out.

Tested with Python 3.12.9, langgraph 1.2.12, langchain-core 1.6.5, langgraph-checkpoint-postgres 3.1.2, langchain-openai 1.6.6 (construction only, no calls), pydantic 2.13.5, FastAPI 0.141.1, structlog 26.1.0, pytest 9.1.1 and PostgreSQL 16.2 on 2026-09-27.