ai-agents-warden 1.0.0.2
Agent guardrail for AI agents — guardrails and policy enforcement. Block dangerous actions, audit everything. The agent guardrail framework for OpenAI, Anthropic, LangChain, CrewAI, Google ADK, browser-use, MCP. Runtime action blocking, NLP intent detection, tamper-proof audit logs.
Requires Python: >=3.11
Keywords: agent-guardrail, agent-guardrails, guardrail, guardrails, ai-guardrail, ai-guardrails, llm-guardrail, llm-guardrails, agentgate, ai-agents-warden, ai-agents, ai-security, ai-safety, agent-security, agent-safety, agent-firewall, agent-policy, policy-engine, runtime-enforcement, mcp, mcp-gateway, mcp-server, model-context-protocol, tool-governance, agent-runtime, llm-agents, ai-automation, agent-observability, compliance, audit-trail, browser-automation, openai, anthropic, langchain, crewai, google-adk, browser-use, agentic-ai

# AgentGate — The Agent Guardrail Framework for AI

[![PyPI](https://img.shields.io/pypi/v/ai-agents-warden?color=blue&label=PyPI)](https://pypi.org/project/ai-agents-warden/)
[![Downloads](https://img.shields.io/pypi/dm/ai-agents-warden?color=green&label=Downloads)](https://pypi.org/project/ai-agents-warden/)
[![Python 3.11+](https://img.shields.io/pypi/pyversions/ai-agents-warden)](https://pypi.org/project/ai-agents-warden/)
[![License: Proprietary](https://img.shields.io/badge/License-Proprietary-red)](#licensing)

> **The #1 agent guardrail for AI agents.** Add guardrails to any AI agent in 3 lines of Python. Block dangerous actions, enforce policies, audit everything.

**AgentGate** is an agent guardrail framework that stops AI agents from doing things they shouldn't — before they happen.

Your AI agent wants to click "Pay $5,000"? **Blocked.** Call `delete_database` via MCP? **Blocked.** Export customer PII? **Requires your approval first.** Every action is recorded in a tamper-proof audit trail.

> **Not guardrails-ai.** That project validates LLM text outputs. AgentGate is an **agent guardrail** — it enforces policies on agent **actions** — clicks, tool calls, browser interactions, MCP requests. If you're searching for "agent guardrail", "AI guardrails", or "guardrail for AI agents", this is the right tool.

---

## Who Is This For?

| You are... | AgentGate helps you... |
|---|---|
| **A developer** building AI agents | Add policy enforcement to your agent in 3 lines of code |
| **A team lead** deploying agents at work | Control what agents can do, prove what they did |
| **A solo user** running OpenClaw / browser-use | Protect yourself from agent mistakes on banking, shopping, admin sites |
| **A business** using AI automation | Meet compliance requirements with tamper-proof audit trails |

---

## Install

```bash
pip install ai-agents-warden
```

Optional extras:

```bash
pip install ai-agents-warden[nlp]   # NLP intent detection (~67MB model, CPU-only, zero API keys)
pip install ai-agents-warden[mcp]   # MCP tool-call governance
pip install ai-agents-warden[all]   # Everything
```

---

## How It Works (30-Second Version)

```
Your AI Agent decides to do something
    ↓
AgentGate checks it against your policies (< 1ms)
    ↓
ALLOWED  → action proceeds normally
BLOCKED  → agent is told "no" and picks an alternative
APPROVAL → you decide whether to allow it
    ↓
Everything is recorded in a tamper-proof evidence trail
```

---

## Quick Start

### Option A: I'm a developer — give me code

**Step 1.** Start the AgentGate sidecar (one terminal):

```bash
agentgate sidecar --policy finance-safe
```

**Step 2.** In your agent code, check before every action:

```python
from agentgate.sidecar.client import AgentGateClient

gate = AgentGateClient()  # connects to localhost:18790

result = gate.check_click("https://bank.com", "Pay Now")
# result.allowed → False
# result.reason  → "Element blocked: payment action"

result = gate.check_navigate("https://bank.com/invoices")
# result.allowed → True
```

**Step 3.** That's it. Every action is now policy-checked and evidence-recorded.

### Option B: I use Playwright — wrap my browser

```python
from agentgate import SecureBrowser

async with SecureBrowser(policy="finance-safe") as browser:
    page = await browser.new_page()
    await page.goto("https://example.com")    # Allowed
    await page.click("button#pay")            # Blocked automatically
```

### Option C: I'm not using Python — REST API

```bash
# Start AgentGate
agentgate sidecar --policy finance-safe

# Check any action from any language
curl -X POST http://localhost:18790/api/v1/evaluate \
  -H "Content-Type: application/json" \
  -d '{"action_type":"click", "url":"https://bank.com", "element_text":"Pay Now"}'

# Response: {"allowed": false, "decision": "block", "reason": "Element blocked"}
```

Works from JavaScript, Go, Rust, TypeScript, C# — anything that can POST JSON.

### Option D: I use OpenClaw

```bash
pip install ai-agents-warden
agentgate sidecar --policy finance-safe
cp -r skills/agentgate ~/.openclaw/workspace/skills/
# Done. OpenClaw now checks every action through AgentGate.
```

---

## Complete Feature Reference

### 1. Policy Engine — Control what agents do

Write simple YAML rules. AgentGate enforces them on every action.

```yaml
name: finance-safe
description: Block payments, allow browsing
version: "1.0"

# Global settings
block_all_downloads: false
block_all_uploads: false
block_js_evaluation: true        # Block JavaScript execution
block_clipboard: true            # Block clipboard access
max_actions_per_minute: 60       # Rate limit (0 = unlimited)

# Domain restrictions
allowed_domains: []              # Empty = all allowed
blocked_domains:
  - malware.example.com

rules:
  # Navigation rules — glob patterns
  - action: navigate
    block:
      - pattern: "*/admin/*"
      - pattern: "*/payments/new*"

  # Click rules — keyword or intent-based matching
  - action: click
    block_elements:
      - text_contains: ["Pay", "Transfer", "Delete"]
        risk: high
      - intent: payment          # Smart NLP-powered matching
        risk: high
      - selector: "button[type='submit']"
        on_pages: "*/payments/*"
    require_approval:
      - text_contains: ["Submit", "Confirm"]
        risk: medium
      - intent: confirm
        risk: medium
    block_risk_above: high       # Auto-block anything above this risk

  # Type rules — block sensitive form fields
  - action: type
    block_elements:
      - selector: "input[type='password']"
      - intent: sensitive_field
```

#### Complete YAML Schema

| Field | Type | Description |
|---|---|---|
| `name` | `string` | **Required.** Policy name |
| `description` | `string` | Human-readable description |
| `version` | `string` | Policy version |
| `enabled` | `bool` | Whether the policy is active (default: `true`) |
| `default_decision` | `allow\|block\|require_approval\|log_only` | Default when no rule matches (default: `allow`) |
| `block_all_downloads` | `bool` | Block all file downloads |
| `block_all_uploads` | `bool` | Block all file uploads |
| `block_js_evaluation` | `bool` | Block JavaScript execution (default: `true`) |
| `block_clipboard` | `bool` | Block clipboard copy/paste |
| `max_actions_per_minute` | `int` | Rate limit, 0 = unlimited |
| `allowed_domains` | `[string]` | Whitelist domains (empty = all) |
| `blocked_domains` | `[string]` | Blacklist domains |

**Rule fields:**

| Field | Type | Description |
|---|---|---|
| `action` | `string` | Action type: `navigate`, `click`, `type`, `select`, `submit`, `download`, `upload`, `copy`, `screenshot`, `scroll`, `hover`, `drag`, `key_press`, `evaluate_js` |
| `block` | `[{pattern}]` | Block URLs matching glob patterns |
| `allow` | `[{pattern}]` | Allow URLs matching glob patterns |
| `block_elements` | `[element_rule]` | Block elements matching rules (see below) |
| `require_approval` | `[element_rule]` | Require human approval for matching elements |
| `block_risk_above` | `none\|low\|medium\|high\|critical` | Auto-block above this risk level |
| `require_approval_risk_above` | `none\|low\|medium\|high\|critical` | Auto-require approval above this risk |

**Element rule fields:**

| Field | Type | Description |
|---|---|---|
| `text_contains` | `[string]` | Match if element text contains any of these strings |
| `intent` | `string` | Match by semantic intent: `payment`, `delete`, `admin`, `confirm`, `modify`, `share`, `export`, `upload`, `messaging`, `authentication`, `sensitive_field`, `financial`, `system_destructive` |
| `selector` | `string` | CSS selector to match |
| `element_tag` | `string` | HTML tag to match |
| `element_type` | `string` | Input type to match |
| `on_pages` | `string` | Only apply on pages matching this glob |
| `risk` | `none\|low\|medium\|high\|critical` | Risk level for this rule |
| `phonetic` | `bool` | Enable phonetic matching for typos (default: `true`) |
| `semantic` | `bool` | Enable NLP semantic matching (default: `true`) |

#### Built-in Policy Presets

| Preset | Description | Rate Limit |
|---|---|---|
| `finance-safe` | Block payments/admin, approve confirmations, intent-based matching | 60/min |
| `readonly` | Block all mutations — downloads, uploads, JS, clipboard | — |
| `permissive` | Allow most actions, block only critical risk | — |
| `procurement-safe` | Block payments/delete, approve cart/quote | 120/min |
| `support-agent` | Block auth/delete/financial, approve messaging/modify | 90/min |

**Using presets in code:**

```python
from agentgate import PolicyEngine
from agentgate.policy.engine import create_finance_safe_policy

engine = PolicyEngine()
engine.load_policy(create_finance_safe_policy())
```

**Using presets from CLI:**

```bash
agentgate sidecar --policy finance-safe
agentgate sidecar --policy readonly
```

**Loading from YAML file:**

```python
from agentgate.policy.engine import Policy, PolicyEngine

# From file
policy = Policy.from_yaml("./policies/my-policy.yaml")

# From YAML string
policy = Policy.from_yaml_string(yaml_text)

# Load multiple from a directory
engine = PolicyEngine()
engine.load_policies_from_dir("./policies/")
```

**Evaluating multiple policies (most restrictive wins):**

```python
engine = PolicyEngine()
engine.load_policy(Policy.from_yaml("base.yaml"))
engine.load_policy(Policy.from_yaml("extra.yaml"))

# evaluate() checks ALL loaded policies — the most restrictive decision wins
decision = engine.evaluate(action_event)
```

**Python API — PolicyEngine:**

```python
from agentgate.policy.engine import PolicyEngine, ActionEvent, ActionType

engine = PolicyEngine()
engine.load_policy(create_finance_safe_policy())

event = ActionEvent(
    action_type=ActionType.CLICK,
    url="https://bank.com/invoices",
    element_text="Pay Now",
    selector="button#pay",
    agent_id="my-agent",
)

decision = engine.evaluate(event)
# decision.decision    → Decision.BLOCK
# decision.risk_level  → RiskLevel.HIGH
# decision.reason      → "Element blocked: payment intent detected"
# decision.policy_name → "finance-safe"
# decision.matched_rule → "click:block_element"
```

---

### 2. NLP Intent Detection — Understands meaning, not just keywords

3-layer pipeline:

1. **Keywords** — 200+ synonyms across 13 categories (payment, delete, admin, etc.)
2. **Phonetic matching** — catches typos ("Chekout" → "checkout", "Paiment" → "payment")
3. **Semantic NLP** — sentence embeddings understand novel phrases ("Process Transaction" → payment intent)

Runs entirely on CPU, ~2ms per check, zero API keys.

```python
from agentgate.policy.engine import classify_risk, ActionEvent, ActionType

event = ActionEvent(
    action_type=ActionType.CLICK,
    element_text="Process Transaction",
)

risk = classify_risk(event)
# risk → RiskLevel.HIGH (detected payment intent via NLP)
```

**13 Intent Categories:**

| Intent | Examples | Default Risk |
|---|---|---|
| `payment` | Pay, Transfer, Wire, Checkout, Process Transaction | high |
| `delete` | Delete, Remove, Purge, Destroy, Drop | high |
| `admin` | Admin, Settings, Configure, Manage Users | high |
| `authentication` | Login, Sign In, Reset Password, Change Credentials | high |
| `financial` | Invoice, Billing, Credit Card, Account Balance | high |
| `system_destructive` | Format, Wipe, Shutdown, Terminate | critical |
| `confirm` | Submit, Confirm, Approve, Accept, OK | medium |
| `modify` | Edit, Update, Change, Rename | medium |
| `share` | Share, Invite, Publish, Broadcast | medium |
| `export` | Export, Download, Extract, Backup | medium |
| `upload` | Upload, Import, Attach, Send File | medium |
| `messaging` | Send, Reply, Forward, Message | medium |
| `sensitive_field` | Password, SSN, Credit Card, Secret Key | high |

---

### 3. Evidence Vault — Tamper-Proof Audit Trail

Every action is recorded in a SHA-256 hash-chained evidence trail. Each record links to the previous via its hash — like a blockchain. If any record is tampered with, the chain breaks.

#### Where Audit Files Are Stored

```
./agentgate_evidence/                     ← Default root directory (configurable)
├── ses_a1b2c3d4e5f67890/                ← Session directory (UUID-based)
│   ├── evidence.jsonl                    ← Append-only evidence records
│   ├── summary.json                      ← Session summary (after end)
│   ├── export.json                       ← Exported evidence (on demand)
│   └── screenshots/                      ← Captured screenshots
│       ├── 000001.png
│       ├── 000002.png
│       └── ...
├── ses_f9e8d7c6b5a43210/                ← Another session (different agent)
│   ├── evidence.jsonl
│   ├── summary.json
│   └── screenshots/
└── ...
```

**Key storage details:**

| Aspect | Detail |
|---|---|
| **Default directory** | `./agentgate_evidence` (relative to working dir) |
| **Configurable via** | CLI: `--evidence-dir`, Python: `evidence_dir=` parameter |
| **Session ID format** | `ses_` + 16 random hex chars (UUID4) — e.g. `ses_a1b2c3d4e5f67890` |
| **Evidence format** | JSONL (one JSON record per line, append-only) |
| **Screenshots** | PNG files, named by sequence number (`000001.png`, `000002.png`, …) |
| **Summary format** | JSON with statistics, written on session end |
| **Sensitive data** | Auto-redacted: passwords, credit cards, tokens, SSNs |

#### Multi-Agent Audit Isolation (No Clashing!)

AgentGate is designed for concurrent multi-agent use. Audit files **never clash** between agents:

- **Each session gets its own UUID directory** — `ses_{uuid4().hex[:16]}` = 16 hex chars = 2⁶⁴ possible values = zero collision risk
- **Each agent gets its own session** — even in multi-agent orchestrations, every agent has a separate `EvidenceSession` with its own directory
- **Append-only JSONL per session** — each session writes to its own `evidence.jsonl` inside its own isolated directory. No shared files between agents
- **Async-safe writes** — records are written via `aiofiles` with immediate flush. Multiple async agents can record simultaneously without race conditions
- **Sidecar auto-creates sessions** — when the sidecar REST API receives a request with a new `agent_id`, it automatically creates a fresh session directory for that agent
- **Hash-chain per session** — each session maintains its own independent SHA-256 chain starting from `"genesis"`, so chains never interfere

```python
# Multi-agent example — each agent gets isolated evidence
from agentgate.sessions import MultiAgentSessionManager, AgentRole

manager = MultiAgentSessionManager(evidence_dir="./agentgate_evidence")
manager.register_agent("researcher", role=AgentRole.READER)
manager.register_agent("writer",     role=AgentRole.WRITER)
manager.register_agent("deployer",   role=AgentRole.EXECUTOR)

orch = await manager.start_orchestration("crewai-task-42")

# Each agent now has its own evidence session directory:
#   ./agentgate_evidence/ses_<uuid1>/   ← researcher's audit trail
#   ./agentgate_evidence/ses_<uuid2>/   ← writer's audit trail
#   ./agentgate_evidence/ses_<uuid3>/   ← deployer's audit trail
#
# They never overlap. Evaluations are recorded to the correct session automatically.
```

#### Evidence Record Contents

Each record in `evidence.jsonl` contains:

```json
{
  "record_id": "a1b2c3d4e5f6",
  "sequence": 1,
  "session_id": "ses_a1b2c3d4e5f67890",
  "agent_id": "my-agent",
  "timestamp": 1739808000.123,
  "timestamp_iso": "2026-02-17T12:00:00.123Z",
  "action_type": "click",
  "url": "https://bank.com/invoices",
  "selector": "button#pay",
  "element_text": "Pay Now",
  "value_redacted": "",
  "decision": "block",
  "risk_level": "high",
  "matched_rule": "click:block_element",
  "reason": "Element blocked: payment intent detected",
  "policy_name": "finance-safe",
  "dom_snapshot_hash": "",
  "screenshot_hash": "sha256:...",
  "previous_hash": "genesis",
  "record_hash": "sha256:..."
}
```

#### Python API — EvidenceVault

```python
from agentgate.evidence.vault import EvidenceVault

vault = EvidenceVault("./agentgate_evidence")

# Start a session
session = await vault.start_session(agent_id="my-agent")

# Record actions (called automatically by SecureBrowser / sidecar)
record = await session.record(
    event=action_event,
    decision=policy_decision,
    screenshot=screenshot_bytes,    # Optional — saved as PNG
    dom_snapshot=html_string,       # Optional — hashed for integrity
)

# End session — writes summary.json
summary = await session.end()
# summary.total_actions, .actions_blocked, .chain_valid, etc.

# Later — retrieve and verify
records = await vault.get_session_records("ses_abc123")
is_valid = await vault.verify_session("ses_abc123")
summary = await vault.get_session_summary("ses_abc123")
all_sessions = await vault.list_sessions()
export_path = await vault.export_session("ses_abc123", output_path="./report.json")
await vault.delete_session("ses_abc123")  # Use with caution
```

#### CLI — Sessions

```bash
agentgate sessions list                                    # List all sessions
agentgate sessions list --dir ./my_evidence                # Custom evidence directory
agentgate sessions verify ses_a1b2c3d4e5f67890             # Verify chain integrity
agentgate sessions replay ses_a1b2c3d4e5f67890             # Replay in terminal
agentgate sessions export ses_a1b2c3d4e5f67890             # Export as JSON
agentgate sessions export ses_a1b2c3d4e5f67890 -o report.json  # Export to file
```

---

### 4. Sidecar REST API — Full Reference

Start the sidecar:

```bash
agentgate sidecar --policy finance-safe --port 18790 --evidence-dir ./agentgate_evidence
```

All CLI options:

| Option | Default | Description |
|---|---|---|
| `--port` | `18790` | Sidecar port |
| `--host` | `127.0.0.1` | Bind host (localhost only by default) |
| `--policy` | `None` | Policy preset name or YAML file path |
| `--policy-dir` | `None` | Directory of YAML policy files |
| `--evidence-dir` | `./agentgate_evidence` | Where audit files are stored |
| `--agent-id` | `openclaw` | Default agent ID for requests |

#### Endpoints

| Method | Path | Description |
|---|---|---|
| `GET` | `/api/v1/health` | Health check — version, uptime, session/policy counts |
| `POST` | `/api/v1/evaluate` | **Core** — evaluate an action against policies |
| `POST` | `/api/v1/evaluate/batch` | Batch evaluate multiple actions at once |
| `POST` | `/api/v1/record` | Manually record an action in the evidence vault |
| `POST` | `/api/v1/sessions` | Start a new evidence session |
| `GET` | `/api/v1/sessions` | List all sessions |
| `GET` | `/api/v1/sessions/{id}` | Get session details |
| `POST` | `/api/v1/sessions/{id}/end` | End a session, save evidence |
| `GET` | `/api/v1/sessions/{id}/records` | Get all evidence records |
| `GET` | `/api/v1/sessions/{id}/verify` | Verify chain integrity |
| `POST` | `/api/v1/tokens` | Issue a capability token |
| `POST` | `/api/v1/tokens/validate` | Validate a token against an action |
| `POST` | `/api/v1/tokens/{id}/revoke` | Revoke a token |
| `GET` | `/api/v1/tokens` | List all active tokens |
| `GET` | `/api/v1/policies` | List loaded policies |
| `POST` | `/api/v1/policies/load` | Load a policy from YAML string |
| `GET` | `/api/v1/license/status` | Get license status |
| `POST` | `/api/v1/license/activate` | Activate a license key |
| `POST` | `/api/v1/heartbeat` | Heartbeat ping (for monitoring) |
| `GET` | `/api/v1/heartbeat/status` | Heartbeat status |

#### Evaluate Request/Response

**Request (`POST /api/v1/evaluate`):**

```json
{
  "action_type": "click",
  "url": "https://bank.com/invoices",
  "selector": "button#pay",
  "element_text": "Pay Now",
  "element_tag": "button",
  "element_type": "submit",
  "element_id": "pay",
  "element_classes": ["btn", "btn-primary"],
  "value": "",
  "agent_id": "my-agent",
  "session_id": "",
  "token_id": null,
  "metadata": {}
}
```

**Response:**

```json
{
  "allowed": false,
  "decision": "block",
  "risk_level": "high",
  "reason": "Element blocked: payment intent detected",
  "policy_name": "finance-safe",
  "matched_rule": "click:block_element",
  "action_fingerprint": "a1b2c3d4...",
  "record_id": "e5f6g7h8i9j0",
  "timestamp": 1739808000.123
}
```

**Valid `action_type` values:** `navigate`, `click`, `type`, `select`, `submit`, `download`, `upload`, `copy`, `screenshot`, `scroll`, `hover`, `drag`, `key_press`, `evaluate_js`

#### Batch Evaluate

```bash
curl -X POST http://localhost:18790/api/v1/evaluate/batch \
  -H "Content-Type: application/json" \
  -d '[
    {"action_type":"navigate", "url":"https://bank.com/admin"},
    {"action_type":"click", "url":"https://bank.com", "element_text":"Pay"},
    {"action_type":"click", "url":"https://bank.com", "element_text":"View"}
  ]'
# Returns: array of evaluate responses
```

#### Issue Token

```bash
curl -X POST http://localhost:18790/api/v1/tokens \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "my-agent",
    "issuer": "admin",
    "purpose": "invoice review",
    "ttl_minutes": 30,
    "domains": ["bank.com"],
    "url_patterns": ["*/invoices/*"],
    "exclude_text": ["Pay", "Delete"],
    "max_actions_total": 50,
    "max_risk": "medium"
  }'
```

#### Load Policy at Runtime

```bash
curl -X POST http://localhost:18790/api/v1/policies/load \
  -H "Content-Type: application/json" \
  -d '{"yaml": "name: temp-block\nrules:\n  - action: click\n    block_risk_above: low"}'
```

---

### 5. Python SDK — AgentGateClient

#### Synchronous Client

```python
from agentgate.sidecar.client import AgentGateClient

# Create client
gate = AgentGateClient(
    host="127.0.0.1",       # Sidecar host
    port=18790,              # Sidecar port
    agent_id="my-agent",     # Your agent's ID
    timeout=5.0,             # HTTP timeout in seconds
    token_id=None,           # Optional capability token
)

# Context manager
with AgentGateClient() as gate:
    ...

# Convenience methods — each returns GateDecision
gate.check_navigate("https://bank.com/admin")
gate.check_click("https://bank.com", "Pay Now", selector="button#pay")
gate.check_type("https://bank.com", "input#amount", "5000")
gate.check_submit("https://bank.com", "Submit Payment")
gate.check_download("https://bank.com/statements.pdf")
gate.check_upload("https://bank.com/documents")
gate.check_js("https://bank.com", "document.cookie")

# Generic evaluate (any action type)
result = gate.evaluate("drag", url="https://app.com", element_text="Slider")

# GateDecision fields:
#   .allowed           → bool
#   .decision          → "allow" | "block" | "require_approval" | "log_only"
#   .risk_level        → "none" | "low" | "medium" | "high" | "critical"
#   .reason            → str
#   .policy_name       → str
#   .matched_rule      → str
#   .action_fingerprint → str
#   .record_id         → str (evidence record ID)

# Token management
token = gate.issue_token(
    purpose="invoice review",
    ttl_minutes=30,
    domains=["bank.com"],
    exclude_text=["Pay", "Delete"],
    max_actions=50,
)
# token.token_id, token.expires_at, token.remaining_ttl

gate.revoke_token(token.token_id, reason="task complete")

# Session management
session_id = gate.start_session(purpose="invoice audit")
gate.end_session(session_id)
sessions = gate.list_sessions()
is_valid = gate.verify_session(session_id)

# Health
health = gate.health()
# health.status, .version, .uptime_seconds, .active_sessions, .policies_loaded

gate.is_running()  # True/False
```

#### Async Client

```python
from agentgate.sidecar.client import AsyncAgentGateClient

async with AsyncAgentGateClient(agent_id="async-agent") as gate:
    result = await gate.check_click("https://bank.com", "Pay Now")
    result = await gate.check_navigate("https://bank.com/admin")
    result = await gate.check_type("https://bank.com", "#amount", "5000")
    result = await gate.check_js("https://bank.com", "alert('hi')")

    health = await gate.health()
    is_up = await gate.is_running()
```

---

### 6. SecureBrowser — Playwright Integration

Wraps Playwright with automatic policy enforcement on every browser action.

```python
from agentgate import SecureBrowser
from agentgate.interceptor.browser import ActionBlockedError

async with SecureBrowser(
    policy="finance-safe",         # Preset name or Policy object
    policy_dir="./policies/",      # Also load all YAMLs from this dir
    evidence_dir="./agentgate_evidence",
    agent_id="browser-agent",
    headless=True,
    capture_screenshots=True,      # Auto-capture screenshots for evidence
    capture_dom=False,             # Capture DOM snapshots
    browser_type="chromium",       # "chromium", "firefox", or "webkit"
    on_block=lambda d: print(f"BLOCKED: {d.reason}"),
    on_approval_needed=lambda d: print(f"NEEDS APPROVAL: {d.reason}"),
) as browser:
    page = await browser.new_page()

    # These are policy-enforced automatically:
    await page.goto("https://example.com")        # navigate → checked
    await page.click("button#view")               # click → checked
    await page.fill("input#search", "invoices")   # type → checked
    await page.type("input#field", "hello")        # type → checked
    await page.press("Enter")                      # key_press → checked
    await page.select_option("select#filter", "A") # select → checked
    await page.set_input_files("input#upload", f)  # upload → checked
    content = await page.evaluate("document.title") # evaluate_js → checked
    await page.screenshot(path="shot.png")          # screenshot → checked
    await page.mouse.wheel(0, 300)                  # scroll → checked

    # Safe read-only methods (NOT intercepted):
    title = await page.title()
    url = page.url
    content = await page.content()
    element = await page.query_selector("div#info")
    elements = await page.query_selector_all("a")
    text = await page.text_content("h1")
    attr = await page.get_attribute("a", "href")
    visible = await page.is_visible("button#pay")
    checked = await page.is_checked("input#agree")

    # Handle blocked actions:
    try:
        await page.click("button#pay")
    except ActionBlockedError as e:
        print(f"Blocked: {e}")
        print(f"Risk: {e.decision.risk_level}")

# After close, summaries are auto-saved to evidence dir
```

---

### 7. Capability Tokens — Scoped, Time-Limited Agent Permissions

Give agents limited, auditable permissions that auto-expire.

```python
from agentgate.tokens.capability import CapabilityToken, Scope, TokenStore
from agentgate.policy.engine import ActionType, RiskLevel

# Create a token
token = CapabilityToken.create(
    agent_id="finance-bot",
    issuer="admin@company.com",
    purpose="Q4 invoice review",
    scopes=[
        Scope(
            domains=["bank.com", "accounting.company.com"],     # Only these domains
            exclude_text=["Pay", "Transfer", "Delete"],         # Block these elements
            exclude_urls=["*/admin/*", "*/settings/*"],         # Block these URLs
            max_risk=RiskLevel.MEDIUM,                          # Max risk allowed
        ),
        Scope(
            action=ActionType.NAVIGATE,                         # Scope per action type
            url_patterns=["*/invoices/*", "*/reports/*"],       # Only these URLs
        ),
        Scope(
            read_only=True,                                     # No mutations at all
        ),
    ],
    ttl_minutes=30,              # Auto-expires in 30 minutes
    max_actions_total=100,       # Max 100 actions then expires
    max_actions_per_minute=20,   # Rate limit within token
    signing_key="your-secret",   # HMAC signing key
)

# Validate against an action
result = token.validate(action_event)
# result.valid               → bool
# result.reason              → str
# result.token_id            → str
# result.remaining_ttl_seconds → float

# Use the token (increment action counter)
token.use()

# Check state
token.is_expired       # bool
token.remaining_ttl    # float (seconds)
token.actions_used     # int
token.revoked          # bool

# Revoke
token.revoke(by="admin", reason="task complete")

# Signature verification
token.verify_signature("your-secret")  # True/False

# Serialize
json_str = token.to_json()
token = CapabilityToken.from_json(json_str)

# Token Store — manage multiple tokens
store = TokenStore()
store.store(token)
store.get("tok_abc123")
store.get_for_agent("finance-bot")
store.revoke("tok_abc123", by="admin", reason="done")
store.revoke_all_for_agent("finance-bot")
store.cleanup_expired()
store.active_count     # int
store.total_count      # int
```

**Via REST API:**

```bash
# Issue token
curl -X POST http://localhost:18790/api/v1/tokens \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "my-agent",
    "purpose": "invoice review",
    "ttl_minutes": 30,
    "domains": ["bank.com"],
    "exclude_text": ["Pay", "Delete"],
    "max_actions_total": 50
  }'

# Validate token
curl -X POST http://localhost:18790/api/v1/tokens/validate \
  -d '{"token_id":"tok_abc123","action_type":"click","url":"https://bank.com","element_text":"View"}'

# Revoke token
curl -X POST http://localhost:18790/api/v1/tokens/tok_abc123/revoke?reason=done

# List all tokens
curl http://localhost:18790/api/v1/tokens
```

---

### 8. MCP Gateway — Guard Tool Calls

Intercept and policy-check MCP tool calls to Stripe, GitHub, databases, etc.

#### MCP Server — Expose AgentGate as MCP Tools

```bash
# stdio transport (for Claude Desktop, Cursor, etc.)
agentgate mcp serve

# HTTP transport
agentgate mcp serve --transport streamable-http --port 8100

# With policies
agentgate mcp serve --policy finance-safe --mcp-policy mcp-finance-safe
```

**Exposed MCP tools:**

| Tool | Description |
|---|---|
| `evaluate_action` | Check if a browser action is allowed |
| `evaluate_mcp_tool` | Check if an MCP tool call is allowed |
| `classify_text_risk` | Classify risk level of any text |
| `list_policies` | List loaded browser + MCP policies |
| `health` | Health check |

#### MCP Proxy — Sits Between Your Agent and MCP Servers

```bash
agentgate mcp proxy --config mcp_servers.yaml --mcp-policy mcp-finance-safe
```

**Config file (`mcp_servers.yaml`):**

```yaml
servers:
  - name: stripe
    command: npx
    args: ["-y", "@stripe/mcp"]
  - name: github
    url: http://localhost:8200/mcp
    transport: streamable-http
  - name: database
    command: python
    args: ["-m", "my_db_mcp"]
```

**MCP Policy YAML:**

```yaml
name: mcp-finance-safe
description: Block payment and billing tools
version: "1.0"

default_decision: allow
max_calls_per_minute: 30
max_arg_length: 10000

blocked_servers: []
allowed_servers: []
blocked_tools:
  - "stripe/create_payment*"
  - "stripe/create_refund*"
  - "*/delete_*"
  - "*/drop_*"
allowed_tools: []

tool_rules:
  - server_pattern: "stripe"
    tool_pattern: "create_*|update_*|delete_*"
    intent: payment
    risk: high
    decision: block
    reason: "Payment operations blocked"

  - server_pattern: "*"
    tool_pattern: "delete_*|drop_*|truncate_*"
    intent: delete
    risk: critical
    decision: block
    reason: "Destructive operations blocked"

  - server_pattern: "database"
    tool_pattern: "execute_query"
    arg_patterns:
      query: "DROP|DELETE|TRUNCATE|ALTER"
    risk: critical
    decision: block
    reason: "Destructive SQL blocked"
```

**MCP Policy Presets:**

| Preset | Description |
|---|---|
| `mcp-default` | Block dangerous tools, allow rest (60/min) |
| `mcp-readonly` | Only allow read/get/list/search/find/fetch/query/describe/count tools |
| `mcp-finance-safe` | Block payment/billing tools (30/min) |
| `mcp-devops-safe` | Block destructive infra ops, require approval for deploy/rollback |

**Pre-check a tool call from CLI:**

```bash
agentgate mcp check stripe create_payment
agentgate mcp check github delete_repo --policy mcp-readonly
agentgate mcp check database drop_table --args '{"table":"users"}'
agentgate mcp policies   # list all MCP presets
```

**Python API — MCPPolicyEngine:**

```python
from agentgate.mcp.policy_engine import MCPPolicyEngine, MCP_PRESET_POLICIES
from agentgate.mcp.types import MCPToolCall

engine = MCPPolicyEngine()
engine.load_policy(MCP_PRESET_POLICIES["mcp-finance-safe"]())

tool_call = MCPToolCall(
    server_name="stripe",
    tool_name="create_payment",
    arguments={"amount": 5000, "currency": "usd"},
    agent_id="my-agent",
)

decision = engine.evaluate(tool_call)
# decision.allowed       → False
# decision.decision      → "block"
# decision.risk_level    → "high"
# decision.reason        → "Payment operations blocked"
# decision.server_name   → "stripe"
# decision.tool_name     → "create_payment"
# decision.tool_fingerprint → "sha256:..."
```

---

### 9. MCP Plugin System — Extend the Pipeline

Add custom hooks that run before, after, or around MCP tool evaluations.

#### Plugin Pipeline (execution order)

```
1. Transform plugins    → modify the tool call (sanitize args, normalize)
2. Validator plugins    → check arguments (SQL injection, PCI data, etc.)
3. Pre-evaluate hooks   → custom pre-checks (business rules, schedules)
4. Core engine          → policy YAML evaluation
5. Post-evaluate hooks  → override decisions (escalation, exceptions)
6. Audit plugins        → logging/alerting (cannot change decisions)
```

#### Plugin Types

| Type | Purpose | Can Block? | Can Modify? |
|---|---|---|---|
| `PolicyPlugin` | Pre/post evaluation hooks | ✅ | ✅ |
| `TransformPlugin` | Modify tool calls before evaluation | ❌ | ✅ args |
| `ValidatorPlugin` | Check arguments for dangerous patterns | ✅ | ❌ |
| `AuditPlugin` | Logging/alerting after decisions | ❌ | ❌ |

#### Writing Custom Plugins

```python
from agentgate.mcp.plugin import (
    PolicyPlugin, TransformPlugin, AuditPlugin, ValidatorPlugin,
    PluginDecision, PluginContext, PluginPriority, PluginRegistry,
)

# Custom policy plugin — block tool calls on weekends
class NoWeekendsPlugin(PolicyPlugin):
    name = "no-weekends"
    description = "Block all tool calls on weekends"
    priority = PluginPriority.NORMAL

    def pre_evaluate(self, tool_call, context):
        import datetime
        if datetime.datetime.now().weekday() >= 5:
            return PluginDecision(
                block=True,
                reason="Tool calls are blocked on weekends",
            )
        return None  # No opinion — pass through

# Custom validator — PCI compliance
class PCIComplianceValidator(ValidatorPlugin):
    name = "pci-compliance"
    priority = PluginPriority.CRITICAL

    def validate(self, tool_call, context):
        import re
        for key, val in tool_call.arguments.items():
            if re.search(r'\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}', str(val)):
                return PluginDecision(
                    block=True,
                    reason=f"PCI: credit card data in argument '{key}'",
                    risk_override="critical",
                )
        return None

# Custom audit plugin — webhook alerts
class WebhookAuditPlugin(AuditPlugin):
    name = "slack-alerts"

    def on_decision(self, tool_call, decision, context):
        if not decision.allowed:
            # Send to Slack, PagerDuty, etc.
            requests.post(WEBHOOK_URL, json={
                "text": f"🚫 Blocked: {tool_call.tool_name} — {decision.reason}"
            })

# Custom transform plugin — sanitize arguments
class SanitizeArgsTransform(TransformPlugin):
    name = "sanitize-args"

    def transform(self, tool_call, context):
        # Strip whitespace from all string arguments
        for key, val in tool_call.arguments.items():
            if isinstance(val, str):
                tool_call.arguments[key] = val.strip()
        return tool_call

# Register plugins
registry = PluginRegistry()
registry.register(NoWeekendsPlugin())
registry.register(PCIComplianceValidator())
registry.register(WebhookAuditPlugin())
registry.register(SanitizeArgsTransform())

# Use with MCP engine
engine = MCPPolicyEngine(plugin_registry=registry)

# Plugin management
registry.unregister("no-weekends")
registry.get_plugin("pci-compliance")
registry.list_plugins()    # List all with metadata
registry.plugin_count      # int
registry.clear()           # Remove all
```

#### Built-in Plugins

| Plugin | Type | Priority | Description |
|---|---|---|---|
| `sql-injection-guard` | Validator | CRITICAL | Detects SQL injection patterns (DROP, DELETE, UNION SELECT, etc.) |
| `sensitive-data-guard` | Validator | CRITICAL | Detects credit cards, SSNs, API keys in arguments |
| `per-server-rate-limit` | Policy | HIGH | Per-server rate limiting (customizable limits per server) |
| `decision-logger` | Audit | AUDIT | Logs all MCP decisions with timestamps |
| `arg-normalizer` | Transform | NORMAL | Strips whitespace, normalizes argument keys |

```python
# Quick: get a secure registry with all built-in plugins
from agentgate.mcp.plugin import create_secure_registry

registry = create_secure_registry()
# Includes: sql-injection-guard, sensitive-data-guard, decision-logger, arg-normalizer

engine = MCPPolicyEngine(plugin_registry=registry)
```

#### Plugin Priority Order

| Priority | Value | Use For |
|---|---|---|
| `CRITICAL` | 0 | Security checks (run first) |
| `HIGH` | 10 | Important business rules |
| `NORMAL` | 50 | Standard plugins |
| `LOW` | 90 | Optional enhancements |
| `AUDIT` | 100 | Logging/alerting (run last) |

---

### 10. Multi-Agent Sessions — Govern Agent Teams

When multiple agents work together (CrewAI, LangGraph, AutoGen), AgentGate provides unified governance with per-agent isolation.

#### Agent Roles

| Role | Allowed Actions | Blocked Actions | Rate |
|---|---|---|---|
| `READER` | navigate, scroll, hover, screenshot | click, type, submit, download, upload, JS, copy, drag | 60/min |
| `WRITER` | all except JS | evaluate_js | 30/min |
| `ADMIN` | all | none | unlimited |
| `EXECUTOR` | click, type, submit, navigate, select, scroll | evaluate_js, download, upload | 45/min |
| `OBSERVER` | screenshot, scroll | everything else | 30/min |
| `CUSTOM` | all (configure manually) | none | unlimited |

#### Usage

```python
from agentgate.sessions import MultiAgentSessionManager, AgentRole, AgentStatus

manager = MultiAgentSessionManager(
    evidence_dir="./agentgate_evidence",
    default_policy="finance-safe",           # Optional default policy
)

# Register agents with roles
manager.register_agent(
    agent_id="researcher",
    role=AgentRole.READER,
    display_name="Research Bot",
    max_actions_per_minute=30,               # Override role default rate
    allowed_action_types=["navigate", "screenshot"],  # Override role defaults
    blocked_action_types=["click"],
    metadata={"team": "finance"},
)
manager.register_agent("writer", role=AgentRole.WRITER)
manager.register_agent("admin-bot", role=AgentRole.ADMIN)

# Start orchestrated session
orch = await manager.start_orchestration(
    name="quarterly-report",
    max_total_actions=500,       # Max actions across ALL agents combined
    collective_rate_limit=100,   # Max actions/min across ALL agents combined
    require_sequential=False,    # Allow parallel (True = one at a time)
)

# Evaluate actions with 7-step pipeline:
#   1. Registration check    — is the agent registered?
#   2. Role restrictions     — allowed/blocked action types for role
#   3. Per-agent rate limit  — individual rate limiting
#   4. Collective rate limit — total across all agents
#   5. Max total actions     — hard cap across orchestration
#   6. Sequential lock       — if require_sequential is True
#   7. Policy engine         — YAML policy eval