Chio/Docs
LOGIN · JOIN

BuildOrchestrators

LangGraph

Use chio-langgraph to evaluate LangGraph node dispatches and map human approval interrupts to Chio's pending_approval verdict.


Why LangGraph Through Chio

chio already ships a LangChain integration that wraps chio tools as BaseTool instances. LangChain provides tool integration; LangGraph provides orchestration. Chio evaluates node dispatches to prevent a supervisor from granting unapproved write access, a replay from using stale capabilities, or an approval interrupt from bypassing policy. Each evaluation produces a signed receipt.

LangGraph aloneLangGraph + Chio
Graph structure defines control flowCapability tokens scope what each node can do
Human-in-the-loop is UX-drivenApproval is a kernel verdict; the graph pauses via interrupt()
Tool calls via LangChain toolsEvery node dispatch produces a signed ChioReceipt
Subgraphs inherit control flow onlySubgraph scopes are enforced against the parent ceiling

Install

The package requires langgraph>=0.2,<1 and pulls in chio-sdk-python, the distribution that provides the chio_sdk module every snippet on this page imports, together with chio-adapter-base, which supplies the redaction policy described below. The chio sidecar runs next to the process running the graph.

sdks/python/chio-langgraph/README.md:12-14bash
uv pip install chio-langgraph
# or
pip install chio-langgraph

Public API

chio-langgraph exports eleven names from its package root. This is all of them.

NameResponsibility
chio_nodeWrap a LangGraph node callable so each dispatch evaluates through the sidecar before the wrapped body runs. A deny verdict raises ChioLangGraphError.
chio_approval_nodeWrap 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=...).
ChioGraphConfigGraph-level wiring: the ChioClient, the workflow scope ceiling, per-node scopes, and (for nested subgraphs) the parent ceiling.
enforce_subgraph_ceilingValidate 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, ApprovalResolutionWire shapes for the HITL approval flow: the request handed to interrupt() and the reviewer answer that comes back.
ApprovalDispatcherType alias for an optional async hook that receives the ApprovalRequestPayload before the pause, for posting it to an approval service. Leave it unset and the payload is surfaced through interrupt() only.
ApprovalPolicy, ApprovalPolicyDecisionType aliases for the local pause rule: (state, config) returning either a bool or an ApprovalRequestPayload that overrides the default payload fields.
ChioLangGraphError, ChioLangGraphConfigErrorError 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. The subset relation is checked by the SDK, at registration time and again when a node is wrapped. The kernel does not re-check it: provision() mints each node token independently against POST /v1/capabilities with that node's own scope, so the kernel sees a token, not a parent-child relation. Treat the ceiling as an SDK-side invariant and keep scope construction in one place.

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
    redaction_policy: RedactionPolicy | None = None,  # defaults to RedactionPolicy.chio_default()
)

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.


Argument Redaction

Before the dispatch is evaluated, the wrapper runs chio_adapter_base.redact.redact_args over the node's state-derived parameters. The default RedactionPolicy.chio_default() stubs protected body fields, for example chio_file_write.content and chio_file_edit.patch, to {"omitted": true, "byte_count": N}. The sidecar and any downstream receipt consumer see only the byte count, never the raw value; the wrapped node body still receives the original, unredacted arguments. Pass redaction_policy= on either chio_node or chio_approval_node to widen or narrow what is stubbed; a custom policy replaces the default.

Redaction changes what parameter_hash can distinguish

Because protected bytes never reach the sidecar, the parameter_hash for chio_file_write / chio_file_edit is uniform across calls. For per-call forensics, combine byte_count with the path and the receipt id. Capability constraints over raw payload bytes cannot be enforced at the sidecar in the redacted shape; enforce those client-side before invoking the node.

Approval Nodes

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

  1. Evaluate the dispatch through the sidecar. A deny raises immediately.
  2. If the kernel returns pending_approval, or if a local approval_policy requires approval, build an ApprovalRequestPayload and pause the graph via interrupt().
  3. 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).

The payload carries no arguments at all. Its fields are approval_id, policy_id, subject_id, capability_id, tool_server, tool_name, action, parameter_hash, summary, expires_at and created_at, plus an optional callback_hint and triggered_by. A reviewer sees the hash, not the bytes, and redaction is what the hash is computed over, so a change of redaction_policy moves the hash rather than revealing content. If an approver must inspect what they are approving, render it out of band and reference it from summary or callback_hint.


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

Graph Patterns

Supervisor / Worker Delegation

When a supervisor dispatches to a worker, the capability narrows in place. Configure a per-node scope on ChioGraphConfig for each worker, and override the capability id at the config configurable hook when a dispatch needs to run under a different one. provision() mints a token per node scope, and enforce_subgraph_ceiling validates the subset relation at wrap time. Capability resolution runs runtime override first, then the node token, then the workflow token, so a configurable override outranks the node scope for that dispatch.

Checkpoint and Receipt Alignment

LangGraph checkpoints graph state at each node; Chio receipts record each dispatch. Correlate the two through ChioReceipt.id, which the sidecar returns on an allow and which ChioLangGraphError carries on a deny. Write that id into your own checkpoint metadata at the end of the node body and a thread id resolves to its receipts.

Scoped Toolkit Injection

ChioToolkit (from chio-langchain) is constructed with a capability id and discovers tools via get_tools(server_id=...). Construct one toolkit per node scope rather than sharing a single wide one across the graph, so that each node holds only the capability it should present. The listing itself is not the boundary: get_tools filters on server_id alone and returns every tool the sidecar reports for that server, stamping the toolkit capability id onto each one. Narrowing the capability changes what the kernel admits when a tool is called, not what the node can see.


Package Layout

chio-langgraph supplies the node and approval-node wrappers and ChioGraphConfig. It sits on two chio packages: chio-sdk-python for the HTTP client to the sidecar, and chio-adapter-base for argument redaction.

sdks/python/chio-langgraph/pyproject.toml25-30toml
dependencies = [
    "chio-sdk-python>=0.1.0",
    "chio-adapter-base>=0.2.0,<0.3",
    "langgraph>=0.2,<1",
    "pydantic>=2.5,<3",
]

chio-langchain, which supplies ChioTool and ChioToolkit, is a sibling rather than a layer underneath. It rests on the same two packages and neither integration imports the other.

sdks/python/chio-langchain/pyproject.toml26-30toml
dependencies = [
    "chio-sdk-python>=0.1.0",
    "chio-adapter-base>=0.2.0,<0.3",
    "langchain-core>=0.2,<1",
]

So teams that only need tool wrapping inside otherwise-unchanged LangChain agents can use chio-langchain on its own, and installing chio-langgraph does not pull it in.


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