Introduction

Vous avez un agent Python. Il fonctionne sur votre machine, dans un notebook ou dans un script. Vous voudriez le faire tourner en production de manière fiable, sécurisée, sans que vos données ne quittent votre infrastructure.

C’est exactement ce pour quoi Apollia OS a été construit.

Ce book vous guide de l’installation à un projet multi-agent complet en production. Pas de théorie abstraite. Chaque concept est introduit au moment où vous en avez besoin, ancré dans un agent que vous écrivez vous-même.

Ce qu’est Apollia OS

Apollia OS est un runtime Rust pour l’exécution souveraine d’agents IA autonomes. Il joue le rôle de couche d’exécution entre votre code Python et la production :

Vos agents Python (decorator-first, typed, async)
            ↓
      Apollia OS Runtime
   ─────────────────────────
   Isolation   │  Garde-fous
   Outils      │  Audit trail
   Mémoire     │  HITL
   A2A         │  Triggers
            ↓
     Votre machine, zéro cloud obligatoire

Le contrat est minimal : vous écrivez une classe Python décorée avec @agent, vous annotez vos méthodes avec @skill ou @on_message, et le runtime sait charger, valider, isoler, et invoquer votre agent depuis la CLI, l’API REST, l’app Desktop, ou un autre agent via A2A.

from apollia import agent, skill, DomainError
from apollia.types import Ctx

@agent(name="hello", version="0.1.0", description="Greets people warmly.")
class Hello:
    @skill("hello.greet", description="Greet a person by name.")
    async def greet(self, name: str, ctx: Ctx) -> dict:
        if not name.strip():
            raise DomainError("EMPTY_NAME", "name must not be empty")
        return {"message": f"Bonjour, {name} !"}

Pas de fichier de configuration, pas d’héritage de classe, pas de boilerplate. La signature de la méthode devient un schéma JSON, les exceptions deviennent des résultats typés, et l’agent est immédiatement utilisable.

Pourquoi pas un SaaS ou un framework ?

Les solutions existantes couvrent bien deux cas extrêmes :

Les frameworks (LangGraph, CrewAI, AutoGen) vous aident à écrire la logique d’un agent. Ils ne gèrent ni l’isolation, ni les garde-fous runtime, ni l’audit. Vous construisez vous-même la couche opérationnelle.
Les SaaS (LangServe Cloud, Dify, Modal) gèrent l’infrastructure pour vous, mais vos données transitent par leurs serveurs. Pour une PME européenne ou tout contexte où la souveraineté des données compte, ce n’est souvent pas acceptable.

Apollia OS occupe l’espace entre les deux. Il s’exécute sur votre machine, en un seul binaire, sans Docker, sans Kubernetes, sans compte cloud obligatoire.

Critère	Apollia OS	LangServe	Dify	CrewAI
Exécution	Local (binaire)	Cloud	Cloud / self-hosted	Framework
Données	Restent sur la machine	Transitent	Transitent / local	Dépend
Isolation outils	Sandbox natif	Non	Non	Non
Garde-fous runtime	StepBudget + circuit breakers	Non	Basique	Non
Framework-agnostic	Oui (SDK typé indépendant)	LangChain only	Format propre	CrewAI only

Quand l’utiliser :

Vous voulez livrer des agents IA à des PME et garder leurs données sur leur machine.
La souveraineté des données est un critère (RGPD, données sensibles).
Vous voulez des garde-fous (budget de steps, isolation, audit, HITL) sans les écrire vous-même.

Quand ne pas l’utiliser :

Vous voulez un SaaS entièrement managé et la souveraineté n’est pas une contrainte.
Vous cherchez un framework de raisonnement (chaîne de prompts, graphes). Apollia n’en est pas un. Il s’appuie sur des LLM existants et expose le pattern ReAct comme utilitaire (apollia.react).
Vous avez besoin d’un cluster multi-nœuds dès maintenant. Apollia est single-node sur la v0.1.

À qui s’adresse ce book

Le développeur Python qui livre des agents en production

Vous prototypez avec un notebook ou un script. Vous voulez passer en production sans réécrire votre logique ni assembler vous-même une stack d’observabilité.

→ Démarrez par l’Installation, puis les 4 quickstarts (Partie I). Vous aurez un agent fonctionnel en moins d’une heure.

Le développeur qui construit de zéro

Vous démarrez un nouveau projet et voulez adopter les bonnes pratiques dès le début.

→ Lisez ce book de façon linéaire. Les quickstarts (Partie I) montrent chaque pattern de bout en bout, puis les Parties II à VII couvrent chaque brique en détail.

Le tech lead ou l’architecte

Vous évaluez Apollia pour votre équipe ou votre client.

→ Lisez cette introduction, parcourez la Partie III (Ctx Protocol), la Partie V (gestion d’erreurs), et la Partie VIII (runtime Rust). Les annexes C (principes) et F (ADRs) résument l’architecture.

Comment est organisé ce book

Le book suit le pattern de The Rust Programming Language : des chapitres courts, code-first, qui s’enchaînent.

Partie I, Premiers pas. L’installation, puis quatre quickstarts (un par type d’agent : conversationnel, worker, director, orchestré). À la fin, vous avez quatre agents fonctionnels et vous savez lequel choisir pour quel besoin.

Partie II, Les décorateurs. Le contrat du SDK en quatre symboles : @agent, @skill, @on_message, @orchestrated. Ce que chacun fait, ce qu’il valide au load, ce qu’il interdit.

Partie III, Le protocole Ctx. Les quatorze services injectés dans chaque agent (ctx.llm, ctx.memory, ctx.tools, ctx.a2a, ctx.events, etc.). Un chapitre par service, signatures + exemples + erreurs typiques.

Partie IV, Design LLM-friendly. Comment annoter vos skills (Annotated, examples, TypedDict) pour qu’un LLM moyen génère des payloads valides du premier coup. Trois techniques, trois chapitres.

Partie V, Gestion des erreurs. Le modèle d’exceptions : DomainError pour les erreurs métier connues, NeedHumanInput pour les pauses HITL. Le boundary du SDK fait le reste.

Partie VI, Tests. Le mock apollia.testing.mock(MyAgent) et les assertions qui vont avec. Tests d’agents sans démarrer le runtime, sans LLM live, sans I/O réseau.

Partie VII, Outillage. Les deux commandes CLI dédiées aux auteurs : apollia inspect (validation statique du manifeste) et apollia new (scaffolding d’un nouveau projet d’agent).

Partie VIII, Le runtime Rust. Ce qui se passe sous le capot : acteurs Tokio, supervisor, API REST, sandbox, triggers, application Desktop, adaptateurs LangGraph/CrewAI. Utile pour l’opérateur et l’architecte.

Partie IX, Projet capstone. Un projet multi-agent end-to-end (workers spécialisés + director) qui consolide tout ce que vous avez appris. Vous reconstituez une mini-application de production.

Conventions

Les blocs de code sont exécutables tels quels :

# Commandes shell, à lancer dans votre terminal
apollia inspect agents/my-worker/agent.py

# Code Python : fichier agent complet ou extrait
from apollia import agent, skill
from apollia.types import Ctx

@agent(name="my-worker", version="0.1.0", description="…")
class MyWorker:
    @skill("my.skill", description="…")
    async def my_skill(self, value: str, ctx: Ctx) -> dict:
        return {"echo": value}

Les termes introduits pour la première fois apparaissent en gras. L’Annexe B (Glossaire) en donne la définition formelle.

Pré-requis

Python 3.10+
Rust (pour compiler depuis les sources), ou téléchargez le binaire précompilé
Un LLM accessible : llama.cpp en local (bundled), Ollama, ou une clé API (Anthropic, OpenAI, …)

Pas de Docker. Pas de Kubernetes. Pas de compte cloud.

Commençons.

→ Installation

Installation

Quickstart : agent conversationnel

Quickstart : agent worker

Quickstart : agent director

Quickstart : agent orchestré

Le décorateur `@agent`

Un agent Apollia est une classe Python décorée par @agent. Le décorateur fait trois choses au moment où le module est importé :

il valide la classe (signature __init__, présence d’au moins un handler) ;
il construit le manifeste statique (__apollia_manifest__) ;
il instancie la classe et expose l’instance comme attribut agent du module.

L’auteur n’écrit ni manifest(), ni agent = MyClass(), ni d’héritage de base. Tout est dérivé de la classe et de ses méthodes décorées.

Exemple minimal

from apollia import agent, skill
from apollia.types import Ctx

@agent(
    name="hello",
    version="0.1.0",
    description="A tiny hello-world agent.",
)
class Hello:
    @skill("hello.greet", description="Greet a person by name.")
    async def greet(self, name: str, ctx: Ctx) -> dict:
        return {"message": f"Bonjour, {name} !"}

Trois éléments suffisent : une classe, un appel @agent(...), et au moins une méthode décorée (@skill, @on_message ou @orchestrated). Le runtime peut maintenant charger ce fichier, lire module.agent.__apollia_manifest__, et router des invocations vers agent.greet(...).

Paramètres

@agent n’accepte que des arguments nommés (keyword-only). Trois sont obligatoires, les autres sont optionnels.

Obligatoires

Paramètre	Type	Rôle
`name`	`str`	Identifiant unique de l’agent (utilisé en CLI, A2A, registre).
`version`	`str`	Version sémantique (`"0.1.0"`, `"1.2.3"`).
`description`	`str`	Phrase courte affichée dans `apollia inspect`, l’UI Desktop et les outils tiers.

Les trois doivent être des chaînes non vides. Sinon, AgentConfigError au load.

Optionnels : dépendances déclarées

Paramètre	Type	Effet
`datasources`	`tuple[str, ...]`	Liste des YAML attendus dans `datasources/` (cf. Partie III).
`templates`	`tuple[str, ...]`	Liste des templates Jinja2 attendus dans `templates/`.
`secrets`	`tuple[str, ...]`	Liste des secrets attendus dans `ctx.secrets.get(...)`.
`tools_required`	`tuple[str, ...]`	Outils natifs dont l’agent dépend (`("file_read", "bash_executor")`).
`packages`	`tuple[str, ...]`	Packages PyPI dont l’agent dépend, installés au boot dans le venv isolé.

Le comportement diffère selon la ressource :

datasources, templates, secrets : gating strict côté SDK Python. Tout accès à un nom non déclaré (ctx.datasources.get("x")) lève une erreur immédiate. Ces trois listes sont la définition exhaustive de ce que l’agent peut lire.
tools_required : déclaration de disponibilité validée au boot par le runtime Rust. Si un outil listé est absent du catalogue, l’agent ne démarre pas (RequiredToolMissing). C’est l’application du principe fail-fast au démarrage. Le gating runtime au moment de l’appel relève du moteur de permissions (cf. chapitre 35), pas du manifeste seul.
packages : déclaration des dépendances PyPI installées dans le venv isolé de l’agent.

Bonne pratique : déclarer explicitement tout outil que l’agent appelle, même si le moteur de permissions est plus permissif. Cela permet au boot de signaler immédiatement un outil manquant, à apollia inspect de présenter un manifeste lisible, et à l’opérateur de comprendre la surface d’attaque d’un agent installé.

Optionnels : métadonnées et garde-fous

Paramètre	Type	Effet
`tags`	`tuple[str, ...]`	Tags libres pour discovery (`("file", "pdf")`).
`agent_type`	`str \| None`	Taxonomie (`"worker"`, `"system"`, …).
`memory_namespace`	`str \| None`	Préfixe d’isolation mémoire (par défaut `name`).
`shared_memory_namespaces`	`tuple[str, ...]`	Namespaces mémoire partagés en lecture/écriture.
`user_memory_write`	`bool`	Autorise les écritures sur `ctx.profile` (par défaut `False`).
`step_budget`	`dict \| None`	Override du `StepBudget` runtime (`{"max_steps": 30}`).

