Aegis — AI Agent Governance

import aegis
aegis.auto_instrument()

# That's it. Every LangChain, CrewAI, OpenAI Agents SDK, OpenAI API,
# and Anthropic API call in your application now passes through:
#   - Prompt injection detection (blocks attacks)
#   - Toxicity detection (blocks harmful content)
#   - PII detection (warns on personal data exposure)
#   - Prompt leak detection (warns on system prompt extraction)
#   - Full audit trail (every call logged)

AEGIS_INSTRUMENT=1 python my_agent.py

Your code                          Aegis layer (invisible)
---------                          -----------------------
chain.invoke("Hello")       -->    [input guardrails] --> LangChain --> [output guardrails] --> response
Runner.run(agent, "query")  -->    [input guardrails] --> OpenAI SDK --> [output guardrails] --> response
crew.kickoff()              -->    [task guardrails]  --> CrewAI     --> [tool guardrails]   --> response
client.chat.completions()   -->    [input guardrails] --> OpenAI API --> [output guardrails] --> response

from aegis.instrument import auto_instrument, patch_langchain, status, reset

# Instrument only specific frameworks
auto_instrument(frameworks=["langchain", "openai_agents"])

# Customize behavior
auto_instrument(
    on_block="warn",       # "raise" (default), "warn", or "log"
    guardrails="default",  # or "none" for audit-only mode
    audit=True,            # log every call
)

# Instrument a single framework
patch_langchain()

# Check what's instrumented
print(status())
# {"active": True, "frameworks": {"langchain": {"patched": True, ...}}, "guardrails": 4}

# Clean removal — restore all original methods
reset()

pip install agent-aegis

import aegis
aegis.auto_instrument()
# Every LangChain/CrewAI/OpenAI/Anthropic call is now governed.

import aegis
aegis.init()
# Discovers aegis.yaml, activates policy engine, audit logging, cost tracking.

import aegis
aegis.patch_openai()    # Only OpenAI calls
aegis.patch_anthropic() # Only Anthropic calls

# Or use the decorator for custom functions
@aegis.guard
def my_agent_function():
    ...

aegis init  # Creates aegis.yaml with sensible defaults

# aegis.yaml
guardrails:
  pii: { enabled: true, action: mask }
  injection: { enabled: true, action: block, sensitivity: medium }

policy:
  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 }
      risk_level: high
      approval: approve
    - name: no_deletes
      match: { type: "delete*" }
      risk_level: critical
      approval: block

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"),
        ])
        results = await runtime.execute(plan)

asyncio.run(main())

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

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.

runtime = Runtime(executor=MyExecutor(), policy=Policy.from_yaml("policy.yaml"))
result = await runtime.run_one(Action("read", "crm"))

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}

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

# aegis.yaml
guardrails:
  pii:
    enabled: true
    action: mask
  injection:
    enabled: true
    action: block
    sensitivity: medium

policy = Policy.from_yaml("policies/crm-agent.yaml")

pip install agent-aegis                   # Core — includes auto_instrument() for all frameworks
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

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)

from aegis.adapters.langchain import AegisMiddleware

middleware = AegisMiddleware(policy=Policy.from_yaml("policy.yaml"))
# Blocked calls return a ToolMessage explaining the policy violation
# Allowed calls proceed normally

from aegis.adapters.langchain import LangChainExecutor, AegisTool

executor = LangChainExecutor(tools=[DuckDuckGoSearchRun()])
runtime = Runtime(executor=executor, policy=Policy.from_yaml("policy.yaml"))

from agents import function_tool
from aegis import Policy
from aegis.adapters.openai_agents import (
    create_aegis_input_guardrail,
    create_aegis_output_guardrail,
)

policy = Policy.from_yaml("policy.yaml")
input_guard = create_aegis_input_guardrail(policy=policy, fail_closed=True)
output_guard = create_aegis_output_guardrail(policy=policy)

@function_tool(
    tool_input_guardrails=[input_guard],
    tool_output_guardrails=[output_guard],
)
def web_search(query: str) -> str:
    """Search the web -- Aegis evaluates before AND after execution."""
    return do_search(query)

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)

from aegis.adapters.crewai import enable_aegis_guardrail

# One line — every tool call now goes through Aegis policy
provider = enable_aegis_guardrail(runtime=my_runtime)

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))

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")

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")])

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

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"}'

pip install 'agent-aegis[mcp]'
aegis-mcp-server --policy policy.yaml

{ "mcpServers": { "aegis": { "command": "uvx", "args": ["--from", "agent-aegis[mcp]", "aegis-mcp-server"] }}}

{ "mcpServers": { "aegis": { "command": "uvx", "args": ["--from", "agent-aegis[mcp]", "aegis-mcp-server"] }}}

{ "servers": { "aegis": { "command": "uvx", "args": ["--from", "agent-aegis[mcp]", "aegis-mcp-server"] }}}

{ "mcpServers": { "aegis": { "command": "uvx", "args": ["--from", "agent-aegis[mcp]", "aegis-mcp-server"] }}}

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"

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

