langchain-forge 0.2.0

Forge Verify + Execute middleware for LangGraph/LangChain — verify tool calls and track executions with cryptographic receipts

langchain-forge

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

---

Why Forge?

AI agents make autonomous decisions, but not every decision should execute unchecked. Forge 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-forge

This 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-openai

---

Prerequisites: Create a Policy

Before 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 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 veritera to get the policy management SDK. See the full policy docs.

---

Quick Start

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 [e-mail]")]})
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.

---

Tutorial: Building a Verified RAG Agent

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.

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 Forge Middleware

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}"),
)

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 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.

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 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.

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

result = agent.invoke({
    "messages": [(
        "user",
        "Find the Q1 revenue data and email a summary to [e-mail]"
    )]
})

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

Expected output:

human: Find the Q1 revenue data and email a summary to [e-mail]
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.

---

Integration Patterns

ForgeVerifyMiddleware -- Automatic Verification (Recommended)

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.

forge_verify_tool() -- Explicit Verification

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).

---

Configuration Reference

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://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.

---

How It Works

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.

---

Working with LangGraph StateGraph

If you are building a custom LangGraph StateGraph instead of using create_react_agent, you can add Forge 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 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")]
})

Reusable Verification Helper

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}

---

Error Handling

Forge API Unreachable

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.

Missing API Key

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."

Logging

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

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 = ForgeVerifyMiddleware(
    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 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.

---

V2: Forge Execute -- Cryptographic Execution Receipts

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.

Quick Start

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 [e-mail]")]})

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.

Manual Receipts

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']}")

Using V1 + V2 Together

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)

ForgeExecuteMiddleware Reference

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://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": ...}.

---

Other Forge Integrations

Forge provides middleware for all major agent frameworks:

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

---

Links

• Forge Documentation
• Forge Dashboard
• API Reference
• GitHub: langchain-forge

---

License

MIT -- Forge by Veritera AI