Skip to main content

LangGraph Integration

Enforce Control Zero policies on LangGraph state machines, tool nodes, and conditional edges.

Overview​

LangGraph models agent workflows as directed graphs where nodes execute tools, call LLMs, or make decisions. Control Zero integrates at the tool node level: every wrapped or explicitly guarded tool invocation is checked against your policies before execution, and a deny stops the tool logic from running. LangGraph's state machine can then route the flow to a human-in-the-loop node or an error handler.

Installation​

pip install controlzero langchain langgraph langchain-openai

Setup​

Tool-Level Enforcement​

The most direct integration wraps each LangGraph tool with a guard() call:

from controlzero import Client
from langchain_core.tools import tool

cz = Client(api_key="cz_live_your_api_key_here")


@tool
def delete_user(user_id: str) -> str:
    """Delete a user from the database."""
    cz.guard("delete_user", args={"user_id": user_id}, raise_on_deny=True)
    # ... your delete logic ...
    return f"Deleted user {user_id}"


@tool
def search_users(query: str) -> str:
    """Search for users in the database."""
    cz.guard("search_users", args={"query": query}, raise_on_deny=True)
    return f"Found users matching: {query}"

Callback Handler Approach​

Reuse the LangChain callback handler for audit logging:

from controlzero.integrations.langchain import ControlZeroCallbackHandler

handler = ControlZeroCallbackHandler(cz)

Pass the handler to your LLM to record activity; enforce with guarded nodes or tools, or explicit cz.guard() calls.

What Gets Enforced​

LangGraph EventPolicy ActionPolicy Resource
Tool node executiontool.calltool/{tool_name}
LLM call within a nodellm.generatemodel/{model_name}
Conditional edge (custom)agent.routeroute/{edge_name}

Full Example: Agent with Human-in-the-Loop Fallback​

This example builds a LangGraph agent that routes to a human approval node when a tool call is denied by policy:

from typing import Annotated, TypedDict

from controlzero import Client
import controlzero
from controlzero.integrations.langchain import ControlZeroCallbackHandler
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

# --- State definition ---
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    needs_approval: bool


# --- Control Zero setup ---
cz = Client(api_key="cz_live_your_api_key_here")
handler = ControlZeroCallbackHandler(cz)


# --- Tools ---
@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    return f"Web results for: {query}"


@tool
def write_to_database(record: str) -> str:
    """Write a record to the production database."""
    return f"Wrote record: {record}"


@tool
def delete_records(table: str) -> str:
    """Delete all records from a database table."""
    return f"Deleted all records from {table}"


tools = [search_web, write_to_database, delete_records]


# --- Nodes ---
llm = ChatOpenAI(model="gpt-5.4", callbacks=[handler]).bind_tools(tools)


def agent_node(state: AgentState) -> AgentState:
    """Call the LLM to decide the next action."""
    response = llm.invoke(state["messages"])
    return {"messages": [response], "needs_approval": False}


def tool_node_with_policy(state: AgentState) -> AgentState:
    """Execute tool calls with policy enforcement."""
    last_message = state["messages"][-1]

    results = []
    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]

        try:
            cz.guard(tool_name, args=tool_args, raise_on_deny=True)
            # Policy allowed -- execute the tool
            tool_fn = {t.name: t for t in tools}[tool_name]
            result = tool_fn.invoke(tool_args)
            results.append(
                ToolMessage(content=str(result), tool_call_id=tool_call["id"])
            )

        except controlzero.PolicyDeniedError as e:
            # Policy denied -- route to human approval
            results.append(
                ToolMessage(
                    content=f"POLICY DENIED: {e.decision.reason}. Requires human approval.",
                    tool_call_id=tool_call["id"],
                )
            )
            return {"messages": results, "needs_approval": True}

    return {"messages": results, "needs_approval": False}


def human_approval_node(state: AgentState) -> AgentState:
    """Placeholder for human-in-the-loop approval."""
    return {
        "messages": [
            HumanMessage(content="Action requires human approval. Pausing workflow.")
        ],
        "needs_approval": False,
    }


# --- Routing ---
def should_continue(state: AgentState) -> str:
    if state.get("needs_approval"):
        return "human_approval"

    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return "end"


# --- Build the graph ---
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node_with_policy)
graph.add_node("human_approval", human_approval_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    "human_approval": "human_approval",
    "end": END,
})
graph.add_edge("tools", "agent")
graph.add_edge("human_approval", END)

app = graph.compile()

# --- Run ---
result = app.invoke({
    "messages": [HumanMessage(content="Search for Q4 revenue data")],
    "needs_approval": False,
})

Example Policy​

Allow search tools but require approval for database writes and block deletes:

{
  "name": "langgraph-agent-policy",
  "rules": [
    {
      "effect": "allow",
      "action": "llm:generate",
      "resource": "model/gpt-5.4"
    },
    {
      "effect": "allow",
      "action": "tool:call",
      "resource": "tool/search_web"
    },
    {
      "effect": "deny",
      "action": "tool:call",
      "resource": "tool/write_to_database"
    },
    {
      "effect": "deny",
      "action": "tool:call",
      "resource": "tool/delete_records"
    }
  ]
}

What happens at runtime:

  • search_web: ALLOWED.
  • write_to_database: DENIED. The graph routes to the human_approval node.
  • delete_records: DENIED. The graph routes to the human_approval node.

Using GovernedStateGraph​

GovernedStateGraph wraps LangGraph's StateGraph to automatically intercept tool calls at the graph level. All tools registered on the graph are governed without per-node guard() calls.

from controlzero import Client
from controlzero.integrations.langchain import GovernedStateGraph
from langgraph.graph import StateGraph, END
from typing import TypedDict

cz = Client(api_key="cz_live_your_key_here")

class AgentState(TypedDict):
    messages: list
    next_action: str

# GovernedStateGraph wraps StateGraph -- same API, governance built in
graph_builder = GovernedStateGraph(AgentState, client=cz)

graph_builder.add_node("agent", run_agent)
graph_builder.add_node("tools", run_tools)
graph_builder.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
graph_builder.add_edge("tools", "agent")
graph_builder.set_entry_point("agent")

graph = graph_builder.compile()

Use this when you want governance across all nodes without adding cz.guard() to each one. For per-node policy control, use direct cz.guard() calls as shown above.

LangGraph-Specific Patterns​

Subgraph Enforcement​

When using LangGraph subgraphs, pass a separate agent_id per subgraph for fine-grained policy control:

cz.guard(
    tool_name,
    method="tool.call",
    context={"agent_id": "research-subgraph"},
    raise_on_deny=True,
)

Checkpointed Workflows​

LangGraph supports checkpointing for long-running workflows. Control Zero audit logs provide a parallel record of every policy decision, even across checkpoint restores. Use the correlation ID to trace decisions:

cz.guard(
    tool_name,
    method="tool.call",
    context={"correlation_id": "workflow-abc-step-3"},
    raise_on_deny=True,
)

Streaming with Policy Enforcement​

In streaming mode, tool calls are enforced only when the tool or node is wrapped with an enforcing guard.

Local Policy File​

Generate a LangChain/LangGraph-specific policy template:

controlzero init --template langchain

This creates a controlzero.yaml that blocks destructive database operations, filesystem writes to sensitive paths, and outbound network to arbitrary hosts, while allowing read-heavy tools like search, fetch, and lookup.

Next Steps​