agno-forge

agno-forge

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:

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

---

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 returned — APPROVED (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

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.

---

Resources

• Forge Documentation
• Forge Dashboard
• Policy Configuration Guide
• Agno Documentation

---

License

MIT — Forge by Veritera AI