LangGraph adapter

Two integration patterns for LangGraph:

1. as_governed_node — wrap a single Python callable as a governed graph node (operates on the graph's state dict)
2. govern_graph_tools — wrap a dict of tool functions as governed LangChain BaseTool instances for use inside a ToolNode

Use pattern 1 when you have a custom node that calls one tool. Use pattern 2 when you're using LangGraph's prebuilt ToolNode to dispatch many tools.

Install

pip install kitelogik langgraph

langgraph is not a hard dependency — only imported when you call govern_graph_tools(). (Pattern 1 doesn't even import LangGraph; it just produces a function with the right signature.)

Setup

from kitelogik import OPAClient, PolicyGate, SessionContext

gate = PolicyGate(opa_client=OPAClient())
context = SessionContext(
    session_id="sess_001",
    user_role="support_agent",
    session_scopes=["read_customer", "approve_refund_under_100"],
)

Pattern 1 — as_governed_node

Returns an async function with signature (state: dict) -> dict. Reads state["args"], runs the governance pipeline, executes the underlying function on allow, and writes either state["result"] (allow) or state["result"] = "[BLOCKED] …" plus state["blocked"] = True (deny).

from kitelogik.adapters.langgraph import as_governed_node
from langgraph.graph import StateGraph, END

async def search_docs(query: str, limit: int = 10) -> str:
    return f"Found {limit} results for {query}"

governed_search = as_governed_node(
    name="search_docs",
    fn=search_docs,
    gate=gate,
    context=context,
)

graph = StateGraph(dict)
graph.add_node("search", governed_search)
graph.add_edge("search", END)
graph.set_entry_point("search")
app = graph.compile()

result = await app.ainvoke({"args": {"query": "refund policy", "limit": 5}})
# result["result"]   = "Found 5 results for refund policy"   (on allow)
# result["blocked"]  = True                                  (on deny)

Pattern 2 — govern_graph_tools (with ToolNode)

Wraps a dict of tool functions as governed LangChain BaseTool instances. Internally this delegates to the LangChain adapter's as_governed_tool, so you get the same Pydantic args_schema inference and the same [BLOCKED] … denial format.

from kitelogik.adapters.langgraph import govern_graph_tools
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph

def search(query: str) -> str:
    return f"Results for: {query}"

def lookup_customer(customer_id: str) -> str:
    return f"Customer {customer_id}: Acme Corp, plan=enterprise"

tools = govern_graph_tools(
    {"search": search, "lookup_customer": lookup_customer},
    gate=gate,
    context=context,
)

# Drop into a standard LangGraph agent setup:
graph = StateGraph(...)
graph.add_node("tools", ToolNode(tools=tools))
# ... rest of the graph as normal

Every tool dispatched by ToolNode runs through the policy gate before the underlying function executes.

What happens on a deny

Pattern	Denial shape
as_governed_node	Sets state["result"] = "[BLOCKED] <error>" and state["blocked"] = True. The graph continues to the next node — your edges can branch on state["blocked"] to route differently.
govern_graph_tools	Returns the string "[BLOCKED] <error>" from the tool. ToolNode packages this as a ToolMessage and feeds it back to the agent loop, exactly like LangChain.

Routing on deny in as_governed_node

A common pattern — branch the graph on a denial:

def route_after_search(state: dict) -> str:
    if state.get("blocked"):
        return "ask_human"
    return "summarise"

graph.add_conditional_edges("search", route_after_search, {
    "ask_human": "human_review_node",
    "summarise": "summary_node",
})

Function signatures

as_governed_node(
    name: str,
    fn: Callable,
    gate: PolicyGate,
    context: SessionContext,
    action: str | None = None,
    sanitize: bool = True,
) -> Callable                  # async (state: dict) -> dict

govern_graph_tools(
    tools: dict[str, Callable],   # {tool_name: fn}
    gate: PolicyGate,
    context: SessionContext,
    sanitize: bool = True,
) -> list[BaseTool]               # for ToolNode(tools=…)

Source

kitelogik/adapters/langgraph.py on GitHub.