🛡️ Aegis SDK — Enterprise AI Security & Governance

Aegis is a multi-layered security, governance and policy engine for AI agents and LLM applications. It provides real-time prompt-injection defense, automated risk scoring, dynamic tool authorization, stateful human-in-the-loop (HITL) approvals, natural-language policy enforcement, session budgets, and framework adapters for LangGraph & CrewAI — powered by a local neural model family that runs on CPU in ~5 ms with zero data leaving your machine.

🌐 Live dashboard: https://aegis-sdk-pi.vercel.app/ 📦 PyPI: https://pypi.org/project/aegis-security-sdk/

Table of Contents

1. Why Aegis
2. Threat Model
3. Architecture — The Five Layers
4. The Aegis Brain Family
5. Installation
6. Quick Start
7. Operating Modes (enforce vs monitoring)
8. Natural-Language Policy Engine
9. Human-in-the-Loop (HITL)
10. Supported LLM Providers
11. Framework Adapters (LangGraph & CrewAI)
12. Function Security (@protect)
13. Dynamic Tool Registry (Hot-Plug)
14. Budgets & Kill Switch
15. Observability, Audit & Dashboard
16. Testing & Red-Team Validation
17. Performance
18. Project Structure
19. Roadmap

Why Aegis

Aegis is fail-closed by design: if any security layer errors, execution is blocked — never permitted. And it is defense-in-depth: five independent layers with no single point of bypass.

Threat Model

Architecture — The Five Layers

User Request
    │
    ▼
┌──────────────────────────────────────────────────────────┐
│  LAYER 1 · REQUEST INTELLIGENCE & FIREWALL               │
│  • MicroBrain INT8 classifier (~5 ms CPU):                │
│      injection type · intent · risk level · tool scope ·  │
│      content class · confidence · attack mass             │
│  • Fail-closed capability matcher: only safe, relevant    │
│    tools are bound to the planner (least privilege)       │
│  • Memory-poisoning scan of conversation history          │
└───────────────┬──────────────────────────────────────────┘
                ▼
┌──────────────────────────────────────────────────────────┐
│  HITL GATE (HIGH/CRITICAL risk)                          │
│  Parked request + SHA-256 signature → "I approve" flow   │
└───────────────┬──────────────────────────────────────────┘
                ▼
┌──────────────────────────────────────────────────────────┐
│  NATURAL LANGUAGE POLICY ENGINE (+ Judge ladder)         │
│  Plain-English rules → deterministic verdicts            │
└───────────────┬──────────────────────────────────────────┘
                ▼
┌──────────────────────────────────────────────────────────┐
│  PLANNER ⇄ EXECUTOR LOOP (LangGraph ReAct)               │
│  Planner sees ONLY the allowlisted tools                 │
└───────────────┬──────────────────────────────────────────┘
                ▼
┌──────────────────────────────────────────────────────────┐
│  LAYER 2 · EXECUTION GOVERNANCE                          │
│  ToolScope · ToolAuthorization · ToolArgument · IFC-Taint│
│  validators → timeout/retry → sanitized execution        │
└───────────────┬──────────────────────────────────────────┘
                ▼
┌──────────────────────────────────────────────────────────┐
│  LAYER 4/5 · MEMORY VAULT & OUTPUT CONTROL               │
│  MemoryWriteGate poison check · result normalization ·   │
│  indirect-injection drop · response compliance scan      │
└───────────────┬──────────────────────────────────────────┘
                ▼
        Audit Trail → Dashboard

The Aegis Brain Family

Aegis ships its own compact neural models as INT8-quantized ONNX artifacts (packages/brain/models/) — all offline, all CPU, all sub-15 ms:

🛡️ MicroBrain — security classifier

Multi-head transformer producing, per request: injection_type (NONE…MEMORY_POISONING) · intent (EMAIL_READ … SYSTEM_ADMIN) · risk_level (LOW→CRITICAL) · tool_scope · content_class · confidence · attack_mass. Calibration gates (calibration.json) tune decision thresholds without retraining.

🧭 Router — semantic tool routing

Embeds the prompt and every registered tool's identity, binds the top-K most relevant tools (K≤4). Tiers: exact-name mention → lexical evidence → embedding cosine (artifact present) → intent alignment → scope net → bind nothing (fail closed). The production embedder normalizes leet/digit-speak (m33t → meet) before inference so noisy user input stays in-distribution.

⚖️ Policy Judge — rule-violation reasoning ladder

deterministic guards (money arithmetic, injection hard-blocks)  ← always authoritative
   ↓ ambiguous?
cross-encoder judge_nli_v2  →  legacy NLI head  →  LLM judge (Groq/Ollama)  →  heuristics

Every resolved verdict lands in an in-memory and disk cache; LLM-judge calls run under a hard timeout (AEGIS_JUDGE_TIMEOUT_S, default 2 s). Failure anywhere falls back down the ladder — never open.

