Skip to content

Quickstart

This page builds an agent in four steps. Every snippet is a real file from the repository's examples/ folder and is executed by the test suite.

1. A tool and a model

"""Quickstart: a tool-using agent in ~25 lines. Runs offline with a scripted model."""

import asyncio

from aaa.algos import ToolLoop
from aaa.llm import FakeLLMClient, LLMHandler
from aaa.llm.fake import text, tool_call
from aaa.runtime import Budget, Engine
from aaa.tools import ToolHandler, ToolRegistry, tool


@tool
def multiply(a: int, b: int) -> int:
    """Multiply two integers."""
    return a * b


def scripted_model(request):  # stands in for a real LLM (see "LLM gateways" to swap it)
    last = request.messages[-1]
    if last.role == "user":  # first turn: the model asks for the tool
        return tool_call("multiply", a=17, b=23)
    return text(f"17 x 23 = {last.content}")  # second turn: it answers using the tool result


async def main() -> None:
    registry = ToolRegistry([multiply])
    async with Engine() as engine:
        LLMHandler(FakeLLMClient(scripted_model)).attach(engine)  # who answers LLM requests
        ToolHandler(registry).attach(engine)  # who runs tool calls
        outcome = await engine.run(
            ToolLoop(registry),
            {"task": "What is 17 times 23?"},
            budget=Budget(max_steps=20, deadline_s=30),
        )
    print(outcome.kind.value, "-", outcome.value["text"])


asyncio.run(main())

Run it:

python examples/quickstart.py

success - 17 x 23 = 391

What happened, line by line:

  1. @tool turned a plain function into a tool. The JSON schema the model sees is generated from the type hints and the docstring becomes the description.
  2. ToolRegistry holds the tools an agent may call.
  3. Engine() is the runtime. It owns an event log, a message bus and the bookkeeping for budgets and deadlines.
  4. LLMHandler(...).attach(engine) registers who answers LLM requests. ToolHandler(...).attach(engine) registers who runs tool calls. Algorithms never call a model or a tool directly; they emit request events and react to the results (see the mental model).
  5. engine.run(algorithm, input, budget=...) starts a run and waits for its outcome.

The scripted model

FakeLLMClient(scripted_model) calls your function with each LLMRequest and expects an LLMResponse back. text(...) and tool_call(...) are helpers that build those. It makes examples and tests deterministic. Replace it with a real client to go live:

from aaa.llm import build_client, gateways

LLMHandler(build_client(gateways.portkey("gpt-4o-mini"))).attach(engine)

2. Look at what actually happened

A run is a list of events. After the run, read them:

"""A run is just a list of events. Print the log of a finished run."""

import asyncio

from aaa.algos import ToolLoop
from aaa.llm import FakeLLMClient, LLMHandler
from aaa.llm.fake import text, tool_call
from aaa.runtime import Engine
from aaa.tools import ToolHandler, ToolRegistry, tool


@tool
def multiply(a: int, b: int) -> int:
    """Multiply two integers."""
    return a * b


def scripted_model(request):
    last = request.messages[-1]
    return tool_call("multiply", a=17, b=23) if last.role == "user" else text(f"= {last.content}")


async def main() -> None:
    registry = ToolRegistry([multiply])
    async with Engine() as engine:
        LLMHandler(FakeLLMClient(scripted_model)).attach(engine)
        ToolHandler(registry).attach(engine)
        run = await engine.start(ToolLoop(registry), {"task": "What is 17 times 23?"})
        outcome = await run.result()
        for event in await engine.store.read(run.run_id):
            print(event.seq, event.type)
    print("outcome:", outcome.kind.value)


asyncio.run(main())

0 aaa.run.started
1 aaa.llm.requested
2 aaa.llm.completed
3 aaa.tool.requested
4 aaa.tool.started
5 aaa.tool.completed
6 aaa.llm.requested
7 aaa.llm.completed
8 aaa.run.completed
outcome: success

This list is the run. The event catalog documents every type.

3. Bound it

Budget(max_steps=20, deadline_s=30) above is enforced by the engine, not by the model's good behavior. Exceeding a limit ends the run with a distinct outcome (budget_exceeded, timeout), never an exception.

Budget(max_steps=20, max_tokens=50_000, max_cost=0.50, deadline_s=60)

4. Make it durable

Pass a durable store and the same code survives restarts:

from aaa.events import SQLiteEventStore

async with Engine(store=SQLiteEventStore("agent.db")) as engine:
    ...

See Persistence and recovery for resuming after a crash.

Next steps