aeko

The intelligent core of the Aether ecosystem — a multi-agent system that reads a company's greenhouse gas (GHG) inventory, explains it, and turns it into a concrete improvement plan.

Aether helps industrial companies cut their greenhouse gas emissions. Their employees fill in the company's GHG inventory; Aeko is what reads that inventory and answers with pollutant analysis, green gas alternatives and a prioritized improvement plan — plus a chatbot for the day-to-day ESG questions that come up along the way.

This package is the reusable engine behind that product. It is a library, not a service: it exposes a small facade over a LangChain/LangGraph agent graph so any backend — an API, a worker, a notebook — can embed it.

🇧🇷 Um resumo em português está no final deste documento: Resumo em português.

What's new in 2.0

Version 2 reshapes the boundary between the SDK and the API that consumes it. The agent system is the same; how you hand work to it changed, and every change is breaking.

If you are coming from 1.x, read Migrating from 1.x — it maps every removed name to its replacement.

Table of contents

• What the SDK does
• The agent system
• Requirements
• Installation
• Quickstart
• Integrating into other systems
  • Design rules that shape the integration
  • 1. Configure once, at startup
  • 2. Register your tools
  • 3. The conversational flow
  • 4. The inventory flow
  • 5. Full example: a stateless FastAPI service
  • 6. Error handling
• Migrating from 1.x
• API reference
• Development
• Resumo em português
• License

What the SDK does

Aeko exposes two independent entry points into the same agent graph:

Both share the same agents, the same registered tools and the same configuration. The difference is where the run enters the graph and how it ends.

The agent system

Eight agents, each with its own prompt, persona and tools. Names are in Portuguese because they are also the routing keys the graph and set_tools() use — pass them exactly as written, accents included.

The four cheap agents (classify, consolidate, review) run on fast_model; the four specialist analysts run on slow_model. Both are configurable, and each model is registered with the other as its fallback — a provider hiccup degrades a run rather than failing it, which is also why a turn can report a model you did not ask for.

Agents hand off to each other by ending their answer with a literal Next agent: <name> line, which the graph reads and the SDK strips before the text reaches you. It is protocol between agents, never part of an answer.

Conversational flow — enters at the router, always passes the guardrail:

send_message()
      │
      ▼
  Roteador ──────────────────────────▶ FAQ ─────────────────────────▶ answer
      │
      ├──▶ Analista de Poluentes ─────┐
      ├──▶ Analista de Gases Verdes ──┼──▶ Orquestrador ──▶ Guardrail de Saída
      │                               │                            │
      └──▶ Coordenador de Melhoria ───┴───────────▶ answer   approved? ──▶ answer
                                                              │
                                                       rejected (up to 3x)
                                                              │
                                                              └──▶ back to Roteador

The guardrail can send a draft back to the router 3 times; a fourth rejection ends the run with no answer at all — AekoMessageResponse.message.output is "" and approved is False. Always check it.

Inventory flow — enters at the inventory analyst, ends at a terminal node, so it never passes the guardrail:

analyze()
      │
      ▼
  Análista de inventários ──▶ Analista de Poluentes ⇄ Analista de Gases Verdes
                                                              │
                                                              ▼
                                        Coordenador de Melhoria Contínua ──▶ plan

A run that entered here is confined to those analysts: an agent naming any other successor — or naming itself — is routed to the coordinator instead. Without that, one stray handoff would kill the run after every analyst had already been paid for.

Requirements

• Python 3.11 (>=3.11,<3.12)
• A Google Gemini API key

Runtime dependencies (langchain-core, pydantic, langchain-classic, langgraph, langchain-google-genai) are installed with the package.

Installation

pip install aeko

Distribution and import name are the same: aeko. (The GitHub repository is aeko-sdk.)

Quickstart

from aeko import (
    Aeko,
    AekoInventoryAnalyzer,
    AekoMessenger,
    AekoSession,
    AekoUser,
    AekoUserMemory,
)

# 1. Configure the SDK once, for the whole process.
Aeko.config("YOUR_GEMINI_API_KEY")

