agentguard-tech 0.10.0

Runtime security for AI agents — policy engine, audit trail, and kill switch

agentguard-tech

Runtime security for AI agents — policy engine, audit trail, and kill switch.

Overview

AgentGuard gives AI agents production-grade guardrails:

• 🛡️ Policy evaluation — check every tool call before execution
• 📋 Audit trail — tamper-evident hash chain of every action
• 🔴 Kill switch — instantly halt all agents
• 🔍 Audit verification — cryptographically verify the audit chain
• ⚡ Zero dependencies — pure Python stdlib, works anywhere

---

Installation

pip install agentguard-tech

Requires Python 3.8+. No external dependencies.

---

Quick Start

from agentguard import AgentGuard

guard = AgentGuard(api_key="ag_your_api_key")

# Evaluate an agent action before executing it
decision = guard.evaluate(
    tool="send_email",
    params={"to": "[e-mail]", "subject": "Hello"}
)

if decision["result"] == "allow":
    print("Action allowed, risk score:", decision["riskScore"])
    # proceed with tool execution
elif decision["result"] == "block":
    print("Action blocked:", decision["reason"])
elif decision["result"] == "require_approval":
    print("Waiting for human approval...")
elif decision["result"] == "monitor":
    print("Action monitored (allowed but logged):", decision["reason"])

---

API Reference

AgentGuard(api_key, base_url=...)

Create a client instance.

guard = AgentGuard(
    api_key="ag_your_api_key",
    base_url="https://api.agentguard.tech"  # optional, default shown
)

---

evaluate(tool, params=None) → dict

Evaluate a tool call against your policy. Call this before every tool execution.

decision = guard.evaluate("read_file", {"path": "/data/report.csv"})
# Returns:
# {
#   "result": "allow",          # allow | block | monitor | require_approval
#   "riskScore": 5,             # 0-1000
#   "reason": "Matched allow-read rule",
#   "durationMs": 1.2,
#   "matchedRuleId": "allow-read"  # optional
# }

Integration pattern:

def safe_tool_call(tool_name, tool_func, **params):
    decision = guard.evaluate(tool_name, params)
    if decision["result"] in ("allow", "monitor"):
        return tool_func(**params)
    elif decision["result"] == "block":
        raise PermissionError(f"Blocked by policy: {decision['reason']}")
    elif decision["result"] == "require_approval":
        raise PermissionError("Awaiting human approval")

---

get_usage() → dict

Get usage statistics for your tenant.

usage = guard.get_usage()
print(usage)
# {
#   "requestsToday": 142,
#   "requestsThisMonth": 3891,
#   "plan": "pro",
#   "limits": { "requestsPerDay": 10000 }
# }

---

get_audit(limit=50, offset=0) → dict

Get audit trail events with pagination.

audit = guard.get_audit(limit=100, offset=0)
for event in audit["events"]:
    print(f"{event['timestamp']} | {event['tool']} | {event['decision']}")

---

kill_switch(active) → dict

Activate or deactivate the global kill switch.

# Emergency halt — stop all agents immediately
guard.kill_switch(True)

# Resume operations
guard.kill_switch(False)

---

verify_audit() → dict

Verify the cryptographic integrity of the audit hash chain.

result = guard.verify_audit()
if result["valid"]:
    print("Audit chain is intact")
else:
    print(f"Chain broken at event index: {result['invalidAt']}")

---

create_webhook(url, events, secret=None) → dict

Register a webhook endpoint to receive AgentGuard events.

webhook = guard.create_webhook(
    url="https://example.com/hooks/agentguard",
    events=["action.blocked", "killswitch.activated"],
    secret="my-signing-secret",  # optional
)
print("Webhook ID:", webhook["id"])

---

list_webhooks() → dict

List all webhook subscriptions for your tenant.

result = guard.list_webhooks()
for wh in result["webhooks"]:
    print(wh["id"], wh["url"])

---

delete_webhook(webhook_id) → dict

Delete a webhook subscription.

guard.delete_webhook("wh_abc123")

---

create_agent(name, policy_scope=None) → dict

Register a new agent with AgentGuard.

agent = guard.create_agent(
    name="email-agent",
    policy_scope={"allowedTools": ["send_email", "read_inbox"]},  # optional
)
print("Agent ID:", agent["id"])

---

list_agents() → dict

List all registered agents for your tenant.

result = guard.list_agents()
for a in result["agents"]:
    print(a["id"], a["name"])

---

delete_agent(agent_id) → dict

Delete a registered agent.

guard.delete_agent("ag_abc123")

---

list_templates() → dict

List all available policy templates.

result = guard.list_templates()
for t in result["templates"]:
    print(t["name"], t["description"])

---

get_template(name) → dict

Get a specific policy template by name.

template = guard.get_template("strict")
print(template["rules"])

---

apply_template(name) → dict

Apply a policy template to your tenant.

guard.apply_template("strict")

---

set_rate_limit(window_seconds, max_requests, agent_id=None) → dict

Create a rate limit rule.

# Tenant-wide: max 100 requests per 60 seconds
limit = guard.set_rate_limit(window_seconds=60, max_requests=100)

# Scoped to a specific agent
guard.set_rate_limit(window_seconds=60, max_requests=20, agent_id="ag_abc123")

---

list_rate_limits() → dict

List all rate limit rules for your tenant.

result = guard.list_rate_limits()
for rl in result["rateLimits"]:
    print(rl["id"], rl["windowSeconds"], rl["maxRequests"])

---

delete_rate_limit(limit_id) → dict

Delete a rate limit rule.

guard.delete_rate_limit("rl_abc123")

---

get_cost_summary(agent_id=None, from_date=None, to_date=None, group_by=None) → dict

Get a cost summary for your tenant with optional filters.

