LangGraph

A small safety monitor you call from your own graph nodes: check the prompt before the model node runs, the response after it, and tool input and output around tool execution. Fits any StateGraph without changing its shape.

Status: Generally available · Evaluates: Prompts, Responses, Tool calls · Vendor: LangChain

Prompt, response, tool input, tool output, from graph state.

Note: This example uses the v1 WonderFenceClient interface of the WonderFence SDK.

Setup

pip install langgraph langchain-core wonderfence-sdk

Configuration: ALICE_API_KEY, ALICE_APP_ID

Example

imports (langgraph_hooks_simple.py)

from langchain_core.language_models.base import BaseLanguageModel
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from langchain_google_genai import ChatGoogleGenerativeAI
from langgraph.graph import END, StateGraph

from wonderfence_sdk.client import WonderFenceClient
from wonderfence_sdk.models import Actions, AnalysisContext

safety monitor: prompt + tool input checks

class WonderFenceSafetyMonitor:
    """
    Monitor that integrates WonderFence SDK for safety checks.

    Provides methods to:
    - Check prompt safety before model calls
    - Check response safety after model calls
    - Check tool input safety before tool calls
    - Check tool output safety after tool calls
    """

    def __init__(self, wonderfence_client: WonderFenceClient) -> None:
        """
        Initialize the WonderFence safety monitor.

        Args:
            wonderfence_client: WonderFenceClient client for safety evaluation
        """
        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"),
            provider="langgraph",
            platform="python",
        )

    def check_prompt_safety(self, state: AgentState) -> None:
        """
        Check prompt safety before model invocation.

        Args:
            state: Current agent state with messages
        """
        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

        # Extract and validate content
        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)

        # Evaluate prompt safety
        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')}")
            elif evaluation.action == Actions.MASK:
                logger.info("   🎭 MASKED: Content modified for safety")
                if hasattr(latest_message, "content") and evaluation.action_text:
                    latest_message.content = evaluation.action_text

        except Exception as e:
            logger.error(f"   ❌ Safety check failed: {e}", exc_info=True, stack_info=True)
            raise

    def check_tool_input_safety(self, state: AgentState, tool_name: str, tool_input: dict[str, Any]) -> dict[str, Any]:
        """
        Check tool input safety before tool execution.

        Args:
            state: Current agent state
            tool_name: Name of the tool being called
            tool_input: Input parameters for the tool
        """
        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

graph nodes: call_model

    def call_model(state: AgentState) -> AgentState:
        """Call the model with safety checks."""
        # Before model call - check prompt safety
        monitor.check_prompt_safety(state)

        messages = state["messages"]
        try:
            # Call the actual model
            response = model.invoke(messages)

            # After model call - check response safety
            response = monitor.check_response_safety(state, response)

            # Add response to messages
            state["messages"].append(response)

        except Exception as e:
            logger.error(f"❌ Model call failed: {e}")
            # Create error response
            error_response = AIMessage(content=f"I encountered an error: {e!s}")
            state["messages"].append(error_response)

        return state

wiring

def create_simple_langgraph_agent() -> Any:
    """
    Create a simple LangGraph agent with WonderFence safety checks.

    Returns:
        Compiled LangGraph agent
    """
    # Step 1: Initialize WonderFenceClient client
    client = WonderFenceClient(
        provider="langgraph",
        platform="python"
    )

    # Step 2: Create safety monitor with WonderFence client
    safety_monitor = WonderFenceSafetyMonitor(client)

    # Step 3: Create LangChain model
    model = ChatGoogleGenerativeAI(
        model="gemini-2.5-flash"
    )

    # Step 4: Create nodes with safety checks
    nodes = create_agent_nodes(safety_monitor, model, [calculator])

    # Step 5: Create the graph
    workflow = StateGraph(AgentState)

    # Add nodes
    workflow.add_node("agent", nodes["agent"])
    workflow.add_node("tools", nodes["tools"])

    # Set entry point
    workflow.set_entry_point("agent")

    # Add conditional edges
    workflow.add_conditional_edges(
        "agent",
        nodes["should_continue"],
        {"tools": "tools", "end": END}
    )
    workflow.add_edge("tools", "agent")

    # Compile the graph
    app = workflow.compile()
    logger.info("✅ LangGraph agent created with WonderFence safety checks")

    return app

Good to know

check_response_safety and check_tool_output_safety mirror the two checks shown, using evaluate_response_sync. call_tool wraps tool.invoke() with the tool input and output checks.

Links

• LangGraph docs