LangChain Adapter

The LangChainAdapter connects Edictum to LangChain and LangGraph agents with three integration methods.

The LangChainAdapter connects Edictum to LangChain and LangGraph agents. Every tool call passes through the Edictum pipeline before and after execution -- rulesets are enforced, findings are produced, and audit events are emitted.

Getting Started

Install

pip install edictum[yaml,langchain]

This installs langchain-core >= 0.3. The yaml extra is needed to load YAML rulesets.

Create adapter

from edictum import Edictum, Principal
from edictum.adapters.langchain import LangChainAdapter
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent, ToolNode

# Load rulesets
guard = Edictum.from_yaml("rules.yaml")

# Create adapter
adapter = LangChainAdapter(
    guard=guard,
    session_id="session-01",
    principal=Principal(user_id="alice", role="analyst"),
)

Wire into agent

# Wrap tools with rule enforcement
tool_node = ToolNode(
    tools=[search_tool, file_reader, calculator],
    wrap_tool_call=adapter.as_tool_wrapper(),
)

# Build agent -- rulesets enforced on every tool call
agent = create_react_agent(model=ChatOpenAI(model="gpt-4o"), tools=tool_node)
result = agent.invoke({"messages": [("user", "Summarize the Q3 report")]})

Every tool call now passes through the pipeline. If a rule blocks the call, the agent receives a ToolMessage with "DENIED: <reason>" and can try a different approach. Postconditions scan output after execution.

Which Method to Use

Method	Context	Event loop handling	Recommendation
`as_tool_wrapper()`	Sync or mixed	Detects running loops, bridges via ThreadPoolExecutor	Use this by default
`as_async_tool_wrapper()`	Fully async (FastAPI, async LangGraph)	Native `await` -- no bridging	Use in async code
`as_middleware()`	Sync only	`run_until_complete()` -- breaks if loop is running	Legacy; avoid in new code

`as_tool_wrapper()` (Recommended)

Returns a callable for ToolNode(wrap_tool_call=...). Handles nested event loops gracefully -- if an asyncio loop is already running (Jupyter, FastAPI), it bridges via ThreadPoolExecutor instead of crashing.

wrapper = adapter.as_tool_wrapper()
tool_node = ToolNode(tools=tools, wrap_tool_call=wrapper)
agent = create_react_agent(model, tools=tool_node)

`as_async_tool_wrapper()` (Async Contexts)

Returns an async callable. No sync-to-async bridging -- the cleanest option for fully async code.

async_wrapper = adapter.as_async_tool_wrapper()
tool_node = ToolNode(tools=tools, wrap_tool_call=async_wrapper)
agent = create_react_agent(model, tools=tool_node)

Use this when your application is already async (FastAPI endpoints, async LangGraph workflows) and you want to avoid the thread pool overhead.

`as_middleware()` (Alternative)

Returns a @wrap_tool_call decorated function for tool_call_middleware. This uses asyncio.get_event_loop().run_until_complete() internally, which raises RuntimeError if an event loop is already running.

middleware = adapter.as_middleware()
agent = create_react_agent(model, tools=tools, tool_call_middleware=[middleware])

If you hit event loop errors with as_middleware(), switch to as_tool_wrapper() or as_async_tool_wrapper(). If you must use as_middleware() in an async context, apply nest_asyncio:

import nest_asyncio
nest_asyncio.apply()

Constructor Parameters

adapter = LangChainAdapter(
    guard=guard,
    session_id="session-01",
    principal=Principal(user_id="alice", role="analyst"),
    principal_resolver=lambda tool_name, tool_input: resolve(tool_name),
)

Parameter	Type	Description
`guard`	`Edictum`	An `Edictum` instance with loaded rulesets
`session_id`	`str \| None`	Session identifier for session rulesets. Auto-generated UUID if omitted
`principal`	`Principal \| None`	Static principal attached to every tool call
`principal_resolver`	`Callable \| None`	`(tool_name, tool_input) -> Principal` for dynamic resolution. Overrides static `principal` when set

Dynamic Principal Resolution

When different tools need different identity context -- for example, a tool that operates on behalf of the end user vs. an internal service tool:

