LangGraph

LangGraph is the stateful orchestration layer for multi-agent systems built on LangChain. It adds graphs, cycles, conditional routing, human-in-the-loop interrupts, and checkpointing on top of tool calls. The chio-langgraph package wraps each node so a node dispatch is evaluated through the Chio sidecar before the wrapped body runs, and it bridges LangGraph's interrupt() mechanism to chio's pending_approval verdict for policy-driven HITL.

Why LangGraph Through chio

chio already ships a LangChain integration that wraps chio tools as BaseTool instances. LangChain is the tool layer. LangGraph is the orchestration layer, and orchestration is where most of the abuse surface lives. A supervisor that can silently hand a worker write access, a replay that resumes with stale capabilities, a human approval modelled as UX rather than policy. chio turns each of those into a policy decision with a signed receipt.

LangGraph alone	LangGraph + chio
Graph structure defines control flow	Capability tokens scope what each node can do
Human-in-the-loop is UX-driven	Approval is a kernel verdict; the graph pauses via `interrupt()`
Tool calls via LangChain tools	Every node dispatch produces a signed `ChioReceipt`
Subgraphs inherit control flow only	Subgraph scopes are enforced against the parent ceiling

Public Surface

chio-langgraph exports a small set of primitives. Everything else (delegation helpers, checkpoint adapters, toolkit injection) is roadmap work, not shipped surface. See the callout below.

Name	Responsibility
`chio_node`	Wrap a LangGraph node callable so each dispatch evaluates through the sidecar before the wrapped body runs. A deny verdict raises `ChioLangGraphError`.
`chio_approval_node`	Wrap a node that must await human approval. Posts a request, pauses the graph via `interrupt()`, and resumes when the caller supplies a decision via `Command(resume=...)`.
`ChioGraphConfig`	Graph-level wiring: the `ChioClient`, the workflow scope ceiling, per-node scopes, and (for nested subgraphs) the parent ceiling.
`enforce_subgraph_ceiling`	Validate that a per-node scope is a subset of the current graph ceiling. Called at wrap time; safe to call eagerly from user code.
`ApprovalRequestPayload`, `ApprovalResolution`	Wire shapes for the HITL approval flow.
`ChioLangGraphError`, `ChioLangGraphConfigError`	Error types. The config error surfaces at wrap time; the runtime error surfaces on a deny verdict.

Architecture

Each node wrapped with chio_node carries a capability token minted for its declared scope. Before the wrapped body runs, the wrapper calls evaluate_tool_call on the ChioClient, using tool_server="langgraph" and tool_name=<node_name>. From the sidecar's perspective, a node dispatch is a tool call against a virtual tool server; scope enforcement, receipt signing, and approval guards all work the same way.

rendering…

Each node runs under a capability minted for its scope; the sidecar evaluates the dispatch, signs a receipt, and returns allow, deny, or pending_approval.

Node-Level Scoping

Each node's scope must be a subset of the graph's effective ceiling. ChioGraphConfig enforces this on registration; chio_node re-checks at wrap time via enforce_subgraph_ceiling, so misconfiguration surfaces during graph construction, not at first invocation.

Graph-Level Configuration

Build a ChioGraphConfig with the chio client, the workflow-level ceiling, and a per-node scope map. Call provision() before running the graph so a capability token is minted for each node.

python

from chio_sdk import ChioClient
from chio_sdk.models import ChioScope, Operation, ToolGrant
from chio_langgraph import ChioGraphConfig, chio_node
from langgraph.graph import StateGraph, START, END

SERVER_ID = "demo-srv"

def scope_for(*tools: str) -> ChioScope:
    return ChioScope(
        grants=[
            ToolGrant(
                server_id=SERVER_ID,
                tool_name=name,
                operations=[Operation.INVOKE],
            )
            for name in tools
        ]
    )

chio = ChioClient("http://127.0.0.1:9090")

config = ChioGraphConfig(
    chio_client=chio,
    workflow_scope=scope_for("search", "browse", "write"),
    node_scopes={
        "researcher": scope_for("search", "browse"),
        "writer": scope_for("write"),
    },
    subject="agent:my-pipeline",
    ttl_seconds=3600,
)

# Mint a capability token for the workflow and each node.
await config.provision()

Scopes narrow, never widen

Per-node scopes and subgraph ceilings must be subsets of workflow_scope. When a graph runs as a subgraph, its parent_ceiling is the authoritative bound. This is checked by the SDK at registration time and again by the kernel on every dispatch, so a bug in graph definition cannot escalate privileges.

The Node Wrapper

chio_node wraps a LangGraph node callable with chio capability enforcement. It preserves sync and async shapes and both the (state) and (state, config) LangGraph arities.

python

from chio_langgraph import chio_node
from typing import TypedDict

class AgentState(TypedDict, total=False):
    messages: list[dict]
    draft: str

def researcher(state: AgentState) -> dict:
    # ...actual agent work here...
    return {"messages": state.get("messages", []) + [{"role": "assistant", "content": "..."}]}

def writer(state: AgentState) -> dict:
    return {"draft": "..."}

graph = StateGraph(AgentState)
graph.add_node(
    "researcher",
    chio_node(researcher, scope=scope_for("search", "browse"), config=config),
)
graph.add_node(
    "writer",
    chio_node(writer, scope=scope_for("write"), config=config),
)
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", END)

app = graph.compile()

Signature:

python

chio_node(
    fn,
    *,
    scope: ChioScope,
    config: ChioGraphConfig,
    name: str | None = None,        # defaults to fn.__name__
    tool_server: str = "langgraph", # sidecar tool_server identifier
)

