LangGraph

A node you wrap with chio_node becomes a tool call that the kernel admits before the node body runs. LangGraph keeps the graph, its state, and its checkpoints. Chio holds a capability for each node and signs a receipt for every dispatch, refusals included.

The adapter sends each dispatch to a Chio sidecar beside the graph process. A second route, the process-backed tool node, runs model-chosen tools inside a durable Chio process and resumes them with their original outcome after a worker restart.

The sidecar treats a node dispatch as a call against a virtual tool server named langgraph. Scope checks, guards, approvals, and receipt signing therefore apply to a node exactly as they apply to any other tool. The kernel refuses a dispatch that claims write access the supervisor never granted, a replayed dispatch carrying a stale capability, and an approval interrupt that would bypass policy.

1. Install the adapter beside your graph

chio-langgraph requires langgraph>=0.6,<2. It pulls in the Chio Python SDK, which provides the sidecar client, and chio-adapter-base, which supplies argument redaction. Run the Chio sidecar next to the process that runs the graph.

You should see LangGraph adapter ready. The kernel check starts when you wrap a node.

2. Give the graph a ceiling and each node a scope

Build a ChioGraphConfig with the sidecar client, the workflow ceiling, and one scope per node. provision() mints a capability token for the workflow and one for each node; call it before the graph runs.

import asyncio

from chio import ChioClient, ChioScope, Operation, ToolGrant
from chio_langgraph import ChioGraphConfig

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.
asyncio.run(config.provision())

Every node scope must be a subset of the workflow ceiling. ChioGraphConfig checks the relation when a scope is registered, and chio_node checks it again at wrap time through enforce_subgraph_ceiling. A scope outside the ceiling raises ChioLangGraphConfigError while the graph is built, before its first run.

The SDK owns that subset check. provision() mints each node token separately against POST /v1/capabilities with the node's own scope, so the kernel sees each token on its own. Keep scope construction in one place.

3. Wrap each node that reaches a tool

chio_node calls evaluate_tool_call with tool_server="langgraph" and tool_name set to the node name, under the node's capability. Only an allow runs the wrapped body.

from typing import TypedDict

from chio_langgraph import chio_node
from langgraph.graph import END, START, StateGraph


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


def researcher(state: AgentState) -> dict:
    # ...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()

The wrapper keeps sync and async nodes and both LangGraph arities, (state) and (state, config). When the runtime config carries configurable["chio_capability_id"], that id outranks the node token for that dispatch. A supervisor uses this to hand a child subgraph a narrower capability through LangGraph's own config propagation.

4. Pause a node until a person approves it

chio_approval_node evaluates the dispatch first, and a deny raises at once. A pending_approval verdict becomes LangGraph's interrupt(), carrying an ApprovalRequestPayload.

from chio_langgraph import chio_approval_node


async def publish_price(state: AgentState) -> dict:
    # ...the side effect here...
    return {"published": True}


graph.add_node(
    "publish_price",
    chio_approval_node(
        publish_price,
        scope=scope_for("publish_price"),
        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="Publish sku-301 at 80 percent off",
    ),
)

The approver's decision comes back through Command(resume=...). An approved resolution runs the body, and any other outcome raises ChioLangGraphError. Keep approver lists, escalation, and timeouts in the kernel's approval policy; the local approval_policy callable only forces an extra pause, for example in a debug build.

The payload carries the parameter_hash of the redacted arguments, never the arguments. When an approver must read the content, render it out of band and reference it from summary or callback_hint.

5. Run the graph and keep each receipt with its checkpoint

A deny, or a node whose token was never provisioned, raises ChioLangGraphError from the wrapper before the body runs.

import logging

from chio_langgraph import ChioLangGraphError

logger = logging.getLogger(__name__)

