OTP com Python e FastAPI: verificação por WhatsApp e SMS passo a passo

Implementar OTP com Python é um dos jeitos mais rápidos de colocar verificação de telefone em produção: o FastAPI cuida das rotas e da validação de entrada, e uma API de envio cuida de entregar o código pelo WhatsApp ou por SMS. O que sobra para você é a parte que importa para o produto: quem pode pedir código, quantas vezes, e o que acontece depois que ele confere.

Este tutorial monta exatamente isso, do zero: um projeto FastAPI com dois endpoints (POST /otp/send e POST /otp/verify), um cliente HTTP para a API da aut2fa, estado em Redis com expiração, limites por telefone e por IP, testes com chave de teste e um roteiro de deploy.

O público é quem já escreve Python no dia a dia. Se as palavras async, venv e pip ainda são novidade, vale passar antes por um tutorial de Python para iniciantes e voltar aqui com o ambiente pronto.

O fluxo: solicitar, enviar e validar

Todo fluxo de OTP com Python (ou com qualquer outra linguagem) tem três momentos, e o desenho abaixo não muda de framework para framework:

1. Solicitar. O app ou o site chama o seu backend com o telefone. O backend checa limites e decide se pode enviar.
2. Enviar. O backend chama a API de OTP, que gera o código, escolhe o canal (WhatsApp primeiro, SMS como alternativa) e devolve um id de envio com o prazo de expiração. O código nunca passa pelo seu servidor.
3. Validar. A pessoa digita o código; o backend envia id e código para a API e só considera o telefone confirmado quando a resposta traz valid: true.

Dois detalhes definem a segurança do desenho. A chave da API fica só no backend: a API não libera CORS, e um app que chamasse a API direto exporia a chave a qualquer um que descompilasse o binário. E o id do envio fica guardado no servidor, associado ao telefone, para que o cliente não precise conhecê-lo.

Quem prefere Node encontra o mesmo fluxo, com envio por SMS e armazenamento próprio do código, no guia de SMS OTP com Node.js. A versão com Express e WhatsApp está em verificar número de telefone no cadastro com Node.js.

Projeto FastAPI e dependências

Crie o ambiente virtual e instale o necessário:

python -m venv .venv && source .venv/bin/activate
pip install "fastapi[standard]" httpx "redis>=5" python-dotenv

A estrutura fica pequena de propósito:

otp-fastapi/
├── app/
│   ├── main.py      # rotas e regras
│   ├── aut2fa.py    # cliente HTTP da API de OTP
│   └── store.py     # estado da verificação em Redis
├── .env             # AUT2FA_API_KEY=ak_test_... e REDIS_URL
└── tests/test_otp.py

No .env, use uma chave de teste (ak_test_) durante o desenvolvimento. Com ela, nenhuma mensagem é entregue, o código é sempre 000000 e nada é cobrado, mas o contrato HTTP é idêntico ao de produção.

O python-dotenv carrega esse arquivo em desenvolvimento: chame load_dotenv() no topo de app/main.py, antes de importar o cliente, porque ele lê AUT2FA_API_KEY no momento do import. Em produção, as variáveis vêm do próprio serviço e o .env nem existe.

Endpoint de envio com a API da aut2fa

O cliente HTTP, em app/aut2fa.py, encapsula as duas chamadas que o projeto usa. Os campos vêm da documentação de POST /v1/send e POST /v1/send/{id}/verify:

import os
import httpx

BASE_URL = "https://api.aut2fa.com"
HEADERS = {
    "Authorization": f"Bearer {os.environ['AUT2FA_API_KEY']}",  # ak_test_... ou ak_live_...
    "Content-Type": "application/json",
}


class RateLimited(Exception):
    def __init__(self, retry_after: int):
        self.retry_after = retry_after


async def enviar_codigo(telefone: str, idempotency_key: str) -> dict:
    # Pede o envio do código. A API escolhe o canal e devolve o id do envio.
    async with httpx.AsyncClient(base_url=BASE_URL, timeout=10) as client:
        resp = await client.post(
            "/v1/send",
            json={"to": telefone, "locale": "pt-BR"},
            headers={**HEADERS, "Idempotency-Key": idempotency_key},
        )
    if resp.status_code == 429:
        raise RateLimited(int(resp.headers.get("Retry-After", "60")))
    resp.raise_for_status()
    return resp.json()  # {"id": "snd_...", "status": "queued", "expires_at": "...", ...}


async def verificar_codigo(send_id: str, codigo: str) -> dict:
    # Confere o código digitado. Só prossiga se valid for True.
    async with httpx.AsyncClient(base_url=BASE_URL, timeout=10) as client:
        resp = await client.post(
            f"/v1/send/{send_id}/verify",
            json={"code": codigo},  # sempre string, nunca número
            headers=HEADERS,
        )
    resp.raise_for_status()
    return resp.json()  # {"valid": bool, "verification": "...", "attempts_left": int}

A rota de envio, em app/main.py, valida o formato do telefone com Pydantic, aplica os limites locais e só então chama a API. O cabeçalho Idempotency-Key protege contra envio duplicado se a conexão cair no meio de uma retentativa:

