Webhooks com FastAPI para CRM

Webhooks são uma das formas mais práticas de ligar formulários, landing pages, CRMs, ferramentas de automação e sistemas internos. Em vez de rodar um script manual para buscar dados, você cria um endpoint HTTP que recebe eventos quando algo acontece: um lead novo entrou, um negócio mudou de etapa, um pagamento foi aprovado, uma tarefa foi criada ou uma campanha gerou conversão.

Para quem trabalha ou quer trabalhar com Python em CRM, RevOps, marketing operations, dados ou automação, saber montar um webhook confiável é uma habilidade muito mais próxima do mercado do que apenas consumir uma API pronta. Vagas recentes de Python no Brasil já combinam FastAPI, REST, HubSpot, Salesforce, n8n, SQL e automação de processos. Este guia mostra um projeto pequeno, seguro e demonstrável: receber leads por webhook, validar o payload, deduplicar por e-mail, registrar logs e preparar a integração com CRM.

Se você ainda não criou APIs com Python, leia primeiro API com FastAPI. Se o seu foco é a visão de negócio, veja também Python para Automação de CRM e Marketing, Python para RevOps e CRM e projetos de portfólio Python para conseguir vaga.

Quando usar webhook

Webhook faz sentido quando outro sistema precisa avisar sua aplicação em tempo quase real. Alguns exemplos comuns:

• formulário de landing page envia um lead para sua API;
• n8n recebe um evento e chama um serviço Python para validar regras;
• HubSpot ou Salesforce avisa que um contato mudou de estágio;
• checkout envia uma compra aprovada para atualizar o CRM;
• ferramenta de suporte cria uma tarefa para vendas quando um ticket vira oportunidade;
• sistema interno notifica que uma conta precisa de onboarding.

O ponto principal é não depender de alguém exportar CSV ou clicar em “sincronizar”. O evento chega, sua aplicação valida e decide o que fazer.

Estrutura do projeto

Um projeto enxuto para portfólio pode ter esta estrutura:

crm-webhook/
  pyproject.toml
  .env.example
  app/
    main.py
    models.py
    crm.py
    security.py
  tests/
    test_webhook.py

Instale as dependências:

python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn httpx pytest

Em produção você provavelmente usaria PostgreSQL, Redis, fila e observabilidade. Para começar, vamos manter a lógica em memória e deixar claro no README como evoluir.

Modelo do evento

Use Pydantic para validar o payload antes de qualquer regra de negócio. Isso evita código cheio de if campo in data e deixa o contrato documentado no /docs do FastAPI.

from pydantic import BaseModel, EmailStr, Field


class LeadEvento(BaseModel):
    email: EmailStr
    nome: str = Field(min_length=2, max_length=120)
    empresa: str | None = Field(default=None, max_length=120)
    telefone: str | None = Field(default=None, max_length=30)
    origem: str = Field(default="landing-page", max_length=80)
    consentimento: bool = False
    utm_campaign: str | None = Field(default=None, max_length=120)

O campo consentimento é importante porque CRM e marketing lidam com dados pessoais. Mesmo em um projeto de estudo, registre se o contato aceitou comunicação e use dados fictícios nos exemplos.

Protegendo o endpoint

Um erro comum é publicar um webhook aberto sem nenhuma autenticação. Para um projeto simples, um token compartilhado no cabeçalho já reduz abuso acidental.

import os
from fastapi import Header, HTTPException, status


WEBHOOK_TOKEN = os.environ.get("WEBHOOK_TOKEN", "dev-token")


def validar_token(x_webhook_token: str | None = Header(default=None)) -> None:
    if x_webhook_token != WEBHOOK_TOKEN:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="token inválido",
        )

Em produção, prefira assinatura HMAC quando a ferramenta de origem suportar. HubSpot, Stripe e outros provedores costumam assinar eventos para provar que a requisição veio deles. O token simples é suficiente para entender o fluxo, mas não deve ser tratado como segurança completa.

Endpoint FastAPI

Agora crie o endpoint que recebe o lead, valida duplicidade e chama o cliente de CRM. Aqui a função enviar_para_crm está isolada para facilitar testes.

import logging
from fastapi import Depends, FastAPI, status


app = FastAPI(title="CRM Webhook", version="1.0.0")
logger = logging.getLogger("crm_webhook")
emails_processados: set[str] = set()


