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.

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 identity, a run trace, the tools you invoked, and the context you retrieved. Each field is optional — add what you have. The verified identity always comes from your API key; a caller-supplied actor.agent_id is recorded as a claimed sub-agent until an admin verifies it.

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.
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.
client.requests.create(
    request_type="approval",
    title="Refund $4,200 to customer 8831",
    source={"integration": "langgraph"},
    actor={"agent_id": "billing-agent", "agent_name": "Billing Agent"},  # claimed sub-agent
    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.