Skip to main content

Governed AI Agent

Build an AI agent that evaluates every action through ARQERA's governance engine before executing it. Low-risk actions proceed automatically; high-risk actions are escalated or blocked.

The complete source code is available at examples/governed-agent.

Overview​

AI agents take real-world actions: sending emails, processing payments, deleting data, deploying models. Each action has different risk profiles. ARQERA evaluates every action and returns a verdict:

Agent plans action → ARQERA evaluates → PROCEED: execute
                                       → ESCALATE: queue for approval
                                       → BLOCK: reject

Prerequisites​

pip install arqera
export ARQERA_API_KEY="ak_your_key_here"

Implementation​

from arqera import ArqeraClient

client = ArqeraClient()

def execute_governed_action(action: str, description: str, **context) -> dict:
    """Execute an action only if governance allows it."""

    # Step 1: Check governance
    result = client.governance.evaluate(
        action,
        description=description,
        context=context,
    )

    # Step 2: Act on the verdict
    if result.passed:
        # Safe to execute
        execution = client.ara.execute(
            action,
            description=description,
            context=context,
        )
        return {"status": "executed", "action_id": execution.id}

    elif result.needs_approval:
        return {
            "status": "pending_approval",
            "explanation": result.explanation,
            "evaluations": [
                {"law": e.law_name, "result": e.result.value, "reason": e.reason}
                for e in result.evaluations
                if e.result.value in ("fail", "warn")
            ],
        }

    else:  # blocked
        return {
            "status": "blocked",
            "explanation": result.explanation,
        }

Task Examples​

Here is a set of tasks that demonstrate how governance handles different risk profiles:

TASKS = [
    {
        "action": "email.send",
        "description": "Send weekly status report to the team",
        "context": {"risk_level": "low", "is_irreversible": False, "recipient_count": 8},
    },
    {
        "action": "report.generate",
        "description": "Generate Q4 revenue dashboard from analytics data",
        "context": {"risk_level": "low", "is_irreversible": False},
    },
    {
        "action": "payment.process",
        "description": "Process $4,200 vendor payment to Acme Corp",
        "context": {"risk_level": "medium", "is_irreversible": True, "amount_usd": 4200},
    },
    {
        "action": "user.data.delete",
        "description": "Delete all user data for accounts inactive > 2 years",
        "context": {"risk_level": "high", "is_irreversible": True, "affected_users": 1847},
    },
    {
        "action": "model.deploy",
        "description": "Deploy updated fraud detection model to production",
        "context": {"risk_level": "high", "is_irreversible": False, "model_version": "v2.3.1"},
    },
    {
        "action": "database.drop",
        "description": "Drop staging database to free up storage",
        "context": {"risk_level": "high", "is_irreversible": True, "database": "staging_v1"},
    },
]

Expected Results​

ActionRiskIrreversibleExpected Verdict
email.sendLowNoPROCEED
report.generateLowNoPROCEED
payment.processMediumYesESCALATE
user.data.deleteHighYesBLOCK
model.deployHighNoESCALATE
database.dropHighYesBLOCK

Full Working Example​

"""Governed AI Agent — governs every action before executing."""
import os
import sys
from arqera import ArqeraClient

TASKS = [
    {"action": "email.send", "description": "Send weekly status report",
     "context": {"risk_level": "low", "is_irreversible": False}},
    {"action": "payment.process", "description": "Process $4,200 vendor payment",
     "context": {"risk_level": "medium", "is_irreversible": True, "amount_usd": 4200}},
    {"action": "user.data.delete", "description": "Delete inactive user accounts",
     "context": {"risk_level": "high", "is_irreversible": True, "affected_users": 1847}},
    {"action": "database.drop", "description": "Drop staging database",
     "context": {"risk_level": "high", "is_irreversible": True}},
]

def main():
    client = ArqeraClient(api_key=os.environ.get("ARQERA_API_KEY"))
    stats = {"proceed": 0, "escalate": 0, "block": 0}

    for task in TASKS:
        result = client.governance.evaluate(
            task["action"],
            description=task["description"],
            context=task["context"],
        )

        risk = task["context"]["risk_level"]
        print(f"\nTask: {task['action']} (risk: {risk})")
        print(f"  {result.verdict.upper()} - {result.explanation}")

        if result.passed:
            stats["proceed"] += 1
            action = client.ara.execute(
                task["action"],
                description=task["description"],
                context=task["context"],
            )
            print(f"  Executed (action_id: {action.id})")
        elif result.needs_approval:
            stats["escalate"] += 1
            print("  Queued for human approval")
        elif result.blocked:
            stats["block"] += 1
            print("  Action rejected")

    print(f"\nSummary: {stats['proceed']} approved, "
          f"{stats['escalate']} escalated, {stats['block']} blocked")
    client.close()

if __name__ == "__main__":
    main()

Run It​

cd examples/governed-agent
pip install -r requirements.txt
python main.py

Integration Patterns​

Wrapper Function​

def governed(action: str, description: str, execute_fn, **context):
    """Execute a function only if governance allows."""
    result = client.governance.evaluate(action, description=description, context=context)

    if result.passed:
        return execute_fn()
    elif result.needs_approval:
        raise PendingApproval(result.explanation)
    else:
        raise GovernanceBlocked(result.explanation)

# Usage
governed(
    "payment.process",
    "Process $4,200 payment",
    lambda: stripe.charges.create(amount=420000, currency="usd"),
    risk_level="medium",
    is_irreversible=True,
)

Batch Governance​

# Evaluate multiple actions in sequence
for task in agent_plan:
    result = client.governance.evaluate(
        task["action"],
        description=task["description"],
        context=task["context"],
    )
    if not result.passed:
        print(f"Plan blocked at step: {task['action']}")
        break
    execute(task)

Next Steps​