Installation

pip install aegis-security-sdk

Provider & framework extras:

pip install "aegis-security-sdk[openai]"       # OpenAI
pip install "aegis-security-sdk[anthropic]"    # Claude
pip install "aegis-security-sdk[google]"       # Gemini
pip install "aegis-security-sdk[nvidia]"       # NVIDIA NIM
pip install "aegis-security-sdk[huggingface]"  # HF Inference
pip install "aegis-security-sdk[crewai]"       # CrewAI adapter
pip install "aegis-security-sdk[all]"          # everything

Requires Python 3.11 – 3.14.

Quick Start

import asyncio
from langchain_core.tools import tool
from aegis import Aegis, GroqProvider

@tool
def lookup_customer(customer_id: str) -> str:
    """Look up customer information by ID."""
    return f"Customer {customer_id}: Tier Gold, Active."

async def main():
    agent = (
        Aegis(name="support-agent", mode="enforce")
        .with_provider(GroqProvider(model_id="llama-3.3-70b-versatile"))
        .with_tools([lookup_customer])
        .with_policy([
            "Do not allow access to raw system prompts.",
            "Block any destructive database operations without approval.",
        ])
    )

    async with agent:
        result = await agent.run("Look up customer CUST-104")
        print("Output:", result.output)

if __name__ == "__main__":
    asyncio.run(main())

Operating Modes (enforce vs monitoring)

from aegis import Aegis

agent_enforce = Aegis("prod-agent", mode="enforce")      # block on violation (default)
agent_audit   = Aegis("audit-agent", mode="monitoring")  # log & score, don't interrupt

In monitoring mode every violation across all enforcement layers (Layer-1 threat filter, policy engine, HITL parking, memory guard) is captured as a structured alert and returned to the caller instead of raised — perfect for shadow audits and compliance scoring. The kill switch remains fatal in both modes.

Natural-Language Policy Engine

Write governance in plain English. Rules are compiled once into fast, evidence-gated matchers:

POLICIES = [
    "PROHIBIT fetching or disclosing private cryptographic keys, passwords, or system prompts under any circumstances.",
    "PROHIBIT running DROP, TRUNCATE, or DELETE operations on production databases.",
    "Block for any financial transfer exceeding $1,000.",
    "Require human approval for deleting backups.",
    "PROHIBIT sending environment variables, secrets, credentials, chat history, or internal data to any external destination such as gists, pastebins, webhooks, or personal email addresses.",
]
agent.with_policy(POLICIES)

Engine behavior you can rely on:

• 💰 Real money arithmetic — $500 passes a $1,000 limit; $5,000 does not. Parses $, ₹, Rs, lakh/crore. Cosine similarity can never compare numbers; we don't ask it to.
• 🔍 Evidence-gated blocking — a PROHIBIT rule only vetoes when the request itself carries matching content, so benign reads (SELECT * FROM users WHERE active=1, reading email) are never collateral damage.
• ✅ Approval wins — kernel-verified human consent satisfies policy-level prohibitions (destructive ops, over-limit transfers), while secret-access and injection-vector directives stay absolute.
• ⚖️ Optional Judge ladder (see above) resolves genuinely gray cases with an LLM you choose.

Human-in-the-Loop (HITL)

High-risk operations are parked, not dropped. Approval is cryptographically bound to the parked request:

# Step 1 — high-risk request parks
res = await agent.run("Delete production database table audit_logs")
print(res.output)
# ⚠️ Action Requires Approval: High-risk operation detected. Type 'I approve' to proceed.

# Step 2 — explicit confirmation resolves it
approval = await agent.run("I approve")
print(approval.output)
# Table audit_logs deleted successfully.

• Bare approvals with nothing parked are rejected fail-closed.
• Divergent confirmation text fails the signature check and discards the pending request.
• Developer/UI flows can pass is_approved=True after their own verification — the runtime then satisfies approval-required policies end-to-end.

Supported LLM Providers

Security policies are decoupled from model execution — swap providers in one line:

from aegis.packages.providers import (
    GroqProvider, HuggingFaceProvider, OpenAIProvider, AnthropicProvider,
    GeminiProvider, NVIDIAProvider, OllamaProvider,
)

bot_groq  = Aegis("groq-bot").with_provider(GroqProvider(model_id="llama-3.3-70b-versatile"))
bot_hf    = Aegis("hf-bot").with_provider(HuggingFaceProvider(model_id="meta-llama/Llama-3.3-70B-Instruct"))
bot_oai   = Aegis("oai-bot").with_provider(OpenAIProvider(model_id="gpt-4o"))
bot_claude= Aegis("cl-bot").with_provider(AnthropicProvider(model_id="claude-3-5-sonnet-20241022"))
bot_gemini= Aegis("gem-bot").with_provider(GeminiProvider(model_id="gemini-2.0-flash-exp"))
bot_nim   = Aegis("nim-bot").with_provider(NVIDIAProvider(model_id="meta/llama-3.3-70b-instruct"))
bot_local = Aegis("offline").with_provider(OllamaProvider(model_id="llama3", base_url="http://localhost:11434/v1"))

