Routing, path & query
TL;DR — Dans FastAPI, une route est une fonction Python décorée (
@app.get("/items/{id}")) où chaque paramètre est typé : FastAPI lit la signature, décide si un paramètre vient du chemin ({id}dans l'URL), de la query string (?limit=10), du corps ou des dépendances, le valide via Pydantic, le convertit (string →int/UUID/enum), et génère l'OpenAPI automatiquement. La règle mentale tient en une phrase : un nom de paramètre qui apparaît entre accolades dans le path est un path param ; tout le reste qui est un scalaire est un query param. Le reste de cette leçon transforme cette règle en réflexes de senior — ordre des routes, validation avecAnnotated[... , Query(...)],EnumvsLiteral, pièges de production (énumération de routes, injection, cardinalité des params) — et la branche directement sur le cas qui t'intéresse : exposer un agent LLM (Anthropicclaude-opus-4-8) derrière une API HTTP propre.
🧠 Mental model
Si tu viens de NestJS, tu connais déjà le concept sous une autre forme. En Nest tu écris :
@Get(':id')
findOne(@Param('id') id: string, @Query('verbose') verbose: boolean) {}Tu décores chaque argument pour dire d'où il vient. FastAPI inverse la charge : il devine l'origine à partir du nom et du type, et tu n'annotes explicitement que pour ajouter de la validation. C'est plus magique, donc il faut comprendre l'algorithme de résolution sinon tu te fais surprendre.
L'analogie : le routeur est un agent des douanes
Imagine chaque requête HTTP comme un voyageur qui arrive à la frontière. Le routeur FastAPI est un douanier qui pose trois questions, dans cet ordre :
- « Ton nom est-il écrit sur le panneau d'entrée ? » → si le nom du paramètre apparaît dans le template du chemin (
/agents/{agent_id}), c'est un path param. Obligatoire par construction (l'URL ne matche pas sinon). - « Es-tu un objet simple sans badge spécial ? » → un scalaire (
int,str,bool,float,Enum) qui n'est pas dans le path devient un query param. Optionnel s'il a une valeur par défaut, obligatoire sinon. - « As-tu un passeport complexe (un modèle) ou un laissez-passer interne ? » → un modèle Pydantic devient le body ; un
Depends(...)est résolu par l'injection de dépendances.
Requête : GET /agents/ag_42/messages?limit=20&model=opus
│
┌──────────────────────────┼───────────────────────────┐
▼ ▼ ▼
path: {agent_id} query: limit, model (pas de body sur GET)
"ag_42" ── str 20 ── int (validé ≥1) "opus" ── Enum ModelChoice
│ │ │
└──────────┬───────────────┴────────────┬───────────────┘
▼ ▼
validation Pydantic OpenAPI auto-généré
│ (/docs, /openapi.json)
▼
handler(agent_id, limit, model) ← types Python natifsLe point clé que les ex-PHP/TS sous-estiment : FastAPI ne te donne jamais une string brute si tu as typé int. À l'intérieur du handler, limit est un int, validé, ou la requête a déjà été rejetée en 422 avant que ton code ne tourne. C'est le même contrat que les pipes de validation Angular/Nest, mais gratuit et déclaratif.
Le cœur : déclarer des routes correctement
Une route minimale, typée bout en bout
from fastapi import FastAPI
app = FastAPI(title="Agent Gateway")
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok"}async def est l'idiomatique : les handlers FastAPI tournent dans une boucle d'événements, et tout ton travail I/O (appel au LLM, base de données, cache) sera await-able. Tu peux écrire def (sync) — FastAPI le pousse alors dans un threadpool pour ne pas bloquer la boucle — mais pour un service qui appelle des API distantes, async est le bon défaut. On y revient en production.
Path param : typé, donc converti et validé
@app.get("/agents/{agent_id}")
async def get_agent(agent_id: int) -> dict[str, int | str]:
# agent_id est DÉJÀ un int ici. /agents/abc → 422 automatique.
return {"agent_id": agent_id, "kind": "agent"}/agents/42 → agent_id == 42 (un int). /agents/abc → réponse 422 Unprocessable Entity avec un corps JSON détaillant l'erreur, sans que ton handler soit appelé. Tu n'écris aucun int(...), aucun try/except. C'est la différence de fond avec Express/Laravel où tu reçois toujours du texte.
Query param : présence d'une valeur par défaut = optionnel
@app.get("/agents/{agent_id}/messages")
async def list_messages(
agent_id: int,
limit: int = 20, # query, optionnel (défaut 20)
cursor: str | None = None, # query, optionnel (peut être absent)
include_thinking: bool = False, # query, bool : ?include_thinking=true
) -> dict[str, object]:
return {
"agent_id": agent_id,
"limit": limit,
"cursor": cursor,
"include_thinking": include_thinking,
}Trois choses de senior à retenir :
str | None = None(syntaxe Python 3.12) signifie « optionnel, peut être omis ».str = Nonesans le| Noneest un bug de typage : le défaut viole le type. Mets toujours le| Nonequand le défaut estNone.boolest parsé intelligemment :?include_thinking=true|1|yes|on→True. C'est pratique mais source de confusion pour les consommateurs ; documente la casse attendue.- Un query param sans défaut est obligatoire :
async def f(q: str)→ omettre?q=renvoie 422.
La façon idiomatique de valider : Annotated[..., Query(...)]
Voici le réflexe moderne (FastAPI ≥ 0.95, et la seule forme à enseigner aujourd'hui). On enrichit le type avec des contraintes sans casser la signature :
from typing import Annotated
from fastapi import FastAPI, Path, Query
app = FastAPI()
@app.get("/agents/{agent_id}/messages")
async def list_messages(
agent_id: Annotated[int, Path(ge=1, description="Identifiant interne de l'agent")],
limit: Annotated[int, Query(ge=1, le=100, description="Taille de page")] = 20,
q: Annotated[str | None, Query(min_length=2, max_length=120)] = None,
tags: Annotated[list[str] | None, Query()] = None, # ?tags=a&tags=b → ["a", "b"]
) -> dict[str, object]:
return {"agent_id": agent_id, "limit": limit, "q": q, "tags": tags}Annotated[T, Query(...)] est supérieur à l'ancienne forme limit: int = Query(20, ...) parce que :
- Le défaut reste un vrai défaut Python (
= 20), donc la fonction est appelable normalement dans tes tests unitaires sans passer par FastAPI. Annotatedest composable et relisible par les linters/IDE.list[str]en query gère nativement la répétition (?tags=a&tags=b) — impossible à exprimer proprement autrement.
❌ La façon « presque juste » qui se trompe (à ne pas faire) :
python# ANTI-PATTERN — défaut mutable et validation manuelle dans le corps @app.get("/agents/{agent_id}/messages") async def list_messages(agent_id, limit="20", tags=[]): # noqa limit = int(limit) # tu refais le travail de FastAPI if limit < 1 or limit > 100: # validation manuelle = bug en attente raise HTTPException(400, "bad limit") ...Problèmes :
agent_idnon typé (aucune conversion ni doc),limit="20"reçu en string et reconverti à la main,tags=[]est un défaut mutable partagé entre requêtes (le classique piège Python), et la validation est impérative donc divergente de l'OpenAPI. FastAPI rejette d'ailleurstags=[]comme défaut — mais l'esprit est là : ne réimplémente jamais ce que la signature typée te donne gratuitement.
Enum et Literal : contraindre les valeurs à un ensemble fini
Quand un query param ne doit prendre qu'un jeu de valeurs connu — par exemple le choix du modèle Anthropic — n'utilise pas un str libre. Utilise un Enum (ou Literal) : FastAPI le valide, le documente dans /docs sous forme de menu déroulant, et rejette tout le reste en 422.
from enum import Enum
class ModelChoice(str, Enum):
opus = "claude-opus-4-8"
sonnet = "claude-sonnet-4-6"
haiku = "claude-haiku-4-5"
@app.get("/agents/{agent_id}/quote")
async def quote(
agent_id: int,
model: ModelChoice = ModelChoice.sonnet, # query, contraint
) -> dict[str, str]:
# model.value == "claude-sonnet-4-6" si non précisé
return {"agent_id": str(agent_id), "model": model.value}?model=gpt-4 → 422. ?model=claude-opus-4-8 → ok. Literal["fast", "balanced", "max"] est l'alternative légère quand tu n'as pas besoin d'une classe réutilisable.
L'ordre des routes compte : le piège le plus fréquent
FastAPI évalue les routes dans l'ordre de déclaration, première qui matche gagne. Donc une route statique doit être déclarée avant une route paramétrée qui pourrait l'absorber.
# ✅ BON ordre
@app.get("/agents/active") # statique d'abord
async def list_active() -> dict[str, str]:
return {"filter": "active"}
@app.get("/agents/{agent_id}") # paramétrée ensuite
async def get_agent(agent_id: int) -> dict[str, int]:
return {"agent_id": agent_id}# ❌ MAUVAIS ordre — /agents/active n'est JAMAIS atteint
@app.get("/agents/{agent_id}") # capture "active" → tente int("active") → 422
async def get_agent(agent_id: int): ...
@app.get("/agents/active") # route morte
async def list_active(): ...Avec agent_id: int, le bug devient un 422 bruyant (« active n'est pas un int »). Avec agent_id: str, c'est pire : silencieux, list_active est simplement jamais appelée. Règle : du plus spécifique au plus générique.
Path complet (segments avec /)
Par défaut un path param ne capture pas les /. Pour router sur un chemin de fichier ou un identifiant hiérarchique, utilise le convertisseur :path :
@app.get("/files/{file_path:path}")
async def read_file(file_path: str) -> dict[str, str]:
# GET /files/skills/agent/SKILL.md → file_path == "skills/agent/SKILL.md"
return {"path": file_path}⚠️ Tout endpoint :path qui touche le système de fichiers est une porte à path traversal (../../etc/passwd). On sécurise ça dans la section production.
🤖 Application : router vers un agent LLM
Tout ce qui précède devient concret dès que la route sert un agent. On va construire une petite gateway qui appelle Claude. Le path identifie quel agent, les query params pilotent comment on l'appelle (modèle, effort, streaming), le body porte le message.
Installe le SDK officiel :
pip install "anthropic>=0.40" fastapi "uvicorn[standard]"Le client comme dépendance (singleton, pas par requête)
On crée un seul AsyncAnthropic au démarrage et on l'injecte. Recréer le client à chaque requête réinitialise le pool de connexions HTTP et tue la latence.
from contextlib import asynccontextmanager
from typing import Annotated
from anthropic import AsyncAnthropic
from fastapi import Depends, FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# Démarrage : un client partagé, pooling HTTP réutilisé
app.state.anthropic = AsyncAnthropic() # lit ANTHROPIC_API_KEY
yield
# Arrêt : ferme proprement les connexions
await app.state.anthropic.close()
app = FastAPI(title="Agent Gateway", lifespan=lifespan)
# La dépendance lit le client partagé depuis l'état de l'app via la Request.
from fastapi import Request
def anthropic_client(request: Request) -> AsyncAnthropic:
return request.app.state.anthropic
ClientDep = Annotated[AsyncAnthropic, Depends(anthropic_client)]ClientDep est un alias Annotated réutilisable — tu le poses comme type sur n'importe quel handler et l'injection se fait toute seule. C'est l'équivalent FastAPI d'un provider Nest injecté par token.
Endpoint synchrone : path + query + body, retour structuré
from enum import Enum
from pydantic import BaseModel, Field
class ModelChoice(str, Enum):
opus = "claude-opus-4-8"
sonnet = "claude-sonnet-4-6"
haiku = "claude-haiku-4-5"
class Effort(str, Enum):
low = "low"
medium = "medium"
high = "high"
class ChatRequest(BaseModel):
prompt: str = Field(min_length=1, max_length=10_000)
class ChatResponse(BaseModel):
agent_id: str
model: str
text: str
input_tokens: int
output_tokens: int
@app.post("/agents/{agent_id}/chat", response_model=ChatResponse)
async def chat(
agent_id: Annotated[str, Path(min_length=1, max_length=64)],
body: ChatRequest,
client: ClientDep,
model: Annotated[ModelChoice, Query()] = ModelChoice.sonnet,
effort: Annotated[Effort, Query()] = Effort.medium,
max_tokens: Annotated[int, Query(ge=1, le=16_000)] = 4_000,
) -> ChatResponse:
message = await client.messages.create(
model=model.value,
max_tokens=max_tokens,
thinking={"type": "adaptive"},
output_config={"effort": effort.value},
messages=[{"role": "user", "content": body.prompt}],
)
text = "".join(b.text for b in message.content if b.type == "text")
return ChatResponse(
agent_id=agent_id,
model=message.model,
text=text,
input_tokens=message.usage.input_tokens,
output_tokens=message.usage.output_tokens,
)Lis bien comment chaque source de donnée est désambiguïsée par le type seul :
| Paramètre | Origine | Pourquoi |
|---|---|---|
agent_id | path | nom présent dans {agent_id} |
body | body | type = modèle Pydantic |
client | dépendance | Annotated[..., Depends(...)] |
model, effort, max_tokens | query | scalaires/Enum, hors path, non-modèles |
thinking={"type": "adaptive"} est le réglage recommandé sur claude-opus-4-8 et la famille 4.6+ (le budget_tokens est supprimé et renvoie une 400). L'effort se règle via output_config.effort — jamais via un budget de tokens.
Endpoint streaming : un query param qui change tout
Pour une UX de type chat, on veut streamer les tokens. C'est un cas d'école où un query param booléen (?stream=true) bascule le type de réponse — du JSON vers du Server-Sent Events. Le routing reste identique ; seul le retour change.
import json
from fastapi import HTTPException
from fastapi.responses import StreamingResponse
@app.post("/agents/{agent_id}/chat/stream")
async def chat_stream(
agent_id: Annotated[str, Path(min_length=1, max_length=64)],
body: ChatRequest,
client: ClientDep,
model: Annotated[ModelChoice, Query()] = ModelChoice.sonnet,
max_tokens: Annotated[int, Query(ge=1, le=64_000)] = 8_000,
) -> StreamingResponse:
async def event_source():
try:
async with client.messages.stream(
model=model.value,
max_tokens=max_tokens,
thinking={"type": "adaptive"},
messages=[{"role": "user", "content": body.prompt}],
) as stream:
async for text in stream.text_stream:
# Format SSE : "data: <payload>\n\n"
yield f"data: {json.dumps({'delta': text})}\n\n"
final = await stream.get_final_message()
usage = {
"input_tokens": final.usage.input_tokens,
"output_tokens": final.usage.output_tokens,
}
yield f"data: {json.dumps({'done': True, 'usage': usage})}\n\n"
except Exception as exc: # le client a peut-être coupé, ou l'API a échoué
yield f"data: {json.dumps({'error': str(exc)})}\n\n"
return StreamingResponse(
event_source(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)Deux détails de production cachés ici : X-Accel-Buffering: no empêche nginx de tamponner le flux (sinon le client reçoit tout d'un coup à la fin), et le try/except à l'intérieur du générateur est obligatoire — une exception levée après que les headers 200 OK sont partis ne peut plus devenir un 500, elle ne peut qu'être signalée dans le flux. Le streaming est aussi la défense recommandée contre les timeouts HTTP quand max_tokens est grand.
Boucle d'outils (tool-use) : pourquoi le routing reste trivial
Un agent « qui agit » exécute une boucle tool-use : le modèle demande un outil, ton serveur l'exécute, renvoie le résultat, recommence jusqu'à end_turn. Côté HTTP, c'est une seule route ; toute la complexité est dans la boucle, pas dans le routing. Le path identifie l'agent, un query param peut plafonner les itérations.
TOOLS = [
{
"name": "get_weather",
"description": "Donne la météo actuelle d'une ville. À appeler "
"quand l'utilisateur demande la météo ou la température.",
"input_schema": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
}
]
def run_tool(name: str, args: dict[str, object]) -> str:
if name == "get_weather":
return f"Il fait 21°C et ensoleillé à {args.get('city')}."
return f"Outil inconnu : {name}"
@app.post("/agents/{agent_id}/act")
async def act(
agent_id: Annotated[str, Path(min_length=1, max_length=64)],
body: ChatRequest,
client: ClientDep,
model: Annotated[ModelChoice, Query()] = ModelChoice.opus,
max_steps: Annotated[int, Query(ge=1, le=10)] = 5, # garde-fou anti-boucle
) -> dict[str, object]:
messages: list[dict[str, object]] = [
{"role": "user", "content": body.prompt}
]
for _ in range(max_steps):
resp = await client.messages.create(
model=model.value,
max_tokens=4_000,
thinking={"type": "adaptive"},
tools=TOOLS,
messages=messages,
)
if resp.stop_reason != "tool_use":
text = "".join(b.text for b in resp.content if b.type == "text")
return {"agent_id": agent_id, "text": text}
messages.append({"role": "assistant", "content": resp.content})
results = []
for block in resp.content:
if block.type == "tool_use":
out = run_tool(block.name, block.input)
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": out,
})
messages.append({"role": "user", "content": results})
raise HTTPException(status_code=409, detail="max_steps atteint sans réponse")max_steps en query param est un garde-fou explicite : il transforme une boucle potentiellement infinie en un échec borné et observable (409). C'est exactement le genre de levier que tu veux exposer côté API plutôt que coder en dur.
Sortie structurée : messages.parse()
Quand tu veux que l'agent renvoie du JSON validé (extraction, classification), passe par les sorties structurées natives plutôt qu'un prompt « renvoie du JSON ». Le path identifie toujours l'agent ; un query param peut sélectionner le schéma attendu.
class Contact(BaseModel):
name: str
email: str
company: str | None = None
@app.post("/agents/{agent_id}/extract", response_model=Contact)
async def extract_contact(
agent_id: Annotated[str, Path(min_length=1)],
body: ChatRequest,
client: ClientDep,
) -> Contact:
message = await client.messages.parse(
model="claude-opus-4-8",
max_tokens=1_000,
messages=[{"role": "user", "content": body.prompt}],
output_format=Contact, # le SDK le traduit en output_config.format
)
return message.parsed_output # déjà un Contact validé⚙️ En production
Failure modes (modes de défaillance)
Ordre de routes inversé. Le bug N°1, déjà vu. En
str, il est silencieux. Mitigation : tests d'intégration qui frappent chaque route statique, ou préfère des path params typés (int,UUID) pour que les collisions deviennent des 422 bruyants.422 vs 400. Une erreur de validation FastAPI renvoie 422, pas 400. Beaucoup de fronts (et de devs) attendent 400. Décide d'une convention et, si besoin, surcharge le
RequestValidationErrorhandler pour remapper. Ne laisse pas le front deviner.Path traversal sur
:path. Un endpoint/files/{p:path}qui lit le disque sans normaliser laisse passer../. Résous le chemin réel et vérifie qu'il reste sous une racine autorisée :pythonfrom pathlib import Path as FsPath from fastapi import HTTPException ROOT = FsPath("/srv/data").resolve() def safe_path(rel: str) -> FsPath: target = (ROOT / rel).resolve() if not target.is_relative_to(ROOT): # Python 3.9+ raise HTTPException(status_code=404, detail="not found") return targetErreurs LLM remontées en 500 bruts. Le SDK lève des exceptions typées (
anthropic.RateLimitError,APIStatusError,OverloadedError). Mappe-les vers des statuts HTTP cohérents au lieu de laisser fuiter une stacktrace :pythonimport anthropic from fastapi import Request from fastapi.responses import JSONResponse @app.exception_handler(anthropic.APIStatusError) async def handle_anthropic(request: Request, exc: anthropic.APIStatusError): status = 429 if isinstance(exc, anthropic.RateLimitError) else 502 return JSONResponse(status_code=status, content={"error": exc.type})
Performance
asyncpartout sur le chemin chaud. Un handlerdef(sync) qui appelle le LLM bloque un worker de threadpool pour toute la durée de l'inférence (souvent plusieurs secondes, voire minutes au plus haut effort). Sous charge, tu épuises le pool et la latence explose. AvecAsyncAnthropic+async def, un seul worker gère des centaines de requêtes en attente d'I/O.- Client singleton + pooling. Vu plus haut : un
AsyncAnthropiccréé danslifespan, jamais par requête. - Cardinalité des query params. Chaque combinaison de query params est une variante de cache HTTP et une ligne de métrique potentielle. Un
?ts=<timestamp>involontaire fait exploser le cache. Pour le LLM spécifiquement : un prompt système stable doit rester byte-identique d'une requête à l'autre pour que le prompt caching Anthropic (cache_control) fonctionne — n'injecte pas de date ou d'ID dans le préfixe. response_modela un coût. FastAPI re-valide ta sortie contre le modèle déclaré. C'est une garantie de contrat précieuse, mais sur un endpoint à très haut débit qui renvoie déjà des objets sûrs, tu peux retourner unResponsedirect pour économiser la sérialisation double.
Sécurité
- Valide la cardinalité et les bornes des query params (
ge,le,max_length). Unmax_tokensnon borné est une facture ouverte ; unlimitnon borné est un déni de service sur ta DB. Toujours plafonner. - N'expose jamais d'identifiant interne séquentiel en path param (
/agents/{int}) si l'énumération est sensible —42invite à essayer43. Préfère unUUIDou un ID opaque (ag_...) pour les ressources multi-tenant, et vérifie l'autorisation, pas seulement l'existence. - Ne mets jamais de secret dans la query string. Les query params finissent dans les logs d'accès, l'historique du navigateur, les en-têtes
Referer. Une clé API passée en?api_key=est une fuite. Auth par header (Authorization: Bearer). - L'
agent_iddu path n'est pas une autorisation. Qu'un agent existe ne veut pas dire que cet appelant y a droit. La validation de format (Pydantic) et l'autorisation (qui a accès) sont deux couches distinctes.
Observabilité
- Loggue
request_iddu SDK. Chaque réponse Anthropic porte unrequest_id; mets-le dans tes logs structurés pour tracer un appel de bout en bout côté Anthropic. - Trace le
usage.message.usage.input_tokens/output_tokens(+cache_read_input_tokens) sont ta facture en direct. Émets-les en métriques par route et par modèle — c'est ainsi que tu repères une route qui dérive en coût. - Distingue 4xx de validation et 5xx d'inférence dans tes dashboards. Un pic de 422 = un client mal codé ; un pic de 529 (
overloaded) = l'API Anthropic sature, retry avec backoff (le SDK le fait déjà par défaut,max_retries=2).
Les tradeoffs du senior
- Magie vs explicite. La résolution par type est concise mais implicite. Sur une grosse codebase, l'alias
Annotatedréutilisable (ClientDep,LimitQuery = Annotated[int, Query(ge=1, le=100)]) redonne de l'explicite sans verbosité. EnumvsLiteral.Enumquand la valeur est réutilisée/partagée (modèles, rôles) ;Literalpour un drapeau local jetable. Les deux valident et documentent ; choisis selon la réutilisation.- Un endpoint streaming + un non-streaming, ou un seul avec
?stream=. Deux routes = contrats OpenAPI clairs et types de retour honnêtes. Une route avec drapeau = surface plus petite mais retour polymorphe (mal typé). En général, deux routes vieillissent mieux.
🏋️ Exercices
Exercice 1 — Le piège de l'ordre (implémenter)
Objectif. Écris deux routes : GET /models/recommended (statique, renvoie claude-opus-4-8) et GET /models/{model_id} (renvoie l'ID reçu). Démontre par un test que l'ordre de déclaration change le comportement, puis corrige.
Indice/Solution. Déclare d'abord la paramétrée avec model_id: str et vérifie que /models/recommended tombe dans le mauvais handler (model_id == "recommended"). Échange l'ordre, statique en premier. Bonus : type model_id: ModelChoice et observe que la collision devient un 422 au lieu d'un silence. La leçon : spécifique avant générique, et typer fort rend les bugs bruyants.
Exercice 2 — Pagination validée bout en bout (implémenter → durcir)
Objectif. Sur GET /agents/{agent_id}/messages, ajoute limit (1–100, défaut 20), offset (≥0, défaut 0) et order: Literal["asc", "desc"] (défaut "desc"). Tout via Annotated[..., Query(...)]. Renvoie un objet {items, limit, offset, order, has_more}.
Indice/Solution. limit: Annotated[int, Query(ge=1, le=100)] = 20, idem offset avec ge=0, order: Annotated[Literal["asc","desc"], Query()] = "desc". Teste les bornes : ?limit=0 et ?limit=101 doivent renvoyer 422 ; ?order=sideways aussi. Le but est de ne jamais écrire un seul if.
Exercice 3 — Gateway de chat avec choix de modèle (implémenter, AI)
Objectif. Implémente POST /agents/{agent_id}/chat qui prend un body {prompt}, un query model: ModelChoice (défaut sonnet) et effort: Effort, appelle client.messages.create(..., thinking={"type":"adaptive"}, output_config={"effort": effort.value}), et renvoie un ChatResponse typé avec le usage.
Indice/Solution. Reprends l'endpoint chat de la leçon. Vérifie qu'?model=gpt-4 renvoie 422 (l'Enum te protège) et que le défaut sonnet s'applique sans query. Assure-toi de ne pas passer budget_tokens — c'est une 400 sur ces modèles ; l'effort se règle uniquement via output_config.effort.
Exercice 4 — Streaming SSE qui survit à une coupure (durcir, AI)
Objectif. Implémente POST /agents/{agent_id}/chat/stream en SSE. Garantis que (a) si l'appel LLM échoue après le 200 OK, le client reçoit un événement {"error": ...} et non un crash, (b) si le client se déconnecte, le serveur n'imprime pas de stacktrace bruyante.
Indice/Solution. Mets tout le corps du générateur dans un try/except qui yield un événement d'erreur. Pour la déconnexion client, FastAPI lève une asyncio.CancelledError dans le générateur quand le socket ferme — laisse-la se propager (ne la « except Exception » pas en silence avec un yield, car le socket est mort). Teste en coupant curl -N à mi-flux.
Exercice 5 — Boucle tool-use bornée, puis casse-la (implémenter → casser → réparer, AI)
Objectif. Implémente POST /agents/{agent_id}/act avec la boucle tool-use et un query max_steps (1–10). Étape « casser » : crée un outil qui renvoie toujours un résultat qui pousse le modèle à re-demander un outil (boucle). Vérifie que max_steps borne bien et renvoie 409. Étape « réparer » : ajoute une métrique du nombre d'étapes réellement consommées et loggue-la.
Indice/Solution. La boucle de la leçon avec for _ in range(max_steps). Pour la casser, un outil get_time dont la description encourage à toujours revérifier — observe le 409. Pour réparer/observer, compte les itérations et émets steps_used ; un agent qui touche souvent max_steps signale soit un outil mal décrit, soit un max_steps trop bas.
Exercice 6 — Sécuriser un endpoint :path (casser → réparer)
Objectif. Expose GET /skills/{skill_path:path} qui lit un fichier sous /srv/skills. D'abord nai͏f (lecture directe), prouve le path traversal avec /skills/../../etc/passwd. Puis durcis avec resolve() + is_relative_to.
Indice/Solution. Version naïve : open(f"/srv/skills/{skill_path}"). Exploit : %2e%2e%2f ou ../ selon le proxy. Fix : la fonction safe_path de la section sécurité, qui résout le chemin absolu et rejette tout ce qui sort de ROOT avec un 404 (pas un 403 — ne révèle pas que le fichier existe ailleurs).
🎤 En entretien
Q : Comment FastAPI décide-t-il qu'un paramètre est un path param, un query param ou un body ? R : Par le nom et le type. Nom présent dans le template du chemin → path ; scalaire/Enum hors path → query ; modèle Pydantic → body ; Depends(...) → dépendance. Aucune annotation explicite n'est requise pour la désambiguïsation — elle ne sert qu'à ajouter de la validation.
Q : Quel statut renvoie FastAPI sur une erreur de validation, et pourquoi est-ce parfois un piège ? R : 422, pas 400. Le piège : beaucoup de clients et de gateways attendent 400 ; il faut soit aligner le contrat documenté, soit surcharger le handler de RequestValidationError pour remapper en 400 si l'équipe l'exige.
Q : Pourquoi l'ordre de déclaration des routes peut-il introduire un bug silencieux, et comment t'en prémunir ? R : La première route qui matche gagne ; une route paramétrée déclarée avant une statique l'absorbe. En str c'est silencieux. Prémunition : déclarer du plus spécifique au plus générique, et typer fort les path params (int/UUID) pour que les collisions deviennent des 422 bruyants plutôt que des routes mortes.
Q : Pour servir un agent LLM, pourquoi des handlers async et un client unique, et comment câbler le streaming ? R : L'inférence est de l'I/O longue ; async def + AsyncAnthropic libèrent le worker pendant l'attente, là où un handler sync épuiserait le threadpool sous charge. Un seul client (créé dans lifespan, injecté par Depends) réutilise le pool de connexions. Le streaming se câble avec client.messages.stream(...) enveloppé dans une StreamingResponse(media_type="text/event-stream"), en gérant les erreurs dans le générateur puisque les headers 200 sont déjà partis.