Exemple : worker avec dépendances et gating

from apollia import agent, skill, DomainError
from apollia.types import Ctx

@agent(
    name="pdf-worker",
    version="1.0.0",
    description="Read, extract and parse PDF files.",
    packages=("pypdf>=4",),
    tags=("file", "pdf"),
    agent_type="worker",
    datasources=("supported_mime_types",),
)
class PdfWorker:
    @skill("pdf.read_text", description="Extract text from a PDF file.")
    async def read_text(self, path: str, ctx: Ctx) -> dict:
        if not path.endswith(".pdf"):
            raise DomainError("UNSUPPORTED", f"Not a PDF: {path}")
        return {"text": _extract(path), "pages": _count(path)}

Vous découvrirez le détail de chaque mécanisme dans les parties suivantes : @skill au chapitre 7, ctx.datasources au chapitre 15, DomainError au chapitre 22.

Auto-instanciation (ADR-107)

À l’import, @agent exécute cls() et expose l’instance via module.agent. Pas de ligne agent = MyClass() à écrire ni à oublier. Le bridge PyO3 récupère cette instance par getattr(module, "agent").

Trois conséquences pratiques :

__init__ doit fonctionner sans argument. Toute config se lit depuis ctx au premier appel, ou via des constantes de classe.
Une seule classe @agent par module. Tenter d’en exposer deux lève AgentConfigError. Pour deux agents, deux fichiers.
L’import a un effet de bord. from agents.my_worker import MyWorker instancie déjà l’agent. Pour des tests purs sans instance, utilisez apollia.testing.mock.

Contraintes validées au load

Le décorateur applique le principe fail-fast : toute incohérence lève AgentConfigError à l’import, pas à la première invocation.

name, version, description non vides.
Chaque tuple de strings ne contient que des chaînes non vides.
memory_namespace est None ou une chaîne non vide.
step_budget est None ou un dict.
__init__ accepte zéro argument requis (en dehors de self).
La classe expose au moins un des trois handlers : @skill, @on_message ou @orchestrated.
@orchestrated est mutuellement exclusif avec @skill et @on_message.
Un seul @on_message par classe.

Toute violation est rattrapée par apollia inspect (cf. chapitre 27) avant déploiement.

Inspection après import

Le décorateur stocke deux attributs lisibles sur la classe :

meta = MyWorker.__apollia_agent_meta__
# {'name': 'pdf-worker', 'version': '1.0.0', 'description': '...', 'packages': ('pypdf>=4',), ...}

manifest = MyWorker.__apollia_manifest__
# {'name': '...', 'version': '...', 'skills': [...], 'on_message': None, ...}

C’est ce que lit apollia inspect, ce que sérialise le bridge PyO3, et ce sur quoi le runtime Rust appuie ses garde-fous.

ADRs

ADR-098 : Decorator-first comme contrat unifié
ADR-107 : Auto-instanciation et exposition au module

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

Le décorateur `@skill`

Un skill est une capacité atomique qu’un agent expose à l’extérieur. Il est appelable par la CLI, l’API REST, l’app Desktop, ou un autre agent via A2A. @skill marque une méthode async d’une classe @agent pour la déclarer comme telle.

La signature de la méthode est le schéma I/O. Pas de manifeste à écrire à la main : le SDK introspecte les annotations et produit le JSON Schema (cf. Partie IV).

Exemple minimal

from apollia import agent, skill
from apollia.types import Ctx

@agent(name="echo", version="0.1.0", description="Echo agent.")
class Echo:
    @skill("echo.say", description="Echo a value back to the caller.")
    async def say(self, value: str, ctx: Ctx) -> dict:
        return {"echoed": value}

Trois exigences :

la méthode est async def ;
elle prend ctx: Ctx en argument ;
elle retourne un dict JSON-sérialisable, ou lève une exception typée (cf. Partie V).

Signature

def skill(
    skill_id: str,
    *,
    description: str = "",
    requires_approval: bool = False,
    dangerous: bool = False,
    examples: list[dict] | None = None,
) -> Callable[[F], F]:

`skill_id`

Identifiant unique de la skill, en dot.snake_case minuscule. La regex appliquée est ^[a-z][a-z0-9_]*(\.[a-z][a-z0-9_]*)*$. Quelques exemples valides :

"greet"
"pdf.read_text"
"web.search_and_extract"
"file.list_directory"

Et quelques invalides qui lèvent AgentConfigError au load :

"Pdf.ReadText" (majuscules)
"pdf-read" (tirets)
"pdf..read" (segment vide)
"" (chaîne vide)

Le dot.namespace permet de regrouper plusieurs skills d’un même domaine. Convention recommandée : préfixer par un nom court qui rappelle l’agent ou la famille (pdf.*, web.*, chart.*).

`description`

Phrase courte affichée à l’invocant (autre agent, LLM, opérateur humain). Si elle est absente, le manifeste retombe sur la première ligne du docstring de la méthode. Toujours fournir l’une des deux : c’est ce que voit le LLM quand il décide d’appeler la skill.

`examples`

Liste optionnelle de payloads d’exemple. Chaque entrée est un dict JSON-sérialisable représentant un appel réaliste. Le tool descriptor LLM les exposera tel quel (cf. chapitre 20).

@skill(
    "pdf.read_text",
    description="Extract text from a PDF, optionally limited to a page range.",
    examples=[
        {"path": "/tmp/report.pdf"},
        {"path": "/tmp/report.pdf", "page_range": "1-10"},
    ],
)
async def read_text(
    self,
    path: str,
    page_range: str | None = None,
    ctx: Ctx = None,
) -> dict: ...

Le SDK ne valide pas les exemples contre le schéma inféré : cela créerait une dépendance circulaire au moment de la construction du manifeste. L’auteur est responsable de la cohérence.

`requires_approval` et `dangerous`

requires_approval=True : déclenche une pause HITL avant l’invocation. Le runtime suspend, demande à l’opérateur de valider, puis reprend (cf. chapitre 23).
dangerous=True : marque la skill comme potentiellement destructive ; affiche un bandeau d’avertissement dans l’UI Desktop et la CLI interactive.

Les deux flags sont indépendants. Un pdf.delete_pages typique est à la fois dangerous=True et requires_approval=True.

Plusieurs skills sur le même agent

Une classe @agent peut exposer autant de skills que de méthodes décorées. C’est le pattern standard d’un worker (cf. chapitre 3).

@agent(name="pdf-worker", version="1.0.0", description="PDF utilities.", agent_type="worker")
class PdfWorker:
    @skill("pdf.read_text", description="Extract text from a PDF.")
    async def read_text(self, path: str, ctx: Ctx) -> dict: ...

    @skill("pdf.count_pages", description="Count the pages of a PDF.")
    async def count_pages(self, path: str, ctx: Ctx) -> dict: ...

    @skill("pdf.extract_metadata", description="Read PDF metadata (author, title).")
    async def metadata(self, path: str, ctx: Ctx) -> dict: ...

apollia inspect listera les trois skills avec leur schéma inféré et leurs exemples.

Contraintes validées au load

@skill lève AgentConfigError à l’import si :

skill_id n’est pas une chaîne non vide ;
skill_id ne respecte pas la regex dot.snake_case ;
description n’est pas une chaîne ;
examples n’est pas une list[dict] ;
la méthode décorée n’est pas async def ;
@skill est appliqué deux fois sur la même méthode.

Le @agent parent enforce en plus :

pas de doublon de skill_id à travers les méthodes (deux @skill("pdf.read_text") ⇒ erreur) ;
mutuelle exclusion avec @orchestrated (cf. chapitre 9).

Anti-patterns

Ne pas rendre la méthode non-async :

# CASSE au load
@skill("foo.bar")
def bar(self, x: int, ctx): ...   # ⇒ AgentConfigError "must be 'async def'"

Ne pas retourner un AIPResult à la main :

# OK : retournez juste un dict
async def bar(self, x: int, ctx) -> dict:
    return {"value": x * 2}

# NON : AIPResult est interne au SDK
async def bar(self, x: int, ctx):
    return AIPResult.completed({"value": x * 2}).to_dict()  # ⇒ erreur runtime

Le boundary du dispatcher emballe automatiquement votre dict en AIPResult.completed et trappe vos exceptions en AIPResult.failed (cf. chapitre 22). Vous n’avez jamais à manipuler AIPResult directement.

Ne pas réutiliser le même skill_id pour plusieurs méthodes : AgentConfigError au load.

ADRs

ADR-098 : Decorator-first
ADR-099 : Signature inference comme schéma I/O
ADR-109 : AIPResult interne au SDK

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

Le décorateur `@on_message`

@on_message marque la méthode qui pilote le mode conversationnel d’un agent, c’est-à-dire l’interaction libre avec un humain dans l’app Desktop ou dans le chat CLI. C’est le pendant naturel de @skill : @skill répond à des invocations machine (CLI, A2A) avec des payloads structurés ; @on_message répond à du texte humain en flux.

Un agent peut combiner les deux. Un assistant peut exposer un @skill("inbox.summary") pour les autres agents et un @on_message pour discuter avec l’utilisateur, sur la même classe.

Exemple minimal

from apollia import agent, on_message
from apollia.types import Ctx, Message

@agent(
    name="apollia-guide",
    version="0.1.0",
    description="Friendly product coach for new Apollia users.",
)
class ApolliaGuide:
    SYSTEM_PROMPT = "You are a helpful product coach for Apollia OS."

    @on_message
    async def chat(
        self,
        message: str,
        history: list[Message],
        ctx: Ctx,
    ) -> str:
        full = ""
        async for token in ctx.llm.stream(
            messages=[
                {"role": "system", "content": self.SYSTEM_PROMPT},
                *history,
                {"role": "user", "content": message},
            ],
        ):
            ctx.events.emit_token(token)
            full += token
        return full

À chaque tour de chat, le runtime appelle agent.chat(message, history, ctx). La méthode décide elle-même comment construire la réponse : passer par ctx.llm.stream() (cas standard), interroger une mémoire, appeler des outils via ctx.tools.call(...), ou orchestrer une boucle ReAct via apollia.react(...) (cf. chapitre 4).

Signature

@on_message n’accepte aucun argument. C’est le décorateur le plus simple du SDK.

@on_message
async def chat(self, message: str, history: list[Message], ctx: Ctx) -> str: ...

La méthode doit :

être async def ; sinon AgentConfigError au load ;
accepter message: str, history: list[Message] et ctx: Ctx ;
retourner une str (la réponse finale, même si elle a déjà été streamée token par token).

`message`

La nouvelle entrée utilisateur, déjà décodée en str.

`history`

L’historique de la conversation sous forme de liste de Message (cf. apollia.types.Message). Chaque entrée a au moins un role ("user" ou "assistant") et un content. Le runtime tronque l’historique selon la stratégie de la session (sliding window, summarization).

`ctx`

Le Ctx complet, avec ses 14 services. En conversationnel, vous utiliserez surtout ctx.llm (génération), ctx.events.emit_token(...) (streaming UI), ctx.memory (rappels), parfois ctx.tools ou ctx.a2a (cf. Partie III).

Streaming

Le contrat de retour est une string finale, mais la valeur ajoutée du mode conversationnel est l’affichage incrémental. Émettez chaque token via ctx.events.emit_token(token) :

