Request body & validation

TL;DR — Dans FastAPI, le corps d'une requête se valide en déclarant un modèle Pydantic v2 comme paramètre typé de ton endpoint : FastAPI lit le body, le parse, le valide, le coerce et te livre un objet Python déjà propre — ou renvoie un 422 structuré sans que tu écrives une ligne. Tu n'écris jamais de if request.json["x"] is None. Tu décris la forme attendue (types, contraintes, valeurs par défaut) et le framework fait respecter le contrat à la frontière. Pour un dev qui sert des agents LLM, c'est le mécanisme qui transforme un payload HTTP brut ({"messages": [...], "effort": "high"}) en une requête Anthropic typée et sûre — et qui rejette les inputs hostiles avant qu'ils n'atteignent un appel facturé à 5 $/Mtok.

---

🧠 Mental model

Pense à un endpoint FastAPI comme à un poste de douane typé. La requête HTTP arrive avec une valise (le body JSON). Tu ne fouilles pas la valise toi-même : tu donnes au douanier (FastAPI) la fiche descriptive de ce qui est autorisé (le modèle Pydantic). Le douanier ouvre, vérifie chaque article contre la fiche, convertit ce qui doit l'être ("42" → 42), refoule ce qui est non conforme (422), et te tend un colis garanti conforme que tu peux consommer les yeux fermés.

La distinction clé que les ex-PHP/Express ratent : en Express/NestJS sans pipe, tu reçois req.body brut et tu valides à la main, à l'intérieur du handler. En FastAPI, la validation est déclarative et à la frontière. Le handler ne s'exécute jamais avec un input invalide.

                      ┌─────────────── FastAPI (Starlette + Pydantic) ───────────────┐
   HTTP POST          │                                                              │
   body: {raw JSON} ──┼──▶ parse JSON ──▶ valide vs modèle ──▶ coerce ──▶ instancie  │
                      │         │              │                              │       │
                      │       400 si           422 si                    objet typé   │
                      │     JSON cassé      non conforme                      │       │
                      └──────────────────────────────────────────────────────┼───────┘
                                                                              ▼
                                                          async def handler(payload: MyModel)
                                                          # payload est DÉJÀ valide ici

Le corollaire mental : le type annotation n'est pas de la documentation, c'est de l'exécution. payload: ChatRequest ne décrit pas ce que tu espères, ça impose ce qui passera.

---

Le cœur : déclarer, ne pas vérifier

La façon idiomatique

Un modèle Pydantic v2 = une class qui hérite de BaseModel, avec des champs typés. FastAPI détecte qu'un paramètre d'endpoint est un BaseModel et en déduit automatiquement que c'est le corps de la requête.

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class ChatRequest(BaseModel):
    prompt: str = Field(min_length=1, max_length=8_000)
    model: str = "claude-opus-4-8"
    max_tokens: int = Field(default=1024, ge=1, le=128_000)
    temperature: float | None = None  # None = laisser le défaut serveur


class ChatResponse(BaseModel):
    text: str
    model: str
    output_tokens: int


@app.post("/chat", response_model=ChatResponse)
async def chat(payload: ChatRequest) -> ChatResponse:
    # payload est garanti : prompt non vide, max_tokens dans [1, 128000].
    # Aucune vérification manuelle ici. C'est tout le point.
    ...

Ce que FastAPI a fait gratuitement :

• lu le Content-Type: application/json et parsé le body ;
• vérifié que prompt est une str de 1 à 8000 caractères ;
• vérifié que max_tokens est un entier entre 1 et 128000 ;
• appliqué les valeurs par défaut (model, max_tokens) si absentes ;
• généré le schéma OpenAPI (/docs) à partir de la classe ;
• renvoyé un 422 détaillé si quoi que ce soit cloche.

Exemple de 422 renvoyé automatiquement pour {"prompt": "", "max_tokens": 999999} :

