Skip to main content

EYDII Verify middleware for LangGraph/LangChain — verify every tool call before execution

Project description

langchain-eydii

PyPI version License: MIT Python 3.10+

EYDII middleware for LangGraph & LangChain -- verify every tool call before execution.


Why EYDII?

AI agents make autonomous decisions, but not every decision should execute unchecked. EYDII sits between your agent's intent and the actual tool call, verifying each action against your security policies in real time. If the action violates a policy, it never executes -- the agent receives a denial message and can explain the restriction to the user, all without crashing or breaking the conversation flow.


Install

pip install langchain-eydii

This installs langchain-eydii along with its dependencies: the EYDII Python SDK (veritera) and langchain-core.

For a complete agent setup, you will also need LangGraph and an LLM provider:

pip install langchain-eydii langgraph langchain-openai

Prerequisites: Create a Policy

Before using EYDII with LangGraph, create a policy that defines what your agent is allowed to do. You only need to do this once:

from veritera import Eydii

eydii = Eydii(api_key="vt_live_...")  # Get your key at id.veritera.ai

# Create a policy from code
eydii.create_policy_sync(
    name="finance-controls",
    description="Controls for financial and data operations",
    rules=[
        {"type": "action_whitelist", "params": {"allowed": ["payment.send", "balance.check", "query.read"]}},
        {"type": "amount_limit", "params": {"max": 10000, "currency": "USD"}},
    ],
)

# Or generate one from plain English
eydii.generate_policy_sync(
    "Allow payments under $10,000, read-only database queries, and balance checks. Block all deletions and bulk exports.",
    save=True,
)

A default policy is created automatically when you sign up — it blocks dangerous actions like database drops and admin overrides. You can use it immediately with policy="default".

Tip: pip install veritera to get the policy management SDK. See the full policy docs.


Quick Start

Add EYDII verification to any LangGraph agent in three lines:

import os
from langgraph.prebuilt import create_react_agent, ToolNode
from langchain_core.tools import tool
from eydii_langgraph import EydiiVerifyMiddleware

os.environ["VERITERA_API_KEY"] = "vt_live_..."
os.environ["OPENAI_API_KEY"] = "sk-..."

@tool
def send_payment(amount: float, recipient: str) -> str:
    """Send a payment to a recipient."""
    return f"Sent ${amount} to {recipient}"

@tool
def check_balance(account_id: str) -> str:
    """Check account balance (read-only)."""
    return f"Account {account_id}: $12,340.00"

# Three lines -- every tool call now goes through EYDII
middleware = EydiiVerifyMiddleware(policy="finance-controls")  # create this policy first (see above) -- or use "default"
tools = [send_payment, check_balance]
tool_node = ToolNode(tools, wrap_tool_call=middleware.wrap_tool_call)

agent = create_react_agent(
    model="gpt-4.1",
    tools=tool_node,
)

# Eydii verifies the tool call before it executes
result = agent.invoke({"messages": [("user", "Send $500 to vendor@acme.com")]})
print(result["messages"][-1].content)

If the send_payment call is approved by your policy, it executes normally. If denied, the agent receives a message like "Action 'send_payment' denied by EYDII: Amount exceeds single-transaction limit" and relays the restriction to the user.


Tutorial: Building a Verified RAG Agent

This walkthrough builds a realistic LangGraph agent that retrieves documents, queries a database, and sends emails -- with EYDII middleware catching any unauthorized operations along the way.

Step 1: Define Your Tools

from langchain_core.tools import tool

@tool
def search_documents(query: str) -> str:
    """Search the internal knowledge base for relevant documents."""
    # In production, this calls your vector store (Pinecone, Weaviate, etc.)
    return (
        "Found 3 documents:\n"
        "1. Q1 Revenue Report (confidential)\n"
        "2. Product Roadmap 2026\n"
        "3. Employee Handbook v4.2"
    )