import uuid
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field

load_dotenv()  # carrega o .env em desenvolvimento, antes de importar o cliente
from app import aut2fa, store  # noqa: E402

app = FastAPI(title="Verificação de telefone")


class SendIn(BaseModel):
    telefone: str = Field(pattern=r"^\+[1-9]\d{7,14}$")  # formato internacional: +5531987654321


@app.post("/otp/send", status_code=202)
async def otp_send(body: SendIn, request: Request):
    ip = request.client.host if request.client else "desconhecido"
    if not await store.pode_enviar(body.telefone, ip):
        raise HTTPException(429, "Aguarde antes de pedir outro código")
    try:
        envio = await aut2fa.enviar_codigo(body.telefone, idempotency_key=str(uuid.uuid4()))
    except aut2fa.RateLimited as e:
        raise HTTPException(429, "Limite de envios atingido", headers={"Retry-After": str(e.retry_after)})
    await store.guardar(body.telefone, envio["id"])
    return {"expires_at": envio["expires_at"]}  # o cliente usa isso para o contador; nada de id ou código

Repare que a resposta ao cliente carrega apenas expires_at. O id do envio fica no Redis, e o código só existe no celular da pessoa.

Endpoint de validação

A rota de validação recupera o id pelo telefone, chama a API e interpreta os três campos da resposta. Um HTTP 200 não significa sucesso: valid pode ser false com verification em pending (errou, ainda há tentativas), locked (esgotou as tentativas), expired (passou do prazo) ou verified (código já usado).

class VerifyIn(BaseModel):
    telefone: str = Field(pattern=r"^\+[1-9]\d{7,14}$")
    codigo: str = Field(min_length=4, max_length=10, pattern=r"^\d+$")


@app.post("/otp/verify")
async def otp_verify(body: VerifyIn):
    send_id = await store.obter_send_id(body.telefone)
    if not send_id:
        raise HTTPException(400, {"erro": "sem_codigo_ativo"})  # peça um novo envio

    resultado = await aut2fa.verificar_codigo(send_id, body.codigo)

    if not resultado["valid"]:
        if resultado["verification"] in ("expired", "locked", "verified"):
            await store.limpar(body.telefone)  # encerrado: o próximo passo é um novo envio
        raise HTTPException(400, {
            "erro": "codigo_invalido",
            "estado": resultado["verification"],
            "tentativas_restantes": resultado["attempts_left"],
        })

    await store.limpar(body.telefone)
    # telefone confirmado: grave no banco e crie a sessão do usuário aqui
    return {"ok": True}

Antes de chamar a API, vale normalizar o que veio do cliente: remover espaços e qualquer caractere que não seja dígito, porque códigos colados do WhatsApp às vezes trazem espaço ou quebra de linha. O Pydantic acima rejeita essas variações com 422; se preferir ser tolerante, limpe o valor em um field_validator e mantenha a regra de só dígitos depois da limpeza.

O cliente recebe o estado e as tentativas restantes, o que permite mensagens claras na tela: "Código incorreto, restam 3 tentativas" é bem diferente de "Código expirado, peça um novo". Em nenhum caso o backend tenta "ajudar" revelando parte do código ou aceitando variações.

Expiração, tentativas e rate limit

Parte dessas regras já vem da API, e o seu backend só precisa respeitá-las. A outra parte é sua.

O que a API já faz:

• Expira o código no instante expires_at devolvido no envio. Depois disso, a verificação retorna expired.
• Bloqueia o código após 5 erros de verificação, com verification: "locked".
• Recusa envios acima dos limites da conta e, quando habilitados no painel, da aplicação e do destinatário, com HTTP 429 e Retry-After.

O que fica com o seu backend:

• Cooldown por telefone entre dois envios, para o botão "reenviar" não virar uma metralhadora.
• Teto por hora por telefone e por IP, a primeira linha de defesa contra abuso do formulário.
• TTL do estado, para que um id antigo suma do Redis junto com o prazo do código.

O módulo app/store.py resolve os três com contadores de janela fixa:

import os
from redis.asyncio import Redis

redis = Redis.from_url(os.environ.get("REDIS_URL", "redis://localhost:6379/0"))

COOLDOWN_S = 60        # intervalo mínimo entre envios para o mesmo telefone
MAX_TEL_HORA = 5       # envios por telefone, por hora
MAX_IP_HORA = 20       # envios por IP, por hora (ajuste para redes compartilhadas)
TTL_S = 10 * 60        # o id do envio expira junto com o código


async def _dentro_do_limite(chave: str, maximo: int) -> bool:
    total = await redis.incr(chave)
    if total == 1:
        await redis.expire(chave, 3600)
    return total <= maximo


async def pode_enviar(telefone: str, ip: str) -> bool:
    if await redis.exists(f"otp:cooldown:{telefone}"):
        return False
    ok_tel = await _dentro_do_limite(f"otp:hora:tel:{telefone}", MAX_TEL_HORA)
    ok_ip = await _dentro_do_limite(f"otp:hora:ip:{ip}", MAX_IP_HORA)
    if not (ok_tel and ok_ip):
        return False
    await redis.set(f"otp:cooldown:{telefone}", 1, ex=COOLDOWN_S)
    return True