# 2. Chat. The documents come straight out of your database, on every request.
user = AekoUser.model_validate(db.user.find_one({"_id": user_id}))
session = AekoSession.model_validate(db.session.find_one({"_id": session_id}))
memories = [AekoUserMemory.model_validate(document)
            for document in db.user_memory.find({"id_user": user_id})]

messenger = AekoMessenger(user, memories)
reply = messenger.send_message(
    "What is the difference between scope 1 and scope 2?", session
)
print(reply.message.output)  # the answer
print(reply.agents_called)   # e.g. ['FAQ']
print(reply.approved)        # guardrail verdict

# `reply.message` is one entry of `session.messages`, ready to append.
db.session.update_one(
    {"_id": session_id}, {"$push": {"messages": reply.message.model_dump()}}
)

# 3. Analyze an inventory.
analyzer = AekoInventoryAnalyzer()
analyzer.set_context("Last report: 12,400 tCO2e, scope 1 dominated by the boiler fleet.")

plan = analyzer.analyze(inventory_markdown, id_external_inventory=502)
db.improvement_plan.insert_one(plan.model_dump(by_alias=True, exclude={"id"}))

Integrating into other systems

Design rules that shape the integration

Three deliberate design choices decide how you wire Aeko into your backend. Read these before writing the integration — they explain every recommendation below.