@tool
def query_database(sql: str) -> str:
    """Run a read-only SQL query against the analytics database."""
    # In production, this connects to your database
    return "| customer_id | total_spend |\n|-------------|-------------|\n| C-1001 | $45,200 |\n| C-1002 | $38,750 |"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to a recipient."""
    return f"Email sent to {to}: '{subject}'"

@tool
def delete_records(table: str, condition: str) -> str:
    """Delete records from a database table."""
    return f"Deleted records from {table} where {condition}"

Step 2: Configure EYDII Middleware

import os
from eydii_langgraph import EydiiVerifyMiddleware

os.environ["VERITERA_API_KEY"] = "vt_live_..."

middleware = EydiiVerifyMiddleware(
    policy="rag-agent-policy",
    agent_id="rag-support-agent",
    fail_closed=True,                         # block if EYDII API is unreachable
    skip_actions=["search_documents"],         # read-only search is always allowed
    on_blocked=lambda action, reason: print(f"BLOCKED: {action} -- {reason}"),
    on_verified=lambda action, result: print(f"APPROVED: {action}"),
)

Step 3: Build the Agent

from langgraph.prebuilt import create_react_agent, ToolNode

tools = [search_documents, query_database, send_email, delete_records]
tool_node = ToolNode(tools, wrap_tool_call=middleware.wrap_tool_call)

agent = create_react_agent(
    model="gpt-4.1",
    tools=tool_node,
)

Step 4: Run It -- Approved Scenario

result = agent.invoke({
    "messages": [("user", "What were our top customers by spend last quarter?")]
})

for msg in result["messages"]:
    print(f"{msg.type}: {msg.content}")

Expected output:

human: What were our top customers by spend last quarter?
ai: [calls search_documents]          # skipped (in skip_actions)
tool: Found 3 documents: ...
ai: [calls query_database]            # APPROVED by EYDII
APPROVED: query_database
tool: | customer_id | total_spend | ...
ai: Based on the analytics data, your top customers last quarter were
    C-1001 ($45,200) and C-1002 ($38,750).

The search_documents call is skipped (configured as a read-only action). The query_database call is verified and approved by EYDII.

Step 5: Run It -- Denied Scenario

result = agent.invoke({
    "messages": [("user", "Delete all records from the customers table older than 2020")]
})

for msg in result["messages"]:
    print(f"{msg.type}: {msg.content}")

Expected output:

human: Delete all records from the customers table older than 2020
ai: [calls delete_records]            # DENIED by EYDII
BLOCKED: delete_records -- Destructive database operations require manual approval
tool: Action 'delete_records' denied by EYDII: Destructive database operations
      require manual approval
ai: I'm unable to delete those records directly. Your organization's security
    policy requires manual approval for destructive database operations.
    Please submit a deletion request through your admin portal.

Eydii intercepted the delete_records call, checked it against the rag-agent-policy, and returned a denial. The tool never executed. The agent received the denial as a ToolMessage and explained the restriction to the user naturally.

Step 6: Mixed Scenario -- Some Calls Approved, Some Denied

result = agent.invoke({
    "messages": [(
        "user",
        "Find the Q1 revenue data and email a summary to external-partner@gmail.com"
    )]
})

for msg in result["messages"]:
    print(f"{msg.type}: {msg.content}")

Expected output:

human: Find the Q1 revenue data and email a summary to external-partner@gmail.com
ai: [calls search_documents]          # skipped (read-only)
tool: Found 3 documents: ...
ai: [calls send_email]                # DENIED by EYDII
BLOCKED: send_email -- Cannot send confidential financial data to external addresses
tool: Action 'send_email' denied by EYDII: Cannot send confidential financial
      data to external addresses
ai: I found the Q1 Revenue Report, but I'm unable to email it to an external
    address. Your security policy restricts sending confidential financial
    documents outside the organization. You can share it through your
    approved file-sharing platform instead.

The document search succeeded, but Eydii caught the attempt to email confidential data to an external address. The agent handled both outcomes gracefully.


Integration Patterns

EydiiVerifyMiddleware -- Automatic Verification (Recommended)

The middleware intercepts every tool call automatically. No changes to your tools or agent logic required.

from eydii_langgraph import EydiiVerifyMiddleware
from langgraph.prebuilt import ToolNode, create_react_agent

middleware = EydiiVerifyMiddleware(
    policy="my-policy",
    agent_id="my-agent",
)

tools = [tool_a, tool_b, tool_c]
tool_node = ToolNode(tools, wrap_tool_call=middleware.wrap_tool_call)

agent = create_react_agent(
    model="gpt-4.1",
    tools=tool_node,
)

When to use: Most cases. You want a security layer that works regardless of what the LLM decides to do.

eydii_verify_tool() -- Explicit Verification

Creates a LangChain tool that the LLM calls explicitly to verify actions. The LLM decides when verification is needed.

from eydii_langgraph import eydii_verify_tool

verify = eydii_verify_tool(
    policy="my-policy",
    agent_id="my-agent",
)

agent = create_react_agent(
    model="gpt-4.1",
    tools=[send_payment, read_balance, verify],
)

When the LLM calls eydii_verify, it receives a structured response:

# Approved
APPROVED: compliant | proof_id: fp_abc123 | latency: 42ms

# Denied
DENIED: Amount exceeds policy limit | proof_id: fp_def456

When to use: When you want the LLM to reason about verification decisions, or when you need to verify actions that are not LangChain tools (e.g., API calls made inside a tool).


Configuration Reference

Parameter Type Default Description
api_key str VERITERA_API_KEY env var Your EYDII API key. Starts with vt_live_ (production) or vt_test_ (sandbox).
base_url str https://id.veritera.ai EYDII API endpoint. Override for self-hosted deployments.
agent_id str "langgraph-agent" Identifier for this agent in Eydii audit logs and dashboards.
policy str None Policy name to evaluate actions against. When None, the default policy for your API key is used.
fail_closed bool True When True, actions are denied if the EYDII API is unreachable. When False, actions are allowed through on API failure.
timeout float 10.0 HTTP request timeout in seconds for the EYDII API call.
skip_actions list[str] [] Tool names that bypass verification entirely. Use for read-only or low-risk tools.
on_verified Callable None Callback function (action: str, result) -> None called when an action is approved.
on_blocked Callable None Callback function (action: str, reason: str) -> None called when an action is denied.

How It Works

User message
    |
    v
LLM decides to call a tool
    |
    v
EydiiVerifyMiddleware.wrap_tool_call()
    |
    +---> Is tool in skip_actions?
    |         YES --> Execute tool normally
    |         NO  --> Call Eydii /v1/verify
    |                     |
    |                     +---> APPROVED
    |                     |         --> Execute tool normally
    |                     |         --> Call on_verified callback
    |                     |
    |                     +---> DENIED
    |                     |         --> Return ToolMessage with denial reason
    |                     |         --> Call on_blocked callback
    |                     |         --> Tool NEVER executes
    |                     |
    |                     +---> API ERROR
    |                               --> fail_closed=True?  --> Deny
    |                               --> fail_closed=False? --> Execute tool
    v
LLM receives tool result (or denial message)
    |
    v
LLM responds to user

Each verification call sends the following to EYDII:

  • action -- the tool name (e.g., "send_email")
  • agent_id -- which agent is making the call
  • params -- the tool arguments as a dictionary
  • policy -- which policy to evaluate against

EYDII evaluates the action and returns a verdict with a proof_id for audit trail purposes.


Working with LangGraph StateGraph

If you are building a custom LangGraph StateGraph instead of using create_react_agent, you can add EYDII verification directly into your graph nodes.

Basic StateGraph Integration

import os
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from veritera import Eydii

os.environ["VERITERA_API_KEY"] = "vt_live_..."
os.environ["OPENAI_API_KEY"] = "sk-..."

# --- State ---
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# --- Tools ---
@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    return f"Results for '{query}': ..."

@tool
def send_slack_message(channel: str, text: str) -> str:
    """Post a message to a Slack channel."""
    return f"Message posted to #{channel}"

tools = [search_web, send_slack_message]
tools_by_name = {t.name: t for t in tools}

# --- Eydii client ---
eydii = Eydii(
    api_key=os.environ["VERITERA_API_KEY"],
    fail_closed=True,
)

# --- Nodes ---
llm = ChatOpenAI(model="gpt-4.1").bind_tools(tools)

def call_model(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def call_tools(state: AgentState) -> dict:
    """Execute tool calls with EYDII verification."""
    outputs = []
    for tool_call in state["messages"][-1].tool_calls:
        name = tool_call["name"]
        args = tool_call["args"]

        # Verify through EYDII before executing
        try:
            result = eydii.verify_sync(
                action=name,
                agent_id="custom-graph-agent",
                params=args,
                policy="slack-bot-policy",
            )
        except Exception:
            outputs.append(ToolMessage(
                content=f"Action '{name}' blocked -- verification unavailable.",
                tool_call_id=tool_call["id"],
            ))
            continue

        if result.verified:
            # Approved -- execute the tool
            tool_result = tools_by_name[name].invoke(args)
            outputs.append(ToolMessage(
                content=str(tool_result),
                tool_call_id=tool_call["id"],
            ))
        else:
            # Denied -- return denial to the LLM
            outputs.append(ToolMessage(
                content=f"Action '{name}' denied by EYDII: {result.reason}",
                tool_call_id=tool_call["id"],
            ))

    return {"messages": outputs}

def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

# --- Graph ---
graph = StateGraph(AgentState)
graph.add_node("model", call_model)
graph.add_node("tools", call_tools)
graph.add_edge(START, "model")
graph.add_conditional_edges("model", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "model")

app = graph.compile()

# --- Run ---
result = app.invoke({
    "messages": [HumanMessage(content="Post our Q1 results to #general on Slack")]
})

Reusable Verification Helper

For larger graphs with multiple tool-calling nodes, extract EYDII verification into a helper:

from veritera import Eydii
from langchain_core.messages import ToolMessage

eydii = Eydii(api_key="vt_live_...", fail_closed=True)

def verify_and_execute(tool_call: dict, tools_map: dict, policy: str) -> ToolMessage:
    """Verify a tool call through EYDII, then execute or deny."""
    name = tool_call["name"]
    args = tool_call["args"]

    try:
        result = eydii.verify_sync(
            action=name,
            agent_id="my-agent",
            params=args,
            policy=policy,
        )
    except Exception:
        return ToolMessage(
            content=f"Action '{name}' blocked -- verification unavailable.",
            tool_call_id=tool_call["id"],
        )

    if result.verified:
        output = tools_map[name].invoke(args)
        return ToolMessage(content=str(output), tool_call_id=tool_call["id"])

    return ToolMessage(
        content=f"Action '{name}' denied by EYDII: {result.reason}",
        tool_call_id=tool_call["id"],
    )

Then use it in any node:

def call_tools(state: AgentState) -> dict:
    outputs = [
        verify_and_execute(tc, tools_by_name, policy="my-policy")
        for tc in state["messages"][-1].tool_calls
    ]
    return {"messages": outputs}

Error Handling

EYDII API Unreachable

When the EYDII API cannot be reached (network issues, timeouts), behavior depends on fail_closed:

# Fail closed (default) -- deny the action when Eydii is unreachable
middleware = EydiiVerifyMiddleware(policy="my-policy", fail_closed=True)

# Fail open -- allow the action through when Eydii is unreachable
middleware = EydiiVerifyMiddleware(policy="my-policy", fail_closed=False)

Recommendation: Use fail_closed=True for production. Use fail_closed=False only during development when you want to test agent behavior without Eydii blocking calls.

Missing API Key

If no API key is provided and VERITERA_API_KEY is not set, EydiiVerifyMiddleware raises a ValueError at initialization -- not at runtime. This ensures misconfigured agents fail fast.

try:
    middleware = EydiiVerifyMiddleware(policy="my-policy")
except ValueError as e:
    print(e)
    # "EYDII API key required. Pass api_key= or set VERITERA_API_KEY env var."

Logging

The middleware logs all verification decisions. Enable debug logging to see approval/denial details:

import logging

logging.basicConfig(level=logging.DEBUG)
logging.getLogger("eydii_langgraph").setLevel(logging.DEBUG)

Log output:

DEBUG:eydii_langgraph:Eydii APPROVED: query_database (proof=fp_abc123)
WARNING:eydii_langgraph:Eydii DENIED: delete_records -- Destructive operations blocked
ERROR:eydii_langgraph:Eydii verify error for send_email: Connection timed out

Custom Callbacks for Monitoring

Use on_verified and on_blocked to integrate with your observability stack:

import json

def log_to_datadog(action: str, result) -> None:
    # Send approved actions to your monitoring system
    print(json.dumps({"action": action, "proof_id": result.proof_id, "status": "approved"}))

def alert_on_block(action: str, reason: str) -> None:
    # Alert when actions are blocked
    print(json.dumps({"action": action, "reason": reason, "status": "denied"}))

middleware = EydiiVerifyMiddleware(
    policy="production-policy",
    on_verified=log_to_datadog,
    on_blocked=alert_on_block,
)

Environment Variables

Variable Required Description
VERITERA_API_KEY Yes (unless passed directly) Your EYDII API key. Get one at veritera.ai/dashboard.
OPENAI_API_KEY For OpenAI models Required if using gpt-4.1 or other OpenAI models via langchain-openai.
ANTHROPIC_API_KEY For Anthropic models Required if using Claude models via langchain-anthropic.

Other Eydii Integrations

Eydii provides middleware for all major agent frameworks:

Package Framework PyPI
langchain-eydii LangGraph / LangChain pip install langchain-eydii
openai-eydii OpenAI Agents SDK pip install openai-eydii
crewai-eydii CrewAI pip install crewai-eydii
llamaindex-eydii LlamaIndex pip install llamaindex-eydii
veritera Python SDK (core) pip install veritera

Links


License

MIT -- Eydii by Veritera AI

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

langchain_eydii-0.1.1.tar.gz (17.4 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

langchain_eydii-0.1.1-py3-none-any.whl (12.4 kB view details)

Uploaded Python 3

File details

Details for the file langchain_eydii-0.1.1.tar.gz.

File metadata

  • Download URL: langchain_eydii-0.1.1.tar.gz
  • Upload date:
  • Size: 17.4 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.1

File hashes

Hashes for langchain_eydii-0.1.1.tar.gz
Algorithm Hash digest
SHA256 c92c04e3a35e373ab036cf24b96a58c45ec616797fb92169b932b7bbb5c9abcc
MD5 166f36cb26f1deb0271f6e227a6d3a77
BLAKE2b-256 bc4ffd4e684304a32cba4b05c0524393a32b01acdf484778e8a0f94da97feb3a

See more details on using hashes here.

File details

Details for the file langchain_eydii-0.1.1-py3-none-any.whl.

File metadata

File hashes

Hashes for langchain_eydii-0.1.1-py3-none-any.whl
Algorithm Hash digest
SHA256 5abfd1347fefcce6eb87e0af78afc8e9e61d8dae77ead7dedc66c046c0e3ea90
MD5 26eb6a5ca9a95ed9908d56e61e3ee829
BLAKE2b-256 f4fd27ef550c20f17a0271015974294a5d58883ac3a59726f29a0e7d29d8a6fd

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page