@on_message
async def chat(self, message, history, ctx) -> str:
    full = ""
    async for token in ctx.llm.stream(messages=[...]):
        ctx.events.emit_token(token)
        full += token
    return full

L’app Desktop et la CLI affichent les tokens en temps réel ; la valeur retournée est consignée dans l’historique. Les détails du streaming sont au chapitre 17.

Combiner `@skill` et `@on_message`

Aucun conflit : ce sont des entrées orthogonales. La même classe peut servir deux usages.

from apollia import agent, skill, on_message
from apollia.types import Ctx, Message

@agent(name="inbox", version="1.0.0", description="Email triage assistant.")
class Inbox:
    @skill("inbox.summary", description="Summarize unread emails.")
    async def summary(self, since: str, ctx: Ctx) -> dict:
        return {"emails": [...]}

    @on_message
    async def chat(self, message: str, history: list[Message], ctx: Ctx) -> str:
        ...

inbox.summary est invocable depuis un autre agent ou la CLI ; chat reçoit les messages humains côté UI. Les deux partagent la mémoire, les datasources et les secrets de l’agent.

Contraintes validées au load

@on_message lève AgentConfigError si :

la méthode n’est pas async def ;
la méthode n’est pas un callable.

@agent enforce en plus :

au plus un @on_message par classe (deux ⇒ AgentConfigError) ;
mutuelle exclusion avec @orchestrated.

Vous n’avez pas besoin de retomber sur l’absence de @on_message : si l’agent n’expose pas de handler conversationnel, le runtime renvoie une erreur claire à l’utilisateur qui tente de chatter avec lui. Inutile d’ajouter un stub vide.

Anti-patterns

Ne pas rendre la méthode non-async :

# CASSE au load
@on_message
def chat(self, message, history, ctx): ...   # ⇒ AgentConfigError

Ne pas déclarer plusieurs @on_message :

@on_message
async def chat_v1(self, ...): ...

@on_message
async def chat_v2(self, ...): ...   # ⇒ AgentConfigError au @agent

Ne pas retourner autre chose qu’une str. Le runtime journalise la valeur en historique ; un objet non-sérialisable casse la session.

ADRs

ADR-098 : Decorator-first
ADR-040 : Pattern agent conversationnel

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

Le décorateur `@orchestrated`

@orchestrated marque la classe entière comme agent orchestré : le runtime Rust (moteur ORIA, pour Observer, Reasoner, Actor) prend en charge la boucle d’exécution. L’auteur ne fournit que le system_prompt et, en option, un hook de post-traitement.

C’est la voie la plus déclarative du SDK. ORIA construit un plan à partir du system_prompt, l’exécute étape par étape, gère les appels d’outils, et restitue les résultats. L’auteur ne pilote ni la boucle ni le dispatch. Il décrit l’intention et laisse le runtime jouer.

Exemple minimal

from apollia import agent, orchestrated

@agent(
    name="email-triage",
    version="0.1.0",
    description="Sort and route incoming emails.",
)
@orchestrated(
    system_prompt="""
    Sort and route incoming emails using the available tools.

    For each email, choose one of three actions: archive, reply with a
    short acknowledgement, or flag for human review. The tools are
    `inbox.list`, `inbox.archive`, `inbox.reply`, `inbox.flag`.
    """
)
class EmailTriage:
    def on_plan_complete(self, step_results: dict) -> str:
        return "\n\n".join(s.get("text", "") for s in step_results.values())

Pas de méthode @skill ni de @on_message. La classe se résume à la description orchestrée et à un hook optionnel.

Signature

def orchestrated(*, system_prompt: str) -> Callable[[C], C]:

Un seul argument, obligatoire et keyword-only :

Paramètre	Type	Rôle
`system_prompt`	`str`	Description orchestrée : les instructions naturelles que ORIA utilise pour planifier et exécuter. Doit être non vide.

Le décorateur retourne la classe inchangée. Il y attache simplement la configuration {"system_prompt": ...} lue plus tard par @agent.

Hook optionnel : `on_plan_complete`

Si la classe définit une méthode on_plan_complete(self, step_results: dict) -> str, ORIA l’appelle avec les résultats de chaque étape une fois le plan terminé. Le retour est la réponse finale présentée à l’utilisateur.

Par défaut (sans hook), ORIA concatène les textes des étapes dans l’ordre. Vous overridez si vous voulez formater, filtrer, ou agréger différemment.

def on_plan_complete(self, step_results: dict) -> str:
    counts = {"archived": 0, "replied": 0, "flagged": 0}
    for step in step_results.values():
        action = step.get("action")
        if action in counts:
            counts[action] += 1
    return f"Archived {counts['archived']}, replied {counts['replied']}, flagged {counts['flagged']}."

Mutuelle exclusion avec `@skill` et `@on_message`

@orchestrated est exclusif des deux autres handlers. La classe est entièrement pilotée par ORIA, donc :

pas de @skill dans la classe, sinon AgentConfigError au load ;
pas de @on_message non plus ;
inversement, si vous voulez le moindre @skill ou @on_message, retirez @orchestrated.

# CASSE au load
@agent(name="bad", version="0.1.0", description="…")
@orchestrated(system_prompt="…")
class Bad:
    @skill("foo.bar")
    async def bar(self, ctx): ...   # ⇒ AgentConfigError "mutuellement exclusif"

La règle est cohérente : ORIA gère son propre dispatcher ; cohabiter avec d’autres serait ambigu sur quelle entrée prendre.

ORIA en bref

ORIA (Observer, Reasoner, Actor) est le moteur de plan dynamique côté Rust. Il fonctionne en trois temps :

Observer : lit le system_prompt, les outils disponibles (tools_required de @agent), et la tâche initiale.
Reasoner : appelle le LLM pour produire un plan structuré (étapes, dépendances).
Actor : exécute chaque étape via les outils, peut replanifier si une étape échoue (max 2 replanifications par défaut).

Le détail (cache de plans, observations par étape, intégration HITL) est dans la Partie VIII et le wiki.

Contraintes validées au load

@orchestrated lève AgentConfigError si :

system_prompt n’est pas une chaîne non vide ;
la cible décorée n’est pas une classe.

@agent enforce en plus :

exclusivité avec @skill / @on_message ;
présence d’au moins un handler (un agent qui n’aurait ni skill, ni on_message, ni orchestrated lève AgentConfigError).

Anti-patterns

Ne pas appeler vous-même ORIA dans la classe : c’est le runtime qui pilote.

Ne pas muter system_prompt dynamiquement à chaque tour. Le manifeste est statique : ce que vous écrivez au décorateur est ce qui sera utilisé. Pour ajuster le contexte à l’exécution, exploitez ctx.datasources (cf. chapitre 15).

Ne pas définir un __init__ qui prend des arguments. La règle d’auto-instanciation de @agent (cf. chapitre 6) s’applique également ici.

ADRs

ADR-022 : ORIA mode orchestré (architecture)
ADR-035 : Observation par étape
ADR-098 : Decorator-first

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

Vue d’ensemble du protocole Ctx

À chaque invocation d’un agent, le runtime injecte un objet ctx dans la méthode appelée. C’est l’unique surface par laquelle l’agent accède au backend Apollia : LLM, mémoire, outils, autres agents, secrets, événements. Tout passe par là.

Ctx n’est pas une classe mais un typing.Protocol. L’agent dépend d’un contrat, pas d’une implémentation. En production, le runtime injecte un objet PyO3 piloté par le Rust ; en test, apollia.testing.mock injecte un faux ctx qui implémente la même surface (cf. chapitre 24).

from apollia import agent, skill
from apollia.types import Ctx

@agent(name="demo", version="0.1.0", description="Demo agent.")
class Demo:
    @skill("demo.do", description="Pattern d'utilisation de ctx.")
    async def do(self, query: str, ctx: Ctx) -> dict:
        response = await ctx.llm.chat(system="You are helpful.", user=query)
        ctx.logger.info("llm done", extra={"latency_ms": response.latency_ms})
        await ctx.memory.record(f"queried: {query}", importance=0.4)
        return {"answer": response.content}

Les 14 services

ctx expose exactement 14 services. Aucun attribut « magique » au niveau racine en dehors de cette liste.

Service	Type	Rôle	Chapitre
`ctx.llm`	`LlmProxy`	Génération, streaming, embeddings	11
`ctx.memory`	`MemoryInterface`	Mémoire épisodique, sémantique, procédurale, FTS5	12
`ctx.tools`	`ToolProxy`	Outils natifs Apollia et serveurs MCP	13
`ctx.a2a`	`A2AInterface`	Appel d’autres agents, discovery	14
`ctx.datasources`	`DatasourcesInterface`	YAML déclarés dans le manifeste	15
`ctx.templates`	`TemplatesInterface`	Templates Jinja2 du package	15
`ctx.secrets`	`SecretsInterface`	Lecture seule de credentials	16
`ctx.events`	`EventsInterface`	Streaming UI, observabilité ReAct	17
`ctx.logger`	`logging.Logger`	Logging stdlib piped vers Rust tracing	17
`ctx.budget`	`BudgetView`	Vue lecture seule du StepBudget	17
`ctx.notify`	`NotifyInterface`	Notifications desktop, webhook	18
`ctx.profile`	`ProfileInterface`	Profil utilisateur canonique	18
`ctx.workspace`	`WorkspaceContext`	Règles et sections d’`APOLLIA.md`	18
`ctx.stt`	`SttInterface`	Speech-to-Text	18

À cela s’ajoute la fonction libre apollia.react(ctx, system, user, tools=..., max_steps=...) qui n’est pas un service de ctx mais utilise plusieurs services en interne (cf. chapitre 14).

Pourquoi un Protocol

Trois bénéfices concrets :

Autocomplete IDE. Tapez ctx. dans n’importe quel IDE qui comprend Python : les 14 services apparaissent, chacun avec ses méthodes typées.
mypy --strict passe sur un agent moyen. Aucun # type: ignore ni getattr(...) défensif.
Mock testing trivial. Le mock fourni par apollia.testing.mock implémente le Protocol sans héritage. Vous pouvez aussi écrire votre propre class FakeLlm qui n’expose que les deux méthodes que votre test utilise.

Le runtime Rust est tenu d’exposer exactement ces 14 services. Toute dérive entre la définition Python et l’implémentation Rust est détectée au boot par le validateur (cf. chapitre 27).

Catégoriser pour mémoriser

Les 14 services se rangent dans 5 catégories :

IA et raisonnement : ctx.llm, plus la fonction libre apollia.react.
Mémoire et profil utilisateur : ctx.memory, ctx.profile.
Outils et A2A : ctx.tools, ctx.a2a.
Données et contenu de l’agent : ctx.datasources, ctx.templates, ctx.secrets.
Observabilité et I/O annexes : ctx.events, ctx.logger, ctx.budget, ctx.notify, ctx.workspace, ctx.stt.

Cette taxonomie est non-normative. Elle aide juste à savoir où chercher.

Que faire si un service manque

Tous les services sont toujours injectés. Le cas où ctx.notify ou ctx.stt n’existerait pas n’arrive jamais en production : si le runtime ne pouvait pas fournir un service (configuration cassée, version incompatible), il refuserait de démarrer la tâche.

En test, un service peut être absent si votre mock customisé ne l’implémente pas. Dans ce cas, l’appel lève une AttributeError claire au runtime du test. Préférez apollia.testing.mock(MyAgent) qui injecte les 14 services en stubs.

Anti-patterns

Ne pas stocker ctx dans un attribut de l’agent (self.ctx = ctx). Chaque invocation reçoit son propre ctx lié à la tâche en cours ; le réutiliser hors de la tâche d’origine produit des comportements indéterminés (budget incorrect, mémoire fuyant entre tâches, événements émis sur la mauvaise session).