{
  "detail": [
    {
      "type": "string_too_short",
      "loc": ["body", "prompt"],
      "msg": "String should have at least 1 character",
      "input": ""
    },
    {
      "type": "less_than_equal",
      "loc": ["body", "max_tokens"],
      "msg": "Input should be less than or equal to 128000",
      "input": 999999
    }
  ]
}

La façon naïve (à ne PAS faire)

from fastapi import Request

# ❌ Anti-pattern : récupérer le body brut et valider à la main
@app.post("/chat")
async def chat(request: Request):
    data = await request.json()           # peut throw si JSON cassé → 500 non géré
    prompt = data.get("prompt")
    if not prompt:                        # tu réinventes la validation
        return {"error": "prompt required"}, 400   # format d'erreur incohérent
    max_tokens = data.get("max_tokens", 1024)
    if max_tokens > 128000:               # règle métier noyée dans le handler
        max_tokens = 128000               # ou pire : coercion silencieuse
    ...

Pourquoi c'est mauvais, point par point :

• Pas de schéma OpenAPI : /docs est vide, ton frontend Angular ne peut pas générer de client typé.
• Erreurs incohérentes : chaque endpoint invente son format. Ton client ne peut pas les traiter uniformément.
• request.json() peut lever une exception sur un JSON malformé → 500 au lieu d'un propre 400.
• Logique de validation dispersée dans le corps du handler, impossible à réutiliser ou tester isolément.
• Coercion silencieuse (if x > max: x = max) = bugs invisibles. L'appelant croit envoyer 999999, reçoit un comportement différent sans le savoir.

La règle senior : si tu lis request.json() toi-même dans un endpoint FastAPI, tu te bats contre le framework. L'exception légitime existe (proxy transparent, body binaire opaque), mais c'est rare.

---

Pydantic v2 : les outils qui comptent

Contraintes de champ — Field

from pydantic import BaseModel, Field
from typing import Literal


class GenerationConfig(BaseModel):
    effort: Literal["low", "medium", "high", "max"] = "high"
    max_tokens: int = Field(default=1024, ge=1, le=128_000)
    stop_sequences: list[str] = Field(default_factory=list, max_length=4)
    metadata: dict[str, str] = Field(default_factory=dict)

Deux pièges classiques pour un dev venant de TS :

1. Jamais de mutable par défaut nu. stop_sequences: list[str] = [] partage la même liste entre toutes les instances (piège Python universel). Utilise default_factory=list. Pydantic te protège partiellement, mais prends l'habitude.
2. Literal > str quand l'ensemble est fini. effort: Literal["low", ...] rejette "ultra" avec un 422 et documente les valeurs valides dans OpenAPI. effort: str accepte n'importe quoi.

Validation croisée — @model_validator et @field_validator

Quand une règle dépend de plusieurs champs, ou nécessite une logique :

from pydantic import BaseModel, field_validator, model_validator
from typing import Self


class ThinkingRequest(BaseModel):
    prompt: str
    max_tokens: int = 4096
    enable_thinking: bool = False
    thinking_min_tokens: int | None = None

    @field_validator("prompt")
    @classmethod
    def prompt_not_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("prompt must not be blank or whitespace-only")
        return v.strip()

    @model_validator(mode="after")
    def check_thinking_budget(self) -> Self:
        # règle inter-champs : si on demande un budget, il doit tenir dans max_tokens
        if self.thinking_min_tokens is not None:
            if self.thinking_min_tokens >= self.max_tokens:
                raise ValueError("thinking_min_tokens must be < max_tokens")
        return self