1. The SDK never reads the environment. There is no GEMINI_API_KEY fallback, no .env loading. Your application owns its configuration and passes it in through Aeko.config(). Nothing works until it does.
2. Configuration and tools are process-wide. Aeko.config() and AekoMessenger.set_tools() write to a single process-wide runtime, and any write to it drops the agents so the next run rebuilds them. Call them at startup, not per request. (set_tools is a classmethod for exactly this reason: registering tools on one instance would silently rebuild the agents behind every other instance's back.)
3. The SDK keeps no session state at all. The conversation lives in the AekoSession you pass to every send_message() call, and the SDK updates that object in place before returning. Nothing is cached between calls, so any worker can serve any conversation — and the process never accumulates sessions it cannot evict.

1. Configure once, at startup

from aeko import Aeko

Aeko.config(
    api_key=settings.GEMINI_API_KEY,
    fast_model="gemini-3.1-flash-lite",     # router, FAQ, orchestrator, guardrail
    slow_model="gemini-3.5-flash",          # the four specialist analysts
    max_tokens=1024,                        # output cap for chat turns
    report_max_tokens=8192,                 # output cap for the inventory report
)

Every keyword argument is optional and falls back to the default shown. Calling Aeko.config() again applies the change immediately — the agents and the compiled graph are rebuilt on the next run. Aeko.is_configured() tells you whether a key was supplied (useful in a health check); Aeko.reset() restores every default and clears registered tools.

2. Register your tools

Tools are your application's code — a price lookup, an emission-factor table, a search over your own documents. Aeko binds them to the agents you choose.

from langchain_core.tools import tool
from aeko import AekoMessenger, AekoTool

@tool
def consulta_precos(equipamento: str) -> str:
    """Return the average market price for a piece of equipment."""
    return price_service.lookup(equipamento)

AekoMessenger.set_tools({
    "Coordenador de Melhoria Contínua": [
        AekoTool(
            tool=consulta_precos,
            description="Use to price the equipment you recommend, in BRL.",
        ),
    ],
    # A bare LangChain tool works too — its own docstring becomes the description.
    "FAQ": [buscar_norma],
})

Each tool is used twice, from a single declaration: its description is rendered into that agent's # Ferramentas Disponiveis prompt section, and the tool object itself is bound to that agent's executor. The prompt therefore can never advertise a tool the agent is unable to call.

Keys must be agent names from the table above; anything else raises UnknownAgentError, which carries the valid names in .known_agents. AGENT_NAMES is exported for validating input before you get there:

from aeko import AGENT_NAMES  # ('Roteador', 'FAQ', 'Orquestrador', ...)

set_tools() replaces the whole registry, so pass every agent's tools in one call. Appending to an already-registered list is the one write the runtime cannot observe, and the agents would go on using the tools they were built with.

3. The conversational flow

from aeko import AekoMessage, AekoMessenger, AekoSession, AekoUser, AekoUserMemory

# Built once per user — it holds nothing about any conversation.
messenger = AekoMessenger(
    AekoUser(id_external_user=1001, role="ESG analyst",
             usecase="Tracks the boiler fleet's gas substitution."),
    [AekoUserMemory(field="preferred_language", description="Answers in Portuguese")],
)

# Rehydrated per request, from what you persisted.
session = AekoSession(
    id="64b8f0a1c9e1a2b3c4d5e6f3",
    id_user="64b8f0a1c9e1a2b3c4d5e6f1",
    name="Scope 1 review",
    messages=[AekoMessage(input="What is scope 3?", output="Scope 3 covers...")],
)

response = messenger.send_message(
    "Our scope 1 jumped 12% this quarter. Where do I look?", session
)

The two documents your database already holds go in as they are, so there is no separate "history" format to translate: session.messages is the conversation, replayed to the agents oldest first. Nothing is kept between calls — pass the session on every request and any worker can serve it.

Only the 10 most recent turns are replayed (SESSION_HISTORY_USAGE, in aeko.config.messenger). Send the session as long as it is: a conversation of 500 turns costs a run exactly what one of 10 does, and you do not have to slice anything on the way in. What the SDK caps is what the agents read, never what you persist — the turn it answers is appended to the session you passed, and every earlier turn is still there when the call returns.

user.role and user.usecase become the business context every agent reads. The identifiers (_id, id_external_user, id_user) never reach a prompt: they are there so you can correlate documents, and a model can do nothing with them.

send_message() returns an AekoMessageResponse:

Only message belongs in the database; everything else describes how the run reached the answer, or which documents it belongs to, and is deliberately not part of the persisted entry — session.messages[] carries no identifiers of its own. The identifiers are still returned so a response is enough on its own to file and log the answer against the right conversation. The turn itself mirrors the collection exactly:

A single turn crosses both configured models — the router, FAQ, orchestrator and guardrail run on the fast one, the analysts on the slow one — so input_tokens and output_tokens are the whole run's consumption, summed across every agent it called, not one model call's. That is the number that answers "what did this conversation cost".

An answered turn is appended to the AekoSession you passed in, and updated_at is bumped — the same object, updated in place, so you can persist exactly what you handed over. Only a final result is recorded: a turn the guardrail rejected is not appended, because a draft that never reached the user cannot become context for the next question.

User memories. Hand the user_memory documents to the constructor and every one of them is rendered into the same business context the user's role and usecase go into, so every agent of the run reads them:

messenger = AekoMessenger(user, memories)   # list[AekoUserMemory]

Each memory becomes one "- <field>: <description>" line (AekoUserMemory.to_prompt_line()) under a Memórias do usuário: heading, with expires_at and the identifiers left out — a model can act on what was remembered, not on when the row stops being valid. There is no cap here, unlike the conversation: a memory is already the condensed form of something the user told you once, and dropping any of them would silently un-remember it.

Deciding which memories are still valid is your API's job: filter expires_at before handing the list over. A user with no memories is normal — the argument is optional, and no empty section reaches the prompt.

4. The inventory flow

from aeko import AekoInventoryAnalyzer

analyzer = AekoInventoryAnalyzer()

# Optional: a company's first report legitimately has no previous one.
analyzer.set_context("2025 report: 12,400 tCO2e, scope 1 dominated by the boiler fleet.")

plan = analyzer.analyze(inventory_markdown, id_external_inventory=502)

analyze() expects the inventory rendered as Markdown — a table is the natural shape. It runs with report_max_tokens instead of the chat cap, since this flow writes a full report that the chat-sized cap would truncate. id_external_inventory is what ties the resulting plan back to the inventory; the SDK never reads your database, so it cannot be derived here.

It returns an AekoImprovementPlan, mirroring one document of the collection:

The continuous improvement coordinator is instructed — in its scope, its tasks and every one of its few-shot examples — to write exactly those three text fields under three fixed headings, and the SDK reads them straight back out:

## Problema definido
...

## Método
...

## Raciocínio
...

Headings rather than a JSON object, for the same reason the agents route on a literal Next agent: line: this flow runs with the report token cap, and a truncated JSON object is unparseable and costs the whole plan, while a truncated last section still yields the ones written before it. Only those three headings delimit a section, so a subtitle the agent writes mid-answer never cuts its own text short — and case and accents are ignored when matching them, so ## RACIOCINIO is read as ## Raciocínio.

An answer that misses a section is sent back to the coordinator to be rewritten, with the missing headings named and the original request attached, up to four times (PLAN_FORMAT_MAX_RETRIES). The retry happens inside the graph and only re-asks the coordinator — the inventory, pollutant and green gas analysts already ran, and their findings stay in the state. If the format never comes, analyze() raises MalformedAgentOutputError rather than padding the plan with guesses: the SDK will not hand you a plan whose fields it invented. The model is also never given a say in _id or updated_at — only the three content fields are read back from it.

Budget for it: a plan that takes every retry costs five coordinator calls at the report token cap, on top of the analysts. A well-formed answer costs exactly one.

There is no approved field: this flow ends at the continuous improvement coordinator, a terminal node, and never reaches the output guardrail.

5. Full example: a stateless FastAPI service

This is the shape Aeko was designed for: an HTTP API with more than one worker, where conversation history lives in your database and is handed back to the SDK on every request.

from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from aeko import (
    Aeko,
    AekoInventoryAnalyzer,
    AekoMessenger,
    AekoNotConfiguredError,
    AekoSession,
    AekoTool,
    AekoUser,
    AekoUserMemory,
)

from .settings import settings
from .tools import buscar_norma, consulta_precos


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Process-wide setup: exactly once, before the first request.
    Aeko.config(settings.GEMINI_API_KEY, report_max_tokens=12288)
    AekoMessenger.set_tools({
        "FAQ": [buscar_norma],
        "Coordenador de Melhoria Contínua": [
            AekoTool(tool=consulta_precos, description="Price the equipment you recommend."),
        ],
    })
    yield


app = FastAPI(lifespan=lifespan)


class ChatRequest(BaseModel):
    session_id: str
    message: str


@app.post("/chat")
def chat(body: ChatRequest):
    # 1. Read the documents — this worker may never have seen this session before.
    #    The DTOs take them exactly as they come out of the collections.
    session = AekoSession.model_validate(db.session.find_one({"_id": body.session_id}))
    user = AekoUser.model_validate(db.user.find_one({"_id": session.id_user}))

    #    Expiring the memories is yours to do: the SDK renders every one it gets.
    memories = [
        AekoUserMemory.model_validate(document)
        for document in db.user_memory.find({"id_user": user.id, "expires_at": None})
    ]

    messenger = AekoMessenger(user, memories)

    # 2. Run it. Blocking call: keep it off the event loop in production
    #    (`run_in_threadpool`, a task queue, or a sync worker).
    try:
        response = messenger.send_message(body.message, session)
    except AekoNotConfiguredError as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from exc

    # 3. The guardrail can refuse every draft — there is no answer to persist.
    if not response.message.output:
        raise HTTPException(
            status_code=502,
            detail="The output guardrail rejected every draft. Please rephrase.",
        )

    # 4. Persist the turn so the next request (on any worker) can replay it.
    db.session.update_one(
        {"_id": response.id_session},
        {
            "$push": {"messages": response.message.model_dump()},
            "$set": {"updated_at": session.updated_at},
        },
    )

    return {
        "answer": response.message.output,
        "agents": response.agents_called,
        "approved": response.approved,
        "guardrail_retries": response.guardrail_retries,
    }


class InventoryRequest(BaseModel):
    inventory_markdown: str
    id_external_inventory: int
    previous_report: str | None = None


@app.post("/inventory")
def inventory(body: InventoryRequest):
    analyzer = AekoInventoryAnalyzer()

    if body.previous_report:
        analyzer.set_context(body.previous_report)

    plan = analyzer.analyze(
        body.inventory_markdown, id_external_inventory=body.id_external_inventory
    )

    # `exclude={"id"}`: the plan is new, so let MongoDB generate its `_id`.
    db.improvement_plan.insert_one(plan.model_dump(by_alias=True, exclude={"id"}))

    return plan.model_dump(exclude={"id"})

Notes for production:

• send_message() and analyze() are synchronous and slow (several model calls per run, more when the guardrail retries). Don't block an async event loop with them — use a threadpool, a background worker, or a task queue.
• Never call Aeko.config() or set_tools() per request. Both rebuild every agent process-wide; doing it under load throws away warm agents for every concurrent run.
• The SDK holds no session state. Read the AekoSession document and pass it to send_message() on every request; any worker can then serve any conversation.
• Configuration and registered tools are shared mutable process state, so treat startup as the only place that writes them.

6. Error handling

Every error the SDK raises inherits from AekoError, so one except covers the SDK:

from aeko import AekoError

try:
    response = messenger.send_message(text, session)
except AekoError as exc:
    logger.exception("Aeko failed")
    raise HTTPException(status_code=500, detail=str(exc)) from exc

Remember that a rejected answer is not an exception — it is a successful run with an empty message.output and approved=False.

Migrating from 1.x

Every name below changed in 2.0. There is no deprecation shim: the 1.x calls fail at import or call time rather than quietly doing something else.

Every 1.x DTO was a plain dataclass describing a call. Every 2.0 DTO is a Pydantic model mirroring a collection, which is what makes the hand-off to the database lossless.

The shape of the change. In 1.x the messenger owned a process-wide dictionary of sessions, and you announced a conversation to it before using it. In 2.0 the conversation is a document you already have:

# 1.x
messenger = AekoMessenger()
messenger.prepare(session_id, user_info="ESG analyst", history=[...])
result = messenger.send_message("Our scope 1 jumped 12%.")
answer = result.answer

# 2.0
messenger = AekoMessenger(user, memories)                    # who is asking
response = messenger.send_message("Our scope 1 jumped 12%.", session)
answer = response.message.output                             # may be "" — check it

Four consequences worth planning for:

1. You must persist the session. The SDK caches nothing, so the turn appended to session.messages is lost unless you write it back. In exchange, any worker can serve any conversation.
2. user_info is now structured. The free-form string that went into prepare() is AekoUser.role plus AekoUser.usecase, and anything else you were packing into it probably belongs in AekoUserMemory — which is rendered in a section of its own.
3. An answer can be empty. .answer was always populated; .message.output is "" when the guardrail rejected every draft, and that is a successful run, not an exception. approved tells you which happened.
4. The improvement plan is structured. analyze() returns defined_problem, method and reasoning instead of one answer string, and raises MalformedAgentOutputError when the coordinator never produces them — code that used to split the report apart should read the fields directly. context_used is gone; you know whether you called set_context().

API reference

Everything below is importable directly from aeko.

Aeko — configuration facade.

AekoMessenger — conversational entry point.

AekoInventoryAnalyzer — report entry point.

Data objects. Every DTO that crosses the API boundary is a Pydantic model mirroring one MongoDB collection, field for field:

model_validate(document) takes a raw document and model_dump(by_alias=True) gives one back — _id included, under that exact name — so the hand-off is lossless in both directions. AekoMessageResponse (the run's answer plus its metadata) and AekoTool are SDK-only and mirror nothing.

The identifiers (_id, id_external_user, id_user, id_external_inventory) and expires_at are carried across the boundary in both directions and echoed back on AekoMessageResponse — they are how you correlate and log what came out of a run. What they never do is reach a prompt: a model can act on a role or a usecase, not on an ObjectId.

Timestamps. The SDK stamps a timestamp only on a document it produces itself, and never invents one for a document it merely received:

So AekoSession, AekoUser and AekoUserMemory default their timestamps to None rather than to "now": a session read from the database brings its own created_at through untouched, and one built in Python without it would otherwise be handed a creation date that is simply false. The consequence to watch for is on the way back — a new session built in Python and dumped straight into MongoDB writes created_at: null. Set it at insert time, or drop the empty fields:

session.model_dump(by_alias=True, exclude_none=True)

Exceptions: AekoError, AekoNotConfiguredError, MalformedAgentOutputError, UnknownAgentError. Constants: AGENT_NAMES, __version__.

Defaults and limits

Development

python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -r requirements-dev.txt
pip install -e .
pytest

The test suite runs against a scripted fake chat model and never calls Gemini, so no API key is needed to run it.

requirements.txt pins the exact versions the suite is green against; pyproject.toml declares the same dependencies as ranges, so installing the SDK next to other LangChain packages does not deadlock a consumer's resolver.

Releases are published to PyPI by the GitHub Actions workflow in .github/workflows/publish.yml, triggered when a GitHub Release is published. The release tag must match aeko.__version__ or the workflow fails before building.

Resumo em português

aeko é o núcleo inteligente do ecossistema Aether: um sistema multiagente (LangChain + LangGraph) que lê o inventário de gases de efeito estufa de uma empresa, explica os números e devolve um plano de melhoria priorizado — além do chatbot que responde às dúvidas de ESG do dia a dia. É uma biblioteca, não um serviço: qualquer backend Python pode embuti-la.

pip install aeko

from aeko import (Aeko, AekoInventoryAnalyzer, AekoMessenger, AekoSession, AekoUser,
                  AekoUserMemory)

Aeko.config("SUA_CHAVE_GEMINI")          # obrigatório: o SDK não lê variáveis de ambiente

usuario = AekoUser.model_validate(db.user.find_one({"_id": id_usuario}))
sessao = AekoSession.model_validate(db.session.find_one({"_id": id_sessao}))
memorias = [AekoUserMemory.model_validate(documento)
            for documento in db.user_memory.find({"id_user": id_usuario})]

messenger = AekoMessenger(usuario, memorias)
resposta = messenger.send_message("Como reduzo o escopo 1 da nossa caldeira?", sessao)
print(resposta.message.output)
db.session.update_one({"_id": id_sessao},
                      {"$push": {"messages": resposta.message.model_dump()}})

analisador = AekoInventoryAnalyzer()
plano = analisador.analyze(inventario_em_markdown, id_external_inventory=502)
db.improvement_plan.insert_one(plano.model_dump(by_alias=True, exclude={"id"}))

O que mudou na 2.0

A fronteira entre o SDK e a API que o consome foi refeita, e toda mudança quebra compatibilidade: SessionInfo e InventoryAnalysisResponse saíram, prepare() e SessionNotPreparedError sumiram, send_message() passou a receber a sessão e devolver AekoMessageResponse, e analyze() devolve AekoImprovementPlan. O mapa completo de nomes antigos para novos está em Migrating from 1.x.

Quatro pontos que definem a integração

1. Configure no startup, nunca por requisição. Aeko.config() e AekoMessenger.set_tools() alteram um runtime único do processo e reconstroem todos os agentes.
2. As DTOs espelham as collections. AekoUser, AekoUserMemory, AekoSession, AekoMessage e AekoImprovementPlan são modelos Pydantic com os mesmos campos dos documentos, _id incluso: model_validate(documento) entra e model_dump(by_alias=True) volta igual. O SDK não fala com o banco — quem lê e grava é a API.
3. O SDK não guarda sessão. Passe o documento AekoSession em todo send_message(): a conversa é o próprio session.messages, o SDK o atualiza in-place e qualquer worker atende qualquer sessão. Só os 10 turnos mais recentes são reenviados aos agentes — mande a sessão inteira, o corte é do que o modelo lê, nunca do que você persiste. Já as memórias do usuário vão no construtor (AekoMessenger(user, memories)) e são renderizadas todas, sem corte, no contexto que todo agente lê — filtrar as expiradas é com a API.
4. Resposta vazia não é exceção. Se o Guardrail de Saída reprovar todas as tentativas (limite de 3 devoluções), o run termina com message.output == "" e approved is False — verifique sempre antes de persistir. Já um plano de melhoria fora do formato levanta MalformedAgentOutputError — depois de o Coordenador ser convidado a reescrever a resposta até quatro vezes —, em vez de devolver campos inventados.

Os nomes dos agentes (Roteador, FAQ, Orquestrador, Guardrail de Saída, Análista de inventários, Analista de Poluentes, Analista de Gases Verdes, Coordenador de Melhoria Contínua) são também as chaves de roteamento — use-os exatamente como escritos, inclusive acentuação. A seção em inglês acima tem o detalhamento completo, incluindo um exemplo de serviço FastAPI stateless.

License

MIT — see LICENSE.