Metadata-Version: 2.4
Name: pydantic-ai-forge
Version: 0.1.1
Summary: Forge Verify tools and middleware for Pydantic AI — verify every agent action before execution
Author-email: Veritera AI <engineering@veritera.ai>
License: MIT
Project-URL: Homepage, https://forge.veritera.ai
Project-URL: Documentation, https://forge.veritera.ai/docs
Project-URL: Repository, https://github.com/VeriteraAI/pydantic-ai-forge
Keywords: veritera,forge,pydantic-ai,verification,ai-safety
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: veritera>=0.2.0
Requires-Dist: pydantic-ai>=0.1.0

# pydantic-ai-forge

[![PyPI version](https://img.shields.io/pypi/v/pydantic-ai-forge.svg)](https://pypi.org/project/pydantic-ai-forge/)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)
[![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue.svg)](https://www.python.org/downloads/)

**Forge tools and middleware for Pydantic AI -- 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 chain tool calls or operate across multi-step workflows.

---

## Install

```bash
pip install pydantic-ai-forge
```

This installs `forge_pydantic_ai` and its dependencies (`veritera` SDK and `pydantic-ai`).

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

```bash
pip install pydantic-ai-forge pydantic-ai[openai]
```

---

## Prerequisites: Create a Policy

Before using Forge with Pydantic AI, create a policy that defines what your agents are allowed to do. You only need to do this once:

```python
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", "report.generate"]}},
        {"type": "amount_limit", "params": {"max": 10000, "currency": "USD"}},
    ],
)

# Or generate one from plain English
forge.generate_policy_sync(
    "Allow payments under $10,000, balance checks, and report generation. Block all 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](https://github.com/VeriteraAI/forge-python#policies).

---

## Quick Start

```python
import os
from pydantic_ai import Agent
from forge_pydantic_ai import ForgeVerifyTool

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

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

# 2. Create your agent and register the tool
agent = Agent('openai:gpt-4o', system_prompt="You are a financial assistant. Verify actions before executing them.")
agent.tool(forge_tool.verify)

# 3. Run the agent -- it will call forge_verify before sensitive actions
result = await agent.run("Send $500 to vendor@acme.com")
print(result.data)
```

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

---

## Tutorial: Building a Verified Financial Agent

This walkthrough builds a Pydantic AI agent that handles financial operations -- with Forge verifying every action before execution.

### The Problem with Unchecked Tool Calls

Pydantic AI agents can call any tool you register. But some actions should not execute without policy checks:

- **Payments above limits** -- a tool that sends $50,000 when the policy caps single transactions at $10,000.
- **Unauthorized data access** -- an agent querying sensitive tables it should not touch.
- **Destructive operations** -- deletions, bulk exports, or admin actions triggered by a misinterpreted prompt.

Forge solves this by intercepting tool calls **before** they execute. The policy engine runs outside the agent -- the agent cannot bypass, ignore, or modify the rules.

### Step 1 -- Set Up Your Environment

```python
import os
from pydantic_ai import Agent

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

### Step 2 -- Define Your Tools

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

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

async def delete_account(ctx, account_id: str) -> str:
    """Delete a customer account permanently."""
    return f"Account {account_id} deleted"
```

### Step 3 -- Wrap Tools with Forge Verification

```python
from forge_pydantic_ai import forge_tool_wrapper

@forge_tool_wrapper(
    policy="finance-controls",
    agent_id="finance-bot",
    skip_actions=["check_balance"],  # read-only tool, always allowed
    on_blocked=lambda action, reason: print(f"BLOCKED: {action} -- {reason}"),
    on_verified=lambda action, result: print(f"APPROVED: {action}"),
)
async def send_payment(ctx, amount: float, recipient: str) -> str:
    """Send a payment to a recipient."""
    return f"Sent ${amount} to {recipient}"

@forge_tool_wrapper(policy="finance-controls", agent_id="finance-bot")
async def delete_account(ctx, account_id: str) -> str:
    """Delete a customer account permanently."""
    return f"Account {account_id} deleted"
```

### Step 4 -- Build and Run the Agent

```python
agent = Agent(
    'openai:gpt-4o',
    system_prompt="You are a financial assistant. Process transactions safely.",
    tools=[send_payment, check_balance, delete_account],
)

# Approved scenario
result = await agent.run("Check balance for account A-1001 and send $500 to vendor@acme.com")
print(result.data)

# Denied scenario
result = await agent.run("Delete account A-1001")
print(result.data)
```

### What Happens at Runtime

1. **check_balance** -- skipped (in `skip_actions`), executes normally.
2. **send_payment** -- Forge verifies `amount=500, recipient=vendor@acme.com` against `finance-controls` policy. Approved; tool executes.
3. **delete_account** -- Forge checks `account_id=A-1001`. Denied; tool returns `"Action 'delete_account' denied by Forge: Destructive account operations require manual approval"`. The agent relays the restriction to the user.

Every verification produces a `proof_id` that links to a tamper-proof audit record in your Forge dashboard.

---

## Three Integration Approaches

### 1. ForgeVerifyTool -- Explicit Verification Tool

Give the agent a tool it calls to check whether an action is allowed. The agent decides when to verify.

```python
from pydantic_ai import Agent
from forge_pydantic_ai import ForgeVerifyTool

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

agent = Agent('openai:gpt-4o')
agent.tool(forge_tool.verify)
```

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

**When to use:** When you want the agent to reason about when verification is needed.

### 2. forge_tool_wrapper -- Decorator Wrapper

Wraps any Pydantic AI tool with automatic Forge verification. The tool is verified before execution -- no agent reasoning required.

```python
from forge_pydantic_ai import forge_tool_wrapper

@forge_tool_wrapper(
    policy="finance-controls",
    agent_id="finance-bot",
    skip_actions=["check_balance"],
)
async def send_payment(ctx, amount: float, recipient: str) -> str:
    """Send a payment to a recipient."""
    return process_payment(amount, recipient)
```

**When to use:** When you want verification on specific tools without modifying the agent. Best for targeted protection on high-risk tools.

### 3. ForgeMiddleware -- Automatic Verification on All Tools

Wraps `agent.run()` to verify every tool call against Forge policies. No changes to tool definitions required.

```python
from pydantic_ai import Agent
from forge_pydantic_ai import ForgeMiddleware

agent = Agent('openai:gpt-4o', tools=[send_payment, check_balance])

middleware = ForgeMiddleware(
    policy="finance-controls",
    agent_id="finance-bot",
    skip_actions=["check_balance"],
)

# All tool calls are verified automatically
result = await middleware.run(agent, "Send $500 to vendor@acme.com")
```

**When to use:** When you want a blanket security layer across all tools. Best for production deployments where every action must be verified.

---

## 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` | `"pydantic-ai-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 prompt
    |
    v
Pydantic AI Agent decides to call a tool
    |
    v
Forge Verification Layer
    |
    +---> Is tool in skip_actions?
    |         YES --> Execute tool normally
    |         NO  --> Call Forge /v1/verify
    |                     |
    |                     +---> APPROVED
    |                     |         --> Execute tool normally
    |                     |         --> Call on_verified callback
    |                     |
    |                     +---> DENIED
    |                     |         --> Return denial message to agent
    |                     |         --> Call on_blocked callback
    |                     |         --> Tool NEVER executes
    |                     |
    |                     +---> API ERROR
    |                               --> fail_closed=True?  --> Deny
    |                               --> fail_closed=False? --> Execute tool
    v
Agent receives tool result (or denial message)
    |
    v
Agent responds to user
```

Each verification call sends the following to Forge:

- **action** -- the tool name (e.g., `"send_payment"`)
- **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.

---

## Forge Execute -- Receipt Tracking

Track every tool execution with cryptographic receipts using Forge Execute.

### ForgeExecuteHook -- Manual Receipt Emission

```python
from forge_pydantic_ai import ForgeExecuteHook

hook = ForgeExecuteHook(
    task_id="task_abc123",
    agent_id="finance-agent",
)

# Emit a receipt when a tool is used
receipt = hook.on_tool_use("payment.send")
# {"receipt_id": "rx_...", "chain_index": 42}
```

### forge_execute_wrapper -- Automatic Receipt Emission

```python
from forge_pydantic_ai import forge_execute_wrapper

@forge_execute_wrapper(task_id="task_abc123", agent_id="finance-agent")
async def send_payment(ctx, amount: float, recipient: str) -> str:
    """Send a payment to a recipient."""
    return process_payment(amount, recipient)
```

Every successful tool execution automatically emits a signed receipt to the Forge Execute chain.

---

## Combining Verify + Execute

Use both Forge Verify (pre-execution policy check) and Forge Execute (post-execution receipt) together:

```python
from forge_pydantic_ai import forge_tool_wrapper, forge_execute_wrapper

@forge_execute_wrapper(task_id="task_abc123", agent_id="finance-agent")
@forge_tool_wrapper(policy="finance-controls")
async def send_payment(ctx, amount: float, recipient: str) -> str:
    """Send a payment to a recipient."""
    return process_payment(amount, recipient)
```

The flow:
1. **Forge Verify** checks the action against your policy -- blocks if denied.
2. The tool executes (only if approved).
3. **Forge Execute** emits a signed receipt for the completed action.

---

## Links

- [Forge Documentation](https://forge.veritera.ai/docs)
- [Forge Python SDK](https://github.com/VeriteraAI/forge-python)
- [Pydantic AI Documentation](https://ai.pydantic.dev/)
- [API Reference](https://forge.veritera.ai/docs/api)
