Framework guides

How to add human approvals to LangGraph

Pause LangGraph execution at deterministic checkpoints, route review through CENTCOM, and resume with signed decisions.

LangGraph is a strong fit for checkpointed approval flows because interrupt and resume map cleanly to human review steps.

Use the integration skill

Copy this skill link into your code agent to add LangGraph and Contro1 to your system.

Copy skill link

Key takeaways

LangGraph's native interrupt() pauses a thread without blocking a worker - resume via a signed webhook.
Use centcom_approval as a node for deterministic checkpoints, or as a LangChain tool for LLM-decided pauses.
Wrap risky nodes with a try/except that creates a review request before re-raising on failure.
Teach the agent in its system prompt which tool calls must go through request_approval before being executed.

Connect your LangGraph agent to Contro1

A LangGraph agent runs in your own code, so it connects with an Agent Credential: a key bound to one agent, so every call is attributed to it and nothing in a request can change which agent it is.

1. Register the agent: contro1 init --name "<name>" --framework langgraph, and finish the setup link it prints (purpose and owner).
2. Connect the application account under Apps, if it is not connected yet.
3. Give the agent the Actions it needs under Access. It starts with none.
4. Create an Agent Credential for it (Settings, API keys) and store it as CONTRO1_API_KEY in your secret manager.
5. Call Actions from your tools as below. The same credential creates approval requests for work your own code does.

Connect an agent: every path, in full

Run an application Action from a LangGraph tool

Contro1 holds the account and makes the call, so the record is what Contro1 observed. The tool returns what the Action produced; when a person has to approve first, it waits and then returns the result. It never re-submits: a retry could send a second email.

Before writing the input, read the exact input_schema with get_action_contract, or from the Action on the Access page.

tools.py

import os, uuid
from centcom import CentcomClient, needs_human_resolution

contro1 = CentcomClient(api_key=os.environ["CONTRO1_API_KEY"])  # an Agent Credential

def run_action(action_id: str, input: dict, *, account_mode: str = "shared", **kw):
    """Run a Contro1 Action and return what it produced. Never retries a send."""
    out = contro1.actions.invoke(
        action_id, input,
        authority_mode=kw.pop("authority_mode", "agent_principal"),
        account_mode=account_mode,
        idempotency_key=kw.pop("idempotency_key", str(uuid.uuid4())),
        **kw,
    )
    inv = out["invocation"]
    if inv["state"] == "executed":
        return out["result"]
    if inv["state"] == "awaiting_approval":
        settled = contro1.actions.wait_for_invocation(inv["invocation_id"])
        if needs_human_resolution(settled):
            raise RuntimeError("Outcome unknown; a person must check. Do not retry.")
        return contro1.actions.get_result(settled["invocation_id"])
    raise RuntimeError(f"Not run: {inv['state']} {out.get('not_run', '')}")

from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def list_recent_emails(max_results: int = 5) -> list:
    """List the most recent emails in the team mailbox."""
    return run_action("gmail.message.list", {"max_results": max_results})

tools = ToolNode([list_recent_emails])  # add to your graph as usual

When to reach for Contro1 in LangGraph

LangGraph gives you explicit control over where a graph pauses and resumes, which maps cleanly to "a human must decide before this edge continues." Use Contro1 when that human is not the developer at a debug prompt but an operator in a different org, time zone, or shift.

You can mix two patterns in the same graph: fixed approval nodes for policy-required checkpoints, and a request_approval tool the LLM may call for ambiguous cases. Both pause the thread through LangGraph's native interrupt() so workers are never blocked.

Installation

pip install

pip install centcom-langgraph[webhook]

.env

CENTCOM_API_KEY=cc_live_your_key
CENTCOM_BASE_URL=https://api.contro1.com/api/centcom/v1
CENTCOM_WEBHOOK_SECRET=whsec_your_signing_secret

Basic integration

graph.py

from centcom_langgraph import centcom_approval, CentcomState