rules:
  - name: block_harmful_content
    match: { type: "generate*" }
    conditions:
      semantic: "contains harmful, violent, or illegal content"
    risk_level: critical
    approval: block

# Quick start
pip install 'agent-aegis[server]'
aegis serve policy.yaml

# Open http://localhost:8000

# Programmatic access
from aegis.server.app import create_app

app = create_app(
    policy_path="policy.yaml",
    audit_db_path="audit.db",
    enable_dashboard=True,
    anomaly_detector=detector,  # optional
)

from aegis.core.anomaly import AnomalyDetector

detector = AnomalyDetector()

# Feed observed actions to build per-agent behavior profiles
detector.observe(agent_id="agent-1", action_type="read", target="crm")
detector.observe(agent_id="agent-1", action_type="read", target="crm")
detector.observe(agent_id="agent-1", action_type="read", target="crm")

# Detect anomalies: rate spikes, bursts, new actions, unusual targets, high block rates
alerts = detector.check(agent_id="agent-1", action_type="delete", target="prod_db")
# => [Anomaly(type=NEW_ACTION, detail="action 'delete' never seen for agent-1")]

# Auto-generate a policy from observed behavior
learned_policy = detector.generate_policy(agent_id="agent-1")

aegis compliance --type soc2 --output report.json
aegis compliance --type gdpr --output gdpr-report.json
aegis compliance --type governance --days 30

from aegis.core.compliance import ComplianceReporter

reporter = ComplianceReporter(audit_store=runtime.audit_store)
report = await reporter.generate(report_type="soc2", days=90)

print(report.score)        # 87.5
print(report.findings)     # List of findings with severity
print(report.evidence)     # Linked audit log entries

# Show added/removed/modified rules between two policies
aegis diff policy-v1.yaml policy-v2.yaml

# Replay historical actions against the new policy to see impact
aegis diff policy-v1.yaml policy-v2.yaml --replay audit.db

 Rules: 2 added, 1 removed, 3 modified

 + bulk_write_block     CRITICAL/block   (new)
 + pii_access_approve   HIGH/approve     (new)
 - legacy_allow_all     LOW/auto         (removed)
 ~ read_safe            LOW/auto → LOW/auto  conditions changed
 ~ deploy_prod          HIGH/approve → CRITICAL/block  risk escalated
 ~ bulk_ops             MEDIUM/approve   param_gt.count: 100 → 50

 Impact (replayed 1,247 actions):
   23 actions would change from AUTO → BLOCK
    7 actions would change from APPROVE → BLOCK

from aegis.core.trust import TrustChain, AgentIdentity, Capability

# Create a root agent with full capabilities
root = AgentIdentity(
    agent_id="orchestrator",
    capabilities=[Capability("*")],  # glob matching
)

# Delegate a subset of capabilities (intersection semantics)
worker = root.delegate(
    agent_id="data-worker",
    capabilities=[Capability("read:*"), Capability("write:staging_*")],
)

# Worker can only do what both root AND delegation allow
chain = TrustChain()
chain.register(root)
chain.register(worker, parent=root)

# Verify capability at runtime
chain.can(worker, "read:crm")           # True
chain.can(worker, "delete:prod_db")     # False -- not in delegation

# Cascade revocation: revoking parent revokes all children
chain.revoke(root)
chain.can(worker, "read:crm")           # False

from aegis.core.rate_limiter import RateLimiter

limiter = RateLimiter()
limiter.add_rule("agent-*", max_requests=100, window_seconds=60)
limiter.add_rule("agent-untrusted", max_requests=10, window_seconds=60)

limiter.check("agent-untrusted", action_type="write")  # True (allowed)
# After 10 calls in 60s:
limiter.check("agent-untrusted", action_type="write")  # False (rate limited)

from aegis.core.rbac import AccessController, Role

ac = AccessController()
ac.assign_role("alice", Role.ADMIN)
ac.assign_role("bot-1", Role.OPERATOR)

ac.check("alice", "policy:write")   # True
ac.check("bot-1", "policy:write")   # False -- operators can execute, not configure
ac.check("bot-1", "action:execute") # True

from aegis.core.versioning import PolicyVersionStore

store = PolicyVersionStore("./policy-versions.json")
store.commit(policy, message="Initial production policy")
store.tag("v1.0")

# Later...
store.commit(updated_policy, message="Relax read rules")
diff = store.diff("v1.0", "HEAD")   # See what changed
store.rollback("v1.0")               # Revert to tagged version

from aegis.core.tenant import TenantContext, TenantRegistry, TenantIsolation

registry = TenantRegistry()
registry.register("acme-corp", quota={"max_actions_per_hour": 1000})

with TenantContext("acme-corp"):
    # All policy evaluations, audit writes, and rate limits
    # are automatically scoped to this tenant
    result = await runtime.run_one(action)

aegis audit --verify              # Verify chain integrity
aegis audit --export-chain        # Export full hash chain
aegis audit --evidence soc2       # Generate compliance evidence package

