aegis

pip install agent-aegis
aegis scan .

Scanning . for ungoverned AI calls...

  src/agent.py:12     openai.ChatCompletion.create()    NO GUARDRAIL
  src/agent.py:34     langchain.ChatOpenAI.invoke()      NO GUARDRAIL
  src/tools.py:8      anthropic.messages.create()        NO GUARDRAIL
  src/pipeline.py:21  crew.kickoff()                     NO GUARDRAIL

  4 ungoverned AI calls found in 3 files.
  Run `aegis.auto_instrument()` to add guardrails, or create a policy with `aegis init`.

- uses: Acacian/aegis@v0.9.1
  with:
    command: scan
    fail-on-ungoverned: true

import aegis
aegis.auto_instrument()

# That's it. Every LangChain, CrewAI, OpenAI, Anthropic, LiteLLM,
# Google GenAI, Pydantic AI, LlamaIndex, Instructor, and DSPy
# call in your application now passes through:
#   - Prompt injection detection (blocks attacks)
#   - PII detection (warns on personal data exposure)
#   - Prompt leak detection (warns on system prompt extraction)
#   - Toxicity detection (warns — opt-in to block)
#   - 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()

from aegis.core import ActionClaim, ClaimAssessor, DeclaredFields

# Agent declares what it intends to do (untrusted)
claim = ActionClaim(
    declared=DeclaredFields(
        proposed_transition="delete_records",
        target="production_db",
        justification="cleanup old data",
    )
)

# Aegis independently assesses actual impact (6-dimensional)
assessor = ClaimAssessor()
result = assessor.assess(claim)
# result.verdict -> BLOCK
# result.assessed.justification_gap -> 0.385 (agent under-reported impact)

aegis plan current.yaml proposed.yaml --audit-db aegis_audit.db

# Policy Impact Analysis
# =====================
#   Rules: 2 added, 1 removed, 3 modified
#   Impact (replayed 1,247 actions):
#     23 actions would change from AUTO → BLOCK
#      7 actions would change from APPROVE → BLOCK
#
#   CI mode: aegis plan current.yaml proposed.yaml --ci  (exit 1 if breaking)

# Auto-generate test suite from policy
aegis test policy.yaml --generate --generate-output tests.yaml

# Run in CI — exit 1 on failure
aegis test policy.yaml tests.yaml

# Regression test between old and new policy
aegis test new-policy.yaml tests.yaml --regression old-policy.yaml

# .github/workflows/policy-check.yml
- uses: Acacian/aegis@main
  with:
    policy: aegis.yaml
    tests: tests.yaml
    fail-on-regression: true

pip install agent-aegis

import aegis
aegis.auto_instrument()
# All 11 supported frameworks are now governed.

import aegis
aegis.auto_instrument()
# 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