Verifiable Agent Infrastructure
Production infrastructure for long-running autonomous agents with verifiable execution
Aegis provides enterprise-grade infrastructure for autonomous AI agents that need to:
- 🔄 Survive Restarts — Durable state machine with checkpoint/replay
- ✅ Maintain Accountability — Verifiable commitments via GCL integration
- 🛡️ Operate Safely — Constitutional AI policies and guardrails
- 📊 Provide Transparency — Complete audit trails with causal tracing
- 🤖 Coordinate — Multi-agent protocols (Contract Net, Auctions)
In Greek mythology, the aegis was Zeus's shield—a symbol of protection and authority. Aegis brings these qualities to autonomous agents:
| Challenge | Aegis Solution |
|---|---|
| Agents lose state on restart | Durable state machine with automatic checkpointing |
| No way to verify commitments | GCL-based formal commitment verification |
| Insufficient safety guardrails | Constitutional AI principles + policy engine |
| Incomplete audit trails | Structured logging with distributed tracing |
| Agents work in isolation | Multi-agent coordination protocols |
# Clone and install
git clone https://github.com/your-org/aegis.git
cd aegis
pip install -e ".[dev]"
# Start the API server
python -m uvicorn aegis.api.server:app --host 127.0.0.1 --port 8000
# Run the demo
python examples/basic_agent.py┌─────────────────────────────────────────────────────────────────┐
│ AEGIS │
│ Verifiable Agent Infrastructure │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Web UI │ │ REST API │ │ CLI │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ └────────────────┴────────────────┘ │
│ │ │
│ ┌───────────────────────┴───────────────────────┐ │
│ │ Agent Runtime Core │ │
│ │ • Durable State Machine │ │
│ │ • Checkpoint Manager │ │
│ │ • Replay Engine │ │
│ └───────────────────────┬───────────────────────┘ │
│ │ │
│ ┌───────────────────────┴───────────────────────┐ │
│ │ Tool Gateway Layer │ │
│ │ • Policy Engine • Auth Delegator │ │
│ │ • Rate Limiter • Result Cache │ │
│ └───────────────────────┬───────────────────────┘ │
│ │ │
│ ┌───────────────────────┴───────────────────────┐ │
│ │ Commitment & Verification │ │
│ │ • GCL Bridge • Violation Detector │ │
│ │ • Recovery Orchestrator │ │
│ └───────────────────────┬───────────────────────┘ │
│ │ │
│ ┌───────────────────────┴───────────────────────┐ │
│ │ Audit & Tracing │ │
│ │ • Structured Logging • Distributed Tracing │ │
│ │ • Causal Graph Viewer │ │
│ └───────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
Agents maintain state across restarts with automatic checkpointing:
from aegis.state_machine import StateMachine
machine = StateMachine()
await machine.initialize(agent_id="agent-1")
# Process events with full durability
state = await machine.process_event(event)
# Checkpoint at any point
checkpoint_id = await machine.checkpoint()
# Restore from any checkpoint
await machine.restore(checkpoint_id)
# Branch for alternative execution paths
new_session = await machine.branch(checkpoint_id)Secure tool execution with fine-grained policies:
from aegis.tools import ToolGateway, PolicyRule, PolicyAction
gateway = ToolGateway()
# Define security policies
gateway.policy_engine.add_rule(PolicyRule(
name="no_file_delete",
action=PolicyAction.DENY,
tool_pattern="file_*",
argument_conditions={"operation": "delete"},
))
# Execute with policy enforcement
result = await gateway.invoke(
tool_name="file_write",
arguments={"path": "output.txt", "content": "Hello"},
agent_id="agent-1",
)Formal commitment tracking with verification:
from aegis.commitments import CommitmentManager
from datetime import datetime, timezone, timedelta
manager = CommitmentManager()
# Create commitment
commitment = manager.create_commitment(
debtor="agent-1",
creditor="user",
action="deliver_report",
condition="data_collected",
deadline=datetime.now(timezone.utc) + timedelta(hours=2),
)
# Verify against conditions
result = manager.verify_commitment(
commitment.id,
context={"data_collected": True, "report_ready": True},
)Built-in protocols for agent collaboration:
from aegis.multiagent import ContractNetProtocol, Task
cnp = ContractNetProtocol()
# Announce task for bidding
task = Task(
requester_id="manager",
description="Analyze dataset",
required_capability="data_analysis",
)
# Collect bids and award
winner, commitment = await cnp.announce_and_award(task)Principle-based evaluation inspired by Anthropic's approach:
from aegis.constitutional import PolicyEvaluator, CONSTITUTIONAL_PRINCIPLES
evaluator = PolicyEvaluator(principles=CONSTITUTIONAL_PRINCIPLES)
# Evaluate action against principles
result = evaluator.evaluate_action(
action="send_email",
context={"recipient": "user@example.com", "content": "..."},
)
if result.decision == PolicyDecision.DENY:
print(f"Blocked: {result.violations}")Detect and recover from commitment violations:
from aegis.recovery import RecoveryOrchestrator
orchestrator = RecoveryOrchestrator()
# Register commitment for monitoring
orchestrator.register_commitment(commitment)
# Automatic recovery on violation
result = await orchestrator.recover(violation, commitment)
print(f"Recovery: {result.strategy_used} -> {result.status}")aegis/
├── src/aegis/
│ ├── core/ # State, checkpoint, replay
│ ├── state_machine/ # Durable state machine
│ ├── tools/ # Tool gateway, policies, auth
│ ├── commitments/ # GCL integration
│ ├── gcl/ # GCL library bridge
│ ├── multiagent/ # Multi-agent coordination
│ ├── constitutional/ # Constitutional AI policies
│ ├── recovery/ # Violation detection & recovery
│ ├── audit/ # Logging and tracing
│ ├── llm/ # LLM client abstraction
│ ├── api/ # REST API (FastAPI)
│ └── cli/ # Command-line interface
├── tests/ # 303 passing tests
├── examples/ # Demo applications
├── docs/ # Documentation
└── web/ # React web UI
| Document | Description |
|---|---|
| Quickstart | Get running in 5 minutes |
| Architecture | System design and components |
| API Reference | REST API documentation |
| GCL Integration | Commitment verification guide |
| Portfolio Report | Comprehensive project overview |
# Basic agent demo
python examples/basic_agent.py
# Tool usage demo
python examples/tool_usage.py
# Checkpoint and replay demo
python examples/checkpoint_demo.py
# Multi-agent coordination demo
python examples/multiagent_demo.py
# Violation recovery demo
python examples/violation_recovery_demo.py# Run all tests (303 tests)
pytest
# Run with coverage
pytest --cov=aegis --cov-report=html
# Run specific module tests
pytest tests/unit/test_state_machine.py -v# Start development server
python -m uvicorn aegis.api.server:app --reload
# Production with multiple workers
python -m uvicorn aegis.api.server:app --workers 4
# Access API docs
open http://localhost:8000/docsEnvironment variables:
# LLM Configuration
export ANTHROPIC_API_KEY=your-api-key
export DEFAULT_MODEL=claude-sonnet-4-20250514
# Storage
export CHECKPOINT_DIR=./checkpoints
export AUDIT_LOG_DIR=./logs
# API
export API_HOST=0.0.0.0
export API_PORT=8000| Component | Technology |
|---|---|
| Language | Python 3.11+ |
| Data Validation | Pydantic |
| API Framework | FastAPI |
| CLI Framework | Click |
| Web UI | React + TypeScript |
| Testing | pytest + hypothesis |
- Durability First — All state survives restarts
- Verifiable Execution — Every action is auditable
- Policy Enforcement — Security at every layer
- Graceful Recovery — Handle failures automatically
- Extensibility — Easy to add tools, policies, strategies
- Fork the repository
- Create a feature branch
- Make your changes
- Run tests:
pytest - Submit a pull request
MIT License — see LICENSE for details.
- Inspired by Anthropic's work on Constitutional AI
- Built on the Grounded Commitment Learning (GCL) framework
- Designed for production agent deployments
Aegis — Verifiable Agent Infrastructure
Built with ❤️ for the future of autonomous agents