from aegis.core.crypto_audit import CryptoAuditLogger

logger = CryptoAuditLogger(algorithm="sha3-256")
# Each entry is hash-linked to the previous -- any tampering breaks the chain
logger.log(action, decision, result)
assert logger.verify_chain()  # True if untampered

aegis regulatory --framework eu-ai-act    # Gap analysis
aegis regulatory --framework nist-ai-rmf  # NIST mapping
aegis regulatory --all --output report.json

aegis autopolicy "block all deletes on production, allow reads, require approval for writes over $10K"

# Generated output:
version: '1'
defaults:
  risk_level: medium
  approval: approve
rules:
- name: delete_block
  match: { type: "delete*", target: "prod*" }
  risk_level: critical
  approval: block
- name: read_auto
  match: { type: "read*" }
  risk_level: low
  approval: auto

aegis probe policy.yaml

# Aegis Policy Probe — 205 probes
# ==================================================
#   Robustness score: 72/100
#   Findings: 8
#
#   CRITICAL [missing_coverage]
#     Destructive action 'drop' on 'production' is auto-approved
#     -> Add a rule to block or require approval for 'drop' actions
#
#   HIGH [glob_bypass]
#     'bulk_delete' bypasses block rule 'no_deletes' (pattern: 'delete')
#     -> Broaden the glob pattern to 'delete*' or add a rule for 'bulk_delete'

aegis scan ./src/

# Output:
# src/agents/mailer.py:42  openai.ChatCompletion.create()  -- ungoverned
# src/agents/writer.py:18  anthropic.messages.create()     -- ungoverned
# src/tools/search.py:7    langchain tool "web_search"     -- ungoverned
#
# 3 ungoverned calls found. Run `aegis score` for governance coverage.

aegis score ./src/ --policy policy.yaml

# Governance Score: 84/100
#   Governed calls:   21/25 (84%)
#   Policy coverage:  18 rules covering 6 action types
#   Anomaly detection: enabled
#   Audit trail:       enabled
#
# Badge: https://img.shields.io/badge/aegis_score-84-brightgreen

![Aegis Score](https://img.shields.io/badge/aegis_score-84-brightgreen)

aegis/
  instrument/        Auto-instrumentation — monkey-patches LangChain, CrewAI, OpenAI Agents SDK, OpenAI, Anthropic
  core/              Action, Policy engine, Conditions, Risk levels, Retry, JSON Schema
  core/anomaly       Behavioral anomaly detection -- per-agent profiling, auto-policy generation
  core/compliance    Compliance report generator -- SOC2, GDPR, governance scoring
  core/trust         Agent trust chain -- hierarchical identity, delegation, revocation
  core/semantic      Semantic conditions engine -- keyword matching + LLM evaluator protocol
  core/diff          Policy diff & impact analysis -- rule comparison, action replay
  core/rate_limiter  Per-agent/global sliding-window rate limiting
  core/rbac          Role-based access control -- 12 permissions, 5 roles, AccessController
  core/versioning    Policy version control -- commit, diff, rollback, tagging
  core/tenant        Multi-tenant isolation -- context, registry, quota enforcement
  core/crypto_audit  Cryptographic audit chain -- hash-linked tamper-evident logs
  core/replay        Action replay engine -- what-if policy analysis
  core/regulatory    EU AI Act / NIST / SOC2 / ISO 42001 compliance mapper
  core/webhooks      Webhook notifications -- Slack, PagerDuty, JSON
  core/builder       Policy-as-Code SDK -- fluent PolicyBuilder API
  core/autopolicy    Natural language -> YAML policy generation (keyword + LLM)
  core/probe         Adversarial policy testing -- gap detection, bypass attempts
  core/tiers         Enterprise tier system -- feature gating with soft nudge
  core/mcp_security  MCP supply chain security -- poisoning, rug pull, sanitization, trust scoring
  core/mcp_vuln_db   MCP vulnerability database -- CVE matching, version ranges, auto-block
  core/mcp_sbom      MCP server SBOM generation -- tool catalog, vulnerability overlay, JSON export
  core/budget        Cost circuit breaker -- 17 model pricing, loop detection, hierarchical budgets
  core/cost_callbacks  Cross-framework cost tracking -- LangChain, OpenAI, Anthropic, Google
  core/cost_attribution  Multi-agent cost attribution -- delegation trees, subtree rollup
  core/a2a_governance  Agent-to-agent communication governance -- capability gates, content filter
  core/policy_git    Policy-as-code Git integration -- diff, impact analysis, drift detection
  core/otel_export   OpenTelemetry export -- governance events → OTel spans
  core/session_replay  Session replay -- record/replay with retroactive security scanning
  guardrails/        Runtime content guardrails -- PII detection (12 categories), injection detection (10 categories, 85+ patterns)
  specs/agef/        AGEF (Agent Governance Event Format) -- JSON schema for governance events
  specs/agp/         AGP (Agent Governance Protocol) -- communication protocol for agent governance
  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 |
                     scan | score | diff | compliance | regulatory | monitor

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