FastAPI do zero: injeção de dependência, Pydantic v2 e streaming com SSE

Segundo post da série sobre o Konhecia, o agente de suporte com RAG local que estou construindo. No anterior subimos a infraestrutura com Docker Compose. Agora a camada que o usuário toca: a API.

A escolha foi FastAPI. Não por hype — por três características que casam com o problema: é assíncrono de verdade, valida entrada e saída a partir de type hints, e trata streaming como cidadão de primeira classe. Um agente de LLM precisa das três.

ASGI: por que “assíncrono” importa aqui

Frameworks Python tradicionais (Flask, Django clássico) são WSGI: uma requisição ocupa um worker do começo ao fim. Se a requisição passa 8 segundos esperando um LLM responder, o worker fica 8 segundos parado — sem usar CPU, apenas ocupado.

ASGI quebra isso. Com async def, quando o código chega num await de I/O (rede, banco, disco), o event loop solta o controle e atende outra requisição enquanto espera. O mesmo processo aguenta muito mais conexões concorrentes.

Isso é decisivo num agente de RAG, porque quase todo o tempo de uma requisição é espera: embedding da pergunta, busca vetorial, geração pelo LLM, escrita no banco. Praticamente nenhum trabalho de CPU nosso — só I/O.

A regra prática: num async def, toda chamada bloqueante contamina o loop inteiro. Um requests.get() ou um driver de banco síncrono dentro de uma rota async trava todas as requisições, não só a sua. Por isso a stack precisa ser assíncrona de ponta a ponta — httpx no lugar de requests, asyncpg no lugar de psycopg2 síncrono.

Rotas e roteadores

O básico é um decorator sobre uma função:

from fastapi import APIRouter

router = APIRouter()

@router.get("/health")
async def health_check() -> dict:
    return {"status": "ok"}

APIRouter agrupa rotas por domínio (health, chat, documents) em módulos separados, que o main.py monta num prefixo comum como /api/v1. Versionar a API no caminho desde o primeiro dia custa nada e evita uma migração dolorosa depois.

Pydantic v2: o contrato é o código

O FastAPI usa os type hints como fonte de verdade. Você declara um modelo Pydantic e ganha, de graça: validação da entrada, coerção de tipos, mensagem de erro 422 estruturada e a documentação OpenAPI.

from pydantic import BaseModel, Field
import uuid

class ChatRequest(BaseModel):
    message: str = Field(min_length=1)
    session_id: uuid.UUID | None = None

Se chega um JSON sem message, ou com session_id que não é UUID válido, a requisição nunca alcança a sua função. O FastAPI devolve 422 com o caminho exato do campo inválido.

Isso é uma fronteira de segurança, não só conveniência. Toda entrada externa passa por um schema explícito antes de virar objeto Python. Nada de request.json()["x"] e torcer.

Na saída, response_model faz o caminho inverso — e o benefício é o oposto do intuitivo:

@router.get("/chat/history/{session_id}", response_model=list[MessageRead])
async def get_history(session_id: uuid.UUID) -> list[MessageRead]:
    ...

O response_model filtra o que sai. Se o objeto do banco tem campos que o schema não declara, eles não são serializados. É a proteção mais barata que existe contra vazamento acidental de campo interno numa resposta pública.

Depends: injeção de dependência que também é autenticação

Esse é o mecanismo mais subestimado do FastAPI. Depends declara: “antes de rodar esta função, resolva isto aqui”.

O uso óbvio é sessão de banco por requisição:

async def get_history(db: AsyncSession = Depends(get_db)):
    ...

O uso menos óbvio, e mais poderoso, é autenticação. Uma dependência que não retorna nada — só levanta exceção se algo estiver errado:

from fastapi.security import APIKeyHeader
import secrets

_api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

def verify_api_key(
    provided_key: str | None = Depends(_api_key_header),
    settings: Settings = Depends(get_settings),
) -> None:
    if provided_key is None or not secrets.compare_digest(provided_key, settings.api_key):
        raise HTTPException(status_code=401, detail="Chave de API ausente ou inválida.")

Duas coisas para reparar:

secrets.compare_digest, não ==. A comparação == de strings em Python sai no primeiro byte diferente. O tempo de resposta, portanto, vaza quantos caracteres iniciais o atacante acertou — dá para reconstruir a chave byte a byte com requisições suficientes. compare_digest compara em tempo constante. É uma linha, e é a diferença entre um ataque de timing viável e inviável.

Aplique no roteador, não na rota. Este é o ponto que evita a falha mais comum de API:

router = APIRouter(dependencies=[Depends(verify_api_key)])

Toda rota registrada nesse roteador já nasce protegida. Decorar rota por rota funciona até o dia em que alguém adiciona um endpoint e esquece o decorator — e esse endpoint fica público sem ninguém perceber. Segurança por padrão, com opt-out explícito, é sempre melhor que segurança por lembrança.

Streaming: por que SSE e não WebSocket

Um LLM gera texto token a token. Segurar a resposta inteira e mandar de uma vez significa o usuário olhando para uma tela parada por vários segundos. Streaming resolve a percepção de latência — o primeiro token chega quase imediatamente.

Há duas formas. WebSocket é bidirecional, full-duplex, e exige protocolo próprio dos dois lados. Server-Sent Events (SSE) é unidirecional servidor→cliente, roda sobre HTTP normal e reconecta sozinho.

Para uma resposta de chat, o fluxo é unidirecional por natureza: o cliente já mandou a pergunta, agora só recebe. SSE é mais simples e suficiente — passa por proxy reverso sem configuração especial e funciona com curl -N, o que é ótimo para depuração.

O formato do protocolo é texto puro:

event: start
data: {"session_id": "af6894fe-676b-4369-bbb3-80110d1dc19d"}

event: token
data: {"text": "Para"}

event: token
data: {"text": " resetar"}

event: done
data: {"session_id": "af68...", "sources": [...], "grounded": true}

Com sse-starlette, basta um gerador assíncrono que produz dicionários:

from sse_starlette.sse import EventSourceResponse

async def _stream_events(...) -> AsyncIterator[dict]:
    yield {"event": "start", "data": json.dumps({"session_id": str(cid)})}
    async for token in gerar_resposta():
        yield {"event": "token", "data": json.dumps({"text": token})}
    yield {"event": "done", "data": json.dumps({...})}

@router.post("/chat")
async def chat(...) -> EventSourceResponse:
    return EventSourceResponse(_stream_events(...))

A pegadinha do streaming: o erro que não pode virar erro HTTP

Esse detalhe merece o próprio destaque, porque quebra uma intuição básica.

Quando você abre um stream, o cabeçalho HTTP 200 já foi enviado. Se a geração falhar no meio — o LLM caiu, o banco recusou a escrita —, é tarde demais para devolver 500. O status já está na rede.

Só resta emitir um evento de erro dentro do stream e encerrar:

try:
    async for kind, payload in stream_answer(...):
        yield {"event": "token", "data": json.dumps({"text": payload})}
    await db.commit()
    yield {"event": "done", "data": json.dumps({...})}
except Exception:
    logger.exception("Falha ao gerar resposta do chat")
    await db.rollback()
    yield {"event": "error", "data": json.dumps({"detail": "Falha interna ao gerar a resposta."})}

A consequência de projeto é o que importa: faça fora do stream tudo que pode falhar com erro HTTP honesto. Criar ou recuperar a conversa e registrar a pergunta do usuário acontecem antes de abrir o EventSourceResponse. Se algo ali falhar, o cliente recebe um 4xx/5xx normal, tratável. Só o que é irredutivelmente streaming fica dentro do gerador.

E repare no logger.exception seguido de uma mensagem genérica para o cliente: o stack trace vai para o log, o usuário recebe "Falha interna ao gerar a resposta.". Detalhe de exceção em resposta de API é reconhecimento de graça para quem está mapeando o sistema.

O que fica

FastAPI não é “Flask mais rápido”. A mudança de fundo é que o contrato da API vira código verificável — tipos que validam, dependências que compõem, e uma documentação OpenAPI em /docs que não desatualiza porque é gerada do mesmo lugar.

No próximo post: a camada de persistência — SQLAlchemy assíncrono, Alembic e por que o histórico de conversa não vai para o banco vetorial.