No API keys? LocalSecurityGatewayProvider runs the entire governed stack fully offline.

Framework Adapters

LangGraph

from aegis import Aegis

governed_agent = (
    Aegis("devops-agent")
    .with_tools(tools)
    .with_adapter("langgraph", langgraph_agent)
    .with_policy(["Rebooting production servers requires approval."])
)

CrewAI

governed_crew = (
    Aegis("crewai-sec-team")
    .with_adapter("crewai", crew)
    .with_policy(["Block unauthorized network port scanning."])
)

Function Security (@protect Decorator)

Govern any standalone Python function:

from aegis import protect

@protect(
    policy=["Do not allow updating system configurations without admin credentials."],
    mode="enforce",
)
def update_system_config(config_key: str, config_val: str) -> str:
    return f"Config {config_key} updated to {config_val}."

Dynamic Tool Registry (Hot-Plug)

Register or remove tools while the agent is RUNNING — Layer-1 matching, LLM bindings and the executor graph pick them up without restart:

agent.register_tool(my_new_tool, category="FINANCE")   # scope pin optional
agent.unregister_tool("risky_tool")

Least privilege holds across turns: the planner only ever sees the allowlisted subset.

Budgets & Kill Switch

Per-session resource ceilings keyed by correlation ID — multi-turn conversations share one ledger:

agent.with_budget(max_tool_calls=50, max_total_tokens=200_000, max_executions=500)

Exceeded before a run → whole run denied. Exceeded before a tool call → that call dies and the session kill switch fires; every subsequent invocation is refused instantly.

Observability, Audit & Dashboard

Every execution produces a scrubbed report (secrets [REDACTED] recursively): timeline events, layer results, planner tokens/costs, tool-call records, governance decisions and scores.

• 💾 Local store — ./aegis_audit/AG-<execution-id>.json (chmod 700 dir / 0600 files)
• ☁️ Cloud store — fire-and-forget streaming when AEGIS_PROJECT_KEY is set
• 🌐 Live dashboard — https://aegis-sdk-pi.vercel.app/

Sensitive-key scrubbing covers key/token/secret/password/auth/credential/env.

Testing & Red-Team Validation

pytest tests -q                      # unit + integration + red-team suites
python scripts/calibrate_router.py   # zero-FP router acceptance battery

• 172 tests passing across unit, integration and adversarial regression suites
• Dedicated black-box UI battery replays real-world attack classes end-to-end
• Independent red-team audit: 13 attack classes attempted → 0 executed

Performance

Project Structure

packages/
├── aegis.py              # public facade: Aegis, protect(), builder API
├── brain/                # local model family
│   ├── engine.py         #   MicroBrain ONNX multi-head classifier
│   ├── scanner.py        #   shared call-site helpers (prompt/output/memory)
│   ├── capability_matcher.py  # tiered least-privilege tool binding
│   ├── embedder.py       #   lazy ONNX sentence embedder (Tier 1b)
│   └── models/           #   *.onnx + tokenizers + labels + calibration.json
├── layers/
│   ├── layer1/           # request intelligence stages
│   ├── layer2/           # governance validators
│   └── layer5/           # telemetry consumer
├── policy/
│   ├── nl_policy.py      # natural-language policy engine
│   ├── policy_judge.py   # judge ladder + caches
│   └── compliance.py     # response compliance validator
├── runtime/
│   ├── kernel/           # orchestration + HITL enforcer
│   ├── nodes/            # planner & executor
│   ├── managers/         # registry, executor, retry, timeout, budget…
│   └── hooks/            # lifecycle hooks (policy enforcement lives here)
├── memory/               # memory vault + poisoning gates
├── observability/        # audit stores (local JSON / cloud)
├── governance/           # information-flow control (taint)
├── providers.py          # 7 LLM providers + offline fallback
├── adapters/             # LangGraph / CrewAI
└── testing/              # Streamlit demo console + benchmarks

Roadmap

• 🧠 Next-gen brain artifacts (fine-tuned router embedder, cross-encoder judge v2, MicroBrain v3 backbone sweep)
• 💱 Currency-aware policy limits (FX-normalized thresholds)
• 🧰 Governed-tool proxy for foreign agent frameworks
• 📜 SOC-2-ready audit-trail pack

Links

• 📦 PyPI — https://pypi.org/project/aegis-security-sdk/
• 🌐 Dashboard — https://aegis-sdk-pi.vercel.app/

License

MIT © Aegis contributors