Skip to main content

Forge Verify tools and middleware for Agno — verify every agent action before execution

Project description

agno-forge

PyPI version License: MIT Python 3.10+

Forge Verify tools and middleware for Agno — verify every agent action before execution.


Why Forge?

When AI agents act autonomously, you need a way to enforce rules that the agents themselves cannot override. Forge sits between your agents and their actions — every sensitive operation is verified against your policies in real time, with a cryptographic proof trail. No more hoping the system prompt holds; Forge gives you external, tamper-proof verification that works even when agents delegate to other agents.


Install

pip install agno-forge

This installs forge_agno and its dependencies (veritera SDK and agno).


Prerequisites: Create a Policy

Before using Forge with Agno, create a policy that defines what your agents are 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 multi-agent financial operations",
    rules=[
        {"type": "action_whitelist", "params": {"allowed": ["trade.execute", "refund.process", "report.generate"]}},
        {"type": "amount_limit", "params": {"max": 10000, "currency": "USD"}},
    ],
)

# Or generate one from plain English
forge.generate_policy_sync(
    "Allow trades under $10,000, refund processing, and report generation. Block all account deletions and unauthorized data 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

import os
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from forge_agno import ForgeVerifyTool

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

# 1. Create a Forge verification tool
verify = ForgeVerifyTool(policy="finance-controls")  # create this policy first (see above) -- or use "default"

# 2. Give it to your agent
agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[verify],
    instructions=[
        "You are a financial analyst.",
        "ALWAYS verify actions through forge_verify before executing them.",
    ],
)

# 3. Run the agent — every action is verified
agent.print_response("Process the refund for order #12345 for $500")

The agent calls forge_verify before executing sensitive actions. If an action is denied, the agent receives a DENIED response and adjusts its plan.


Three Integration Points

1. ForgeVerifyTool — Agent Tool for Explicit Verification

The most direct integration. Give agents a callable tool they can use to check whether an action is allowed.

from forge_agno import ForgeVerifyTool
from agno.agent import Agent

tool = ForgeVerifyTool(
    policy="finance-controls",
    agent_id="analyst-bot",
    fail_closed=True,
)

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[tool],
)

How the agent uses it:

The agent calls forge_verify(action="payment.create", params='{"amount": 500, "currency": "USD"}') and receives:

  • APPROVED: Allowed | proof_id: fp_abc123 | latency: 42ms — proceed with the action.
  • DENIED: Amount exceeds $200 limit | proof_id: fp_def456 | Do NOT proceed with this action. — the agent adjusts its plan.

Constructor parameters:

Parameter Type Default Description
api_key str VERITERA_API_KEY env var Your Forge API key
base_url str https://veritera.ai Forge API endpoint
agent_id str "agno-agent" Identifier in audit logs
policy str None Policy set to evaluate against
fail_closed bool True Deny when API is unreachable
timeout float 10.0 Request timeout in seconds
skip_actions list[str] None Actions to skip verification for
on_verified Callable None Callback when action is approved
on_blocked Callable None Callback when action is denied

2. forge_wrap_tool() — Decorator for Pre-Execution Verification

Wraps any Agno tool function with automatic Forge verification. The tool is only executed if Forge approves the action.

from forge_agno import forge_wrap_tool
from agno.agent import Agent

@forge_wrap_tool(policy="finance-controls")
def send_payment(amount: float, recipient: str) -> str:
    """Send a payment to a recipient."""
    return process_payment(amount, recipient)

@forge_wrap_tool(policy="finance-controls")
def delete_record(record_id: str) -> str:
    """Delete a database record."""
    return db.delete(record_id)

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[send_payment, delete_record],
)

How it works:

  1. The agent calls send_payment(amount=500, recipient="user@example.com").
  2. The decorator intercepts the call and sends a verification request to Forge with action=send_payment and the function parameters.
  3. If Forge approves, the original function executes normally.
  4. If Forge denies, the function is NOT called and a denial message is returned.

3. ForgeToolkit — Agno Toolkit Class

An Agno Toolkit that exposes verify_action and list_policies as tools the agent can call.

from forge_agno import ForgeToolkit
from agno.agent import Agent

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[ForgeToolkit(policy="production-safety")],
    instructions=[
        "You are a production operations agent.",
        "Always verify actions before executing them.",
    ],
)

Tools provided:

  • verify_action(action, params) — Verify an action against Forge policies. Same behavior as ForgeVerifyTool.
  • list_policies() — List available Forge verification policies.

Execute Receipts

For execution tracking and audit trails, use the Execute integration:

ForgeExecuteHook — Manual Receipt Emission

from forge_agno import ForgeExecuteHook

hook = ForgeExecuteHook(task_id="task_abc...", agent_id="research-agent")

# Call whenever a tool is invoked
receipt = hook.on_tool_use("file_read")
# Returns: {"receipt_id": "rx_...", "chain_index": 1}

forge_execute_wrapper — Automatic Receipt Emission

from forge_agno import forge_execute_wrapper

@forge_execute_wrapper(task_id="task_abc...", agent_id="research-agent")
def send_payment(amount: float, recipient: str) -> str:
    """Send a payment to a recipient."""
    return process_payment(amount, recipient)

# Every successful call automatically emits a signed receipt

Configuration Reference