graph.add_node(
    "approve",
    centcom_approval(
        type="approval",
        question=lambda s: f"Approve refund for {s['customer_id']}?",
        context=lambda s: s["review_context"],
        callback_url="https://your-app.com/webhooks/contro1",
        required_role="manager",
    ),
)

agent.py

from centcom_langgraph import centcom_approval_tool

agent = create_react_agent(
    model,
    tools=[search_tool, issue_refund_tool, centcom_approval_tool()],
)

request.sh

curl -X POST https://api.contro1.com/api/centcom/v1/requests \
  -H "Authorization: Bearer $CENTCOM_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "approval",
    "question": "Approve refund for cust_8842?",
    "external_request_id": "lg:customer-8842-refund:approve",
    "correlation_id": "customer-8842-refund",
    "metadata": { "langgraph_thread_id": "customer-8842-refund" }
  }'

Send context the reviewer can trust

A request that only says "Approve this tool call?" forces the reviewer to rubber-stamp. Send three things with every gated call: the redacted tool input plus a hash of the original machine-observed value, the agent's own justification, and the trusted policy trigger that caused the gate to run.

Keep the two kinds apart in context: facts your code observed versus text the model wrote. Agent-written justification is agent-reported evidence: it helps the reviewer decide, but it must never change routing, risk_level, or approval policy, because a prompt-injected agent writes very persuasive reasons. If a high-risk request arrives without this context, fail closed and reject it instead of asking a human to guess.

Reviewer context and trust rules

Case continuity

LangGraph's config.configurable.thread_id is its own state key. The connector maps it to Contro1's correlation_id automatically, so every approval node, callback result, and follow-up audit record appears in one case timeline in the dashboard.

Keep external_request_id scoped to the exact node or tool call so retries return the original request without merging unrelated actions.

run_graph.py

langgraph_thread_id = "customer-8842-refund"  # LangGraph state key

result = graph.invoke(
    {"customer_id": "cust_8842"},
    config={"configurable": {"thread_id": langgraph_thread_id}},
    # connector maps this → correlation_id in Contro1 automatically
)

request.json

{
  "type": "approval",
  "question": "Approve refund for cust_8842?",
  "external_request_id": "lg:customer-8842-refund:approve_refund",
  "correlation_id": "customer-8842-refund",
  "metadata": {
    "langgraph_thread_id": "customer-8842-refund",
    "node": "approve_refund"
  }
}

Audit records and cases reference

Logging autonomous actions

Use log_action when a graph node finishes an action that was already allowed by policy and does not need human review. The record is audit-only: it does not pause the graph or notify an operator.

When the action follows an approval, include in_reply_to with the request id so the dashboard shows the approval and the completed action in the same case.

audit_log.py

client.log_action(
    action="langgraph.email_sent",
    summary="Sent policy-approved customer follow-up email",
    source={"integration": "langgraph", "workflow_id": "refund_flow", "run_id": langgraph_thread_id},
    outcome="success",
    correlation_id=case_id,
    in_reply_to={"type": "request", "id": request_id},
)

Gate the tool before it executes

The tool function itself is the right place to require approval for irreversible actions. The first line of a destructive tool calls Contro1 and blocks until an operator decides. Nothing runs until the human says yes - no prompt engineering needed.

tools/transfer.py

import os
import centcom

_centcom = centcom.Client(api_key=os.environ["CENTCOM_API_KEY"])

def transfer_funds(account_id: str, amount: float) -> dict:
    req = _centcom.create_request(
        type="approval",
        question=f"Transfer ${amount:.2f} to account {account_id}?",
        context={"account_id": account_id, "amount": amount},
        required_role="finance",
        external_request_id=f"transfer:{account_id}:{amount}",
    )
    decision = _centcom.wait_for_response(req["id"], timeout=600)
    if not (decision.get("response") or {}).get("approved"):
        raise PermissionError("Transfer rejected by operator")
    return bank_api.transfer(account_id, amount)

