LangGraph Integration
AgentTrustNode for LangGraph state machine governance.
Overview
The LangGraph adapter provides AgentTrustNode — a governance node you add to your LangGraph state graph. It validates agent output at a defined point in the graph execution.
Tier: Team+
Why It Matters
LangGraph agents need in-graph governance checkpoints, not just post-hoc validation. AgentTrustNode integrates natively with graph state and routing.
Prerequisites
pip install "agentrust-py[langgraph]"
# Team+ API key requiredStep-by-Step Guide
The tier gate raises `RuntimeError`, and only sometimes
This adapter calls _check_adapter_tier() at construction, which raises a plain
RuntimeError prefixed [AgentTrust] — except TierGateError will not catch it.
The check also short-circuits: it returns without checking anything when
AGENTRUST_ENV is development, dev, demo, or test, when no api_key argument
was passed to the constructor, or when the key does not start with at_. Constructing
the adapter with no key therefore never gates — it will run and simply be rejected by the
gateway later.
1. Import adapter
from agentrust_sdk.adapters.langgraph import AgentTrustNodeAgentTrustNode does not read AGENTRUST_GATEWAY_URL
base_url defaults to http://localhost:8000 in the constructor, so the node ignores
AGENTRUST_GATEWAY_URL — including the value embed_gateway() sets. Pass base_url
explicitly whenever the gateway is not on :8000.
2. Add node to graph
from langgraph.graph import StateGraph
governance = AgentTrustNode(
agent_id="langgraph-agent",
base_url="http://127.0.0.1:8765", # embedded gateway; omit for Edge on :8000
)
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("governance", governance)
graph.add_edge("agent", "governance")
graph.add_conditional_edges("governance", route_on_decision)3. Route on decision
The node merges the full ValidateResponse into state under the key
_agentrust_result — not a bare decision key.
from langgraph.graph import END
def route_on_decision(state):
result = state.get("_agentrust_result") or {}
outcome = (result.get("decision") or {}).get("outcome")
if outcome in ("block", "escalate"):
return "error_handler"
return END4. Handle blocked outputs
With the default block_on_block=True, a block outcome raises BlockedError from
inside the node, so the graph run aborts before your conditional edge is consulted. Set
block_on_block=False if you would rather route on the outcome yourself:
governance = AgentTrustNode(
agent_id="langgraph-agent",
base_url="http://127.0.0.1:8765",
block_on_block=False, # never raise — always route via state
output_key="output", # which state key holds the dict to validate
input_key="input",
user_key="user",
)Examples
See examples/langgraph_agent.py for a complete runnable graph.
from agentrust_sdk.adapters.langgraph import AgentTrustNode
from agentrust_sdk import embed_gateway
gw = embed_gateway()
governance_node = AgentTrustNode(agent_id="research-graph", base_url=gw.url)Best Practices
- Place governance node after LLM/tool nodes, before output delivery
- Use conditional edges to route
blockandescalateoutcomes - For multi-node trust chains, call
AgentTrustClient.validate(parent_envelope_id=…)directly —AgentTrustNodehas noparent_envelope_idparameter and never forwards one - Set meaningful
agent_idper graph deployment
Common Mistakes
- Catching
TierGateErroraround construction — the adapter raises plainRuntimeError - Placing governance node before agent produces output
- Using
auto_instrument()instead ofAgentTrustNodefor graph-level control
Troubleshooting
| Issue | Fix |
|---|---|
RuntimeError: [AgentTrust] LangGraph adapter requires Team tier… | Upgrade to Team tier (this is not a TierGateError) |
| Node not in graph path | Verify edges connect through the governance node |
| Validations hitting the wrong gateway | Pass base_url= — the node ignores AGENTRUST_GATEWAY_URL |
| State missing output | Ensure upstream node writes output to state |