LangGraph / LangChain

Vectimus governs tool calls in LangChain agents and LangGraph workflows. The middleware wraps ToolNode to evaluate every tool call against Cedar policies before execution. A separate MCP interceptor governs external MCP server connections.

Installation

pip install vectimus[langgraph]

Or with uv:

uv add vectimus[langgraph]

Middleware (recommended)

Use VectimusMiddleware with LangGraph’s ToolNode(awrap_tool_call=...) interface to govern every tool call:

from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.tool_node import ToolNode
from vectimus.integrations.langgraph import VectimusMiddleware

# 1. Create the middleware
middleware = VectimusMiddleware(
    observe_mode=False,       # set True to log without blocking
    principal="my-agent",     # identity for the audit trail
)

# 2. Define tools
@tool
def bash(command: str) -> str:
    """Run a shell command and return its output."""
    import subprocess
    return subprocess.check_output(command, shell=True, text=True)

@tool
def file_read(file_path: str) -> str:
    """Read a file and return its contents."""
    return open(file_path).read()

# 3. Wrap tools in a ToolNode with Vectimus
tools = [bash, file_read]
tool_node = ToolNode(tools, awrap_tool_call=middleware)

# 4. Create the agent
model = ChatAnthropic(model="claude-sonnet-4-20250514")
agent = create_react_agent(model=model, tools=tool_node)

When a tool call is denied, the middleware raises a ToolException with the policy violation details. The agent sees the error and can try a different approach.

MCP interceptor

For governing MCP tool calls through MultiServerMCPClient, use create_interceptor:

from langchain_mcp_adapters.client import MultiServerMCPClient
from vectimus.integrations.langgraph import create_interceptor

interceptor = create_interceptor(
    observe_mode=False,
)

client = MultiServerMCPClient(
    {
        "my-server": {
            "command": "npx",
            "args": ["-y", "@my-org/mcp-server"],
        }
    },
    tool_interceptors=[interceptor],
)

The interceptor evaluates each MCP tool call before it reaches the server. Denied calls never leave your process.

How it works

Vectimus maps LangChain tool names to the same canonical action types used by the CLI integrations:

Shell commands are further classified into git_operation, infrastructure and package_operation based on the command content. Shell commands that perform file I/O (redirects, tee, inline scripts like python3 -c "open('f','w').write('x')") are reclassified to file_read or file_write with the file path extracted for Cedar policy matching.

Middleware options

Observe mode

Both the middleware and interceptor support observe mode:

middleware = VectimusMiddleware(observe_mode=True)

Or via environment variable:

export VECTIMUS_OBSERVE=true

Custom policies

Load custom Cedar policies from a directory:

middleware = VectimusMiddleware(
    policy_dir="./my-policies",
)

See Writing policies for the full guide on creating custom Cedar rules.

Audit trail

Every tool call is logged to ~/.vectimus/logs/audit-YYYY-MM-DD.jsonl regardless of the decision. Each evaluation also produces an Ed25519-signed governance receipt stored in .vectimus/receipts/. Receipts provide tamper-evident proof of each decision for audit compliance. The daemon cleans up receipts older than 7 days by default (configurable via [receipts] retention_days in config). Use vectimus verify to validate receipts and vectimus receipts prune for manual cleanup.