Agent Governance Foundation

LangGraph integration

LangGraph has two distinct places policy enforcement can go — a tool-calling node and an arbitrary graph node — and they need different patterns. Both are covered here.

Install

pip install agf-sdk[langgraph]

Requires Python ≥ 3.10.

Set your API key

export AGF_API_KEY=agfk_your_key_here

Generate a key from Settings → API Keys.

Pattern 1 — Tool-calling nodes

If your tools are wired through ToolNode or create_react_agent, you don't need anything LangGraph-specific — AGFGuardedTool already is a BaseTool, so it works unmodified.

import os
from langgraph.prebuilt import ToolNode
from agf import AGFClient
from agf.langchain import AGFGuardedTool
from langchain_community.tools import ShellTool

client = AGFClient(api_key=os.environ["AGF_API_KEY"])

guarded_shell = AGFGuardedTool(
    tool=ShellTool(),
    client=client,
    agent_id="did:agf:agt_01abc",
    action_type="exec:shell",
    resource="local-shell",
)

# AGFGuardedTool is already a BaseTool — ToolNode and create_react_agent
# accept it exactly like any other LangChain tool. No LangGraph-specific
# adapter needed for this case.
tool_node = ToolNode([guarded_shell])

Pattern 2 — Graph nodes

A plain StateGraph.add_node() function isn't a BaseTool, so it can't route through the LangChain adapter. Use guard_node instead — same decorator shape as every other AGF SDK guard.

import os
from agf.client import AGFClient
from agf.keys import build_self_signed_chain
from agf.langgraph import guard_node
from langgraph.graph import StateGraph

client = AGFClient(api_key=os.environ["AGF_API_KEY"])
AGF_AGENT_ID = "did:agf:agt_01abc"


def _refund_chain_provider(state):
    # Built fresh per call — a chain built once expires in 5 minutes.
    return build_self_signed_chain(AGF_PRIVATE_KEY_PEM, AGF_AGENT_ID, "node:issue_refund", "agf")


@guard_node(
    client,
    agent_id=AGF_AGENT_ID,
    action_type="node:issue_refund",
    chain_provider=_refund_chain_provider,
    validate_execution=True,
    report_outcome=True,
)
def issue_refund(state: RefundState) -> dict:
    result = process_refund(state["order_id"])
    return {"result": result}


graph = StateGraph(RefundState)
graph.add_node("issue_refund", issue_refund)
Note: chain_provideris called with the node's own arguments — build the delegation chain fresh on every call, not once at import time. Self-signed chains expire in 5 minutes.

Handling REVIEW_REQUIRED

guard_nodedoesn't catch this itself — there's no LangGraph-native "requires approval" state to map it onto, so it propagates like any other exception from the node:

from agf.exceptions import AGFDeniedError, AGFReviewRequiredError

try:
    graph.compile().invoke({"order_id": "ord_123"})
except AGFDeniedError as e:
    print(f"Denied: {e}")
except AGFReviewRequiredError as e:
    # A common first-call outcome for a freshly-enrolled agent — not an
    # error to work around, a real approval workflow to handle.
    print(f"Approval required: {e.approval_request_id}")

If you have a single dispatcher node

Some agents route every tool call through one generic node that looks up the tool by name at runtime. guard_node's action type is fixed when you apply the decorator, so it can only gate "may this agent call some tool," not which one. For real per-tool authorization here, call the client directly inside the dispatcher instead:

# A single node that dispatches to differently-named tools at runtime
# (a common pattern for hand-rolled tool-calling loops) can't be gated by
# one guard_node call, since its action_type/resource are fixed when the
# decorator is applied. Call decide() directly inside the dispatcher instead:
async def _execute_tool(tool_call: dict):
    decision = await client.decide(
        f"tool:{tool_call['name']}",
        tool_call["name"],
        chain=build_self_signed_chain(AGF_PRIVATE_KEY_PEM, AGF_AGENT_ID, f"tool:{tool_call['name']}", "agf"),
    )
    return await tools_by_name[tool_call["name"]].ainvoke(tool_call["args"])

Related