Skip to content

Using AAA

AAA (Autonomous Agent Architecture) is a Python library for everything around the model in an agent system. It does not ship "an agent class". You pick an algorithm (a ToolLoop, a workflow DAG, a generate-and-verify loop, ...), run it on an engine, and the engine supplies the parts that are easy to get wrong:

AAA gives you So you do not hand-write
Append-only event log as the source of truth ad-hoc state, "what did the agent do?" archaeology
Budget (steps, tokens, cost, deadline) runaway loops and surprise bills
Typed @tools with schemas, timeouts, approval gates, idempotency keys glue code and unsafe retries
SSRF-hardened http_fetch, web_search (SearXNG/Brave), MCP client a fetch tool a web page can turn against you
SQLiteEventStore + engine.recover() resumable, crash-safe runs on one machine
Validators, guardrails, human approval "the prompt says please" as a safety mechanism

It is MIT-licensed and on PyPI: pip install autonomous-agent-architecture. This agent uses it for every stage, so the examples below are real code from this repository.

The 5-step pattern

Every AAA program, including this one, has the same shape.

Plain Python functions become tools. Types and the docstring become the schema the model sees; unknown arguments are rejected.

from aaa.tools import tool

@tool(timeout_s=30, pool="github")
async def gh_search(topic: str, pushed_within_days: int = 30, limit: int = 30) -> list[dict]:
    """Search GitHub repositories by topic, recently pushed, most starred first."""
    ...

An algorithm decides what happens next. Pick the simplest that fits (guide).

from aaa.algos import Node, WorkflowGraph

graph = WorkflowGraph(
    [Node("gh0", "gh_search", args={"topic": "ai-agent"}, on_error="continue"),
     Node("hn0", "hn_search", args={"query": "AI agent"}, on_error="continue"),
     Node("ingest", "ingest_and_score", after=["gh0", "hn0"],
          args=lambda r, inp: {"batches": [r.get("gh0"), r.get("hn0")]})],
    output="ingest",
)

Handlers answer the engine's requests: the LLM, tools, validators. They attach to the engine, which is the only moving part.

from aaa.runtime import Engine
from aaa.llm import LLMHandler, build_client, LLMConfig
from aaa.tools import ToolHandler, ToolRegistry
from aaa.validate import ValidationHandler
from aaa.events import SQLiteEventStore

async with Engine(store=SQLiteEventStore("aaa_events.db")) as engine:
    LLMHandler(build_client(LLMConfig(model="gpt-4o-mini", base_url="https://api.openai.com/v1"))).attach(engine)
    ToolHandler(ToolRegistry([gh_search, hn_search])).attach(engine)
    ValidationHandler({"facts": my_validator}).attach(engine)

from aaa.runtime import Budget

outcome = await engine.run(graph, {}, budget=Budget(deadline_s=300))
outcome.kind.value   # "success" | "no_solution" | "timeout" | "budget_exceeded" | "failed" | ...
outcome.value        # the algorithm's result

Because state is the event log, a restarted process picks up open runs.

await engine.recover([graph, other_algorithm, ...])   # pass every algorithm that might have open runs

Our full engine wiring is runtime.py:

    return ingest_and_score


@asynccontextmanager
async def runtime(s: Settings | None = None, llm_client=None, fetcher: RawFetcher | None = None):
    s = s or Settings()
    s.data_dir.mkdir(parents=True, exist_ok=True)
    repo = Repo(s.botshub_db)
    fetcher = fetcher or RawFetcher(FetchPolicy())
    registry = build_registry(repo, s, fetcher)
    store = SQLiteEventStore(s.events_db)

    def pages_for(ctx: dict) -> dict[str, str]:
        cid = ctx["task"]["candidate_id"] if isinstance(ctx.get("task"), dict) else None
        return {p["url"]: p["text"] for p in repo.pages(cid)} if cid else {}

    client = llm_client or build_client(llm_config(s))
    if any(is_local(m, s.local_prefixes) for m in (s.critic_model, s.embed_model, s.llm_model)):
        client = RoutingClient(client, LocalClient(), s.local_prefixes)
    async with Engine(store=store) as engine:
        LLMHandler(client).attach(engine)
        ToolHandler(registry, pools={"default": 16, "github": 2, "hn": 2, "searxng": 2, "fetch": 6, "db": 1, "browser": 2}).attach(engine)
        ValidationHandler({"facts": make_validator(pages_for), "classify": classify_validator,
                           "default": json_validator(Classification)}).attach(engine)
        try:
            from botshub_agent.algos import make_algos

Where to go next

AAA is alpha (0.x)

Its own status page is deliberately blunt about what is verified. We pin the version and wrap it behind thin interfaces (algos.py, runtime.py, local.py). See status and known gaps.