LangGraph

Inject Cheqpoint approval checkpoints as nodes in LangGraph stateful graphs. Block state transitions until a human reviewer approves each sensitive action.

Prerequisites

Python 3.9+ environment.
cheqpoint and langgraph packages installed.
LangGraph graph defined with a StateGraph.
Cheqpoint Connection Key.

Steps

Define a Cheqpoint approval node function that reads action details from the graph state.
In the node, call client.request_sync() with the action type, summary, and details extracted from state.
If approved, update the state to mark approval and pass any modifiedDetails to the next node.
If rejected, set an error flag in the state and route to a fallback or end node.
Wire the approval node between your plan node and action node using add_conditional_edges.
Compile and run the graph — it will block at the approval node until a decision is received.

Installation

bash

pip install cheqpoint langgraph langchain-core

Sample request payload

json

{
  "action": "deploy_schema_migration",
  "summary": "LangGraph agent attempting to run a database migration",
  "details": {
    "migration": "0042_add_user_roles",
    "tables_affected": ["users", "roles"],
    "estimated_rows": 250000
  },
  "justification": "New RBAC feature requires schema update before release."
}

Sample Cheqpoint response

json

{
  "status": "approved",
  "modifiedDetails": null,
  "decisionNote": "Migration reviewed and approved. Run in off-peak hours."
}

Python — approval node in a LangGraph StateGraph

python

import os
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from cheqpoint import CheqpointClient

cheq = CheqpointClient(api_key=os.environ["CQ_API_KEY"])

class AgentState(TypedDict):
    action: str
    summary: str
    details: dict
    justification: str
    approval_status: str
    effective_details: dict

def approval_node(state: AgentState) -> AgentState:
    result = cheq.request_sync(
        action=state["action"],
        summary=state["summary"],
        details=state["details"],
        justification=state.get("justification"),
        timeout_ms=60_000,  # wait up to 60 s
    )
    effective = result.get("modifiedDetails") or state["details"]
    return {
        **state,
        "approval_status": result["status"],
        "effective_details": effective,
    }

def route_after_approval(state: AgentState) -> Literal["action_node", "rejected_node"]:
    return "action_node" if state["approval_status"] == "approved" else "rejected_node"

def action_node(state: AgentState) -> AgentState:
    run_migration(state["effective_details"])
    return state

def rejected_node(state: AgentState) -> AgentState:
    print(f"Action rejected by reviewer.")
    return state

graph = StateGraph(AgentState)
graph.add_node("approval", approval_node)
graph.add_node("action_node", action_node)
graph.add_node("rejected_node", rejected_node)
graph.set_entry_point("approval")
graph.add_conditional_edges("approval", route_after_approval)
graph.add_edge("action_node", END)
graph.add_edge("rejected_node", END)

app = graph.compile()

Notes

You have full control over what data is passed into the details object to provide human reviewers with sufficient context. Use modifiedDetails to let reviewers adjust parameters before execution.

Tips

For long-running graphs, use request_async() with a callback_url to receive the decision via webhook instead of blocking the graph thread.

Get your Connection Key at cheqpoint.co/signup.