Ne pas importer un service en haut du fichier (from apollia.context.llm import LlmProxy). Ces modules définissent uniquement le Protocol ; ils n’ont aucune implémentation. Utilisez les attributs de ctx.

Ne pas dupliquer une fonctionnalité que ctx expose déjà. Par exemple, ne ré-implémentez pas un cache LLM : ctx.llm peut s’appuyer sur le cache du runtime configuré globalement. De même pour la mémoire, la recherche, et l’A2A.

ADRs

ADR-101 : Ctx exhaustif et typé via Protocol
ADR-098 : Decorator-first (le contrat qui injecte ctx)

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

`ctx.llm`

ctx.llm (Protocol LlmProxy) est l’unique surface d’accès au LLM. Quatre méthodes principales : complete, chat, stream, embed. Le backend utilisé est résolu par le LlmRouter du runtime (local llama.cpp, Ollama, Anthropic, OpenAI, Vertex), invisible côté agent sauf si l’agent veut forcer un backend.

`complete(messages)` : appel one-shot

response = await ctx.llm.complete(
    messages=[
        {"role": "system", "content": "You are a careful assistant."},
        {"role": "user", "content": "Summarize this transcript in 3 bullets."},
    ],
    temperature=0.3,
    max_tokens=500,
)

print(response.content)              # str : le texte final
print(response.latency_ms)            # int : ms de génération
print(response.usage.prompt_tokens)   # int
print(response.usage.completion_tokens)
print(response.usage.cost_usd)        # float : coût estimé

Signature complète :

async def complete(
    self,
    messages: list[dict[str, Any]],
    *,
    backend: str | None = None,
    temperature: float = 0.7,
    max_tokens: int | None = None,
) -> LlmResponse: ...

Argument backend : laissez None pour utiliser le backend par défaut configuré au niveau session. Forcez ("anthropic", "openai", "local", "ollama") quand l’agent a un besoin spécifique (vision, contexte long, latence locale).

LlmResponse expose content, latency_ms, et usage (avec prompt_tokens, completion_tokens, cost_usd).

`chat(system, user)` : raccourci deux messages

Si vous n’avez qu’un system prompt et un message utilisateur, le raccourci chat est plus lisible :

response = await ctx.llm.chat(
    system="You are a French-to-English translator.",
    user="Le chat dort sur le tapis.",
    temperature=0.0,
)
return {"translation": response.content}

Sous le capot, chat assemble [{"role": "system", ...}, {"role": "user", ...}] et appelle complete. Pas de différence de coût ni de performance.

`stream(messages)` : itération token par token

stream retourne un AsyncIterator[str]. Chaque itération produit un fragment de tokens, dans l’ordre. La méthode elle-même est synchrone : c’est l’itération qui est async for.

async for token in ctx.llm.stream(messages=[...]):
    ctx.events.emit_token(token)

Cas d’usage typique : le mode conversationnel (@on_message). Le streaming UI passe par ctx.events.emit_token(...) qui pousse chaque chunk vers l’app Desktop ou la CLI. Le détail des événements est au chapitre 17.

Annulation : si la tâche est annulée par l’utilisateur ou si le budget est épuisé, la prochaine itération lève une exception. Le runtime se charge de fermer proprement la connexion côté backend.

`embed(text)` : vecteur d’embedding

vec = await ctx.llm.embed(text="Le runtime Apollia est local-first.")
# vec : list[float] de longueur 768 (par défaut) ou 1024/1536 selon le backend

Le backend par défaut est un modèle GGUF local bundlé, qui n’envoie rien au réseau. Vous pouvez forcer un backend cloud si vous voulez la même métrique qu’un service tiers.

L’usage typique : peupler ctx.memory avec des entrées vectorisées pour rechercher par similarité plus tard. La recherche FTS5 reste l’option par défaut, l’embedding est utile pour les cas où la similarité sémantique compte plus que les mots-clés.

Choisir le bon backend

ctx.llm ne sait pas, et n’a pas à savoir, lequel des backends est en jeu. Trois patterns sont courants :

Aucun argument backend. Le LlmRouter choisit selon la configuration de la session. C’est l’option par défaut, à privilégier.
backend="local" explicite. Quand l’agent tient à rester offline (souveraineté). Le runtime fail-fast si aucun modèle local n’est configuré.
backend="anthropic" ou autre cloud. Quand l’agent a besoin d’une capacité spécifique (vision, contexte long). Pensez à déclarer la clé via secrets=("anthropic_api_key",) dans @agent (cf. chapitre 16).

Quand utiliser `apollia.react` au lieu de `ctx.llm`

ctx.llm.complete est un appel atomique : un prompt, une réponse. Quand l’agent doit appeler des outils en boucle (raisonner, agir, observer, recommencer), passez par apollia.react(ctx, ...) (cf. chapitre 14). C’est la même chose qu’une boucle manuelle avec ctx.llm, mais le runtime gère le step budget, le formatage des tool calls, et l’arrêt sur convergence.

Vision : images multimodales

Les messages peuvent contenir une liste de blocs au lieu d’une chaîne. Apollia fournit des helpers pour construire des blocs vision :

from apollia import text, image_from_path

messages = [
    {"role": "system", "content": "Describe the image in 2 sentences."},
    {
        "role": "user",
        "content": [
            text("Voici le diagramme du système."),
            image_from_path("/tmp/diagram.png"),
        ],
    },
]
response = await ctx.llm.complete(messages, backend="anthropic")

Seuls les backends cloud (Anthropic, OpenAI) supportent la vision aujourd’hui. Le backend local llama.cpp lève une erreur claire si on tente d’envoyer une image.

Anti-patterns

Ne pas appeler le SDK Anthropic / OpenAI en direct (from anthropic import Anthropic) depuis un agent. Vous perdez l’isolation, le budget, le cache, le routing, et la possibilité de basculer sur un backend local sans toucher au code.

Ne pas boucler en while True sur ctx.llm pour faire du tool calling à la main. Utilisez apollia.react (cf. chapitre 14) qui implémente la boucle proprement avec un step budget.

Ne pas stocker response.content dans une mémoire conversationnelle sans ctx.memory.record ou ctx.memory.remember. Sinon, l’historique vit dans une variable Python qui disparaît à la fin de la tâche.

Ne pas ignorer response.usage.cost_usd. Logguez-le via ctx.logger.info("llm done", extra={"cost_usd": response.usage.cost_usd}) pour suivre la consommation en observabilité.

ADRs

ADR-101 : Ctx exhaustif et typé
ADR-047 : Multi-LLM backend registry
ADR-057 : Prompt caching strategy
ADR-111 : Vision (typage MessageContent)
ADR-112 : Stream cleanup et rename

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

`ctx.memory`

ctx.memory (Protocol MemoryInterface) est la mémoire persistante de l’agent. Trois familles : mémoire épisodique (événements horodatés), mémoire sémantique (paires clé-valeur), mémoire procédurale (recettes apprises). Plus une recherche full-text FTS5 sur l’épisodique.

Le stockage est local, dans une base SQLite scopée par memory_namespace de l’agent (cf. chapitre 6). Aucune donnée ne quitte la machine.

