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.

Provider-object shape (preferred): Agent(provider=ClaudeCode(model="opus", effort="high"))

Flat shape (legacy): Agent(harness="claude-code", model="opus", effort="high", permission="auto")

aegis.config.ClaudeCode

Bases: _ProviderBase

aegis.config.GeminiCLI

Bases: _ProviderBase

aegis.config.OpenCode

Bases: _ProviderBase

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.yaml.

aegis.config.load_config 

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

Load agents + default_agent from .aegis.yaml at root (or the discovered project root). Plain-tuple return for back-compat with pre-YAML call sites.

Queues

aegis.queue.Queue dataclass 

Queue(
    name: str,
    agent_profile: str,
    max_parallel: int,
    provider: str = "",
    model: str = "",
    budgets: list[Budget] = list(),
)

Workflows

aegis.workflow.workflow 

workflow(arg: WorkflowFn | str) -> Any

Register an async workflow.

Supports two forms::

@workflow
async def my_flow(engine): ...

@workflow("custom_name")
async def my_flow(engine): ...

aegis.workflow.WorkflowEngine 

WorkflowEngine(
    *,
    name: str | None = None,
    workflow_id: str | None = None,
    host: str | None = None,
    config: dict | None = None,
    workflow_name: str | None = None,
    workflow_run_id: str | None = None,
    caller_handle: str | None = None,
    bridge=None,
    queue_manager=None,
    inbox_router=None,
    groups_runtime=None,
    groups_wiring=None,
    session_manager=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) and to a WorkflowRunner on the bridge that mediates send-and-await-reply, ask_human, spawn, bash, ledger persistence, and narration. Tracks _spawned_handles for auto-close and _touched_handles for auto-drain at runner exit.

caller_handle instance-attribute 

caller_handle = host_val

spawn async 

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

Spawn a fresh subagent of profile; track for auto-close on workflow exit. alias is the preferred kwarg; handle is a legacy alias accepted for back-compat. Returns the new handle.

send async 

send(
    handle: str,
    prompt: str,
    *,
    timeout: float | None = None,
) -> str

Send prompt as a user-turn to handle (host or subagent); await the next complete assistant message from that handle; return its text.

drain async 

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

Await each touched handle's session to reach state == ready. Legacy primitive kept for back-compat with tdd_step and tests; the new send already blocks until reply, so most new workflows won't need this.

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 spawned subagent. Raises ValueError if handle is the host. Idempotent for unknown handles.

bash async 

bash(
    cmd: str,
    *,
    cwd: str | Path | None = None,
    timeout: float | None = None,
    env: dict | None = None,
) -> "_BashResult"

Async shell. Delegates to bridge.workflow_runner.run_bash when the runner exposes one (test bridges intercept here); otherwise runs the command via asyncio.create_subprocess_shell with cwd defaulting to the project root.

Returns a _BashResult (subprocess.CompletedProcess subclass) so callers may use either attribute access (.returncode/.stdout/.stderr) or dict-style indexing (["exit"]/["stdout"]/["stderr"]) — the latter is what bash_predicate retry-with templates expect.

log 

log(message: str) -> None

Single-line narration. Writes to stderr + JSONL under state_dir.

Kept sync for v1 back-compat with existing workflows; the catalog seeds may also call this without await.

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