When the runtime config carries configurable["chio_capability_id"], that id overrides the token resolved from ChioGraphConfig. This lets a supervisor node hand a narrower capability to a child subgraph via LangGraph's standard config propagation.

Approval Nodes

chio_approval_node bridges LangGraph's interrupt() mechanism to chio's pending_approval verdict. The flow:

Evaluate the dispatch through the sidecar. A deny raises immediately.
If the kernel returns pending_approval, or if a local approval_policy requires approval, build an ApprovalRequestPayload and pause the graph via interrupt().
When the caller resumes with Command(resume=...), normalise the value into an ApprovalResolution. An approved outcome runs the wrapped body; any other outcome raises ChioLangGraphError.

python

from chio_langgraph import chio_approval_node

async def send_email(state: AgentState) -> dict:
    # ...actual side effect here...
    return {"sent": True}

graph.add_node(
    "send_email",
    chio_approval_node(
        send_email,
        scope=scope_for("email-send"),
        config=config,
        # Optional: a local policy can also *require* approval even if the
        # kernel did not flag pending_approval. Default: always require.
        approval_policy=None,
        approval_ttl_seconds=3600,
        summary="Send the drafted customer email",
    ),
)

The payload delivered to interrupt() is the ApprovalRequestPayload as a dict. The resume value can be any of: an ApprovalResolution, a dict with outcome (or approved: bool), the strings "approved", "denied", or "rejected", or a bare boolean. The wrapper normalises all of these.

The kernel owns the policy

Approver lists, escalation rules, and timeouts belong in the kernel approval policy, not hard-coded into graph Python. The local approval_policy callable exists for cases where the node itself wants to force a pause regardless of kernel verdict (for example, debug builds).

Subgraph Isolation

LangGraph supports nested subgraphs. Build a child ChioGraphConfig via subgraph_config(...): the child's parent_ceiling is pinned to the outer graph's effective ceiling, so any node the subgraph registers must attenuate that ceiling.

python

# Outer graph: research + write
outer_config = ChioGraphConfig(
    chio_client=chio,
    workflow_scope=scope_for("search", "browse", "write"),
    node_scopes={"outer_plan": scope_for("search")},
)
await outer_config.provision()

# Inner subgraph: research only. The parent ceiling is carried over.
inner_config = outer_config.subgraph_config(
    workflow_scope=scope_for("search", "browse"),
    node_scopes={
        "search": scope_for("search"),
        "analyze": scope_for("browse"),
    },
    subject="agent:research-subgraph",
)
await inner_config.provision()

Per-node scopes inside the subgraph are validated against the parent ceiling at registration time. A subgraph node that tries to declare a scope outside the ceiling raises ChioLangGraphConfigError during graph construction.

Errors and Deny Handling

A deny verdict from the kernel, or a missing capability token (for example, you forgot to call provision()), raises ChioLangGraphError from the node wrapper. The error carries the node name, the tool server, the guard that produced the deny, a reason string, and the receipt id for audit correlation.

python

from chio_langgraph import ChioLangGraphError

try:
    await app.ainvoke({"messages": []})
except ChioLangGraphError as exc:
    # exc.node_name, exc.guard, exc.reason, exc.receipt_id
    logger.warning(
        "chio denied %s via %s: %s (receipt=%s)",
        exc.node_name, exc.guard, exc.reason, exc.receipt_id,
    )
    raise

Roadmap

Design sketches below are not yet shipped

The sections that follow describe features we intend to add but have not yet shipped. Until they land, none of the referenced types or kwargs (ChioNodeContext, ChioDelegation, ChioCheckpointAdapter, budget= / can_delegate= kwargs on chio_node) exist in chio-langgraph. Track the work before relying on them.

Supervisor / Worker Delegation

When a supervisor dispatches to a worker, the capability should narrow in place. Today the supervisor pattern is expressed by configuring per-node scopes on ChioGraphConfig and (if needed) overriding the capability id at the config configurable hook. A first-class chio_ctx.delegate(...) helper that mints a scoped child capability per dispatch is planned but not yet shipped.

Checkpoint and Receipt Alignment

LangGraph checkpoints graph state at each node; chio receipts record each dispatch. A ChioCheckpointAdapter that annotates checkpoint metadata with the receipt ids produced during the step is planned so you can query receipts by thread id and replay deterministically. Today, correlate via ChioReceipt.id surfaced on ChioLangGraphError or returned from the sidecar on allow.

Scoped Toolkit Injection

Ideally, each node receives a ChioToolkit already filtered to the tools the node scope permits. Today, ChioToolkit (from chio-langchain) is constructed with a capability id and discovers tools via get_tools(server_id=...). A ChioToolkit.from_context(...) classmethod that reads the active node's scope is planned.

Package Layout

The Python packages form a layered dependency chain so you can adopt as much as you need:

text

chio-sdk                 (base HTTP client to the sidecar)
    |
chio-langchain           (tool wrapping: ChioTool, ChioToolkit)
    |
chio-langgraph           (node / approval-node wrappers, ChioGraphConfig)

chio-langgraph does not depend on chio-langchain. Teams that only need tool wrapping inside otherwise-unchanged LangChain agents can use chio-langchain directly.

Next Steps

Temporal · durable workflow integration with per-activity enforcement
Custom Guards · define the approval guard the kernel uses when chio_approval_node pauses
Budgets · attach per-capability ceilings at mint time

PreviousA2A Adapter NextTemporal