try:
    app.invoke({"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

You should see the node name, the guard that refused, the reason, and the receipt id in the log line; the error also names the tool server. On an allow, the sidecar returns the ChioReceipt. Write ChioReceipt.id into your checkpoint metadata at the end of the node body, and a thread id resolves to its receipts.

A process-backed tool node resumes the original tool call

ChioProcessToolNode runs the tools a model chose as native operations of a Chio process. Each call is bound to the graph thread, the assistant-message ID, and the model-generated tool-call ID, so a restarted worker resolves the same operation and receives its original outcome.

from chio_process import ProcessClient
from chio_langgraph import ChioProcessToolNode, ProcessTool

client = ProcessClient(socket_path, credential)
tool_node = ChioProcessToolNode(
    client,
    [
        ProcessTool(
            "publish_report",
            "reports",
            "publish",
            "Publish a report.",
            {
                "type": "object",
                "properties": {"report": {"type": "string"}},
                "required": ["report"],
            },
        ),
    ],
    namespace="research-workflow-v1",
)
graph.add_node("tools", tool_node)

Provision the native host, the process capability, and the tool server first. The trusted host passes a private socket path and credential to the graph worker; keep both out of graph state and tracing metadata. Persist the complete assistant tool-call message in LangGraph's checkpoint before this node runs, and restart with the same thread and message IDs.

A denial, an incomplete result, or a transport error stops the graph. A fresh model decision would be a new operation, so recovery resumes the saved message instead. The node mediates tools; model calls and direct Python effects stay in the graph process.

Read more in Calls and recovery.

The example graph publishes once across a worker exit

Download langgraph.zip and extract it. Use Linux, Chio, Python 3.11+, and uv, and run commands from langgraph/. The manifest declares the graph, checkpoint, and Chio dependencies, and the default planner needs no provider account.

{
  "schema": "chio.process.run.v2",
  "max_parallel": 1,
  "workers": [
    {
      "process": "reporter",
      "command": [
        ".venv/bin/python",
        "graph-worker.py"
      ],
      "cwd": ".",
      "input": {
        "threadId": "report-42",
        "prompt": "Publish a short report",
        "receipts": ".chio/receipts.ndjson",
        "graphDb": ".chio/graph.db"
      },
      "max_attempts": 2,
      "timeout_seconds": 120
    }
  ]
}

uv sync
chio process init --config host.json --state .chio --local-tools
chio process run --state .chio --plan process.json

The normal plan completes in one attempt. The recovery plan's first attempt exits after the tool runs and before LangGraph receives the tool message. The replacement resumes the saved assistant message and the original publication.

chio process init --config host-recovery.json --state .chio-recovery --local-tools
chio process run --state .chio-recovery --plan process-recovery.json
chio receipt verify --input .chio-recovery/receipts.ndjson \
  --trusted-kernel-pubkey .chio-recovery/kernel.pub

You should see two reporter attempts, one row in .chio-recovery/reports.db, and the original receipt verified against the kernel key. ToolMessage.artifact["chio"] keeps the native response and receipt outside the model-visible tool content.

graph-worker.py separates planning, tool execution, checkpointing, and continuation. durability="sync" commits the assistant message before the tool node runs.

On restart, the same SQLite graph database and thread_id select the saved continuation, and invoke(None, config, durability="sync") resumes it; sending the first user request again starts a new input. The native operation key joins namespace, thread identity, assistant-message ID, and tool-call ID, so keep all four.

"""A complete durable graph with deterministic or configured-provider planning."""

import json
import os
import sqlite3
import sys
from pathlib import Path
from chio_process import ProcessClient
from chio_langgraph import ChioProcessToolNode, ProcessTool, ChioProcessToolError
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import START, END, MessagesState, StateGraph


def build_graph(client, settings, on_receipts_saved):
    tools = ChioProcessToolNode(
        client,
        [
            ProcessTool(
                "publish_report",
                "reports",
                "publish",
                "Publish one bounded report.",
                {
                    "type": "object",
                    "properties": {"report": {"type": "string"}},
                    "required": ["report"],
                    "additionalProperties": False,
                },
            )
        ],
        namespace="report-graph-v1",
    )
    model = None
    if settings.get("provider"):
        from langchain_openai import ChatOpenAI

        # OPENAI_API_KEY is operator environment, never graph state.
        model = ChatOpenAI(
            model=settings["model"], base_url=settings["endpoint"], max_retries=0
        ).bind_tools(tools.model_schemas())

    def plan(state):
        if model is not None:
            return {"messages": [model.invoke(state["messages"])]}
        if any(isinstance(m, ToolMessage) for m in state["messages"]):
            return {"messages": [AIMessage(content="Report published.", id="finished")]}
        return {
            "messages": [
                AIMessage(
                    content="",
                    id="report-plan",
                    tool_calls=[
                        {
                            "name": "publish_report",
                            "id": "publish-report",
                            "args": {"report": "Three inputs total ten."},
                        }
                    ],
                )
            ]
        }

    def execute(state, config):
        result = tools.invoke(state, config)
        for message in result["messages"]:
            with Path(settings["receipts"]).open("a", encoding="utf-8") as output:
                output.write(message.artifact["chio"]["receipt_json"] + "\n")
                output.flush()
                os.fsync(output.fileno())
        on_receipts_saved()
        return result

    graph = StateGraph(MessagesState)
    graph.add_node("model", plan)
    graph.add_node("tools", execute)
    graph.add_edge(START, "model")
    graph.add_conditional_edges(
        "model", lambda state: "tools" if state["messages"][-1].tool_calls else END
    )
    graph.add_edge("tools", "model")
    return graph


def run_worker(bootstrap, on_receipts_saved=lambda: None):
    if bootstrap["schema"] != "chio.process.worker-bootstrap.v1":
        raise RuntimeError("Unsupported bootstrap")
    connection, settings = bootstrap["connection"], bootstrap["input"]
    client = ProcessClient(connection["socket_path"], connection["credential"])
    graph = build_graph(client, settings, on_receipts_saved)
    config = {"configurable": {"thread_id": settings["threadId"]}}
    with sqlite3.connect(settings["graphDb"], check_same_thread=False) as database:
        app = graph.compile(checkpointer=SqliteSaver(database))
        retained = app.get_state(config)
        initial = (
            None
            if retained.values
            else {"messages": [HumanMessage(content=settings["prompt"], id="request")]}
        )
        try:
            result = app.invoke(initial, config, durability="sync")
        except ChioProcessToolError as error:
            if error.receipt_json:
                with Path(settings["receipts"]).open("a") as output:
                    output.write(error.receipt_json + "\n")
            raise
        print(json.dumps({"complete": True, "text": result["messages"][-1].content}))


if __name__ == "__main__":
    run_worker(json.load(sys.stdin))

A provider node that fails before its own response checkpoint leaves a model recovery question for the application, because the tool node journals tools and leaves inference to the graph. Model-call recovery journals the provider response for an AI SDK worker.

Two routes put the kernel inside a graph

Choose the node adapter to authorize existing Python nodes and pause them for approval. Choose the process tool node when a model-chosen tool must resume with its original outcome.

In LangGraph alone, graph structure defines control flow and human review is a UX step. Either route adds capability tokens that scope each node or tool, approval as a kernel verdict, subgraph scopes checked against the parent ceiling, and a signed ChioReceipt for every dispatch. For LangChain agents without a graph, chio-langchain wraps Chio tools as BaseTool instances.

Adapter reference

def 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,  # RedactionPolicy.chio_default()
): ...

# Outer graph: research + write
outer_config = ChioGraphConfig(
    chio_client=chio,
    workflow_scope=scope_for("search", "browse", "write"),
    node_scopes={"outer_plan": scope_for("search")},
)
asyncio.run(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",
)
asyncio.run(inner_config.provision())

dependencies = [
    "chio-sdk-python>=0.1.0",
    "chio-adapter-base>=0.2.0,<0.3",
    "langgraph>=0.2,<1",
    "pydantic>=2.5,<3",
]

dependencies = [
    "chio-sdk-python>=0.1.0",
    "chio-adapter-base>=0.2.0,<0.3",
    "langchain-core>=0.2,<1",
]

Next: carry one grant across a durable workflow