Config Source Required Example
API key VERITERA_API_KEY env var or api_key= parameter Yes vt_live_abc123
Base URL base_url= parameter No https://veritera.ai
Policy policy= parameter No (but recommended) "finance-controls"
Agent ID agent_id= parameter No "my-agno-agent"
Fail closed fail_closed= parameter No (default: True) True or False
Timeout timeout= parameter No (default: 10.0) 30.0

How It Works

┌─────────────────────────────────────────────────────────┐
│                   Your Agno Agent                        │
│                                                         │
│  ┌───────────────────────────────────────────────┐      │
│  │              Agent                            │      │
│  │  tools=[forge_verify, send_payment, ...]      │      │
│  └──────────────────┬────────────────────────────┘      │
│                     │                                   │
│         ┌───────────┼───────────┐                       │
│         │           │           │                       │
│    ┌────▼────┐ ┌────▼────┐ ┌────▼────┐                 │
│    │ Verify  │ │ Wrapped │ │ Toolkit │                 │
│    │  Tool   │ │  Tool   │ │  Tools  │                 │
│    └────┬────┘ └────┬────┘ └────┬────┘                 │
│         │           │           │                       │
└─────────┼───────────┼───────────┼───────────────────────┘
          │           │           │
          ▼           ▼           ▼
    ┌─────────────────────────────────────────┐
    │            Forge Verify API             │
    │                                         │
    │  Policy Engine  │  Audit Trail  │ Proof │
    └─────────────────────────────────────────┘
  1. Agent calls tool — The verification tool sends a request to the Forge API.
  2. Forge evaluates — The policy engine checks the action and parameters against your defined policies.
  3. Result returnedAPPROVED (with proof ID) or DENIED (with reason and proof ID).
  4. Agent decides — On approval, the agent proceeds. On denial, the agent adjusts its plan.
  5. Audit trail recorded — Every verification produces a proof_id linking to a permanent, tamper-proof record.

Error Handling

The package handles three failure modes:

1. Forge API Unreachable

Controlled by fail_closed:

# fail_closed=True (default) — deny when Forge is down
tool = ForgeVerifyTool(policy="controls", fail_closed=True)
# Agent receives: "DENIED: Verification unavailable — ConnectionError(...). Action blocked (fail-closed)."

# fail_closed=False — allow when Forge is down (use for non-critical paths)
tool = ForgeVerifyTool(policy="controls", fail_closed=False)

2. Invalid Parameters

If the agent passes malformed JSON as params, the tool wraps it safely:

# Agent calls: forge_verify(action="test", params="not valid json")
# Tool parses it as: {"raw": "not valid json"} and proceeds with verification

3. Wrapped Tool Denials

When a wrapped tool is denied, the function is not called:

@forge_wrap_tool(policy="finance-controls")
def send_payment(amount: float, recipient: str) -> str:
    return process_payment(amount, recipient)

# If denied: returns "DENIED: Action 'send_payment' blocked by Forge: Amount exceeds limit"
# The process_payment function is never called

All errors are logged via Python's logging module under the forge_agno logger:

import logging
logging.getLogger("forge_agno").setLevel(logging.DEBUG)

Environment Variables

Variable Required Description
VERITERA_API_KEY Yes (unless passed via api_key=) Your Forge API key. Get one at forge.veritera.ai/dashboard.
OPENAI_API_KEY Yes (if using OpenAI models) Your OpenAI key for the underlying language model.

You can also pass the API key directly to avoid environment variables:

tool = ForgeVerifyTool(api_key="vt_live_...", policy="my-policy")

Other Forge Integrations

Forge works across the major agent frameworks. Use the same policies and audit trail regardless of which framework you choose.

Framework Package Install
CrewAI crewai-forge pip install crewai-forge
OpenAI Agents SDK openai-forge pip install openai-forge
LangGraph langgraph-forge pip install langgraph-forge
LlamaIndex llamaindex-forge pip install llamaindex-forge
Python SDK veritera pip install veritera
JavaScript SDK @veritera/sdk npm install veritera

Resources


License

MIT — Forge 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

agno_forge-0.1.1.tar.gz (10.7 kB view details)

Uploaded Source

Built Distribution

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

agno_forge-0.1.1-py3-none-any.whl (11.5 kB view details)

Uploaded Python 3

File details

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

File metadata

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

File hashes

Hashes for agno_forge-0.1.1.tar.gz
Algorithm Hash digest
SHA256 4af4c05a7128793fdb93ebd73186ecac925ae00ae9d77762c09f948a4dc9ae18
MD5 c479ff872eb365b7ef08b175992aae13
BLAKE2b-256 44c24f925bb1554c5c38ad51a53b6fa921546af82c5827a989f6e5c5d56d0f8d

See more details on using hashes here.

File details

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

File metadata

  • Download URL: agno_forge-0.1.1-py3-none-any.whl
  • Upload date:
  • Size: 11.5 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.1

File hashes

Hashes for agno_forge-0.1.1-py3-none-any.whl
Algorithm Hash digest
SHA256 49d4cae7a4e4138f61e41694aa591711ff067b34d2872535ecdd5f5e22de0fad
MD5 30c8554e7a7614926bdb90791df937cd
BLAKE2b-256 f0d15af6bb38e170b937d15d3319ecb26fa51a34fcc6794ba98c4bd27496bf65

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