aegis-idempotency

aegis-idempotency

AEGIS is the reliability layer that makes AI agents safe to run in production.

Async Python SDK for the AEGIS HITL Reliability Layer — exactly-once execution and human approval for every AI agent action.

---

The problem

AI agents that take real-world actions (send an email, charge a card, modify a database) fail in non-obvious ways in production:

• Retries cause duplicates — two emails sent, two charges fired, two records created.
• No human checkpoint — an agent can act on a hallucination with no way to intercept it.
• No audit trail — when something goes wrong, you have no record of what the agent did and why.

AEGIS solves all three with one pip install.

---

Quickstart — Hosted API (free tier)

pip install aegis-idempotency

import asyncio
from aegis_hitl import AegisClient

# 1. Sign up at https://aegis-idempotency-production.up.railway.app/docs
#    POST /api/v1/accounts/signup → returns your sk_live_... API key

async def main():
    async with AegisClient(
        base_url="https://aegis-idempotency-production.up.railway.app",
        api_key="sk_live_...",   # your AEGIS API key
        thread_id="order-42",   # unique identifier for this agent run
    ) as client:

        # Idempotent — retries won't fire the action twice
        response = await client.approve(
            "idempotency-key-2026-001",
            payload={"action": "send_email", "to": "user@example.com"},
        )

        if response.status_code == 202:
            print("Action approved and queued.")
        elif response.status_code == 409:
            print("Duplicate — already processed. Safe to ignore.")

asyncio.run(main())

Free tier: 100 events/month, no credit card required. View plans →

---

Quickstart — Self-hosted

# 1. Clone and start the AEGIS server
docker compose up        # API :8000 · Redis :6379
alembic upgrade head     # DB migrations

async with AegisClient(
    base_url="http://localhost:8000",
    tenant_id="your-tenant-id",
    thread_id="order-42",
) as client:
    response = await client.approve("idempotency-key-001")

---

Three guarantees

---

API Reference

AegisClient constructor

Methods

Status codes

---

Works with any agent framework

# LangGraph node example
async def review_node(state: AgentState) -> AgentState:
    async with AegisClient(
        base_url="https://aegis-idempotency-production.up.railway.app",
        api_key=AEGIS_KEY,
        thread_id=state["run_id"],
    ) as client:
        r = await client.approve(state["idempotency_key"], payload=state["action"])
        if r.status_code != 202:
            return {**state, "status": "rejected"}
    return {**state, "status": "approved"}

Works with LangGraph, CrewAI, AutoGen, n8n, or plain Python.

---

Testing without a server

import httpx, pytest
from aegis_hitl import AegisClient

@pytest.mark.asyncio
async def test_approve():
    mock_client = httpx.AsyncClient(
        transport=httpx.MockTransport(lambda r: httpx.Response(202, json={"ok": True})),
        base_url="http://test",
    )
    async with AegisClient(api_key="sk_live_test", thread_id="th1", client=mock_client) as client:
        r = await client.approve("key-001")
    assert r.status_code == 202

---

About AEGIS

AEGIS is a production reliability layer for AI agents: exactly-once execution, human approval gates, per-event billing, Prometheus metrics, and tenant isolation — validated with 500 concurrent requests on real PostgreSQL.

• Hosted API: aegis-idempotency-production.up.railway.app/docs
• PyPI: pypi.org/project/aegis-idempotency

License

MIT — see LICENSE.