Aegis
The simplest way to govern AI agent actions. No infra. No lock-in. Just Python.
pip install agent-aegis → YAML policy → governance in 5 minutes. Works with LangChain, CrewAI, OpenAI, Anthropic, MCP, and more.
Try it Live in Your Browser • Quick Start • How It Works • Documentation • Integrations • Contributing
English • 한국어
The Problem
AI agents are getting real-world access. Without governance, a hallucinating agent can:
- Bulk-delete your CRM contacts
- Submit wrong forms to government portals
- Trigger irreversible API calls at 3am
- Run up cloud bills with infinite loops
There's no sudo for AI agents. Until now.
The Solution
Aegis is a Python middleware that sits between your AI agent and the actions it takes. It's not a separate server you have to run -- you import it directly into your agent code and it wraps every action with policy checks, approval gates, and audit logging.
Your Agent Aegis Real World
| | |
|-- "delete all users" ---> | |
| [Policy check] |
| risk=CRITICAL |
| approval=BLOCK |
| |--- X (blocked, logged) -----> |
| | |
|-- "read contacts" ------> | |
| [Policy check] |
| risk=LOW |
| approval=AUTO |
| |--- execute (logged) --------> |
| | |
|-- "bulk update 500" ----> | |
| [Policy check] |
| risk=HIGH |
| approval=APPROVE |
| |--- ask human (Slack/CLI) ---> |
| |<-- "approved" --------------- |
| |--- execute (logged) --------> |
Copy, paste, run — zero config needed:
from aegis import Action, Policy
policy = Policy.from_dict({
"version": "1",
"defaults": {"risk_level": "low", "approval": "auto"},
"rules": [{"name": "block_delete", "match": {"type": "delete_*"},
"risk_level": "critical", "approval": "block"}]
})
safe = policy.evaluate(Action(type="read_users", target="db"))
print(safe.approval) # Approval.AUTO ✅
danger = policy.evaluate(Action(type="delete_users", target="db"))
print(danger.approval) # Approval.BLOCK 🚫
Or with a YAML file — 3 lines:
from aegis import Action, Policy, Runtime
runtime = Runtime(executor=your_executor, policy=Policy.from_yaml("policy.yaml"))
results = await runtime.run_one(Action("write", "salesforce", params={...}))
No servers to deploy. No Kubernetes. No vendor lock-in. One pip install, one YAML file, and your agent has policy checks, human approval gates, and a full audit trail — across any AI provider.
How It Works
Core Concepts
Aegis has 3 key components. You need to understand these to use it:
| Concept | What it is | Your responsibility |
|---|---|---|
| Policy | YAML rules that define what's allowed, what needs approval, and what's blocked. | Write the rules. |
| Executor | The adapter that actually does things (calls APIs, clicks buttons, runs queries). | Provide one, or use a built-in adapter. |
| Runtime | The engine that connects Policy + Executor. Evaluates rules, gates approval, executes, logs. | Create it. Call run_one() or plan() + execute(). |
The Pipeline
Every action goes through 5 stages. This happens automatically -- you just call runtime.run_one(action):
1. EVALUATE Your action is matched against policy rules (glob patterns).
→ PolicyDecision: risk level + approval requirement + matched rule
2. APPROVE Based on the decision:
- auto: proceed immediately (low-risk actions)
- approve: ask a human via CLI, Slack, Discord, Telegram, webhook, or email
- block: reject immediately (dangerous actions)
3. EXECUTE The Executor carries out the action.
Built-in: Playwright (browser), httpx (HTTP), LangChain, CrewAI, OpenAI, Anthropic, MCP
Custom: extend BaseExecutor (10 lines)
4. VERIFY Optional post-execution check (override executor.verify()).
5. AUDIT Every decision and result is logged to SQLite automatically.
Export: JSONL, webhook, or query via CLI/API.
Two Ways to Use
Option A: Python library (most common) -- no server needed.
Import Aegis into your agent code. Everything runs in the same process.
runtime = Runtime(executor=MyExecutor(), policy=Policy.from_yaml("policy.yaml"))
result = await runtime.run_one(Action("read", "crm"))
Option B: REST API server -- for non-Python agents (Go, TypeScript, etc.).
pip install 'agent-aegis[server]'
aegis serve policy.yaml --port 8000
curl -X POST localhost:8000/api/v1/evaluate \
-d '{"action_type": "delete", "target": "db"}'
# => {"risk_level": "CRITICAL", "approval": "block", "is_allowed": false}
Approval Handlers
When a policy rule requires approval: approve, Aegis asks a human. You choose how:
| Handler | How it works | Status |
|---|---|---|
| CLI (default) | Terminal Y/N prompt | Stable |
| Slack | Posts Block Kit message, polls thread replies | Stable |
| Discord | Sends rich embed, polls callback | Stable |
| Telegram | Inline keyboard buttons, polls getUpdates | Stable |
| Webhook | POSTs to any URL, reads response | Stable |
| Sends approval request via SMTP, polls mailbox | Beta | |
| Auto | Approves everything (for testing / server mode) | Stable |
| Custom | Extend ApprovalHandler with your own logic |
Stable |
Audit Trail
Every action is automatically logged to a local SQLite database. No setup required.
aegis audit # View all entries
aegis audit --risk-level HIGH # Filter by risk
aegis audit --tail # Live monitoring (1s poll)
aegis stats # Statistics per rule
aegis audit --format jsonl -o export.jsonl # Export
Quick Start
pip install agent-aegis
1. Generate a policy
aegis init # Creates policy.yaml with sensible defaults
# policy.yaml
version: "1"
defaults:
risk_level: medium
approval: approve
rules:
- name: read_safe
match: { type: "read*" }
risk_level: low
approval: auto
- name: bulk_ops_need_approval
match: { type: "bulk_*" }
conditions:
param_gt: { count: 100 } # Only when count > 100
risk_level: high
approval: approve
- name: no_deletes
match: { type: "delete*" }
risk_level: critical
approval: block
2. Add to your agent
import asyncio
from aegis import Action, Policy, Runtime
from aegis.adapters.base import BaseExecutor
from aegis.core.result import Result, ResultStatus
class MyExecutor(BaseExecutor):
async def execute(self, action):
print(f" Executing: {action.type} -> {action.target}")
return Result(action=action, status=ResultStatus.SUCCESS)
async def main():
async with Runtime(
executor=MyExecutor(),
policy=Policy.from_yaml("policy.yaml"),
) as runtime:
plan = runtime.plan([
Action("read", "crm", description="Fetch contacts"),
Action("bulk_update", "crm", params={"count": 150}),
Action("delete", "crm", description="Drop table"),
])
print(plan.summary())
results = await runtime.execute(plan)
asyncio.run(main())
3. See what happened
aegis audit
ID Session Action Target Risk Decision Result
1 a1b2c3d4... read crm LOW auto success
2 a1b2c3d4... bulk_update crm HIGH approved success
3 a1b2c3d4... delete crm CRITICAL block blocked
Features
| Feature | Description |
|---|---|
| YAML policies | Glob matching, first-match-wins, JSON Schema for validation |
| Smart conditions | time_after, time_before, weekdays, param_gt/lt/eq/contains/matches |
| 4-tier risk model | low / medium / high / critical with per-rule overrides |
| Approval gates | CLI, Slack, Discord, Telegram, email, webhook, or custom |
| Audit trail | SQLite, JSONL export, Python logging, or webhook to external SIEM |
| REST API server | aegis serve policy.yaml -- govern from any language via HTTP |
| MCP adapter | Govern Model Context Protocol tool calls |
| Retry & rollback | Exponential backoff, error filters, automatic rollback on failure |
| Dry-run & simulate | Test policies without executing: aegis simulate policy.yaml read:crm |
| Hot-reload | runtime.update_policy(...) -- swap policies without restart |
| Policy merge | Policy.from_yaml_files("base.yaml", "prod.yaml") -- layer configs |
| Runtime hooks | Async callbacks for on_decision, on_approval, on_execute |
| Type-safe | Full mypy --strict compliance, py.typed marker |
| 9 policy templates | Pre-built for CRM, code, finance, browser, DevOps, healthcare, and more |
| Interactive playground | Try in browser -- no install needed |
| Docker ready | examples/docker/ -- deploy REST API in one command |
Real-World Use Cases
| Scenario | Policy | Outcome |
|---|---|---|
| Finance | Block bulk transfers > $10K without CFO approval | Agents can process invoices safely; large amounts trigger Slack approval |
| SaaS Ops | Auto-approve reads; require approval for account mutations | Support agents handle tickets without accidentally deleting accounts |
| DevOps | Allow deploys Mon-Fri 9-5; block after hours | CI/CD agents can't push to prod at 3am |
| Data Pipeline | Block DELETE on production tables; auto-approve staging | ETL agents can't drop prod data, even if the LLM hallucinates |
| Compliance | Log every external API call with full context | Auditors get a complete trail for SOC2 / GDPR evidence |
Policy Templates
Pre-built YAML policies for common industries. Copy one, customize it, deploy:
| Template | Use Case | Key Rules |
|---|---|---|
crm-agent.yaml |
Salesforce, HubSpot, CRM | Read=auto, Write=approve, Delete=block |
code-agent.yaml |
Cursor, Copilot, Aider | Read=auto, Shell=high, Deploy=block |
financial-agent.yaml |
Payments, invoicing | View=auto, Payments=approve, Transfers=critical |
browser-agent.yaml |
Playwright, Selenium | Navigate=auto, Click=approve, JS eval=block |
data-pipeline.yaml |
ETL, database ops | SELECT=auto, INSERT=approve, DROP=block |
devops-agent.yaml |
CI/CD, infrastructure | Monitor=auto, Deploy=approve, Destroy=block |
healthcare-agent.yaml |
Healthcare, HIPAA | Search=auto, PHI=approve, Delete=block |
ecommerce-agent.yaml |
Online stores | View=auto, Refund=approve, Delete=block |
support-agent.yaml |
Customer support | Read=auto, Respond=approve, Delete=block |
policy = Policy.from_yaml("policies/crm-agent.yaml")
Production Ready
| Aspect | Detail |
|---|---|
| 518 tests, 92% coverage | Every adapter, handler, and edge case tested |
| Type-safe | mypy --strict with zero errors, py.typed marker |
| Performance | Policy evaluation < 1ms; auto-approved actions add < 5ms overhead |
| Fail-safe | Blocked actions never execute; can't be bypassed without policy change |
| Audit immutability | Results are frozen dataclasses; audit writes happen before returning |
| No magic | Pure Python, no monkey-patching, no global state |
Compliance & Audit
Aegis audit trails provide evidence for regulatory and internal compliance:
| Standard | What Aegis provides |
|---|---|
| SOC2 | Immutable audit log of every agent action, decision, and approval |
| GDPR | Data access documentation -- who/what accessed which system and when |
| HIPAA | PHI access trail with full action context and approval chain |
| Internal | Change management evidence, risk assessment per action |
Export as JSONL, query via CLI/API, or stream to external SIEM via webhook. For defense-in-depth with container isolation, see the Security Model guide.
Integrations
Works with the agent frameworks you already use:
pip install langchain-aegis # LangChain (standalone integration)
pip install 'agent-aegis[langchain]' # LangChain (adapter)
pip install 'agent-aegis[crewai]' # CrewAI
pip install 'agent-aegis[openai-agents]' # OpenAI Agents SDK
pip install 'agent-aegis[anthropic]' # Anthropic Claude
pip install 'agent-aegis[httpx]' # Webhook approval/audit
pip install 'agent-aegis[playwright]' # Browser automation
pip install 'agent-aegis[server]' # REST API server
pip install 'agent-aegis[all]' # Everything
LangChain -- govern any LangChain tool with one function call
Option A: langchain-aegis (recommended) — standalone integration package
pip install langchain-aegis
from langchain_aegis import govern_tools
# Add governance to existing tools — no other code changes
governed = govern_tools(tools, policy="policy.yaml")
agent = create_react_agent(model, governed)
Option B: agent-aegis[langchain] — adapter-based
from aegis.adapters.langchain import LangChainExecutor, AegisTool
# Wrap existing LangChain tools with governance
executor = LangChainExecutor(tools=[DuckDuckGoSearchRun()])
runtime = Runtime(executor=executor, policy=Policy.from_yaml("policy.yaml"))
# Or expose governed actions AS LangChain tools
tool = AegisTool.from_runtime(runtime, name="governed_search",
description="Policy-governed search", action_type="search", action_target="web")
OpenAI Agents SDK -- decorator-based governance
from aegis.adapters.openai_agents import governed_tool
@governed_tool(runtime=runtime, action_type="write", action_target="crm")
async def update_contact(name: str, email: str) -> str:
"""Update a CRM contact -- governed by Aegis policy."""
return await crm.update(name=name, email=email)
CrewAI -- governed tools for crews
from aegis.adapters.crewai import AegisCrewAITool
tool = AegisCrewAITool(runtime=runtime, name="governed_search",
description="Search with governance", action_type="search",
action_target="web", fn=lambda query: do_search(query))
Anthropic Claude -- govern tool_use calls
from aegis.adapters.anthropic import govern_tool_call
for block in response.content:
if block.type == "tool_use":
result = await govern_tool_call(
runtime=runtime, tool_name=block.name,
tool_input=block.input, target="my_system")
httpx -- governed REST API calls
from aegis.adapters.httpx_adapter import HttpxExecutor
executor = HttpxExecutor(base_url="https://api.example.com",
default_headers={"Authorization": "Bearer ..."})
runtime = Runtime(executor=executor, policy=Policy.from_yaml("policy.yaml"))
# Action types map to HTTP methods: get, post, put, patch, delete
plan = runtime.plan([Action("get", "/users"), Action("delete", "/users/1")])
MCP (Model Context Protocol) -- govern any MCP tool call
from aegis.adapters.mcp import govern_mcp_tool_call, AegisMCPToolFilter
# Option 1: Govern individual tool calls
result = await govern_mcp_tool_call(
runtime=runtime, tool_name="read_file",
arguments={"path": "/data.csv"}, server_name="filesystem")
# Option 2: Filter-based governance
tool_filter = AegisMCPToolFilter(runtime=runtime)
result = await tool_filter.check(server="filesystem", tool="delete_file")
if result.ok:
# Proceed with actual MCP call
pass
REST API -- govern from any language
pip install 'agent-aegis[server]'
aegis serve policy.yaml --port 8000
# Evaluate an action (dry-run)
curl -X POST http://localhost:8000/api/v1/evaluate \
-H "Content-Type: application/json" \
-d '{"action_type": "delete", "target": "db"}'
# => {"risk_level": "CRITICAL", "approval": "block", "is_allowed": false}
# Execute through full governance pipeline
curl -X POST http://localhost:8000/api/v1/execute \
-H "Content-Type: application/json" \
-d '{"action_type": "read", "target": "crm"}'
# Query audit log
curl http://localhost:8000/api/v1/audit?action_type=delete
# Hot-reload policy
curl -X PUT http://localhost:8000/api/v1/policy \
-H "Content-Type: application/json" \
-d '{"yaml": "rules:\n - name: block_all\n match: {type: \"*\"}\n approval: block"}'
Custom adapters -- 10 lines to integrate anything
from aegis.adapters.base import BaseExecutor
from aegis.core.action import Action
from aegis.core.result import Result, ResultStatus
class MyAPIExecutor(BaseExecutor):
async def execute(self, action: Action) -> Result:
response = await my_api.call(action.type, action.target, **action.params)
return Result(action=action, status=ResultStatus.SUCCESS, data=response)
async def verify(self, action: Action, result: Result) -> bool:
return result.data.get("status") == "ok"
Policy Conditions
Go beyond glob matching with smart conditions:
rules:
# Block writes after business hours
- name: after_hours_block
match: { type: "write*" }
conditions:
time_after: "18:00"
risk_level: critical
approval: block
# Escalate bulk operations over threshold
- name: large_bulk_ops
match: { type: "update*" }
conditions:
param_gt: { count: 100 }
risk_level: high
approval: approve
# Only allow deploys on weekdays
- name: weekday_deploys
match: { type: "deploy*" }
conditions:
weekdays: [1, 2, 3, 4, 5]
risk_level: medium
approval: approve
Available: time_after, time_before, weekdays, param_eq, param_gt, param_lt, param_gte, param_lte, param_contains, param_matches (regex).
Architecture
aegis/
core/ Action, Policy engine, Conditions, Risk levels, Retry, JSON Schema
adapters/ BaseExecutor, Playwright, httpx, LangChain, CrewAI, OpenAI, Anthropic, MCP
runtime/ Runtime engine, ApprovalHandler, AuditLogger (SQLite/JSONL/webhook/logging)
server/ REST API (Starlette ASGI) -- evaluate, execute, audit, policy endpoints
cli/ aegis validate | audit | schema | init | simulate | serve | stats
Why Aegis?
There are many ways to add governance to AI agents. Here's how they compare:
vs. Writing Your Own
| DIY | Aegis | |
|---|---|---|
| Policy engine | Custom if/else per action | YAML rules + glob + conditions |
| Risk model | Hardcoded | 4-tier with per-rule overrides |
| Human approval | Build your own | Pluggable (CLI, Slack, Discord, Telegram, email, webhook) |
| Audit trail | printf debugging | SQLite + JSONL + session tracking |
| Framework support | Rewrite per framework | 7 adapters out of the box |
| Retry & rollback | DIY error handling | Exponential backoff + automatic rollback |
| Type safety | Maybe | mypy strict, py.typed |
| Time to integrate | Days | Minutes |
vs. Platform-Native Guardrails
OpenAI, Google, and Anthropic each ship built-in guardrails — but they only govern their own ecosystem. If your agent calls OpenAI and Anthropic, or uses LangChain and MCP tools, you need one governance layer that works across all of them. That's Aegis.
vs. Enterprise Governance Platforms
Enterprise platforms like centralized control planes need Kubernetes clusters, cloud infrastructure, and procurement cycles. Aegis is a library — pip install and you have governance in 5 minutes. Start with a library, graduate to a platform when you need to.
CLI
aegis init # Generate starter policy
aegis validate policy.yaml # Validate policy syntax
aegis schema # Print JSON Schema (for editor autocomplete)
aegis simulate policy.yaml read:crm delete:db # Test policies without executing
aegis audit # View audit log
aegis audit --session abc --format json # Filter + format
aegis audit --tail # Live monitoring
aegis audit --format jsonl -o export.jsonl # Export
aegis stats # Policy rule statistics
aegis serve policy.yaml --port 8000 # Start REST API server
Roadmap
| Version | Status | Features |
|---|---|---|
| 0.1 | Released | Policy engine, 7 adapters (incl. MCP), CLI, audit (SQLite + JSONL + webhook), conditions, JSON Schema |
| 0.1.3 | Released | REST API server, retry/rollback, dry-run, hot-reload, policy merge, Slack/Discord/Telegram/email approval, simulate CLI, runtime hooks, stats, live tail |
| 0.1.4 | Released | Multi-agent foundations (agent_id, PolicyHierarchy, conflict detection), performance optimizations (compiled globs, batch audit, eval cache), security hardening, MCP/LangChain/CrewAI/OpenAI cookbooks |
| 0.2 | Q2 2026 | Dashboard UI, rate limiting, queue-based async execution |
| 0.3 | Q3 2026 | Agent identity (agent_id in actions), policy hierarchy (org → team → agent), conflict detection |
| 0.4 | Q4 2026 | Multi-agent governance (delegation, chain tracing), centralized policy server, cross-agent audit correlation |
| 1.0 | 2027 | Distributed governance, policy versioning & rollback, multi-tenant REST API |
Contributing
We welcome contributions! Check out:
- Good First Issues -- great starting points
- Contributing Guide -- setup, code style, PR process
- Architecture -- how the codebase is structured
git clone https://github.com/Acacian/aegis.git && cd aegis
make dev # Install deps + hooks
make test # Run tests
make lint # Lint + format check
make coverage # Coverage report
Or jump straight into a cloud environment:
Badge
Using Aegis? Add a badge to your project:
[](https://github.com/Acacian/aegis)
License
MIT -- see LICENSE for details.
Copyright (c) 2026 구동하 (Dongha Koo, @Acacian). Created March 21, 2026.
Built for the era of autonomous AI agents. If Aegis helps you, consider giving it a star -- it helps others find it too.