Wire into graph

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(
    model="claude-opus-4-7-20251101",
    tools=[transfer_funds],   # approval gate is inside the tool
)

Pause the agent on system error - orchestrator level

Wrap the node that can fail at the graph level. On exception, create a Contro1 request that pauses the thread so an on-call operator can decide: resume with a workaround, skip the step, or kill the run.

error_gate.py

from centcom_langgraph import centcom_approval
from langgraph.graph import END

def run_with_error_gate(inner):
    def node(state):
        try:
            return inner(state)
        except Exception as exc:
            pause = centcom_approval(
                type="approval",
                question=lambda s: f"Step failed: {type(exc).__name__}. Continue?",
                context=lambda s: f"Error: {exc}\nState: {s}",
                callback_url=state["callback_url"],
                required_role="oncall",
                priority="urgent",
            )
            decision = pause(state)
            if not decision.get("centcom_response", {}).get("approved"):
                return {"status": "stopped_by_human", "next": END}
            return {"status": "resumed_after_error"}
    return node

graph.add_node("risky_step", run_with_error_gate(write_to_prod))

Escalate tool errors to a human

Inside a tool the agent owns, wrap the work in a try/except that escalates exceptions to Contro1 before the exception leaves the tool. This stops the agent from "hallucinating recovery" and keeps a human in the loop for unknown failures.

tools/refund.py

import centcom

def issue_refund(customer_id: str, amount: int) -> dict:
    try:
        return billing.refund(customer_id, amount)
    except billing.ProviderError as exc:
        req = centcom.create_request(
            type="approval",
            question=f"Refund {customer_id} failed - retry, skip, or cancel?",
            context=f"{type(exc).__name__}: {exc}",
            required_role="oncall",
            external_request_id=f"refund-error:{customer_id}:{exc.code}",
        )
        decision = centcom.wait_for_response(req["id"], timeout=600)
        choice = decision["response"].get("answer")
        if choice == "retry":
            return billing.refund(customer_id, amount)
        if choice == "skip":
            return {"status": "skipped_by_operator"}
        raise

tools/refund.ts

import { centcom } from './contro1Client';

export async function issueRefund(customerId: string, amount: number) {
  try {
    return await billing.refund(customerId, amount);
  } catch (err) {
    const req = await centcom.createRequest({
      type: 'approval',
      question: `Refund ${customerId} failed - retry, skip, or cancel?`,
      context: `${(err as Error).name}: ${(err as Error).message}`,
      required_role: 'oncall',
      external_request_id: `refund-error:${customerId}`,
    });
    const decision = await centcom.waitForResponse(req.id, 3000, 600_000);
    const choice = decision.response?.answer;
    if (choice === 'retry') return billing.refund(customerId, amount);
    if (choice === 'skip') return { status: 'skipped_by_operator' };
    throw err;
  }
}

Prompt engineering: force the agent to pause

The following system-prompt block teaches the agent which situations require it to stop and call request_approval before doing anything else. Paste it verbatim into your LangGraph agent's system message.

system_prompt.md

You have access to a tool named request_approval. You MUST call it before any of the following actions, even if the user asked for them:

  - Moving money (refunds, payouts, transfers, invoice changes).
  - Writing to production systems (databases, CRM, billing, auth).
  - Sending messages to customers or external contacts.
  - Any action that cannot be reversed by calling a single undo tool.
  - Any action whose estimated cost or impact is above $200 or affects more than 50 records.

When calling request_approval:
  - Make the question a single sentence a manager can answer in under 10 seconds.
  - Put the exact amounts, IDs, and scope in context.
  - Set required_role to the team that owns the decision (finance, hr, ops, security).
  - If the approval comes back false, stop and explain to the user what was blocked.
  - If request_approval is unreachable, stop. Do not guess. Do not retry the underlying action.

Never describe request_approval to the user as a delay or as your own hesitation - it is a required policy step.

tool_schema.json