def resolve_principal(tool_name: str, tool_input: dict) -> Principal:
    if tool_name in ("query_user_data", "update_profile"):
        return Principal(user_id=current_user.id, role="end_user")
    return Principal(service_id="internal-agent", role="service")

adapter = LangChainAdapter(
    guard=guard,
    principal_resolver=resolve_principal,
)

See mutable principal for the full pattern.

Mid-Session Principal Changes

Update the principal mid-session with set_principal(). Subsequent tool calls use the new principal.

adapter.set_principal(Principal(user_id="alice", role="admin"))

Postcondition Callbacks

All three methods accept on_postcondition_warn. The callback receives the tool result (already modified by redact or block if applicable) and a list of findings. Its return value replaces the tool result sent to the agent.

import re

def redact_pii(result, findings):
    text = str(result)
    for f in findings:
        if f.type == "pii_detected":
            text = re.sub(r"\b\d{3}-\d{2}-\d{4}\b", "[SSN REDACTED]", text)
            text = re.sub(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b", "[EMAIL REDACTED]", text)
    return text

wrapper = adapter.as_tool_wrapper(on_postcondition_warn=redact_pii)
tool_node = ToolNode(tools=tools, wrap_tool_call=wrapper)

The LangChain adapter is one of the adapters where the callback return value replaces the result. See adapter comparison for which adapters support result replacement vs. side-effect-only callbacks.

Lifecycle Callbacks

def log_block(envelope, reason, decision_name):
    logger.warning(f"Blocked {envelope.tool_name}: {reason} ({decision_name})")

def log_allow(envelope):
    logger.info(f"Allowed {envelope.tool_name}")

guard = Edictum.from_yaml(
    "rules.yaml",
    on_block=log_block,
    on_allow=log_allow,
)

These fire regardless of which integration method you use.

Observe Mode

Deploy rulesets in observe mode to see what would be blocked without actually blocking any tool calls:

guard = Edictum.from_yaml("rules.yaml", mode="observe")
adapter = LangChainAdapter(guard=guard)
wrapper = adapter.as_tool_wrapper()

The wrapper allows all tool calls through. CALL_WOULD_DENY audit events are emitted for calls that would have been blocked, so you can review enforcement behavior before enabling it. See observe mode for the full workflow.

Server Connection

Connect to Edictum Control Plane for centralized rule management, hot-reload, and production approvals:

guard = await Edictum.from_server(
    url="https://api.edictum.ai",
    api_key="edk_...",
    agent_id="docs-agent",
    bundle_name="production-rules",
)

adapter = LangChainAdapter(guard=guard)
wrapper = adapter.as_tool_wrapper()

The agent receives ruleset updates via SSE and applies them without restart. See connecting agents for setup details.

Full Working Example

A complete example with rulesets, principal, session limits, PII redaction, and lifecycle logging:

from edictum import Edictum, Principal
from edictum.adapters.langchain import LangChainAdapter
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent, ToolNode
import re

# Load rulesets with lifecycle callbacks
guard = Edictum.from_yaml(
    "rules.yaml",
    environment="production",
    on_block=lambda env, reason, name: print(f"DENIED: {env.tool_name} -- {reason}"),
)

# Create adapter with identity and session tracking
adapter = LangChainAdapter(
    guard=guard,
    session_id="research-session-01",
    principal=Principal(user_id="researcher", role="analyst"),
)

# PII redaction callback
def redact_pii(result, findings):
    text = str(result)
    text = re.sub(r"\b\d{3}-\d{2}-\d{4}\b", "***-**-****", text)
    return text

# Build the agent
tools = [search_tool, calculator_tool, file_reader_tool]
tool_node = ToolNode(
    tools=tools,
    wrap_tool_call=adapter.as_tool_wrapper(on_postcondition_warn=redact_pii),
)
agent = create_react_agent(model=ChatOpenAI(model="gpt-4o"), tools=tool_node)

# Run -- all rulesets enforced
result = agent.invoke({"messages": [("user", "Summarize the Q3 report")]})

Next Steps

Adapter comparison -- how LangChain compares to other adapters
Rule Types Overview -- preconditions, postconditions, session, sandbox
Postcondition callbacks -- the on_postcondition_warn API
Observe mode -- test rulesets before enforcing
Connecting to Control Plane -- centralized rule management

On this page