@app.post("/webhooks/leads", status_code=status.HTTP_202_ACCEPTED)
def receber_lead(
    evento: LeadEvento,
    _: None = Depends(validar_token),
) -> dict[str, str]:
    email = evento.email.lower()

    if email in emails_processados:
        logger.info("lead_duplicado email=%s origem=%s", email, evento.origem)
        return {"status": "duplicado", "email": email}

    if not evento.consentimento:
        logger.info("lead_sem_consentimento email=%s", email)
        return {"status": "ignorado_sem_consentimento", "email": email}

    enviar_para_crm(evento)
    emails_processados.add(email)
    logger.info("lead_processado email=%s origem=%s", email, evento.origem)
    return {"status": "aceito", "email": email}

Esse exemplo retorna 202 Accepted porque o evento foi recebido e aceito para processamento. Em uma aplicação maior, você colocaria o evento em uma fila e responderia rápido, sem travar o webhook enquanto o CRM externo demora.

Cliente de CRM simulado

Não use token real no repositório. Para portfólio, comece com um cliente que aponta para uma API mockada, ambiente de sandbox ou serviço de teste.

import os
import httpx


def enviar_para_crm(evento: LeadEvento) -> None:
    base_url = os.environ.get("CRM_BASE_URL", "https://example.invalid")
    token = os.environ.get("CRM_API_TOKEN", "token-falso")

    payload = {
        "email": evento.email.lower(),
        "name": evento.nome,
        "company": evento.empresa,
        "phone": evento.telefone,
        "source": evento.origem,
        "utm_campaign": evento.utm_campaign,
    }

    with httpx.Client(timeout=10) as client:
        resposta = client.post(
            f"{base_url.rstrip('/')}/contacts/upsert",
            headers={"Authorization": f"Bearer {token}"},
            json=payload,
        )
        resposta.raise_for_status()

Quando for demonstrar, substitua example.invalid por um mock local, como outro endpoint FastAPI, WireMock ou uma rota de teste. O objetivo é mostrar o contrato e o tratamento de erro, não expor credenciais.

Testando com curl

Rode a aplicação:

uvicorn app.main:app --reload

Envie um evento fictício:

curl -X POST http://127.0.0.1:8000/webhooks/leads \
  -H 'Content-Type: application/json' \
  -H 'X-Webhook-Token: dev-token' \
  -d '{
    "email": "[email protected]",
    "nome": "Ana Souza",
    "empresa": "Empresa Demo",
    "origem": "landing-python-crm",
    "consentimento": true,
    "utm_campaign": "guia-webhooks"
  }'

Teste também os casos ruins: token ausente, e-mail inválido, nome muito curto, contato sem consentimento e envio duplicado. Esses cenários aparecem em entrevistas porque mostram que você pensa além do “caminho feliz”.

Integração com n8n

Uma arquitetura comum combina n8n e Python:

1. n8n recebe evento de formulário, planilha ou ferramenta de anúncio;
2. n8n chama POST /webhooks/leads com o token configurado em variável segura;
3. FastAPI valida dados, consentimento e duplicidade;
4. Python atualiza CRM, banco ou fila;
5. n8n envia notificação para Slack, e-mail ou WhatsApp interno.

Essa divisão é saudável. O n8n fica com orquestração visual e conectores prontos. Python fica com regra testável, versionada e mais fácil de revisar em pull request.

O que colocar no README

Para transformar este projeto em sinal de carreira, documente:

• problema de negócio resolvido;
• exemplo de payload;
• variáveis de ambiente em .env.example;
• como rodar localmente;
• como executar testes;
• decisões de segurança;
• limitações da versão demo;
• próximos passos para produção.

Evite afirmar que o projeto é “LGPD compliant” se você não fez uma revisão jurídica. Prefira dizer que ele demonstra boas práticas técnicas: consentimento explícito no payload, minimização de campos, tokens fora do código e logs sem dados sensíveis desnecessários.

Evoluindo para produção

Depois que a versão local funcionar, os próximos passos são claros:

• trocar deduplicação em memória por PostgreSQL;
• salvar cada evento recebido com status de processamento;
• usar fila para retentativas;
• implementar assinatura HMAC;
• adicionar idempotência por event_id;
• criar métricas de sucesso, falha e latência;
• escrever testes para payload inválido, duplicado e CRM fora do ar.

Se o projeto crescer para uma API interna completa, combine este guia com Docker para Python, testes com pytest e Python e PostgreSQL. Para quem mira vagas, conecte o projeto a páginas reais de vagas Python, ao guia de Python para RevOps e CRM e ao guia de currículo Python para vaga júnior. Um webhook pequeno, bem validado e bem documentado costuma impressionar mais do que um CRUD grande sem contexto de negócio.