Principe directeur (cf. principe #6) : la mémoire est à l’initiative de l’agent. Le runtime n’injecte jamais automatiquement un contexte mémoriel dans le prompt. C’est l’agent qui décide quoi rappeler, quand, et comment l’utiliser.

Mémoire épisodique : `record`, `search`

Un événement horodaté avec une importance entre 0 et 1.

event_id = await ctx.memory.record(
    "Utilisateur a demandé un résumé du rapport Q3.",
    importance=0.7,
    metadata={"file": "report-q3.pdf"},
)

Signature :

async def record(
    self,
    content: str,
    *,
    importance: float = 0.5,
    task_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    expires_in: timedelta | None = None,
) -> str:

Recherche full-text :

hits = await ctx.memory.search("rapport Q3", limit=5)
# hits : list[dict] avec keys 'content', 'importance', 'timestamp', 'metadata'

Le tokenizer FTS5 est configuré unicode61 (accents-insensible, tolère le français). Pour les recherches sémantiques par embedding, combinez avec ctx.llm.embed(...) et stockez le vecteur en metadata.

Mémoire sémantique : `remember`, `recall`, `forget`

Une paire clé-valeur stable et déduplicable.

await ctx.memory.remember(
    "user.preferred_language",
    "français",
    source="explicit_question",
    confidence=1.0,
)

lang = await ctx.memory.recall("user.preferred_language")  # str | None

Trois variantes de lecture :

recall(key) -> str | None : la valeur brute.
recall_entry(key, injection_reason="...") -> dict | None : l’entrée complète (valeur, source, confidence, timestamp). Le injection_reason est tracé pour l’observabilité.
recall_all(limit=100, injection_reason="...") -> list[dict] : toutes les paires, paginées.

Et forget(key) pour supprimer.

Mémoire procédurale : `learn_procedure`, `recall_procedure`

Pour les recettes répétables : une liste d’étapes associée à un déclencheur en langue naturelle.

await ctx.memory.learn_procedure(
    trigger="quand l'utilisateur dit 'résume ma boîte'",
    steps=[
        "appeler inbox.list_unread",
        "filtrer les emails à pièce jointe",
        "résumer chaque thread en 2 phrases",
        "présenter la liste par priorité décroissante",
    ],
)

procedures = await ctx.memory.recall_procedure(trigger="résume ma boîte")
# procedures : list[dict] avec 'steps', 'last_used', 'usage_count'

Cas d’usage : un agent qui apprend les habitudes de l’utilisateur sur la durée. Les étapes sont du texte libre, le LLM les ré-interprète au moment de l’usage.

Recherche FTS5 : pratique

search(query, limit=10) est la voie rapide pour retrouver une trace épisodique :

hits = await ctx.memory.search("réunion direction commerciale", limit=5)
for h in hits:
    print(h["timestamp"], h["importance"], h["content"][:80])

Le score de pertinence est implicite : les hits sont retournés par ordre de pertinence FTS5 décroissant, et l’importance enregistrée à record est un facteur secondaire.

Astuce de recall court : pour un agent conversationnel, recherchez sur le dernier message utilisateur avant d’appeler ctx.llm. Vous évitez des prompts à zéro contexte sans pour autant pré-charger la base entière.

Export et import

Pour la portabilité et les sauvegardes (cf. ADR-066).

snapshot = await ctx.memory.export()
# snapshot : dict JSON-sérialisable contenant épisodique + sémantique + procédurale

# plus tard, sur une autre machine ou après reset
await ctx.memory.import_data(snapshot)

L’import est additif : il ne supprime pas les entrées existantes. Pour repartir d’une base propre, supprimez le fichier SQLite côté runtime avant d’appeler import_data.

Quand utiliser quoi

Besoin	Outil
Tracer un événement utilisateur	`record`
Stocker un fait stable (préférence, paramètre)	`remember`
Retrouver un événement par mots-clés	`search`
Lire un fait nommé	`recall` ou `recall_entry`
Lister tous les faits	`recall_all`
Recette répétable	`learn_procedure` / `recall_procedure`
Profil utilisateur (`user.*`)	`ctx.profile` (chapitre 18)
Sauvegarder / migrer	`export` / `import_data`

Anti-patterns

Ne pas stocker des objets complexes par sérialisation manuelle (pickle, repr). Le storage est str. Utilisez du JSON sérialisé si vraiment besoin, et préférez des entrées atomiques.

Ne pas transformer ctx.memory en logger d’événements. C’est ctx.logger qui sert à tracer ce que fait l’agent. La mémoire sert à se souvenir de ce qui compte pour la prochaine tâche.

Ne pas appeler recall_all() puis filtrer en Python si vous savez ce que vous cherchez. Utilisez recall(key) direct ou search(query) avec un terme précis.

Ne pas écrire des clés sémantiques au préfixe user.* depuis un agent qui n’a pas user_memory_write=True dans son manifeste. Les écritures sont rejetées au runtime (cf. chapitre 18).

ADRs

ADR-007 : Mémoire à l’initiative de l’agent (principe #6)
ADR-054 : Memory episodic consolidation
ADR-066 : Memory export/import format
ADR-070 : Memory namespace project-scoped
ADR-087 : User profile redesign

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

`ctx.tools`

ctx.tools (Protocol ToolProxy) est la surface d’appel des outils natifs Apollia (file_read, bash_executor, file_grep, web_read, …) et des outils MCP exposés par des serveurs externes connectés.

Du point de vue de l’agent, il n’y a pas de différence : tout outil s’appelle par ctx.tools.call(name, input). Le routage (natif vs. MCP) est transparent.

Appeler un outil : `call`

result = await ctx.tools.call(
    "file_read",
    input={"path": "/tmp/report.txt", "max_bytes": 50_000},
)
content = result["content"]

Signature :

async def call(
    self,
    tool_name: str,
    input: dict[str, Any],
) -> dict[str, Any]: ...

L’argument input est un dict (pas des keyword args éparpillés). C’est le payload validé contre l’input_schema de l’outil côté runtime. Le retour est un dict conforme à l’output_schema.

Lister et décrire

names = ctx.tools.list_tools()
# names : list[str], par exemple ["file_read", "file_write", "bash_executor", ...]

descriptor = await ctx.tools.describe("file_read")
# descriptor : dict | None, métadonnées : description, input_schema, output_schema, tags

list_tools ne renvoie que les outils résolus (présents au catalogue, et non bloqués par les permissions de la session). C’est utile pour un agent director qui veut construire dynamiquement la liste d’outils à passer à apollia.react.

Outils natifs et outils MCP

Le nom de l’outil porte le routage :

Outil natif Apollia : "file_read", "bash_executor", "file_grep", etc. Résolu par le ToolRegistry Rust.
Outil MCP : préfixé par mcp:<server>/<name>. Exemple : "mcp:github/list_issues". Le runtime route vers le serveur MCP connecté correspondant.

Du code de l’agent, ça donne :

issues = await ctx.tools.call("mcp:github/list_issues", input={"repo": "owner/name"})

La liste des outils MCP disponibles dépend des serveurs configurés au niveau de la session. L’opérateur les active dans l’app Desktop ou via apollia mcp enable <server>.

Permissions et HITL

Tous les outils ne sont pas appelables en silence. Trois niveaux de gating au runtime :

Disponibilité : l’outil est-il dans le catalogue ? Si non, call lève une erreur UnknownTool.
Permission de session : le moteur de permissions à 3 couches (session / project / global) peut bloquer un outil. L’erreur retournée est tool not allowed for this session.
HITL : un outil marqué requires_approval déclenche une pause pour validation humaine. Le runtime suspend la tâche et la reprend après accord.

Du point de vue de l’agent, ces gates sont transparents : vous appelez, et soit ça passe, soit ça lève. Pas besoin d’interroger les permissions à l’avance.

Bonne pratique : déclarez les outils que votre agent utilise dans @agent(tools_required=("file_read", "bash_executor")). Cela force la validation au boot (fail-fast si un outil manque) et rend votre manifeste lisible (cf. chapitre 6).

Compteur d’appels

count = ctx.tools.tool_call_count()
# int : nombre d'appels d'outils émis par l'agent dans cette tâche

Utile pour des décisions adaptatives (si déjà 20 appels, arrêter de fouiller), même si le runtime applique son propre StepBudget en parallèle (cf. chapitre 17).

Exemple complet : lecture + filtrage

@skill("audit.list_large_files", description="List files larger than a threshold.")
async def list_large_files(self, root: str, min_kb: int, ctx: Ctx) -> dict:
    out = await ctx.tools.call(
        "bash_executor",
        input={"cmd": f"find {root} -type f -size +{min_kb}k"},
    )
    paths = out["stdout"].splitlines()
    descriptors = []
    for p in paths[:50]:
        meta = await ctx.tools.call("file_read", input={"path": p, "max_bytes": 0})
        descriptors.append({"path": p, "bytes": meta["size_bytes"]})
    return {"files": descriptors, "truncated": len(paths) > 50}

L’agent déclare bien tools_required=("bash_executor", "file_read") dans @agent pour bénéficier du fail-fast au boot.

Anti-patterns

Ne pas appeler subprocess.run, open() ou requests en direct depuis un agent. Ces appels contournent la sandbox, le budget, et l’audit trail. Utilisez ctx.tools.call("bash_executor", ...), ctx.tools.call("file_read", ...), ctx.tools.call("web_read", ...).

Ne pas boucler en for x in big_list: await ctx.tools.call(...) sans surveiller ctx.budget. Chaque appel consomme du step budget. Préférez un outil qui traite un batch (file_grep plutôt qu’une boucle de file_read).

Ne pas capturer une exception sur ctx.tools.call et la transformer en réponse “vide”. L’erreur est de l’information : laissez-la remonter, ou re-levez une DomainError typée (cf. chapitre 22).

Ne pas parser le name d’un outil pour décider du routage. Le préfixe mcp: est une convention, mais le routage est interne au runtime. Ne réinventez pas la dispatch.

ADRs

ADR-015 : ToolExecutor trait abstraction
ADR-043 : Décomposition atomique des outils
ADR-044 : Client MCP natif
ADR-061 : Permission engine 3 layers
ADR-082 : Tool governance unifiée
ADR-096 : Tool execution paths convergence

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

`ctx.a2a` et `apollia.react`

ctx.a2a (Protocol A2AInterface) est l’unique surface pour appeler d’autres agents. Quatre méthodes : invoke, discover, list_skills, skill_as_tool.

apollia.react est une fonction libre (pas une méthode de ctx) qui pilote une boucle ReAct : LLM → tool calls → LLM → … → réponse. C’est l’outil principal d’un agent director.

`ctx.a2a.invoke` : appeler une skill d’un autre agent

result = await ctx.a2a.invoke(
    "pdf.read_text",
    input={"path": "/tmp/report.pdf", "page_range": "1-3"},
    timeout_secs=60,
)
text = result["text"]

Signature :

async def invoke(
    self,
    skill_id: str,
    input: dict[str, Any] | None = None,
    *,
    timeout_secs: int = 120,
    **kwargs: Any,
) -> dict[str, Any]: ...

Le skill_id est l’identifiant dot.snake_case déclaré par le worker cible (cf. chapitre 7). Le dict de retour est le payload retourné par la skill, exactement comme si vous l’aviez appelée localement.

Si la skill cible lève une DomainError("CODE", "message"), invoke propage l’erreur. C’est cohérent : votre agent traite les erreurs A2A comme les erreurs locales.

`discover` : lire une SkillCard

card = await ctx.a2a.discover("pdf.read_text")
# card : dict | None, keys : skill_id, name, description, agent_name,
#        input_schema, output_schema

Utile pour décider dynamiquement si une skill existe avant de l’appeler, ou pour afficher sa description dans une UI d’agent director.

`list_skills` : inventaire

skills = await ctx.a2a.list_skills()
# skills : list[dict] de SkillCards, agrégeant toutes les skills installées

Retourne toutes les skills exposées par les agents installés et activés sur la machine. Utile pour scaffolder dynamiquement la liste d’outils d’un director.

`skill_as_tool` : convertir en tool descriptor LLM

descriptor = ctx.a2a.skill_as_tool("pdf.read_text")
# descriptor : dict au format Anthropic / OpenAI tool-use
#   {"name": "pdf.read_text", "description": "...",
#    "input_schema": {...}, "examples": [...]}

C’est la méthode synchrone qui transforme une skill A2A en descripteur compatible avec les tools=[...] que prennent les LLM. Cas d’usage principal : nourrir apollia.react (voir plus bas).

`apollia.react` : boucle ReAct prête à l’emploi

apollia.react est importée directement, pas via ctx :

from apollia import react

Elle pilote une boucle Reason+Act : appel LLM avec system+user et une liste de tools, exécution des tool calls demandés par le LLM, ré-appel LLM avec les résultats, jusqu’à une réponse finale ou épuisement du max_steps.

from apollia import agent, on_message, react
from apollia.types import Ctx, Message

@agent(name="research-director", version="0.1.0", description="Investigate questions.")
class ResearchDirector:
    SYSTEM_PROMPT = "You orchestrate research workers to answer questions thoroughly."

    @on_message
    async def chat(self, message: str, history: list[Message], ctx: Ctx) -> str:
        return await react(
            ctx,
            system=self.SYSTEM_PROMPT,
            user=message,
            tools=[
                ctx.a2a.skill_as_tool("web.search_and_extract"),
                ctx.a2a.skill_as_tool("web.summarize"),
                ctx.a2a.skill_as_tool("research.synthesize"),
            ],
            max_steps=15,
        )

Signature :

async def react(
    ctx: Ctx,
    system: str,
    user: str,
    *,
    tools: list[dict[str, Any]] | None = None,
    max_steps: int = 15,
    temperature: float = 0.3,
    stream: bool = True,
) -> str: ...

tools est une liste de descripteurs au format LLM tool-use. Vous l’assemblez avec ctx.a2a.skill_as_tool(skill_id) pour les workers A2A, ou avec await ctx.tools.describe(name) pour les outils natifs.
max_steps est le plafond de tours LLM. Si dépassé, le runtime LLM lève une erreur.
temperature est informationnelle aujourd’hui (run_tools utilise le default du backend).
stream=True émet un emit_thought à l’entrée de la boucle pour les dashboards d’observabilité.

Le retour est la string finale produite par le LLM. C’est ce que vous renvoyez à l’utilisateur dans un @on_message.

`apollia.react` vs `@orchestrated`

Deux voies pour faire raisonner un agent :

apollia.react : vous gardez le contrôle. Le code Python décide quand entrer dans la boucle, quels outils exposer, quoi faire de la sortie. Vous pouvez chaîner plusieurs react, brancher selon la réponse, faire de la pré et post-procédure.
@orchestrated : vous décrivez l’intention en system_prompt et laissez le moteur ORIA (côté Rust) piloter la boucle. Moins de code, moins de contrôle.

Choix typique :

Director qui exécute un workflow connu en quelques étapes ⇒ apollia.react.
Agent qui doit décider en autonomie complète quelle séquence d’outils utiliser ⇒ @orchestrated.

Erreurs `apollia.react`

react lève une DomainError("REACT_MAX_STEPS", ...) si max_steps <= 0 à l’appel. La boucle interne du LLM peut aussi lever une erreur si elle épuise max_steps sans converger ; cette erreur remonte telle quelle.

L’agent peut catcher localement pour tomber dans un fallback :

try:
    answer = await react(ctx, system="...", user=msg, tools=[...], max_steps=5)
except DomainError as exc:
    if exc.code == "REACT_MAX_STEPS":
        return "Désolé, je n'ai pas pu finir cette analyse dans les temps."
    raise

Anti-patterns

Ne pas appeler await ctx.a2a.skill_as_tool(...). La méthode est synchrone. Le await lèvera TypeError.

Ne pas boucler manuellement sur ctx.llm.complete avec des prompts qui simulent du tool calling. Utilisez apollia.react qui parse les tool calls correctement et gère le step budget.

Ne pas mélanger @orchestrated et apollia.react dans la même classe. Si la classe est décorée @orchestrated, le runtime ne lui injecte pas le dispatcher d’invocation A2A en direct. Choisissez l’un ou l’autre.

Ne pas lister explicitement les skills dans un list_skills côté director si vous savez quelles skills vous utilisez. Codez les skill_as_tool en dur : plus rapide, plus lisible, et le fail-fast au boot signalera une skill manquante.

ADRs

ADR-049 : A2A routing inter-agents
ADR-101 : Ctx exhaustif
ADR-102 : SDK A2A API unifiée
ADR-108 : SDK mailbox A2A suppression (alignement sur ctx.a2a)

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

`ctx.datasources` et `ctx.templates`

Deux services pour découpler le contenu modifiable (YAML, templates) du code de l’agent. L’opérateur ou un autre agent peut ajuster les datasources et les templates sans toucher au Python.

Tous les deux suivent le même principe : déclarés dans le manifeste, résolus au boot, accédés via ctx.<service>.get(...) ou .render(...).

`ctx.datasources` : YAML déclarés

Un fichier YAML par datasource, placé dans datasources/ du package de l’agent. Le runtime parse au boot et garde la valeur en mémoire.

Structure projet :

agents/veille-ia/
├── agent.py
├── datasources/
│   ├── topics.yaml
│   ├── sources.yaml
│   └── competitors.yaml
└── ...

Déclaration dans @agent :

from apollia import agent, skill
from apollia.types import Ctx

@agent(
    name="veille-ia",
    version="0.1.0",
    description="Competitive intelligence cycles.",
    datasources=("topics", "sources", "competitors"),
)
class VeilleIA:
    @skill("veille.run_cycle", description="Run one intelligence cycle.")
    async def run_cycle(self, ctx: Ctx) -> dict:
        topics = ctx.datasources.get("topics")
        sources = ctx.datasources.get("sources")
        return {"covered": len(topics["entries"]), "feeds": len(sources["rss"])}

L’API du Protocol :

def get(self, name: str) -> Any: ...
def list_names(self) -> list[str]: ...

get retourne le YAML parsé : un dict, une list, un scalaire, selon la racine du document.

Gating strict

Si vous appelez ctx.datasources.get("foo") sans avoir déclaré "foo" dans @agent(datasources=(...)), l’appel lève FileNotFoundError. Le manifeste est la liste exhaustive de ce que l’agent peut lire.

C’est volontaire. À tout instant, apollia inspect (cf. chapitre 27) peut afficher l’inventaire complet : « cet agent lit topics.yaml, sources.yaml, competitors.yaml et rien d’autre ».

Le workspace override

Si l’opérateur a placé un APOLLIA.md avec une section # topics au niveau du workspace, ctx.workspace.get("topics") retourne cette section. La convention est que le code prend la valeur du workspace en priorité, puis retombe sur le datasource local :

topics = ctx.workspace.get("topics") or ctx.datasources.get("topics")

Cf. chapitre 18 pour ctx.workspace.

`ctx.templates` : Jinja2 servi par le runtime

Les templates Jinja2 vivent dans templates/ du package. Ils sont rendus côté runtime (moteur minijinja en Rust), pas en Python.

Structure projet :

agents/veille-ia/
├── agent.py
├── templates/
│   ├── weekly-digest.j2
│   └── alert.j2
└── ...

Déclaration :

@agent(
    name="veille-ia",
    version="0.1.0",
    description="…",
    templates=("weekly-digest", "alert"),
)
class VeilleIA:
    @skill("veille.format_digest", description="Render the weekly digest.")
    async def format_digest(self, items: list, week: int, ctx: Ctx) -> dict:
        body = ctx.templates.render(
            "weekly-digest",
            items=items,
            week=week,
            generated_at=datetime.now().isoformat(),
        )
        return {"markdown": body}

API :

def render(self, name: str, **context: Any) -> str: ...
def list_names(self) -> list[str]: ...

Le nom du template n’inclut pas l’extension (weekly-digest, pas weekly-digest.j2).

Sandbox du moteur

Le runtime utilise minijinja qui n’expose pas {% include %}, {% import %}, ni les filtres dangereux par défaut. Utilisez les templates comme de simples formatters :

Boucles {% for item in items %}
Conditions {% if condition %}
Filtres natifs courants ({{ value | upper }}, {{ value | length }})

Les besoins plus avancés (sous-templates) se font en composition au niveau Python (rendre deux templates et concaténer).

Gating identique

Comme pour les datasources, un render("foo") sans "foo" dans templates=(...) lève une erreur. Le manifeste reste la définition exhaustive.

Pattern composite : datasource + template

Cas classique de la veille : assembler un digest à partir d’une source YAML et d’un template :

@skill("veille.format_alert", description="Format a single alert.")
async def format_alert(self, entry: dict, ctx: Ctx) -> dict:
    competitors = ctx.datasources.get("competitors")
    matched = next((c for c in competitors["names"] if c in entry["title"]), None)
    body = ctx.templates.render(
        "alert",
        entry=entry,
        competitor=matched,
        urgency="high" if matched else "normal",
    )
    return {"markdown": body, "urgency": "high" if matched else "normal"}

Le code reste très court. Toute la logique de formatting et de configuration vit hors du Python.

Anti-patterns

Ne pas ouvrir les fichiers à la main (open("datasources/topics.yaml")). Passez par ctx.datasources.get(...) qui résout depuis le bon répertoire selon l’installation (dev local vs. agent installé dans ~/.apollia/agents/).

Ne pas sérialiser des objets Python dans les datasources (yaml.dump(custom_obj)). Restez en types primitifs YAML (dict, list, str, int, float, bool). C’est ce que l’opérateur doit pouvoir lire et modifier sans devoir interpréter du Python.

Ne pas rendre des templates avec du code Python à l’intérieur. La sandbox minijinja ne l’exécutera pas et lèvera une erreur. Les templates sont des formatters, pas du scripting.

Ne pas muter les datasources depuis l’agent. La surface est en lecture seule par contrat. Pour persister du state, c’est ctx.memory (cf. chapitre 12).

ADRs

ADR-056 : Workspace context assembly (APOLLIA.md)
ADR-103 : SDK datasources et templates runtime

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

`ctx.secrets`

ctx.secrets (Protocol SecretsInterface) est la surface en lecture seule pour récupérer les credentials d’un agent : clés d’API, jetons OAuth, mots de passe d’intégrations.

Le stockage est chiffré (AES-256-GCM) dans une base locale gérée par apollia-auth, avec fallback keyring système (macOS Keychain, libsecret sur Linux). Les agents ne voient jamais le storage : ils demandent par nom et reçoivent la valeur ou None.

API

def get(self, key: str) -> str | None: ...
def has(self, key: str) -> bool: ...

Usage :

@skill("weather.fetch", description="Fetch weather for a city.")
async def fetch(self, city: str, ctx: Ctx) -> dict:
    api_key = ctx.secrets.get("openweather_api_key")
    if not api_key:
        raise DomainError("CONFIG", "openweather_api_key not configured")
    url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}"
    response = await ctx.tools.call("web_read", input={"url": url})
    return _parse(response["content"])

get retourne :

la valeur (str) si le secret est configuré et déclaré dans le manifeste,
None si le secret est déclaré mais pas encore configuré (l’agent doit lever une DomainError("CONFIG", ...) ou retomber sur un comportement dégradé).

has(key) est un raccourci pour get(key) is not None. Utile pour brancher tôt :

if not ctx.secrets.has("anthropic_api_key"):
    raise DomainError("CONFIG", "anthropic backend requires anthropic_api_key")

Déclaration dans `@agent`

Les secrets sont gated : un agent ne peut accéder qu’à ceux qu’il a déclarés.

from apollia import agent, skill, DomainError

@agent(
    name="weather-worker",
    version="0.1.0",
    description="Fetch weather data from public APIs.",
    secrets=("openweather_api_key",),
)
class WeatherWorker:
    @skill("weather.fetch")
    async def fetch(self, city: str, ctx: Ctx) -> dict: ...

ctx.secrets.get("not_declared") lève une erreur. Le manifeste est la liste exhaustive des credentials que cet agent peut lire.

apollia inspect (cf. chapitre 27) liste les secrets déclarés et signale ceux qui ne sont pas configurés (avertissement, pas erreur).

Comment l’opérateur configure un secret

L’opérateur n’édite pas un fichier. Trois voies :

App Desktop : écran « Secrets », recherche par nom, saisie de la valeur. Stockage chiffré local.
CLI : apollia secrets set openweather_api_key=... (prompt interactif si la valeur n’est pas passée en argument).
Flow OAuth : pour les intégrations OAuth (Google, Microsoft), l’opérateur clique sur « Connect » dans l’UI ; le runtime gère le flow PKCE et stocke le token.

L’agent ne participe à aucun de ces flows. Il consomme la valeur au runtime via ctx.secrets.get(...).

OAuth et refresh transparent

Pour les credentials OAuth, le runtime gère le refresh automatiquement. Si l’access token a expiré, ctx.secrets.get("google_oauth_access_token") retourne un token rafraîchi (côté apollia-auth, transparent pour l’agent).

L’agent peut donc supposer que la valeur retournée est valide au moment de l’appel. Si le refresh a échoué (refresh token expiré, compte révoqué), get retourne None et c’est à l’agent de gérer l’absence.

Pas d’écriture

Le Protocol n’expose aucune méthode d’écriture. C’est volontaire : un agent compromis ne doit pas pouvoir réécrire les credentials de l’utilisateur. La configuration se fait toujours par un canal séparé (UI, CLI), à l’initiative de l’opérateur.

Pour les écritures vers le profil utilisateur (non sensibles, par exemple user.preferred_language), c’est ctx.profile.set(...) qui est utilisé, et seulement si l’agent a user_memory_write=True dans son manifeste (cf. chapitre 18).

Anti-patterns

Ne pas hardcoder une clé d’API dans le code (API_KEY = "sk-..."). C’est l’erreur de sécurité la plus banale et la plus grave. Utilisez toujours ctx.secrets.get(...).

Ne pas logguer la valeur d’un secret. ctx.logger.info("starting", extra={"key": api_key}) finit dans tracing et potentiellement dans un fichier de log. Si vous voulez vérifier la présence, loggez juste le booléen has.

Ne pas retomber silencieusement quand get retourne None. Levez une DomainError("CONFIG", ...) claire : l’opérateur saura immédiatement qu’il doit configurer le secret.

Ne pas passer le secret en argument à un autre agent via A2A. Chaque agent doit déclarer ses propres secrets et les récupérer lui-même. Cela respecte le gating et l’audit trail.

ADRs

ADR-064 : OAuth2 PKCE keyring
ADR-094 : Linux keyring fallback strategy
ADR-104 : SDK secrets read-only gating

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

`ctx.events`, `ctx.logger`, `ctx.budget`

Trois services regroupés ici parce qu’ils servent l’observabilité de l’agent :

ctx.events émet des événements typés visibles dans l’UI (streaming, raisonnement ReAct, retries).
ctx.logger est un logging.Logger stdlib qui alimente le subscriber Rust tracing.
ctx.budget expose le StepBudget en lecture seule (combien reste-t-il avant que le runtime coupe).

`ctx.events` : événements typés

Quatre méthodes, et seulement ces quatre. Les événements de cycle de vie (task_started, step_completed, etc.) sont émis par le runtime, pas par l’agent.

def emit_token(self, delta: str) -> None: ...
def emit_thought(self, text: str, *, step: int) -> None: ...
def emit_retry(self, *, step: int, reason: str, count: int) -> None: ...
def emit_action_parse_error(
    self,
    *,
    step: int,
    raw: str,
    fatal: bool = False,
) -> None: ...

`emit_token(delta)` : streaming UI

Pousse un fragment de tokens vers l’app Desktop ou la CLI interactive. Le cas typique est @on_message qui boucle sur ctx.llm.stream :

async for chunk in ctx.llm.stream(messages=[...]):
    ctx.events.emit_token(chunk)

L’UI reçoit chaque chunk au fil de l’eau.

`emit_thought(text, step)` : raisonnement ReAct

À utiliser quand l’agent veut tracer un raisonnement intermédiaire. apollia.react émet déjà un emit_thought à l’entrée de boucle. Vous pouvez en émettre d’autres pour documenter une décision :

ctx.events.emit_thought("J'identifie 3 mentions du concurrent X dans le digest.", step=2)

Visible dans les dashboards d’observabilité, utile pour comprendre après coup pourquoi l’agent a pris telle branche.

`emit_retry(step, reason, count)`

Trace une tentative refaite après échec :

try:
    result = await ctx.tools.call("web_read", input={"url": url})
except Exception as exc:
    ctx.events.emit_retry(step=current_step, reason=str(exc), count=1)
    result = await ctx.tools.call("web_read", input={"url": url})  # second essai

`emit_action_parse_error(step, raw, fatal)`

À utiliser quand le parsing d’une réponse LLM échoue (JSON mal formé, action inconnue). apollia.react l’émet automatiquement quand le LLM produit une réponse mal structurée.

Pas d’événement custom

La liste est fermée. Si vous voulez tracer une information métier supplémentaire, utilisez ctx.logger.info(...) avec un extra={...}. Les extra arrivent dans tracing comme des champs structurés, exploitables côté observabilité.

`ctx.logger` : logging stdlib

Un logging.Logger du module stdlib, configuré pour piper les records dans le subscriber tracing Rust.

ctx.logger.info("starting cycle", extra={"topics_count": 12, "sources_count": 38})
ctx.logger.warning("rate limit approached", extra={"backend": "anthropic", "rate": 0.9})
ctx.logger.error("scrape failed", extra={"url": url, "error_code": "TIMEOUT"})

Trois règles :

Pas de format string. Préférez info("event name", extra={...}). Les champs structurés sont récupérables dans les dashboards.
Niveaux standards : debug, info, warning, error. Pas de niveaux custom.
extra est sérialisé. Restez en types primitifs (str, int, float, bool, list, dict). Pas d’objets Python opaques.

Le nom du logger est calculé automatiquement (apollia.agent.<name>). Vous n’avez pas besoin de getLogger(...).

`ctx.budget` : vue lecture seule du StepBudget

Le runtime applique des limites globales à chaque tâche : nombre de steps, nombre d’appels d’outils, temps wall-clock. Ces limites sont des garde-fous non négociables (principe #7). L’agent ne peut pas les contourner. Il peut en revanche les consulter :

@property
def steps_remaining(self) -> int: ...

@property
def tool_calls_remaining(self) -> int: ...

@property
def elapsed_seconds(self) -> float: ...

@property
def wall_clock_remaining(self) -> float | None: ...

Usage typique : décider de la profondeur d’une analyse en fonction du budget restant.

if ctx.budget.steps_remaining < 5:
    ctx.logger.info("budget low, switching to fast path")
    return await self._fast_summary(items, ctx)
return await self._deep_analysis(items, ctx)

wall_clock_remaining est None si la tâche n’a pas de timeout wall-clock configuré.

Pas d’écriture

Le Protocol n’expose aucune méthode pour ajuster le budget. Si vous voulez plus de budget pour une tâche donnée, c’est dans @agent(step_budget={"max_steps": 60}) au manifest level (cf. chapitre 6), pas via ctx.budget.

Quand utiliser quoi

Besoin	Service
Streamer la réponse LLM token par token	`ctx.events.emit_token`
Tracer un raisonnement ou une décision pour les dashboards	`ctx.events.emit_thought`
Tracer une donnée métier structurée pour les logs	`ctx.logger.info(..., extra=...)`
Signaler une erreur applicative	`ctx.logger.error(..., extra=...)` puis `raise DomainError(...)`
Adapter la profondeur selon le budget restant	`ctx.budget.steps_remaining`

Anti-patterns

Ne pas émettre des centaines d’événements emit_token ou emit_thought dans une boucle serrée. L’EventBus est performant mais pas illimité. Aggrégez quand c’est possible (un emit_thought par étape ReAct, pas par sous-action).

Ne pas utiliser print(...) ou logging.getLogger(__name__) à la place de ctx.logger. Ces calls ne sont pas wirés au subscriber Rust et n’apparaîtront pas dans les dashboards.

Ne pas stocker ctx.budget dans une variable et le réutiliser plus tard. La vue évolue à chaque appel d’outil et à chaque step LLM. Lisez-le au moment où vous en avez besoin.

Ne pas essayer d’émettre un événement custom (ctx.events.emit_my_thing). Le Protocol ne le permet pas. Passez par ctx.logger.info(..., extra={...}).

ADRs

ADR-026 : Observabilité complète et timeline
ADR-030 : EventBus + Tauri events
ADR-105 : SDK events types publics
ADR-106 : SDK logger structure

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

`ctx.profile`, `ctx.workspace`, `ctx.stt`, `ctx.notify`

Quatre services regroupés. Aucun ne mérite un chapitre entier, mais chacun est utile dans son contexte.

`ctx.profile` : profil utilisateur canonique

Le profil utilisateur global, stocké au niveau de la machine et lisible par tous les agents. C’est la mémoire « qui est l’utilisateur » (nom, langue préférée, rôle, fuseau horaire, etc.).

@property
def writable(self) -> bool: ...

async def get(self, key: str) -> str | None: ...
async def has(self, key: str) -> bool: ...
async def all(self) -> dict[str, str]: ...
def schema_keys(self) -> list[str]: ...

async def set(self, key: str, value: str) -> None: ...
async def update(self, entries: dict[str, str]) -> None: ...

Lecture (toujours autorisée)

Tout agent peut lire le profil :

name = await ctx.profile.get("user.name")
lang = await ctx.profile.get("user.preferred_language")
all_keys = await ctx.profile.all()

Écriture (gated)

Les méthodes set et update ne fonctionnent que si l’agent a déclaré user_memory_write=True :

@agent(
    name="onboarding",
    version="0.1.0",
    description="…",
    user_memory_write=True,
)
class Onboarding:
    @on_message
    async def chat(self, message: str, history, ctx: Ctx) -> str:
        if "je m'appelle" in message.lower():
            name = _extract_name(message)
            await ctx.profile.set("user.name", name)
            return f"Bonjour {name} !"

Si l’agent appelle set sans user_memory_write=True, le runtime lève une erreur claire à l’invocation. La règle est conservatrice : un seul agent (l’onboarding) devrait avoir l’autorisation d’écriture dans les conditions par défaut.

writable (booléen) permet de vérifier le statut :

if ctx.profile.writable:
    await ctx.profile.set("user.preferred_language", "fr")

Convention `user.*`

Les clés du profil utilisateur sont préfixées user. par convention (user.name, user.preferred_language, user.timezone). C’est la frontière logique avec ctx.memory qui est scopée à l’agent. La règle est documentée dans la mémoire utilisateur globale (cf. principes architecturaux).

`ctx.workspace` : rules et sections d’APOLLIA.md

Quand un workspace contient un fichier APOLLIA.md à sa racine, le runtime le parse au boot et expose le contenu via ctx.workspace. C’est la voie standard pour mettre des règles ou de la configuration partagée entre agents au niveau projet.

@property
def rules(self) -> str | None: ...           # alias de apollia_md
@property
def apollia_md(self) -> str | None: ...      # contenu brut du fichier

def get(self, title: str) -> str | None: ...  # section par titre

@property
def sections(self) -> list[dict[str, str]]: ...  # toutes les sections

Usage typique :

rules = ctx.workspace.rules
if rules:
    system_prompt = f"{base_prompt}\n\nWorkspace rules:\n{rules}"

Ou pour récupérer une section précise :

brief = ctx.workspace.get("Marketing brief")
# brief = contenu de la section "# Marketing brief" dans APOLLIA.md, ou None

Le pattern de fallback workspace puis datasource (cf. chapitre 15) :

topics = ctx.workspace.get("Topics") or ctx.datasources.get("topics")

Le workspace prend priorité quand il est présent. Le datasource sert de défaut embarqué avec l’agent.

sections retourne la liste complète si vous voulez itérer. Chaque entrée est {"title": "...", "body": "..."}.

`ctx.stt` : Speech-to-Text

Surface de transcription audio, backed par apollia-stt (whisper-rs en local par défaut).

async def transcribe(
    self,
    path: str,
    *,
    language: str | None = None,
    backend: str | None = None,
) -> str: ...

async def status(self) -> dict[str, Any]: ...

Usage :

text = await ctx.stt.transcribe("/tmp/meeting.wav", language="fr")

path est un chemin local vers un fichier audio (WAV, MP3, FLAC, M4A). Le moteur whisper-rs est packagé avec le runtime et tourne sur CPU ou GPU selon la configuration. Pas d’appel réseau par défaut.

language est un code ISO 639-1 ("fr", "en"). Si omis, whisper auto-détecte (légèrement plus lent).

status retourne l’état du backend (modèles disponibles, temps moyen, dernière erreur). Utile pour diagnostiquer.

Cas d’usage : un agent qui reçoit un fichier audio en input et doit le transcrire avant analyse LLM.

`ctx.notify` : notifications

Pour pousser une notification à l’utilisateur (desktop, webhook, autres canaux à venir).

async def publish(
    self,
    message: str,
    *,
    severity: str = "info",
    title: str | None = None,
    channel: str | None = None,
) -> None: ...

Usage :

await ctx.notify.publish(
    "Cycle de veille terminé : 3 alertes critiques détectées.",
    severity="warning",
    title="Veille IA",
)

severity : "info", "warning", "error". Le rendu de la notification dans l’app Desktop varie selon le niveau.

channel : None (utilise les canaux configurés au niveau session, par défaut le desktop), "desktop", "webhook". Les webhooks sont configurés au niveau opérateur (URL + headers).

publish est async mais retourne dès que la notification est mise en queue. Pas besoin d’attendre la délivrance pour continuer.

Bonne pratique : utilisez notify pour les événements dignes d’attention humaine (résultat critique d’une veille, échec d’un workflow long). Pour la trace technique courante, c’est ctx.logger ou ctx.events.emit_thought.

Anti-patterns

Ne pas stocker des informations sensibles dans ctx.profile (mots de passe, clés). C’est ctx.secrets.

Ne pas écrire à ctx.profile.set depuis un agent qui n’est pas explicitement le manager du profil. La règle par défaut : seul l’onboarding agent écrit. Les autres lisent.

Ne pas transcrire des fichiers volumineux (> 200 Mo) en bloc avec ctx.stt. Découpez en chunks audio plus petits si la durée dépasse 30 min : whisper-rs est rapide, mais reste séquentiel.

Ne pas spammer ctx.notify. Une notification par tâche significative, pas une par sous-étape. Le canal est respecté quand il reste rare.

ADRs

ADR-024 : Notifications trait + channel JSON
ADR-038 : Global user memory
ADR-041 : STT embarqué (whisper-rs)
ADR-056 : Workspace context assembly
ADR-087 : User profile redesign

(ADRs disponibles prochainement, cf. l’encadré “ADRs et wiki” en introduction.)

Descriptions de paramètres via Annotated

Exemples de payloads

Schémas via TypedDict

DomainError

NeedHumanInput

apollia.testing.mock

Assertions

Suites d’évaluation

apollia inspect

apollia new : scaffolding

Vue d’ensemble du runtime Rust

Acteurs Tokio et Supervisor

API REST et configuration

L’application Desktop

La CLI complète

Adapter LangGraph / CrewAI

Outils, sandbox, permissions

Triggers

Capstone : vue d’ensemble

Capstone : architecture multi-agent

Capstone : implémentation des workers

Capstone : director et résultat

Diagrammes d’Architecture

C4 — Vues architecturales

C4 — Vue Contexte

C4 Context

C4 — Vue Container (16 crates)

C4 Container

C4 — Composants Runtime Core (acteurs Tokio)

C4 Component

Architecture — Application Desktop (Tauri v2 + Svelte 5)

Desktop Architecture

Architecture — Python SDK

SDK Architecture

Machines d’état

Machine d’état — Agent (ProcessState)

ProcessState

Machine d’état — Tâche (TaskState)

TaskState

Machine d’état — Session de chat

ChatSession State

Machine d’état — Circuit Breaker (ResilienceLayer)

Circuit Breaker

Séquences — Démarrage & arrêt

Séquence — Démarrage Supervisor (16 phases)

Supervisor Startup

Séquence — Configuration → Acteurs (démarrage ordonné)

Config to Actors

Séquence — Graceful Shutdown (SIGTERM → drain → exit)

Graceful Shutdown

Séquences — Exécution des tâches

Séquence — Cycle de vie d’une tâche

Task Lifecycle

Séquence — Boucle ORIA (Direct + Orchestré)

ORIA Loop

Séquence — ORIA Mode Orchestré (ActorLoop)

ORIA Orchestrated

Séquence — Boucle ReAct run_tools

ReAct run_tools

Séquence — Bridge AIP (Rust ↔ Python via PyO3)

AIP Bridge

Séquence — HITL Flow complet (approve / reject)

HITL Flow

Séquences — Outils & intégrations

Séquence — Appel outil natif

Tool Call Native

Séquence — Appel outil MCP

Tool Call MCP

Séquence — Cycle de vie session MCP (lazy start → handshake → call)

MCP Session Lifecycle

Séquence — Appel LLM (ctx.llm.chat / complete / stream)

LLM Call

Séquence — Routing Multi-LLM (binding par agent)

Multi-LLM Routing

Séquences — Mémoire & observabilité

Séquence — Mémoire (record + search FTS5)

Memory Usage

Séquence — Timeline Aggregation (5 sources → chronologie unifiée)

Timeline Aggregation

Séquences — Chat & STT

Séquence — Chat Libre (ReAct + streaming + mémoire)

Chat Libre

Séquence — Injection User Memory dans le chat

Chat User Memory

Séquence — Résumé de conversation (context window management)

Conversation Summarize

Séquence — Speech-to-Text (hotkey → transcribe → clipboard)

STT Flow

Séquence — Onboarding conversationnel (5 topics)

Onboarding Flow

Séquences — Triggers & notifications

Séquence — Trigger Fire (Cron / FileWatch / Webhook)

Trigger Fire

Séquence — CRUD Configuration opérationnelle (SQLite)

Config CRUD

Séquence — Dispatch des notifications (event → channel)

Notification Dispatch

A2A Routing

Séquence — Discovery + Invocation A2A (happy path)

A2A Discovery Invoke

Séquence — Garde-fous A2A (max_hops, cycle_detected)

A2A Guards

Séquence — Chaîne A2A complète (A -> B -> C, happy path + CycleDetected)

A2A Full Chain

Séquence — Onboarding v2.1 complet (ADR-086)

Onboarding v2.1

Permissions & Sécurité

Séquence — Moteur de permissions 3 couches (SafeList / PrefixRules / HITL)

Permission Engine

Worker Agents (32)

Séquence — Cycle de vie Worker Agent (manifest → SYSTEM_PROMPT → ReAct)

Worker Agent Lifecycle

Séquence — Installation d’agents (bundled + communautaire)

Agent Install

Annexe B. Glossaire

Annexe C. Principes architecturaux

Les 8 principes non-négociables qui guident chaque décision dans Apollia OS. Chacun a été forgé par un problème réel rencontré dans la phase SaaS précédente ou par l’analyse rigoureuse des besoins du projet.

Principe #1 : Local-first, toujours

Formulation : Aucun octet de données utilisateur ne quitte la machine sans une action explicite du développeur.

Concrètement : runtime Rust entièrement local, mémoire SQLite locale, modèles d’embedding GGUF locaux, audit trail SQLite local, aucun telemetry.

Pourquoi : L’architecture SaaS cloud précédente rendait impossible la souveraineté réelle des données. Les retours des prospects PME étaient clairs : “On veut bien essayer, mais nos données client ne peuvent pas sortir de chez nous.” La solution n’était pas d’améliorer les garanties contractuelles. Il fallait rendre le cloud techniquement inutile.

Conséquence architecturale : Toute dépendance à un service externe est optionnelle et se dégrade gracieusement. SQLite remplace PostgreSQL/Redis/Qdrant.

Principe #2 : Zéro dépendance externe côté runtime

Formulation : Le binaire Apollia OS doit fonctionner sur n’importe quel Linux avec zéro installation préalable.

Concrètement : pas de Docker requis, pas de Node.js, pas de base de données externe, pas de Python côté runtime (PyO3 intègre l’interpréteur), un seul fichier binaire.

Pourquoi : Le projet précédent nécessitait plusieurs services d’infrastructure à déployer et maintenir. Chaque service était une source de friction à l’installation et un composant à maintenir. Pour un runtime qui cible des développeurs individuels et des DSI réticentes, la complexité opérationnelle est un veto commercial.

Conséquence architecturale : cargo install apollia-os suffit. SQLite remplace PostgreSQL + Qdrant + Redis. Les namespaces Linux remplacent Docker. PyO3 intègre Python.

Principe #3 : Contrat minimal, friction zéro

Formulation : Un agent existant doit pouvoir tourner dans Apollia OS avec moins de 10 lignes de code d’adaptation.

Concrètement : AIP supporte le duck typing Python, manifest() et run() suffisent pour un agent minimal, wrappers fournis pour LangGraph et CrewAI.

Pourquoi : Les runtimes qui imposent un framework ont une courbe d’adoption élevée. Apollia OS résout un problème d’infrastructure. Si adopter la solution nécessite de réécrire l’agent, la solution crée autant de travail qu’elle en économise.

Conséquence architecturale : AIP = duck typing. hasattr(agent, 'manifest') and hasattr(agent, 'run') suffit à la validation. La classe de base est optionnelle.

Principe #4 : Fail fast, pas de surprises à l’exécution

Formulation : Toute erreur détectable au démarrage doit être détectée au démarrage, jamais silencieusement au milieu d’une tâche.

Concrètement : validation stricte du manifest à INITIALIZING, résolution des tools_required au démarrage, connexion aux serveurs MCP à INITIALIZING.

Pourquoi : Un agent qui démarre avec succès et plante à la 3ème étape de sa 2ème tâche parce qu’un outil n’est pas disponible est un désastre en production. “Fail fast” transforme des bugs de production en erreurs de configuration détectables avant le déploiement.

Conséquence architecturale : Distinction explicite tools_required vs tools_optional. DEGRADED vs STOPPED pour les outils manquants. L’agent ne passe à ACTIVE que si tout est prêt.

Principe #5 : Un acteur, une responsabilité

Formulation : Le Runtime Core n’est pas un monolithe interne. Chaque responsabilité est un acteur Tokio distinct.

Concrètement : EventBus diffuse uniquement, AgentRegistry inventorie uniquement, TaskRouter dispatche uniquement. Chaque acteur communique par messages, jamais par état partagé.

Pourquoi : Le projet SaaS précédent avait des services qui faisaient trop de choses. Quand un bug apparaissait, il était difficile de déterminer dans quelle couche il se trouvait. Le modèle acteur Tokio force la séparation des responsabilités par construction.

Conséquence architecturale : Pattern mpsc::channel + état interne + JoinHandle Tokio pour chaque acteur. Pas d’état partagé entre acteurs (pas de Arc<Mutex<...>> traversant les frontières).

Principe #6 : La mémoire à l’initiative de l’agent

Formulation : Le runtime n’injecte jamais automatiquement de mémoire dans le contexte d’un agent. C’est toujours l’agent qui décide ce qu’il récupère et comment il l’utilise.

Concrètement : ctx.memory.search est appelé explicitement, le runtime ne pré-charge pas de contexte mémoriel dans le prompt.

Pourquoi : La “mémoire automatique” génère des appels LLM non contrôlés, des coûts imprévisibles, et des comportements difficiles à debugger. Chaque injection automatique ajoute un appel LLM et 1-3 secondes de latence.

Conséquence architecturale : MemoryInterface est une API appelée explicitement. Pas de hook de lifecycle qui injecte automatiquement. La consolidation sera une feature opt-in v1.0, jamais un comportement par défaut.

Principe #7 : Les garde-fous sont non négociables

Formulation : Tout agent, quel que soit son code, est soumis aux limites du StepBudget et du ResilienceLayer.

Concrètement : max_steps, max_tool_calls, et wall_clock_timeout sont appliqués par le runtime Rust. Un agent ne peut pas se soustraire à ces limites depuis son code Python.

Pourquoi : Les boucles infinies et les coûts LLM incontrôlés sont des risques fréquemment observés dans les déploiements d’agents IA. En production PME, ce type d’incident est inacceptable. Le runtime doit être la couche de sécurité sur laquelle on peut compter indépendamment de la qualité du code de l’agent.

Conséquence architecturale : StepBudget implémenté dans ExecutionCoordinator (Rust), pas dans l’agent Python. L’agent reçoit ctx.step_budget en lecture seule pour adapter son comportement proactivement.

Principe #8 : La CLI est pour les humains, l’API est pour les machines

Formulation : La CLI doit être utilisable par un administrateur PME non-développeur. L’API REST doit être exploitable par n’importe quel script bash.

Concrètement : commandes de niveau 1 lisibles sans documentation (start(), stop(), status, run()), --json disponible sur toutes les commandes, TTY auto-détecté, exit codes standards Unix.

Pourquoi : La CLI est la première impression d’Apollia OS. Une CLI cryptique qui suppose une connaissance de l’architecture interne est un obstacle à l’adoption. Les meilleures CLIs techniques (docker, kubectl, git) ont en commun un onboarding progressif et des messages d’erreur actionnables.

Conséquence architecturale : Pattern noun verb cohérent. apollia-os sans argument explique les commandes disponibles. --json global, pas par commande.

Résumé

#	Principe	Résumé
1	Local-first, toujours	Zéro cloud dans le chemin d’exécution
2	Zéro dépendance externe	Un binaire, aucun service requis
3	Contrat minimal	Duck typing, 10 lignes d’adaptation maximum
4	Fail fast	Les erreurs détectables au démarrage le sont au démarrage
5	Un acteur, une responsabilité	Modèle acteur Tokio sans état partagé
6	Mémoire à l’initiative de l’agent	Pas d’injection automatique, coûts prévisibles
7	Garde-fous non négociables	StepBudget et résilience appliqués par le runtime
8	CLI humaine, API machine	Deux audiences, deux interfaces, un seul outil

Annexe D. Roadmap

Annexe E. Vision et positionnement

Annexe F. Index des ADRs

Annexe G. FAQ

Keyboard shortcuts

Apollia OS