LangGraph

Pavri integrates with LangGraph by wrapping the graph's invoke() method with policy enforcement. This provides a pre-invocation governance check on every graph execution.

One-Line Integration

from pavri import PavriConfig, secure

graph = build_langgraph_agent()
graph = secure(graph, PavriConfig(agent_name="orders-graph", plugin="langgraph"))

The secure() call attaches the Pavri runtime to the graph, registers it with the backend, and wraps the execution entry point with enforcement.

Supported Versions

• LangGraph >=0.2.0
• LangChain Core >=0.3.0
• Python >=3.12

How It Works

Plugin Detection

The LangGraphAdapter matches when either:

• You pass plugin="langgraph" in the config
• The target object's module name contains "langgraph"

Hooks Attached

When the runtime has an active policy cache and telemetry emitter, secure() wraps the graph's invoke() method:

Hook	Purpose
invoke() wrapper	Evaluates policy before each graph invocation; raises PolicyDeniedError on denial

The wrapper calls enforce_policy(f"tool:{ClassName}", ctx) using the graph's class name as the policy subject, then delegates to the original invoke() if the policy allows.

Enforcement Flow

# Internal flow — you do not call this directly
# Before each graph.invoke(...) call:
enforce_policy(f"tool:{graph.__class__.__name__}", ctx)
# If allowed, the original invoke() runs normally

This provides a blanket pre-invocation policy check. Per-tool granularity within the graph requires deeper middleware integration that depends on the compiled graph structure; this baseline enforcement prevents unauthorized graph execution entirely.

Policy evaluation respects the configured fail_mode:

• fail-open (default): If policy evaluation fails, the invocation is allowed.
• fail-closed: If policy evaluation fails, the invocation is denied.

When a policy rule denies the action, PolicyDeniedError is raised:

from pavri.core.exceptions import PolicyDeniedError

try:
    result = graph.invoke({"messages": [("user", "hello")]})
except PolicyDeniedError as e:
    print(f"Blocked: {e.reason} (rule: {e.rule_id})")

Configuration

Key PavriConfig options for LangGraph:

from pavri import PavriConfig

config = PavriConfig(
    agent_name="orders-graph",
    plugin="langgraph",
    api_key="fab_...",
    backend_url="https://pavri.example.com",
    org_id="org-acme",
    session_id="session-123",
    team="platform",
    environment="production",
    fail_mode="open",
)

Field	Default	Description
fail_mode	"open"	Whether to allow or deny when policy evaluation fails
telemetry_mode	"full"	Controls telemetry detail level
environment	"development"	Deployment environment tag
policy_sync.poll_interval_seconds	60	How often to poll for policy updates

LangGraph Architecture Notes

LangGraph compiled graphs are immutable — middleware must be defined at graph build time, not added post-compilation. The current Pavri integration works around this by wrapping invoke() on the compiled graph, which provides enforcement without requiring access to the builder.

For deeper integration, the design spec recommends wrapping the StateGraph builder before compilation to inject wrap_tool_call and wrap_model_call middleware. This is planned for a future release.

Graceful Degradation

If the Pavri runtime cannot build an enforcement context (for example, the backend is unreachable and no cached policy is available), the plugin does not wrap invoke(). The graph runs without Pavri enforcement. Check the runtime status to verify:

from pavri.core import get_runtime

runtime = get_runtime(graph)
status = runtime.status()
print(status.connected, status.policy_revision)

Troubleshooting

Graph runs but no policy enforcement

• Verify the backend is reachable and api_key is set.
• Check runtime.status().policy_revision — if 0, the SDK has no cached policy.
• The plugin requires both a policy cache and telemetry emitter.
• Ensure the graph object has a callable invoke attribute. If the target is immutable and invoke cannot be replaced, enforcement will be silently skipped.

PolicyDeniedError raised unexpectedly

• Inspect the exception: e.decision contains the full PolicyDecision with the matched rule_id and reason.
• The policy subject is tool:{ClassName} — check that your policy rules match the graph's class name.
• Check whether fail_mode="closed" is set.

Plugin not detected

• Pass plugin="langgraph" explicitly in the config.
• Ensure the graph object's module includes "langgraph" in its path for auto-detection.

What's Deferred

The following capabilities are planned for Phase 2:

• Log forwarding from the LangGraph plugin
• Trace-link capture for LangSmith (TraceProvider.langsmith)
• Per-tool middleware enforcement via wrap_tool_call (requires builder-level integration)
• Per-model enforcement via wrap_model_call
• Checkpoint and streaming behavior monitoring
• Async callback handler for non-blocking telemetry