async def guardar(telefone: str, send_id: str) -> None:
    await redis.set(f"otp:send:{telefone}", send_id, ex=TTL_S)


async def obter_send_id(telefone: str) -> str | None:
    valor = await redis.get(f"otp:send:{telefone}")
    return valor.decode() if valor else None


async def limpar(telefone: str) -> None:
    await redis.delete(f"otp:send:{telefone}")

Os números são ponto de partida, não dogma. Cinco envios por hora por telefone é generoso para uso legítimo e suficiente para frear scripts; o teto por IP precisa de folga para redes corporativas e operadoras com NAT, por isso é maior.

Se preferir deixar o limite por destinatário com a própria API, habilite-o no painel da aplicação. Lá o destino é normalizado, então formatos diferentes do mesmo número contam como um só, e o excesso volta como 429 com Retry-After, que o seu endpoint já repassa.

Testes e deploy do OTP com Python

Com a chave ak_test_, o teste de integração em tests/test_otp.py roda contra a API real sem custo e sem mensagem entregue:

import pytest
from httpx import ASGITransport, AsyncClient
from app.main import app


@pytest.mark.anyio
async def test_fluxo_completo():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        r = await c.post("/otp/send", json={"telefone": "+5531987654321"})
        assert r.status_code == 202 and "expires_at" in r.json()

        errado = await c.post("/otp/verify", json={"telefone": "+5531987654321", "codigo": "123456"})
        assert errado.status_code == 400
        assert errado.json()["detail"]["tentativas_restantes"] == 4

        certo = await c.post("/otp/verify", json={"telefone": "+5531987654321", "codigo": "000000"})
        assert certo.status_code == 200 and certo.json()["ok"] is True

O teste cobre o caminho feliz, o erro com contagem de tentativas e a limpeza do estado. O marcador anyio vem com o próprio FastAPI, sem plugin extra; se o projeto já usa pytest-asyncio, troque o decorador. Como o Redis é real, limpe as chaves do telefone de teste no início de cada teste ou aponte a REDIS_URL de teste para um banco lógico separado.

Para o deploy, a lista é curta:

1. Servidor: fastapi run app/main.py em desenvolvimento; em produção, uvicorn atrás de um proxy com HTTPS, com mais de um worker.
2. Segredos: AUT2FA_API_KEY e REDIS_URL como variáveis de ambiente do serviço, nunca no repositório. Troque para ak_live_ só no ambiente de produção.
3. IP real: se há proxy ou load balancer, configure o FastAPI (ou o Uvicorn, com --proxy-headers) para ler X-Forwarded-For; sem isso, todo mundo divide o mesmo limite por IP.
4. Observabilidade: registre id do envio, estado e attempts_left, nunca o código ou o telefone completo. Um webhook da API avisa send.delivered, send.verified e send.failed sem polling.

Perguntas frequentes

Como o código chega por WhatsApp em vez de SMS?

A escolha do canal é feita pela API entre os canais permitidos na aplicação, do mais barato ao mais caro. Com WhatsApp e SMS habilitados, o código tenta primeiro o WhatsApp; se o número não tiver conta ou a entrega falhar, vai por SMS. O seu código Python não muda.

Posso usar o SDK Python em vez de chamar a API com httpx?

Pode. O pacote aut2fa (pip install aut2fa) expõe Aut2fa(chave).send(to=...) e .verify(send_id, code=...), com os mesmos campos. O tutorial usou httpx para deixar explícito o que trafega em cada chamada.

Preciso de Redis ou posso guardar o estado em memória?

Em memória funciona para um único processo em desenvolvimento. Com dois workers ou duas instâncias, o id guardado em um não existe no outro, e a verificação falha de forma aleatória. Redis, ou o seu banco com TTL, resolve isso.

O que fazer quando a API responde 429?

Repasse o Retry-After ao cliente e mostre o prazo no botão de reenvio. Ao repetir a chamada, mantenha a mesma Idempotency-Key e o mesmo corpo para não gerar um segundo envio.

Como testar sem gastar saldo?

Use uma aplicação em modo de teste, cuja chave começa com ak_test_. O código é sempre 000000, nenhuma mensagem é entregue e as respostas, inclusive erros e bloqueio por tentativas, são as mesmas de produção.

Próximos passos

Com os dois endpoints no ar, o OTP com Python vira detalhe de infraestrutura e o restante é produto: a tela de código com contador de reenvio, a confirmação do telefone no cadastro e a sessão emitida depois do valid: true. Para ver como fica a chamada de envio sem o FastAPI em volta, o guia de enviar código de verificação pelo WhatsApp via API mostra a requisição crua e os estados de entrega.

Crie uma aplicação de teste, coloque a chave ak_test_ no .env e rode o teste acima: se ele passar, trocar para ak_live_ é a única mudança até a produção.