On this page
Installation
pip install aegis-os
Initialisation
Use the client as a context manager — the session is automatically closed when the block exits. For long-lived services, instantiate once and call client.close() on shutdown.
from aegis_os import AegisClient import os # Context manager — recommended for scripts and agents with AegisClient(api_key=os.environ["AEGIS_API_KEY"]) as aegis: # all calls go here ... # Long-lived service pattern aegis = AegisClient( api_key=os.environ["AEGIS_API_KEY"], timeout=10, # seconds )
Submitting intents
Call aegis.intents.submit() before any AI-driven payment. Returns immediately with the pending intent — decision is asynchronous.
with AegisClient(api_key=os.environ["AEGIS_API_KEY"]) as aegis: intent = aegis.intents.submit( agent_id = os.environ["AEGIS_AGENT_ID"], amount = 12500, currency = "INR", reason = "Monthly SaaS subscription renewal", confidence = 0.97, merchant = "Notion", merchant_category = "saas", metadata = { "invoice_id": "INV-2026-0341" }, ) print(intent.id) # "int_9f3a2b..." print(intent.status) # "pending" if intent.status == "approved": execute_payment(intent.id) elif intent.status == "pending_approval": print("Awaiting finance team approval") else: raise Exception(f"Spend denied: {intent.policy_decision}")
Async usage
For async frameworks (FastAPI, asyncio agents), use AsyncAegisClient instead.
from aegis_os import AsyncAegisClient import asyncio async def run_agent(): async with AsyncAegisClient(api_key="aegis_sk_...") as aegis: intent = await aegis.intents.submit( agent_id = "your-agent-uuid", amount = 5000, currency = "INR", reason = "Cloud compute invoice", confidence = 0.92, ) resolved = await aegis.intents.wait_for_decision(intent.id) return resolved asyncio.run(run_agent())
Polling for a decision
Use wait_for_decision() to block until the policy engine resolves the intent.
resolved = aegis.intents.wait_for_decision( intent.id, poll_interval_s=0.5, # default 0.5s timeout_s=15, # default 15s ) if resolved.status == "approved": execute_payment(resolved.id) elif resolved.status == "denied": raise Exception(f"Denied by policy: {resolved.policy_decision}")
Error handling
The SDK raises typed exceptions. Catch AegisError for API errors and AegisTimeoutError for polling timeouts.
from aegis_os import AegisClient, AegisError, AegisTimeoutError try: intent = aegis.intents.submit( /* ... */ ) except AegisTimeoutError: # Decision not received in time — safe to retry pass except AegisError as e: print(e.status_code, e.message, e.code) raise
All methods
intents.submit(**kwargs)
Submit a new spend intent for policy evaluation.
intents.get(intent_id)
Retrieve the current state of an intent by ID.
intents.list(**filters)
List intents with optional status / agent / date filters.
intents.wait_for_decision(id, ...)
Block until the intent reaches a terminal status.
approvals.list()
List pending approval requests for the current user.
approvals.respond(id, decision)
Approve or reject a pending intent.
agents.list()
List all registered agents in the organisation.
agents.get(agent_id)
Get agent details, risk level, and wallet balance.
AsyncAegisClient — just prefix calls with await.