Add a pre-action check to a LangGraph agent

What the check does

The check returns permit, deny, or escalate with the governing clause and a signed receipt. It is advisory, and the graph route is what refuses execution. The hosted check evaluates the request in memory and keeps no request content.

Code

This is the audited example exactly as published under public/examples/frameworks/.

"""LangGraph route that checks HANRIA before its execution node."""

from __future__ import annotations

import json
from typing import Any, TypedDict
from urllib.error import URLError
from urllib.request import Request, urlopen

from langgraph.graph import END, START, StateGraph

CHECK_URL = "https://check.hanria.ai/v1/check"
MANDATE = {
    "schema_version": "0.2-draft",
    "mandate_id": "read-text-files",
    "purpose": "Permit reads of text files under the example directory.",
    "default": "deny",
    "clauses": [
        {
            "id": "permit-example-read",
            "effect": "permit",
            "match": {
                "kind": ["file"],
                "verb": ["read"],
                "target_prefix": ["/tmp/example/"],
            },
        }
    ],
}


class ToolState(TypedDict, total=False):
    action: dict[str, Any]
    check_url: str
    decision: dict[str, Any]
    executed: bool
    result: str


def check_action(action: dict[str, Any], check_url: str = CHECK_URL) -> dict[str, Any]:
    """Return a HANRIA decision, mapping every transport or response failure to error."""
    body = json.dumps({"mandate": MANDATE, "action": action}).encode()
    request = Request(check_url, data=body, headers={"content-type": "application/json"})
    try:
        with urlopen(request, timeout=10) as response:
            decision = json.load(response)
    except (OSError, URLError, ValueError, json.JSONDecodeError) as error:
        return {"outcome": "error", "reason": f"HANRIA check failed: {error}"}
    if not isinstance(decision, dict) or decision.get("outcome") not in {
        "permit", "deny", "escalate", "error"
    }:
        return {"outcome": "error", "reason": "HANRIA returned an invalid decision"}
    return decision


def check_node(state: ToolState) -> ToolState:
    return {"decision": check_action(state["action"], state.get("check_url", CHECK_URL))}


def route_decision(state: ToolState) -> str:
    return "execute" if state["decision"].get("outcome") == "permit" else "refuse"


def execute_node(state: ToolState) -> ToolState:
    target = state["action"]["operation"]["target"]
    return {"executed": True, "result": f"tool would run for {target}"}


def refuse_node(state: ToolState) -> ToolState:
    decision = state["decision"]
    return {
        "executed": False,
        "result": f"tool refused: {decision.get('outcome', 'error')}: "
        f"{decision.get('reason', 'no reason returned')}",
    }


def build_graph():
    graph = StateGraph(ToolState)
    graph.add_node("check", check_node)
    graph.add_node("execute", execute_node)
    graph.add_node("refuse", refuse_node)
    graph.add_edge(START, "check")
    graph.add_conditional_edges("check", route_decision, {"execute": "execute", "refuse": "refuse"})
    graph.add_edge("execute", END)
    graph.add_edge("refuse", END)
    return graph.compile()


def example_action() -> dict[str, Any]:
    return {
        "schema_version": "0.1-draft",
        "requested_by": {"agent": "langgraph-example"},
        "operation": {"kind": "file", "verb": "read", "target": "/tmp/example/note.txt"},
        "justification": "Read the example note.",
    }


if __name__ == "__main__":
    graph = build_graph()
    print("Constructed HANRIA graph without invoking it.")

Run it

Save langgraph.py, then run this command from the folder where you saved the file.

uv run --with langgraph==1.2.12 python langgraph.py

The command constructs the graph and does not invoke it. In your application, call build_graph().invoke({"action": example_action()}). The action dictionary goes under the "action" key in the graph state. The check node must stay before the execution node.

Handle every outcome

• permit: the route selects the execution node because a mandate clause permits the proposed action.
• deny: the route selects the refusal node. Do not attempt the action through another path.
• escalate: the graph route selects the refusal node. Ask the operator for a decision outside this graph route.
• error: the route selects the refusal node. Treat it as deny, report the reason, and do not retry blindly.

A transport failure, malformed response, or unknown outcome becomes error, so it cannot select the execution node.

References

Try the check · Verify signed receipts · OpenAPI document · Agent-facing summary