LangChain
A LangChain AgentMiddleware subclass. wrap_model_call evaluates the latest human message before the model and the AI message after it; wrap_tool_call evaluates tool arguments before execution and tool output after. Block raises, mask rewrites the message content.
Status: Generally available · Evaluates: Prompts, Responses, Tool calls · Vendor: LangChain
Prompt, response, tool input, tool output. Session and user ids come from agent state.
Note: This example uses the v1
WonderFenceClientinterface of the WonderFence SDK.
Setup
pip install langchain>=0.3.0 langgraph wonderfence-sdk
Configuration: ALICE_API_KEY, ALICE_APP_ID
Example
imports (langchain_hooks_simple.py)
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, AgentState
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from langchain_google_genai import ChatGoogleGenerativeAI
from langgraph.prebuilt.tool_node import ToolCallRequest
from langgraph.types import Command
from wonderfence_sdk.client import WonderFenceClient
from wonderfence_sdk.models import Actions, AnalysisContext
middleware: prompt check
class WonderFenceSafetyMiddleware(AgentMiddleware):
"""
Middleware that integrates WonderFence SDK for safety checks.
Hooks into LangChain agent lifecycle to:
- Check prompt safety before model calls (_before_model)
- Check response safety after model calls (_after_model)
- Check tool input safety before tool calls (_before_tool)
- Check tool output safety after tool calls (_after_tool)
"""
def __init__(self, wonderfence_client: WonderFenceClient):
"""
Initialize the WonderFence safety middleware.
Args:
wonderfence_client: WonderFenceClient client for safety evaluation
default_session_id: Default session identifier if not found in state
default_user_id: Default user identifier if not found in state
"""
super().__init__()
self.client = wonderfence_client
def _generate_wonderfence_context_from_state(self, state: AgentState) -> AnalysisContext:
"""
Generate an analysis context from the agent state.
"""
return AnalysisContext(
session_id=state.get("session_id", str(uuid.uuid4())),
user_id=state.get("user_id", "anonymous"),
)
def _before_model(self, state: AgentState) -> None:
"""Check prompt safety before model execution."""
logger.info("🔍 Checking prompt safety with WonderFence...")
messages = state.get("messages", [])
if not messages:
return
# Find the latest user message
user_messages = [msg for msg in messages if isinstance(msg, HumanMessage)]
if not user_messages:
return
latest_message = user_messages[-1]
if not hasattr(latest_message, "content"):
return
content = getattr(latest_message, "content", None)
if content is None:
return
content_str = str(content).strip()
if not content_str:
return
analysis_context = self._generate_wonderfence_context_from_state(state)
try:
logger.info(f" 🔍 Evaluating prompt safety: {content_str}")
evaluation = self.client.evaluate_prompt_sync(content_str, analysis_context)
logger.info(f" ✅ Prompt safety check: {evaluation.action.name}")
if evaluation.action == Actions.BLOCK:
raise Exception(f"Prompt blocked: {getattr(evaluation, 'explanation', 'Safety violation')}")
if evaluation.action == Actions.MASK:
if hasattr(latest_message, "content") and evaluation.action_text:
latest_message.content = evaluation.action_text # type: ignore[attr-defined]
logger.info(" 🎭 MASKED: Content modified for safety")
except Exception as e:
logger.error(f" ❌ Prompt safety check failed: {e}", exc_info=True, stack_info=True)
raise
middleware: tool input check + wrap_tool_call
def _before_tool(self, tool_name: str, tool_input: dict[str, Any], state: AgentState) -> dict[str, Any]:
"""Check tool input safety before tool execution."""
logger.info(f"🔍 Checking tool '{tool_name}' input safety with WonderFence...")
# Create description of tool usage for evaluation
tool_description = f"Tool '{tool_name}' called with: {tool_input}"
content_str = str(tool_description).strip()
if not content_str:
return tool_input
analysis_context = self._generate_wonderfence_context_from_state(state)
try:
evaluation = self.client.evaluate_prompt_sync(content_str, analysis_context)
logger.info(f" ✅ Tool input safety check: {evaluation.action.name}")
if evaluation.action == Actions.BLOCK:
raise Exception(f"Tool call blocked: {getattr(evaluation, 'explanation', 'Safety violation')}")
elif evaluation.action == Actions.MASK:
logger.info(" 🎭 MASKED: Tool input modified for safety")
return evaluation.action_text or tool_input
return tool_input
except Exception as e:
logger.error(f" ❌ Tool safety check failed: {e}", exc_info=True, stack_info=True)
raise e
def wrap_tool_call(
self,
request: ToolCallRequest,
handler: Any,
) -> Union[ToolMessage, Command]:
"""Wrap tool execution with before/after safety checks."""
tool_name = request.tool_call.get("name", "unknown")
tool_input = request.tool_call.get("args", {})
# Before tool check
modified_tool_input = self._before_tool(tool_name, tool_input, request.state)
request.tool_call.update({"args": modified_tool_input})
# Execute tool
result = handler(request)
# After tool check
if isinstance(result, ToolMessage):
content = result.content
if isinstance(content, str):
self._after_tool(tool_name, content, request.state)
elif isinstance(content, list):
# Handle list content by converting to string
content_str = str(content)
self._after_tool(tool_name, content_str, request.state)
return cast(Union[ToolMessage, Command], result)
# =====================================
# SIMPLE TOOL FOR DEMONSTRATION
# =====================================
wiring
def main() -> None:
"""
Main demo function showing WonderFence integration with LangChain agent.
"""
# Step 1: Initialize WonderFenceClient client
client = WonderFenceClient(
provider="langchain",
platform="python"
)
# Step 2: Create middleware with WonderFence client
safety_middleware = WonderFenceSafetyMiddleware(client)
# Step 3: Create LangChain model
model = ChatGoogleGenerativeAI(
model="gemini-2.5-flash"
)
# Step 4: Create agent with middleware
agent = create_agent(
model=model,
tools=[calculator],
middleware=[safety_middleware] # register the middleware with the agent
)
Good to know
Pass session_id and user_id in invoke() so every verdict is traceable in the Alice dashboard. _after_model and _after_tool mirror the checks shown with evaluate_response_sync.