Note v2 : @field_validator est un @classmethod, et @model_validator(mode="after") reçoit self (l'instance déjà construite) et renvoie Self. C'est différent de Pydantic v1 (@validator, values dict) — n'utilise pas la syntaxe v1, elle est dépréciée.

Composition et modèles imbriqués

Le corps réel d'une requête vers un agent n'est pas plat. Pydantic compose les modèles naturellement :

from pydantic import BaseModel, Field
from typing import Literal


class Message(BaseModel):
    role: Literal["user", "assistant"]
    content: str = Field(min_length=1)


class AgentRequest(BaseModel):
    messages: list[Message] = Field(min_length=1, max_length=100)
    model: str = "claude-opus-4-8"
    max_tokens: int = Field(default=1024, ge=1, le=128_000)

    @field_validator("messages")
    @classmethod
    def first_is_user(cls, v: list[Message]) -> list[Message]:
        # contrainte de l'API Anthropic : le premier message doit être 'user'
        if v[0].role != "user":
            raise ValueError("first message must have role 'user'")
        return v

FastAPI valide récursivement : un content vide dans messages[3] produit "loc": ["body", "messages", 3, "content"]. La précision du chemin d'erreur est un cadeau pour le debugging et le frontend.

---

Connexion concrète : valider, puis appeler un agent Anthropic

Voici le pattern complet qui relie ce chapitre à ta vraie activité — recevoir une requête HTTP validée, et la transformer en appel AsyncAnthropic. La validation à la frontière est ta première ligne de défense économique : un payload rejeté ne déclenche pas d'appel facturé.

import os
from contextlib import asynccontextmanager

from anthropic import AsyncAnthropic
from fastapi import FastAPI, Depends, HTTPException


# --- Cycle de vie : un seul client AsyncAnthropic, réutilisé (pool HTTP) ---
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.anthropic = AsyncAnthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
    yield
    await app.state.anthropic.close()


app = FastAPI(lifespan=lifespan)


def get_client(request) -> AsyncAnthropic:  # injection de dépendance
    return request.app.state.anthropic


@app.post("/agent/chat", response_model=ChatResponse)
async def agent_chat(
    payload: AgentRequest,
    client: AsyncAnthropic = Depends(get_client),
) -> ChatResponse:
    # À ce point, payload est 100 % valide. On mappe vers l'API Anthropic.
    message = await client.messages.create(
        model=payload.model,
        max_tokens=payload.max_tokens,
        thinking={"type": "adaptive"},          # adaptive : le modèle dose son budget
        output_config={"effort": "high"},       # effort de raisonnement (low/medium/high/max)
        messages=[m.model_dump() for m in payload.messages],
    )

    if message.stop_reason == "refusal":
        # Opus 4.8 peut décliner ; surface propre plutôt qu'un crash d'index
        raise HTTPException(status_code=422, detail="request refused by safety policy")

    text = "".join(b.text for b in message.content if b.type == "text")
    return ChatResponse(
        text=text,
        model=message.model,
        output_tokens=message.usage.output_tokens,
    )

Points senior à retenir dans cet exemple :

• Un seul client, partagé via lifespan + Depends. Créer un AsyncAnthropic() par requête détruit le pool de connexions HTTP et la performance. C'est l'équivalent FastAPI d'un singleton injecté en NestJS.
• payload.model_dump() convertit le modèle Pydantic en dict que l'SDK accepte. (.model_dump() est la méthode v2 ; .dict() est v1, déprécié.)
• thinking={"type": "adaptive"} + output_config={"effort": "high"} est l'idiome moderne pour Opus 4.8 : la pensée adaptive laisse le modèle doser son propre budget de raisonnement, et effort (valeurs low/medium/high/max) règle l'intensité. Le mode explicite thinking={"type": "enabled", "budget_tokens": ...} reste valide, mais budget_tokens est un réglage de bas niveau : ne l'expose pas dans ton modèle Pydantic public sans raison — préfère effort comme bouton de contrôle stable.
• stop_reason == "refusal" : vérifie-le avant de lire message.content[0], sinon un refus pré-output (content vide) lève un IndexError.

Structured outputs : Pydantic des deux côtés

L'élégance ultime : le même modèle Pydantic sert à valider l'input HTTP et à contraindre l'output du LLM. L'SDK Anthropic expose messages.parse() qui valide la réponse contre un schéma.

from pydantic import BaseModel


class ExtractedContact(BaseModel):
    name: str
    email: str
    company: str | None = None


class ExtractRequest(BaseModel):
    raw_text: str = Field(min_length=1, max_length=20_000)


@app.post("/extract", response_model=ExtractedContact)
async def extract(
    payload: ExtractRequest,
    client: AsyncAnthropic = Depends(get_client),
) -> ExtractedContact:
    message = await client.messages.parse(
        model="claude-opus-4-8",
        max_tokens=1024,
        messages=[{"role": "user", "content": f"Extract contact: {payload.raw_text}"}],
        output_format=ExtractedContact,   # passe le modèle Pydantic directement
    )
    # message.parsed_output est DÉJÀ une instance typée d'ExtractedContact :
    # le SDK a généré le json_schema, contraint le LLM, puis validé la réponse.
    parsed = message.parsed_output
    if parsed is None:                    # refus / sortie non parsable → frontière propre
        raise HTTPException(status_code=422, detail="could not extract structured contact")
    return parsed

Pydantic devient ainsi le langage commun entre HTTP, ton code, et le LLM. Un seul schéma, trois frontières.

Streaming SSE avec input validé

Quand tu sers des tokens en streaming (le cas par défaut d'un chat d'agent), le modèle d'entrée reste identique — seule la réponse change de forme. La validation à la frontière fonctionne exactement pareil ; c'est la sortie qui devient un flux.

from fastapi.responses import StreamingResponse


@app.post("/agent/stream")
async def agent_stream(
    payload: AgentRequest,
    client: AsyncAnthropic = Depends(get_client),
) -> StreamingResponse:
    async def event_source():
        async with client.messages.stream(
            model=payload.model,
            max_tokens=payload.max_tokens,
            thinking={"type": "adaptive"},
            output_config={"effort": "high"},
            messages=[m.model_dump() for m in payload.messages],
        ) as stream:
            async for text in stream.text_stream:
                yield f"data: {text}\n\n"
            yield "data: [DONE]\n\n"

    return StreamingResponse(event_source(), media_type="text/event-stream")

Le streaming est recommandé dès que max_tokens est grand : il évite les timeouts HTTP côté SDK. payload reste validé exactement comme avant — la validation du body est indépendante de la forme de la réponse.

---

⚙️ En production

Modes de défaillance

Symptôme	Cause	Correctif senior
500 au lieu de 400 sur JSON cassé	Tu lis request.json() à la main	Déclare un modèle Pydantic, laisse FastAPI gérer le parse
Coercion surprise ("true" → True)	Pydantic coerce les types compatibles par défaut	Active model_config = ConfigDict(strict=True) sur les champs sensibles
Champs inconnus ignorés silencieusement	Comportement par défaut = extra="ignore"	model_config = ConfigDict(extra="forbid") → 422 sur clés non déclarées
Appel LLM déclenché sur input pourri	Validation faite après l'appel	La validation Pydantic se fait avant le handler — c'est gratuit, exploite-le
IndexError sur message.content[0]	Refus Anthropic (content vide)	if message.stop_reason == "refusal" avant de lire le content

extra="forbid" : le réglage que tout endpoint d'agent devrait avoir

Par défaut, Pydantic ignore les champs non déclarés. Pour une API qui mappe vers un appel facturé, c'est dangereux : un client qui envoie {"prompt": "...", "budget_tokens": 999999} croit configurer quelque chose, mais le champ est silencieusement jeté.

from pydantic import BaseModel, ConfigDict


class StrictChatRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")  # rejette tout champ inconnu

    prompt: str = Field(min_length=1)
    max_tokens: int = Field(default=1024, ge=1, le=128_000)

Maintenant budget_tokens non déclaré → 422 explicite. Le client apprend immédiatement que son champ n'existe pas, au lieu de débugger un comportement fantôme. Choix de tradeoff senior : extra="forbid" rend ton API rigide (toute évolution du client casse), donc réserve-le aux contrats internes et aux frontières sensibles ; pour une API publique versionnée, extra="ignore" offre plus de souplesse de compatibilité ascendante.

Performance

• Pydantic v2 a son cœur en Rust (pydantic-core). La validation est ~5–50× plus rapide que v1. Ne contourne pas la validation « pour la perf » — elle est négligeable face à une seule latence d'appel LLM (centaines de ms).
• max_length sur les champs texte n'est pas cosmétique. Sans max_length sur prompt, un client peut envoyer 50 Mo de texte → tu le passes à count_tokens/messages.create → facture explosive et risque de dépasser la fenêtre de contexte. Le max_length est un garde-fou économique.
• Le client AsyncAnthropic réutilisé (via lifespan) garde le pool HTTP chaud. Un client par requête = handshake TLS à chaque appel.

Sécurité

• La validation à la frontière EST une mesure de sécurité. Elle borne la taille des inputs (anti-DoS), restreint les valeurs (Literal), et empêche les payloads hostiles d'atteindre la couche LLM.
• N'expose jamais un champ model libre sans Literal ou allow-list. Sinon un client peut router vers un modèle plus cher (claude-fable-5 à 10/50 $/Mtok) que ce que ton budget prévoit. Borne explicitement :

  model: Literal["claude-opus-4-8", "claude-haiku-4-5"] = "claude-opus-4-8"

• Ne mets jamais de secret dans un modèle de requête. Les clés API viennent de l'environnement serveur (os.environ), jamais du body. Un api_key: str dans ChatRequest est une faille béante.
• Le message d'erreur 422 inclut input (la valeur fautive). Sur des données sensibles, configure un handler d'exception custom pour ne pas logger/renvoyer des PII.

Observabilité

• Compte les 422 par loc : un pic sur ["body", "messages"] révèle un client mal formé ou une attaque.
• Logge message.usage (input/output tokens) de chaque appel — c'est ta source de vérité pour la facturation et la détection d'anomalies.
• Ajoute un @app.exception_handler(RequestValidationError) custom pour émettre des erreurs au format de ton org (et y attacher un request_id).

---

🏋️ Exercices

Exercice 1 — Le modèle de base (implémenter)

Objectif. Crée un endpoint POST /summarize qui accepte {"text": "...", "max_words": 50}, valide que text fait entre 1 et 50000 caractères et que max_words est dans [10, 500], puis appelle Opus 4.8 pour résumer.

Indice/Solution. Modèle SummarizeRequest(BaseModel) avec text: str = Field(min_length=1, max_length=50_000) et max_words: int = Field(ge=10, le=500). Dans le handler, construis le prompt f"Summarize in at most {payload.max_words} words: {payload.text}", appelle client.messages.create(...) avec thinking={"type": "adaptive"}, extrais le texte des blocs type == "text".

Exercice 2 — Validation inter-champs (implémenter)

Objectif. Modèle TranslationRequest avec source_lang, target_lang (tous deux Literal["fr", "en", "es", "de"]) et text. Rejette avec un 422 les requêtes où source_lang == target_lang.

Indice/Solution. Utilise @model_validator(mode="after") renvoyant self. Lève ValueError("source and target languages must differ") si égaux. Vérifie que le chemin d'erreur dans la réponse 422 est bien ["body"] (erreur de modèle, pas de champ).

Exercice 3 — Rendre production-grade (durcir)

Objectif. Reprends l'endpoint /agent/chat. Ajoute : (a) extra="forbid", (b) model borné par Literal, (c) un max_length sur la liste messages, (d) la gestion du stop_reason == "refusal", (e) la conversion des exceptions de l'SDK en codes HTTP propres.

Indice/Solution. Pour (e), enveloppe l'appel dans try/except et mappe les exceptions typées : anthropic.RateLimitError → HTTPException(429), anthropic.APIStatusError → relaie e.status_code. N'utilise jamais de string-matching sur le message d'erreur ; utilise les classes typées de l'SDK. Ajoute un Retry-After sur le 429 à partir du header de l'exception si présent.

Exercice 4 — Casser puis réparer : la coercion fantôme (debug)

Objectif. On te donne un modèle Config(BaseModel) avec stream: bool = False. Un client envoie {"stream": "false"} (string). Observe que Pydantic coerce "false" en... False, mais "no" lève une erreur, et "0" donne False. Documente le comportement exact, puis force le mode strict pour que seuls les vrais booléens JSON passent.

Indice/Solution. Pydantic v2 en mode lax coerce les strings booléennes reconnues ("true"/"false"/"yes"/"no"/"0"/"1"). Pour interdire : stream: bool = Field(default=False, strict=True) ou model_config = ConfigDict(strict=True). En mode strict, "false" (string) lève un 422 ; seul le JSON false (booléen) passe. Le piège senior : ne jamais supposer le comportement de coercion — teste-le, ou rends-le strict sur les champs critiques.

Exercice 5 — Streaming sous contrainte (production)

Objectif. Construis POST /agent/stream (SSE) qui valide le même AgentRequest, mais ajoute un champ max_duration_s: int = Field(ge=1, le=60). Coupe le stream proprement si la génération dépasse cette durée, en envoyant un événement data: [TIMEOUT]\n\n.

Indice/Solution. Enveloppe la boucle async for text in stream.text_stream dans un asyncio.timeout(payload.max_duration_s) (Python 3.11+). Sur TimeoutError, yield l'événement timeout puis termine. Attention : le StreamingResponse doit fermer le contexte async with client.messages.stream(...) même en cas de timeout — d'où l'importance de garder le async with à l'intérieur du générateur.

Exercice 6 — Le contrat partagé (architecture)

Objectif. Un même modèle Invoice(BaseModel) doit servir à (a) valider l'upload d'une facture en JSON, et (b) contraindre l'extraction d'une facture par le LLM via messages.parse(). Implémente les deux endpoints en réutilisant exactement la même classe.

Indice/Solution. Définis Invoice une fois. Endpoint /upload : payload: Invoice (FastAPI valide l'input). Endpoint /extract : passe Invoice.model_json_schema() à output_config.format, puis Invoice.model_validate(message.parsed_output). Le piège : les schémas JSON de structured outputs n'acceptent pas certaines contraintes (minLength, maximum...). Pydantic les retire du schéma envoyé et valide côté client — donc Invoice.model_validate() après coup re-applique ces contraintes. Garde les deux validations.

---

🎤 En entretien

Q : Quelle est la différence entre valider le body dans le handler vs déclarer un modèle Pydantic en paramètre ? R : Avec un modèle Pydantic en paramètre, la validation est à la frontière — le handler ne s'exécute jamais avec un input invalide, l'erreur 422 est structurée et cohérente, et le schéma OpenAPI est généré. Valider dans le handler disperse la logique, produit des erreurs incohérentes, et risque un 500 sur JSON malformé.

Q : Pydantic v2 coerce-t-il les types par défaut ? Comment l'empêcher quand c'est dangereux ? R : Oui, mode « lax » par défaut ("42" → 42, "true" → True). On désactive ça champ par champ avec Field(strict=True) ou globalement via ConfigDict(strict=True) — utile sur les champs où une coercion silencieuse masquerait un bug client.

Q : Pourquoi max_length sur un champ texte est-il une décision de production, pas un détail ? R : Sans borne, un client peut envoyer un payload géant qui, transmis tel quel à messages.create, explose la facture (tokens) ou dépasse la fenêtre de contexte. max_length est à la fois un garde-fou anti-DoS et un plafond économique — il rejette le payload avant tout appel facturé.

Q : Comment fais-tu pour qu'un même schéma valide à la fois l'input HTTP et l'output d'un LLM ? R : Un seul modèle Pydantic. En entrée, il est le paramètre typé de l'endpoint (FastAPI valide). En sortie, on passe model_json_schema() à output_config.format de messages.parse(), puis on model_validate() le parsed_output. Attention : structured outputs ne supporte pas toutes les contraintes JSON Schema, donc le model_validate() final les re-applique côté client.