Skip to content

API reference

Auto-generated from the source via mkdocstrings. Only the public surface — from aegis import … — is documented here. Internal modules (drivers, MCP server internals, TUI widgets) are intentionally not exposed.

Configuration

aegis.config.Agent

Bases: BaseModel

An agent profile. Two construction shapes are supported, both equivalent after validation:

Provider-object shape (preferred — per-provider fields are validated): Agent(provider=ClaudeCode(model="opus", effort="high")) Agent(provider=GeminiCLI(model="gemini-3-flash-preview")) Agent(provider=OpenCode(model="opencode/claude-sonnet-4-6"))

Flat shape (legacy — still works; constructs the provider internally): Agent(harness="claude-code", model="opus", effort="high", permission="auto") Agent(harness="gemini", model="gemini-3-flash-preview", permission="full")

Internally a Provider object is always populated post-validation, so drivers / queue config / TUI can read either agent.provider.* or the legacy flat fields uniformly.

aegis.config.ClaudeCode

Bases: _ProviderBase

Anthropic's claude CLI. Has an effort field (low|medium|high|max) that no other provider currently exposes.

aegis.config.GeminiCLI

Bases: _ProviderBase

Google's gemini CLI. Model strings are bare gemini model names (e.g. gemini-3-flash-preview, gemini-3.1-pro). No effort field — Gemini doesn't expose one. Permission maps to its --approval-mode (read=plan, write=auto_edit, full=yolo, auto=default).

aegis.config.OpenCode

Bases: _ProviderBase

OpenCode's opencode CLI. Model strings use the provider/model format opencode expects (e.g. opencode/claude-sonnet-4-6, opencode/gemini-3-flash, opencode/gpt-5). Run opencode models to list what's available on your install.

aegis.config.Permission

Bases: str, Enum

aegis.config.Effort

Bases: str, Enum

aegis.config.ConfigError

Bases: Exception

aegis.config.find_project_root 

find_project_root(start: Path | None = None) -> Path | None

Closest ancestor of start (default cwd) containing .aegis.py.

aegis.config.load_config 

load_config(
    search_paths: Sequence[Path] | None = None,
) -> tuple[dict[str, Agent], str]

Queues

aegis.queue.Queue dataclass 

Queue(name: str, agent_profile: str, max_parallel: int)

Workflows

aegis.workflow.workflow 

workflow(fn: WorkflowFn) -> WorkflowFn

Register an async workflow under fn.__name__.

aegis.workflow.WorkflowEngine 

WorkflowEngine(
    *,
    workflow_name: str,
    workflow_run_id: str,
    bridge,
    queue_manager,
    inbox_router,
    caller_handle: str | None = None,
    state_dir: Path | None = None,
    now: Callable[[], str] = now_iso,
    drain_timeout: float = 30.0,
)

Runtime handle a workflow receives as its first positional argument.

Constructed once per workflow run; bound to live aegis substrate (AppBridge, QueueManager, InboxRouter). Tracks _spawned_handles for auto-close and _touched_handles for auto-drain at runner exit.

caller_handle instance-attribute 

caller_handle = caller_handle

spawn async 

spawn(profile: str, *, handle: str | None = None) -> str

Spawn a long-lived agent through the bridge. Tracks handle for auto-close on workflow exit. Returns the handle.

send 

send(handle: str, message: str) -> None

Enqueue a substrate-tagged message in handle's inbox. Sync, fire-and-forget. Returns immediately; the actual delivery is scheduled as an asyncio task (which inherits the calling context — workflows always run on the aegis event loop).

drain async 

drain(handle: str | None = None) -> None

Await each touched handle's session to reach state == ready. If handle is None, drain all touched handles. Per-handle ceiling of self._drain_timeout; on timeout, log a warning and continue (don't trap workflow shutdown on a hung agent).

delegate async 

delegate(queue: str, payload: str) -> str

Enqueue a one-shot task on the named queue; await the worker's callback; return its final assistant text. Raises WorkflowError on unknown queue, worker failure, or substrate error.

close async 

close(handle: str) -> None

Close a long-lived agent. Idempotent — silent no-op if the handle is unknown to this engine.

bash async 

bash(
    cmd: str,
    *,
    cwd: str | Path | None = None,
    timeout: float | None = None,
    env: dict | None = None,
) -> subprocess.CompletedProcess

Async shell. cwd defaults to project root (find_project_root) or os.getcwd(); timeout=None means wait forever. On timeout, raises WorkflowError after killing the subprocess.

log 

log(message: str) -> None

list_sessions 

list_sessions() -> list[SessionInfo]

list_agents 

list_agents() -> list[str]

aegis.workflow.WorkflowError

Bases: Exception

Expected failure inside a workflow (predicate violated, retry exhausted, etc.). Workflows raise this for clean failure reporting. Plain Exception is treated as an unexpected crash.

aegis.workflow.list_workflows 

list_workflows() -> list[str]

aegis.workflow.get_workflow 

get_workflow(name: str) -> WorkflowFn | None