Quickstart — the capstone example

examples/07_everything_together.py in visvoai-core composes visvoai-ai and visvoai-core into one small product — a tiny "ops assistant" that shows every piece from this site working together:

• the model chosen by facts from the visvoai-ai registry
• tools in three shapes, mixed in one list (plain function / Args: docstring function / lifecycle class)
• every lifecycle-tool call audited into SQLite via ToolPersistence
• ten tools in total — the eight non-core ones indexed in a ToolCatalog, with only the matching ones bound per turn (the two core tools are always bound and never go through retrieval)
• multi-turn memory — "restart it" resolves via a LangGraph checkpointer

Run it

pip install visvoai-core "visvoai-ai[gemini]"
python 07_everything_together.py                       # no API key — scripted model
GEMINI_API_KEY=... python 07_everything_together.py     # same code, live model

Without a key, pick_model() falls back to a scripted FakeMessagesListChatModel that replays a fixed queue of tool calls and answers — same graph, same code path, no network call. With GEMINI_API_KEY set, it picks the cheapest non-lite gemini-2.5-flash deployment from the live registry instead.

What it demonstrates

core = as_tools([service_status, as_tool(restart)])   # always bound
fleet = as_tools([disk_usage, *FLEET])                # bound on demand

catalog = ToolCatalog([(t.name, t.description) for t in fleet])

def retrieve(query: str):
    return catalog.search(query, k=3)

graph = AgentRuntime().build_graph(
    model=model,
    core_tools=core,
    all_tools_map=all_map,
    system_prompt="You are a calm ops assistant. Use tools; be brief.",
    per_round_retrieve=retrieve,
    checkpointer=MemorySaver(),
)

for question in ("is the api healthy in eu-west? and how is our error budget?",
                 "restart it anyway, please"):        # "it" needs memory
    print(await ask(graph, question, thread_id="shift-42"))

The second question — "restart it anyway" — only resolves correctly because the checkpointer carries conversation state across turns under the same thread_id. restart_service is the only tool here that changes anything, and every call to it is recorded in SQLite via ToolPersistence:

rows = sqlite3.connect(audit_db).execute(
    "SELECT tool, status, ms FROM audit").fetchall()

This file is the shape of a real product — everything else from here is your business logic. See the examples ladder for each idea introduced one file at a time, and the full API reference for every signature used above.