# Overall summary
summary = guard.get_cost_summary()

# Filtered by agent and date range
summary = guard.get_cost_summary(
    agent_id="ag_abc123",
    from_date="2024-01-01",
    to_date="2024-01-31",
    group_by="day",
)
print(summary)

---

get_agent_costs() → dict

Get per-agent cost breakdown for your tenant.

costs = guard.get_agent_costs()
for entry in costs["agents"]:
    print(entry["agentId"], entry["totalCost"])

---

get_dashboard_stats() → dict

Get high-level dashboard statistics.

stats = guard.get_dashboard_stats()
print(stats["requestsToday"], stats["blocksToday"])

---

get_dashboard_feed(since=None) → dict

Get the live activity feed for the dashboard.

# All recent events
feed = guard.get_dashboard_feed()

# Only events after a specific timestamp
feed = guard.get_dashboard_feed(since="2024-06-01T00:00:00Z")

for event in feed["events"]:
    print(event["timestamp"], event["type"])

---

get_agent_activity() → dict

Get per-agent activity summary.

activity = guard.get_agent_activity()
for a in activity["agents"]:
    print(a["agentId"], a["requests"], a["blocks"])

---

Framework Integrations

One-liner guards for popular Python AI agent frameworks.

LangChain

from agentguard.integrations import langchain_guard

handler = langchain_guard(api_key="ag_...")

# Pass as a callback to AgentExecutor — every tool call is evaluated automatically:
executor = AgentExecutor.from_agent_and_tools(
    agent=agent,
    tools=tools,
    callbacks=[handler],
)

Blocked tool calls raise AgentGuardBlockError:

from agentguard.integrations import langchain_guard, AgentGuardBlockError

handler = langchain_guard(api_key="ag_...", agent_id="my-agent")
try:
    handler.on_tool_start({"name": "exec"}, '{"cmd": "rm -rf /"}')
except AgentGuardBlockError as e:
    print(f"Blocked: {e.reason}")
    print(f"Try instead: {e.alternatives}")
    print(f"Docs: {e.docs}")

---

OpenAI

from openai import OpenAI
from agentguard.integrations import openai_guard

client = OpenAI(api_key="sk-...")
guarded = openai_guard(client, api_key="ag_...")

# Use `guarded` exactly like the regular OpenAI client:
response = guarded.chat.completions.create(
    model="gpt-4o",
    messages=[...],
    tools=[...],
)

# Check which tool calls were blocked before executing them:
if response._agentguard and response._agentguard["has_blocks"]:
    for decision in response._agentguard["decisions"]:
        if decision["decision"] == "block":
            print(f"Blocked {decision['tool']}: {decision['reason']}")
            print(f"Suggestion: {decision['suggestion']}")

---

CrewAI

from agentguard.integrations import crewai_guard, AgentGuardBlockError

guard = crewai_guard(api_key="ag_...")

# In your agent tool execution hook — raises on block:
try:
    guard.before_tool_execution("send_email", {"to": "[e-mail]"})
    # Safe to execute if we reach here
    result = send_email(to="[e-mail]")
except AgentGuardBlockError as e:
    print(f"Blocked: {e.reason}")
    print(f"Suggestion: {e.suggestion}")

# Batch evaluate (no raise — caller handles blocks):
results = guard.evaluate_batch([
    {"tool": "file_read", "args": {"path": "/data/report.csv"}},
    {"tool": "send_email", "args": {"to": "[e-mail]"}},
])
blocked = [r for r in results if r["decision"] == "block"]
print(f"{len(blocked)}/{len(results)} tool calls blocked")

---

Batch Evaluate

Evaluate multiple tool calls with a single API call:

from agentguard import AgentGuard

guard = AgentGuard(api_key="ag_...")

# The core client supports batch evaluation directly:
from agentguard.integrations import crewai_guard

batch_guard = crewai_guard(api_key="ag_...")
results = batch_guard.evaluate_batch([
    {"tool": "file_read",  "args": {"path": "/data/report.csv"}},
    {"tool": "send_email", "args": {"to": "[e-mail]"}},
    {"tool": "exec",       "args": {"cmd": "rm -rf /"}},
])

for r in results:
    status = "✅" if r["decision"] == "allow" else "🚫"
    print(f"{status} {r['tool']}: {r['decision']}")
    if r["decision"] == "block":
        print(f"   Reason: {r['reason']}")

---

Complete Example — LangChain-style Agent

from agentguard import AgentGuard

guard = AgentGuard(api_key="ag_your_api_key")

def run_tool(name: str, func, **params):
    """Execute a tool with AgentGuard policy enforcement."""
    decision = guard.evaluate(name, params)
    
    result = decision["result"]
    if result == "block":
        raise PermissionError(f"Policy blocked {name}: {decision['reason']}")
    if result == "require_approval":
        raise PermissionError(f"Human approval required for {name}")
    
    # "allow" or "monitor" — proceed
    return func(**params)


# Your tools
def send_email(to: str, subject: str, body: str) -> str:
    # ... send the email
    return f"Email sent to {to}"

def read_file(path: str) -> str:
    with open(path) as f:
        return f.read()


# Use with policy enforcement
content = run_tool("read_file", read_file, path="/data/report.csv")
run_tool("send_email", send_email, to="[e-mail]", subject="Report", body=content)

---

Error Handling

from agentguard import AgentGuard

guard = AgentGuard(api_key="ag_your_key")

try:
    decision = guard.evaluate("dangerous_tool", {"target": "production_db"})
except RuntimeError as e:
    print(f"API error: {e}")
    # RuntimeError: AgentGuard API error: 401 Unauthorized

---

Links

• 🌐 agentguard.tech
• 🎮 Live Demo
• 📦 GitHub
• 📘 npm SDK

License

Business Source License 1.1