centcom-langgraph 0.3.0

Contro1 connector for LangGraph human approvals and human input

centcom-langgraph

Human approval and human-input nodes for LangGraph workflows, powered by Contro1.

Agent Integration Kit

To save time, give your coding agent this skill. It inspects your system, reports governance gaps, and suggests Contro1 integration (optional):

https://contro1.com/agent-kit

Drop a Contro1 request node into any LangGraph graph. The connector uses LangGraph's native interrupt() to pause graphs and resume them when operators approve, reject, or provide requested input in Contro1. The canonical response is delivered back to your app via webhook and used to resume the graph - no thread blocked, fully persistent.

This connector normalizes requests through Contro1 Integration Protocol v1.

Install

pip install centcom-langgraph

# With webhook handler (FastAPI)
pip install centcom-langgraph[webhook]

Environment variables

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

Quick Start

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from centcom_langgraph import centcom_approval, CentcomState

class MyState(CentcomState):
    order_id: str
    approved: bool

graph = StateGraph(MyState)
graph.add_node("approve", centcom_approval(
    type="approval",
    question=lambda s: f"Approve order {s['order_id']}?",
    context=lambda s: f"Order {s['order_id']} needs approval",
    callback_url="https://my-app.com/centcom-webhook",
))

graph.add_edge(START, "approve")
graph.add_edge("approve", END)

app = graph.compile(checkpointer=MemorySaver())
result = app.invoke(
    {"order_id": "ORD-42", "approved": False},
    config={"configurable": {"thread_id": "order-42"}},  # LangGraph's own state key
)

Context the reviewer can trust

Build context inside the context=lambda s: ... callable from what your code already has: the exact state/tool input (machine-observed), and the event that triggered the run. For centcom_tool(), where the model decides whether to request approval, make reason a required tool argument so the model's justification is produced at decision time, not asked for afterward - keep it in a separate agent-reported field. Agent-reported text should never change required_role, risk_level, or approval_policy; if a high-risk node is missing its required machine-observed context, fail closed rather than guessing. Full pattern: https://contro1.com/docs/requests-api

Runtime Flow

1. Your graph reaches centcom_approval(...) and sends a request to CENTCOM.
2. The node calls interrupt(...), so execution pauses and state is checkpointed.
3. An operator answers in the CENTCOM dashboard.
4. CENTCOM sends the signed response payload to your webhook endpoint.
5. Your webhook handler verifies the signature and resumes LangGraph with Command(resume=payload).

Case continuity

LangGraph's config.configurable.thread_id is LangGraph's own state key. The connector maps it to Contro1's correlation_id automatically. Every approval node and audit log entry in the same LangGraph run shares one case timeline in the CENTCOM dashboard.

Use client.log_action to record autonomous actions in the same case:

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,          # provided by the connector from LangGraph thread
    in_reply_to={"type": "request", "id": request_id},  # links back to the approval
)

Control Map preview

For high-risk approval nodes, Control Map can preview whether routing is satisfiable - for example, whether required reviewers are mapped and available. Cache this for 5-15 minutes; do not call it on every graph invocation.

from centcom import CentcomClient

client = CentcomClient()
preview = client.post("/requests/control-map", {
    "approval_requirements": {"required_roles": ["finance"], "required_approvals": 2},
    "approval_policy": {
        "mode": "threshold",
        "required_approvals": 2,
        "separation_of_duties": True,
        "fail_closed_on_timeout": True,
    },
})

if not preview["satisfiable"]:
    # preview["warnings"] lists the setup gap.
    print("Routing setup needed:", preview["warnings"])

Response fields: satisfiable (bool), status (ready | needs_mapping | needs_capacity), warnings, suggested_action.

The approval node remains the gate: continue only after the final signed decision.

API

centcom_approval(**kwargs)

Factory returning a LangGraph node. Parameters accept static values or (state) -> value callables. Supports continuation_mode="decision" | "instruction" and protocol routing metadata propagation.

centcom_tool(**kwargs)

LangChain @tool for agent graphs where the LLM decides when to request approval.

create_webhook_handler(**kwargs)

Async handler that verifies CENTCOM webhooks and resumes LangGraph threads.

CentcomState

TypedDict mixin adding centcom_request_id, centcom_response, centcom_status to your graph state.

Production pattern: Agent Plugin

For teams running multiple graphs with overlapping governance requirements, a thin plugin reduces token overhead and keeps policy consistent:

from datetime import datetime, timedelta
from centcom import CentcomClient

class Contro1Plugin:
    """Wraps Contro1 calls. preview_policy is TTL-cached."""

    def __init__(self, client: CentcomClient, cache_ttl_minutes: int = 10):
        self._client = client
        self._cache: dict = {}
        self._ttl = timedelta(minutes=cache_ttl_minutes)

    def preview_policy(self, approval_requirements: dict, approval_policy: dict) -> dict:
        key = str(sorted(approval_requirements.items()))
        cached = self._cache.get(key)
        if cached and datetime.utcnow() < cached["expires"]:
            return cached["data"]
        result = self._client.post("/requests/control-map", {
            "approval_requirements": approval_requirements,
            "approval_policy": approval_policy,
        })
        self._cache[key] = {"data": result, "expires": datetime.utcnow() + self._ttl}
        return result

    def request_human_review(self, payload: dict) -> dict:
        return self._client.create_protocol_request(payload)

    def log_audit_action(self, payload: dict) -> dict:
        return self._client.log_action(**payload)

    def resume_from_decision(self, case_id: str) -> dict:
        return self._client.get(f"/cases/{case_id}")

Official Resources

• SDK repository: github.com/contro1-hq/centcom-langgraph
• Core Python SDK (centcom): github.com/contro1-hq/centcom
• Official skill file: skills/centcom-langgraph.md
• Microsoft AGT companion: github.com/contro1-hq/contro1-microsoft-agent-governance-toolkit-integration
• Full docs: contro1.com/docs

Governance readiness

For teams operating AI in regulated environments:

• EU AI Act readiness guide
• US AI Governance readiness guide