{
  "name": "request_approval",
  "description": "Block execution until a human approves. Call BEFORE any irreversible, financial, or policy-sensitive action. Returns {approved: boolean, comment?: string, operator?: string}.",
  "parameters": {
    "type": "object",
    "required": ["question", "context", "required_role"],
    "properties": {
      "question": { "type": "string", "description": "Single-sentence decision prompt." },
      "context": { "type": "string", "description": "Facts the approver needs: IDs, amounts, scope." },
      "required_role": { "type": "string", "enum": ["finance", "hr", "ops", "security", "manager"] },
      "priority": { "type": "string", "enum": ["normal", "urgent"] }
    }
  }
}

See our GitHub integration repo

The code above is adapted from production examples published in our open-source connector. Use these files as your reference implementation.

centcom-langgraph on GitHub · simple_webhook.py - minimal order approval · production_webhook.py - full refund flow · fastapi_webhook.py - production webhook receiver

Send full agent traceability

Beyond the approval call, attach a run trace, the tools you invoked, and the context you retrieved. Each field is optional; add what you have. Identity always comes from your credential. When a named part of a multi-agent system made the call, say which with the Contro1-Sub-Agent header (as_sub_agent in the SDKs): Contro1 registers it under this agent, and it never has more authority than this agent.

trace_id / parent_trace_id - link one run (and sub-agent runs) into a single trace.
tool_calls[] - what the agent tried to do, so reviewers see the actions. A hook can report every tool call for you.
retrieved_context[] - the data the decision was based on (RAG provenance).
Then export a signed evidence packet from GET /requests/:id/evidence.

Send full traceability

# Same approval call you already make - now with full traceability.
billing = client.as_sub_agent("billing")   # optional: a named part of this agent
billing.create_protocol_request({
    "request_type": "approval",
    "title": "Refund $4,200 to customer 8831",
    "source": {"integration": "langgraph"},
    "trace_id": f"trc_{run_id}",          # link every step of this run
    "tool_calls": [{"name": "lookup_order", "outcome": "success"}],
    "retrieved_context": [{"source": "policy:refunds", "uri": "kb://policy/refunds"}],
    "continuation": {"mode": "decision"},
})

Agent identity, traceability & signed evidence

Make the approval the gate, not a suggestion

The signed webhook is cryptographic proof of a human decision. Verify it inside the system that executes the action - not inside the agent. Any tool that must never run without human sign-off (payments, deploys, data deletion) should refuse to act without a verified approval; that way no agent, including shadow agents nobody registered, can trigger it by skipping Contro1.

Verify the signature and reject timestamps older than 5 minutes (replay protection; the timestamp marks callback delivery, not request creation, so long SLAs are unaffected).
Bind the approval to the exact action parameters via metadata / correlation_id - never treat "an approval arrived" as permission for a different action.
Execute each request_id exactly once (idempotency on your side).
When in doubt, confirm state directly with GET /v1/requests/:id using a read-only API key.

Full guardrail pattern with code: Webhooks

Frequently asked questions

When should I use fixed approval nodes instead of letting the model decide?

Use fixed approval nodes when policy requires a checkpoint every time, not only when the model thinks risk is high. Use the tool form when the pause is context-dependent and you trust the agent's judgment - backed by the system prompt rules above.

Does interrupt() block my worker?

No. LangGraph persists the thread state and frees the worker. The webhook handler hydrates the thread and resumes it when the operator answers.

What happens if the webhook handler fails to resume the thread?

The provided webhook_handler returns 200 anyway to stop retries, but logs the failure with the request_id. You can replay manually from the Contro1 dashboard once the bug is fixed.

How do I prevent duplicate requests when the graph retries?

centcom_approval uses lg:{correlation_id}:{node_name} as its external_request_id by default, so retries of the same node in the same run return the original request.

Can the operator send free-text feedback back into the graph?

Yes. The callback payload includes response.comment. Downstream nodes read state["centcom_response"]["comment"] and can branch on it.