Forge middleware for LangGraph & LangChain -- verify every tool call before execution.
AI agents are taking real-world actions — sending emails, moving money, accessing databases, calling APIs. But there's no independent system verifying what they do.
The agent frameworks (LangChain, CrewAI, OpenAI Agents SDK) provide no enforcement beyond prompt engineering. When an agent goes rogue, you find out from your customers, not from your monitoring. This is like letting employees process transactions with no audit department, no compliance checks, no separation of duties.
Forge is the independent verification layer for AI agents. It checks every action before execution (Verify) and tracks every execution with cryptographic receipts (Execute). One SDK. Sub-15ms latency. Works with every major agent framework.
Forge never sees your agent's instructions, prompts, code, data, or outputs. It verifies the action type and behavioral pattern only. Your intellectual property stays yours.
Every other tool in this space — Guardrails AI, NeMo Guardrails, LlamaGuard — requires reading your prompts and outputs to make decisions. Forge doesn't. That's not a feature difference — it's a fundamentally different architecture.
Ed25519 asymmetric cryptography. The agent signs receipts with a private key only it holds. Anyone with the public key can verify independently. You don't trust Veritera. You trust math. No shared secrets. No phone-home. The proof stands on its own.
Policies are enforced outside the agent. The agent cannot override, bypass, or modify its own guardrails. This is separation of duties — the same principle that prevents an accountant from approving their own expenses.
Forge works like a home security system. Your security company doesn't know what's inside your house. They don't inventory your jewelry or read your mail. They monitor doors opening, windows breaking, motion where there shouldn't be motion. They protect the pattern — not the contents. Forge does the same for your AI agents.
Not just what we verify — but how we verify it. Patent pending.
These aren't hypothetical. Each is a documented attack pattern against production AI agents.
| Threat | What Happens Without Forge | How Forge Stops It |
|---|---|---|
| Prompt Injection | Injected instructions cause agent to exfiltrate data | Action blocked pre-execution — the unauthorized API call never fires |
| Agent Drift | 5 benign steps cascade into a data breach | Cumulative risk scoring detects the escalation pattern |
| Trust Poisoning | Compromised agent poisons every agent that trusts it | Agent-to-agent attestation — no agent inherits trust without proof |
| Tool Shadowing | Shadow tool mimics your allowlist, steals credentials | Identity-based tool verification at the MCP layer |
| Hallucinated Completion | Agent claims it finished work it never started | Receipt chain is empty — no cryptographic proof, no credit |
| Specification Drift | Agent builds the wrong features convincingly | ZKP probes verify alignment to the actual specification |
| Circular Execution | Agent loops for hours, looks productive, accomplishes nothing | Execution graph analysis detects the cycle |
| Behavioral Drift | Agent quietly stops running tests over weeks | Cross-task behavioral baselines catch the regression |
pip install langchain-forgeThis installs langchain-forge along with its dependencies: the Forge Python SDK (veritera) and langchain-core.
For a complete agent setup, you will also need LangGraph and an LLM provider:
pip install langchain-forge langgraph langchain-openaiBefore using Forge with LangGraph, create a policy that defines what your agent is allowed to do. You only need to do this once:
from veritera import Forge
forge = Forge(api_key="vt_live_...") # Get your key at forge.veritera.ai
# Create a policy from code
forge.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
forge.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 veriterato get the policy management SDK. See the full policy docs.
Add Forge 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 forge_langgraph import ForgeVerifyMiddleware
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 Forge
middleware = ForgeVerifyMiddleware(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,
)
# Forge 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 Forge: Amount exceeds single-transaction limit" and relays the restriction to the user.
This walkthrough builds a realistic LangGraph agent that retrieves documents, queries a database, and sends emails -- with Forge middleware catching any unauthorized operations along the way.
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}"import os
from forge_langgraph import ForgeVerifyMiddleware
os.environ["VERITERA_API_KEY"] = "vt_live_..."
middleware = ForgeVerifyMiddleware(
policy="rag-agent-policy",
agent_id="rag-support-agent",
fail_closed=True, # block if Forge 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}"),
)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,
)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 Forge
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 Forge.
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 Forge
BLOCKED: delete_records -- Destructive database operations require manual approval
tool: Action 'delete_records' denied by Forge: 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.
Forge 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.
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 Forge
BLOCKED: send_email -- Cannot send confidential financial data to external addresses
tool: Action 'send_email' denied by Forge: 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 Forge caught the attempt to email confidential data to an external address. The agent handled both outcomes gracefully.
The middleware intercepts every tool call automatically. No changes to your tools or agent logic required.
from forge_langgraph import ForgeVerifyMiddleware
from langgraph.prebuilt import ToolNode, create_react_agent
middleware = ForgeVerifyMiddleware(
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.
Creates a LangChain tool that the LLM calls explicitly to verify actions. The LLM decides when verification is needed.
from forge_langgraph import forge_verify_tool
verify = forge_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 forge_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).
| Parameter | Type | Default | Description |
|---|---|---|---|
api_key |
str |
VERITERA_API_KEY env var |
Your Forge API key. Starts with vt_live_ (production) or vt_test_ (sandbox). |
base_url |
str |
https://forge.veritera.ai |
Forge API endpoint. Override for self-hosted deployments. |
agent_id |
str |
"langgraph-agent" |
Identifier for this agent in Forge 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 Forge API is unreachable. When False, actions are allowed through on API failure. |
timeout |
float |
10.0 |
HTTP request timeout in seconds for the Forge 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. |
User message
|
v
LLM decides to call a tool
|
v
ForgeVerifyMiddleware.wrap_tool_call()
|
+---> Is tool in skip_actions?
| YES --> Execute tool normally
| NO --> Call Forge /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 Forge:
- 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
Forge evaluates the action and returns a verdict with a proof_id for audit trail purposes.
If you are building a custom LangGraph StateGraph instead of using create_react_agent, you can add Forge verification directly into your graph nodes.
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 Forge
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}
# --- Forge client ---
forge = Forge(
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 Forge verification."""
outputs = []
for tool_call in state["messages"][-1].tool_calls:
name = tool_call["name"]
args = tool_call["args"]
# Verify through Forge before executing
try:
result = forge.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 Forge: {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")]
})For larger graphs with multiple tool-calling nodes, extract Forge verification into a helper:
from veritera import Forge
from langchain_core.messages import ToolMessage
forge = Forge(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 Forge, then execute or deny."""
name = tool_call["name"]
args = tool_call["args"]
try:
result = forge.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 Forge: {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}When the Forge API cannot be reached (network issues, timeouts), behavior depends on fail_closed:
# Fail closed (default) -- deny the action when Forge is unreachable
middleware = ForgeVerifyMiddleware(policy="my-policy", fail_closed=True)
# Fail open -- allow the action through when Forge is unreachable
middleware = ForgeVerifyMiddleware(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 Forge blocking calls.
If no API key is provided and VERITERA_API_KEY is not set, ForgeVerifyMiddleware raises a ValueError at initialization -- not at runtime. This ensures misconfigured agents fail fast.
try:
middleware = ForgeVerifyMiddleware(policy="my-policy")
except ValueError as e:
print(e)
# "Forge API key required. Pass api_key= or set VERITERA_API_KEY env var."The middleware logs all verification decisions. Enable debug logging to see approval/denial details:
import logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("forge_langgraph").setLevel(logging.DEBUG)Log output:
DEBUG:forge_langgraph:Forge APPROVED: query_database (proof=fp_abc123)
WARNING:forge_langgraph:Forge DENIED: delete_records -- Destructive operations blocked
ERROR:forge_langgraph:Forge verify error for send_email: Connection timed out
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 = ForgeVerifyMiddleware(
policy="production-policy",
on_verified=log_to_datadog,
on_blocked=alert_on_block,
)| Variable | Required | Description |
|---|---|---|
VERITERA_API_KEY |
Yes (unless passed directly) | Your Forge 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. |
While V1 (Verify) checks individual actions before they happen, V2 (Execute) monitors entire task executions and provides cryptographic proof that the work was done correctly -- without ever seeing the actual code or output.
Execute works by generating signed receipts at each step of an agent's task. These receipts form a tamper-proof audit trail that proves what happened and in what order, using mathematical proof. The receipts are submitted to Forge Execute, which verifies the behavioral pattern matches expectations -- without needing access to the actual instructions, code, or data.
Each tool call triggers a signed receipt containing the action type, timestamp, and a unique nonce — signed with your agent's Ed25519 private key. The receipt proves what action happened and in what order — without containing any code, data, or output. The server verifies the behavioral pattern, not the content. The receipt is submitted to Forge Execute, which verifies the signature and adds it to the task's receipt chain.
Receipt emission is non-blocking. If Forge Execute is unreachable, the error is logged and your agent continues working. A receipt failure should never stop your agent. This is by design — Verify is fail-closed (blocks on error), Execute is fire-and-forget (logs on error).
Anyone with your agent's public key can verify the receipt chain independently — no need to contact Veritera, no API call required. The Ed25519 signatures are self-contained mathematical proofs.
import os
from langgraph.prebuilt import create_react_agent, ToolNode
from langchain_core.tools import tool
from forge_langgraph import ForgeExecuteMiddleware
os.environ["VERITERA_API_KEY"] = "vt_live_..."
os.environ["OPENAI_API_KEY"] = "sk-..."
@tool
def run_query(sql: str) -> str:
"""Run a database query."""
return f"Query executed: {sql}"
@tool
def send_report(to: str, data: str) -> str:
"""Send a report via email."""
return f"Report sent to {to}"
# Create Execute middleware -- receipts are emitted after each tool call
execute_mw = ForgeExecuteMiddleware(
task_id="task_weekly_analytics",
agent_id="analytics-agent",
)
tools = [run_query, send_report]
tool_node = ToolNode(tools, wrap_tool_call=execute_mw.wrap_tool_call)
agent = create_react_agent(
model="gpt-4.1",
tools=tool_node,
)
result = agent.invoke({"messages": [("user", "Run the weekly analytics report and email it to team@acme.com")]})Every tool call automatically generates a signed receipt after successful execution. Forge Execute verifies the chain of receipts to confirm the task followed the expected behavioral pattern.
For actions that happen outside of tool calls, emit receipts manually:
execute_mw = ForgeExecuteMiddleware(
task_id="task_data_pipeline",
agent_id="etl-agent",
)
# Emit a receipt for a custom action
result = execute_mw.emit_receipt("data_extraction_complete")
print(f"Receipt: {result['receipt_id']}, Chain position: {result['chain_index']}")V1 (Verify) and V2 (Execute) are complementary. V1 checks permission before each action. V2 tracks execution across the entire task. Use both for complete coverage:
from forge_langgraph import ForgeVerifyMiddleware, ForgeExecuteMiddleware
# V1: Policy verification
verify_mw = ForgeVerifyMiddleware(policy="finance-controls")
# V2: Execution receipts
execute_mw = ForgeExecuteMiddleware(
task_id="task_monthly_close",
agent_id="finance-agent",
)
# Chain both middlewares -- verify first, then emit receipt
def combined_wrapper(request, handler):
# V1: Verify the action is allowed
result = verify_mw.wrap_tool_call(request, handler)
# V2: Emit receipt for the completed action
tool_name = getattr(request, "name", None) or "unknown"
execute_mw.emit_receipt(tool_name)
return result
tool_node = ToolNode(tools, wrap_tool_call=combined_wrapper)| Parameter | Type | Default | Description |
|---|---|---|---|
task_id |
str |
Required | Identifier for the task being executed. Links all receipts in the chain. |
agent_id |
str |
Required | Identifier for the agent performing the task. |
api_key |
str |
VERITERA_API_KEY env var |
Your Forge API key. |
signing_key |
str |
Same as api_key |
Key used to sign receipts. Defaults to the API key. |
base_url |
str |
"https://forge.veritera.ai" |
Forge API endpoint. |
Methods:
| Method | Description |
|---|---|
wrap_tool_call(request, handler) |
Wraps a LangGraph tool call. Executes the tool, then emits a signed receipt. Matches the ToolCallWrapper protocol. |
emit_receipt(action_type) |
Manually emit a receipt for a custom action. Returns {"receipt_id": ..., "chain_index": ...}. |
| Package | Framework | What It Does | Install |
|---|---|---|---|
| veritera | Any Python | Core SDK. Verify actions, manage policies, sign and submit receipts. | pip install veritera |
| forge-openai | OpenAI Agents SDK | forge_protect() wrapper + ForgeExecuteGuardrail. Drop-in verification for OpenAI agents. |
pip install forge-openai |
| langchain-forge (this package) | LangChain / LangGraph | ForgeVerifyMiddleware + ForgeExecuteMiddleware. Wraps LangGraph tool nodes. |
pip install langchain-forge |
| crewai-forge | CrewAI | ForgeVerifyTool + guardrails + LLM hooks + ForgeExecuteHook. Multi-agent crew support. |
pip install crewai-forge |
| llama-index-tools-forge | LlamaIndex | ForgeVerifyToolSpec + event handlers + ForgeExecuteHandler. Document agent support. |
pip install llama-index-tools-forge |
| forge-blog | — | Technical articles on AI agent security and trust verification. | — |
MIT -- Forge by Veritera AI