Pular para o conteúdo principal
Documentation

SDK Python

PyPI: coffeemail v0.1.0

Referência completa do pacote oficial coffeemail para Python (PyPI). Suporte síncrono e assíncrono (FastAPI/asyncio), modelos Pydantic v2, retorno seguro data, error e método .unwrap().

Buscando a documentação para Node.js ou TypeScript? Acesse a referência oficial do pacote @coffeemail/node.

SDK Node.js
pip install coffeemail

1. Cliente Síncrono (Django, Flask, Celery)

Django • Flask • Celery

Suporte a clientes síncronos (CoffeeMail) e assíncronos (AsyncCoffeeMail). A API Key é carregada automaticamente da variável de ambiente COFFEEMAIL_API_KEY se omitida.

CoffeeMail(api_key: str | None = None, *, timeout: float = 10.0, locale: Locale = "pt-BR")
sync_client.py
from coffeemail import CoffeeMail # Carrega automaticamente COFFEEMAIL_API_KEY do ambiente:client = CoffeeMail() data, error = client.emails.send({    "from": "noreply@seudominio.com.br",    "to": "cliente@empresa.com",    "subject": "Boas-vindas!",    "html": "<h1>Olá!</h1>",})

2. Cliente Assíncrono (FastAPI, asyncio, Temporal)

FastAPI • asyncio • Temporal

Suporte a clientes síncronos (CoffeeMail) e assíncronos (AsyncCoffeeMail). A API Key é carregada automaticamente da variável de ambiente COFFEEMAIL_API_KEY se omitida.

AsyncCoffeeMail(api_key: str | None = None, *, timeout: float = 10.0, locale: Locale = "pt-BR")
async_client.py
import asynciofrom coffeemail import AsyncCoffeeMail async def main():    client = AsyncCoffeeMail()    data, error = await client.emails.send({        "from": "noreply@seudominio.com.br",        "to": "cliente@empresa.com",        "subject": "Boas-vindas!",        "html": "<h1>Olá via FastAPI!</h1>",    }) asyncio.run(main())

Desempacotamento de Tupla (Recomendado)

Padrão Idiomático Seguro

Todas as operações retornam uma instância de CoffeeMailResponse[T] permitindo desempacotar data, error diretamente ou invocar .unwrap() para lançar exceções capturáveis.

tuple_unpacking.py
data, error = client.emails.get("eml_123") if error:    print(f"Erro [{error.code}]: {error.message}")else:    print(f"Status do email: {data.status}")

Método .unwrap() (Estilo Exceção)

Try / Except

Todas as operações retornam uma instância de CoffeeMailResponse[T] permitindo desempacotar data, error diretamente ou invocar .unwrap() para lançar exceções capturáveis.

unwrap_exception.py
from coffeemail.core.errors import CoffeeMailError try:    email = client.emails.get("eml_123").unwrap()    print(f"Entregue em: {email.delivered_at}")except CoffeeMailError as err:    print(f"Exceção capturada: {err.message}")

Validação Criptográfica de Webhooks

O utilitário Webhooks.verify_signature valida a assinatura HMAC SHA-256 no header x-coffeemail-signature contra adulteração de payloads.

verify_webhook.py
from coffeemail import Webhooks # Em um endpoint FastAPI:@app.post("/webhooks/coffeemail")async def handle_webhook(request: Request):    body = await request.body()    signature = request.headers.get("x-coffeemail-signature")        is_valid = Webhooks.verify_signature(        payload=body,        signature=signature,        secret=os.environ["COFFEEMAIL_WEBHOOK_SECRET"],    )        if not is_valid:        raise HTTPException(status_code=400, detail="Assinatura inválida")            return {"received": True}
client.emails

Emails

client.emailsPython 3.10+

.send()

Assinatura Síncrona:client.emails.send(payload: SendEmailPayload | dict) -> CoffeeMailResponse[SendEmailResponse]Assinatura Assíncrona:await async_client.emails.send(payload: SendEmailPayload | dict) -> CoffeeMailResponse[SendEmailResponse]

Envia um e-mail transacional de forma síncrona ou assíncrona.

Chamada do método
from coffeemail import CoffeeMail client = CoffeeMail() data, error = client.emails.send({    "from": "noreply@seudominio.com.br",    "to": "cliente@example.com",    "subject": "Confirmação do pedido #123",    "html": "<h1>Obrigado!</h1><p>Seu pedido está em processamento.</p>",}) if error:    raise SystemExit(f"Erro ao enviar: {error.message}") print("E-mail enfileirado com sucesso:", data.id)
Retorno esperado (data)data, error = client...
data.json
{  "id": "eml_8f3a2c1b",  "status": "queued",  "queuedAt": "2026-10-02T00:00:00.000Z"}
client.emailsPython 3.10+

.send_batch()

Assinatura Síncrona:client.emails.send_batch(emails: list[SendEmailPayload | dict]) -> CoffeeMailResponse[BatchSendEmailResult]Assinatura Assíncrona:await async_client.emails.send_batch(emails: list[SendEmailPayload | dict]) -> CoffeeMailResponse[BatchSendEmailResult]

Envia múltiplos e-mails em um único lote otimizado.

Chamada do método
send_batch.py
data, error = client.emails.send_batch([    {        "from": "noreply@seudominio.com.br",        "to": "alice@example.com",        "subject": "Seu extrato mensal",        "html": "<p>Olá Alice, seu extrato está disponível.</p>",    },    {        "from": "noreply@seudominio.com.br",        "to": "bob@example.com",        "subject": "Seu extrato mensal",        "html": "<p>Olá Bob, seu extrato está disponível.</p>",    },]) if error:    raise SystemExit(f"Erro no lote: {error.message}") print(f"Total enfileirado: {len(data.items)}")
client.emailsPython 3.10+

.get()

Assinatura Síncrona:client.emails.get(email_id: str) -> CoffeeMailResponse[EmailDetail]Assinatura Assíncrona:await async_client.emails.get(email_id: str) -> CoffeeMailResponse[EmailDetail]

Consulta detalhes, status de entrega e timestamps de um e-mail.

Chamada do método
get.py
data, error = client.emails.get("eml_123456789") if error:    print(f"Falha ao buscar e-mail: {error.message}")else:    print(f"Status: {data.status} | Entregue em: {data.delivered_at}")
Retorno esperado (data)data, error = client...
data.json
{  "id": "eml_8f3a2c1b",  "status": "delivered",  "from": "noreply@seudominio.com.br",  "to": ["cliente@example.com"],  "subject": "Confirmação do pedido #123",  "deliveredAt": "2026-10-02T00:01:15.000Z"}
client.emailsPython 3.10+

.list()

Assinatura Síncrona:client.emails.list(limit: int | None = None, after: str | None = None, status: str | None = None) -> CoffeeMailResponse[ListEmailsResponse]Assinatura Assíncrona:await async_client.emails.list(limit: int | None = None, after: str | None = None, status: str | None = None) -> CoffeeMailResponse[ListEmailsResponse]

Lista e-mails recentes com suporte a paginação e filtros de status.

Chamada do método
list.py
data, error = client.emails.list(limit=20, status="delivered") if error:    raise SystemExit(error.message) for email in data.items:    print(f"{email.id} -> {email.to}: {email.status}")
Retorno esperado (data)data, error = client...
data.json
{  "data": [    {      "id": "eml_8f3a2c1b",      "status": "delivered",      "to": ["cliente@example.com"],      "subject": "Boas-vindas",      "createdAt": "2026-10-02T00:00:00.000Z"    }  ],  "pagination": {    "page": 1,    "pageSize": 20,    "total": 1  }}
client.emailsPython 3.10+

.cancel()

Assinatura Síncrona:client.emails.cancel(email_id: str) -> CoffeeMailResponse[dict[str, object]]Assinatura Assíncrona:await async_client.emails.cancel(email_id: str) -> CoffeeMailResponse[dict[str, object]]

Cancela um e-mail agendado antes do disparo.

Chamada do método
cancel.py
data, error = client.emails.cancel("eml_agendado_123") if error:    print(f"Não foi possível cancelar: {error.message}")else:    print("Envio agendado cancelado com sucesso!")
Retorno esperado (data)data, error = client...
data.json
{  "id": "eml_agendado_123",  "cancelled": true}
client.domains

Domínios

client.domainsPython 3.10+

.create()

Assinatura Síncrona:client.domains.create(name: str) -> CoffeeMailResponse[CreateDomainResponse]Assinatura Assíncrona:await async_client.domains.create(name: str) -> CoffeeMailResponse[CreateDomainResponse]

Registra um novo domínio remetente e retorna registros DNS obrigatórios.

Chamada do método
create.py
data, error = client.domains.create("app.seudominio.com.br") if error:    raise SystemExit(f"Erro ao criar domínio: {error.message}") print(f"Domínio {data.name} registrado. Registros DNS obrigatórios:")for record in data.records:    print(f"{record.type} {record.name} -> {record.value}")
Retorno esperado (data)data, error = client...
data.json
{  "id": "dom_7a8b9c",  "name": "seudominio.com.br",  "status": "pending",  "dnsRecords": [    {      "type": "TXT",      "name": "_coffeemail",      "value": "cm_verify=7a8b9c..."    },    {      "type": "TXT",      "name": "cm._domainkey",      "value": "v=DKIM1; k=rsa; p=MIGfMA0GCS..."    }  ]}
client.domainsPython 3.10+

.verify()

Assinatura Síncrona:client.domains.verify(domain_id: str) -> CoffeeMailResponse[DomainDetail]Assinatura Assíncrona:await async_client.domains.verify(domain_id: str) -> CoffeeMailResponse[DomainDetail]

Executa a verificação ativa dos registros SPF, DKIM, DMARC e MX.

Chamada do método
verify.py
data, error = client.domains.verify("dom_123456") if error:    print(f"Verificação pendente: {error.message}")else:    print(f"Status SPF: {data.spf_status} | DKIM: {data.dkim_status} | Geral: {data.status}")
Retorno esperado (data)data, error = client...
data.json
{  "id": "dom_7a8b9c",  "name": "seudominio.com.br",  "status": "verified",  "spfValid": true,  "dkimValid": true,  "dmarcValid": true}
client.domainsPython 3.10+

.list()

Assinatura Síncrona:client.domains.list() -> CoffeeMailResponse[ListDomainsResponse]Assinatura Assíncrona:await async_client.domains.list() -> CoffeeMailResponse[ListDomainsResponse]

Lista os domínios configurados para a organização.

Chamada do método
list.py
data, error = client.domains.list() if error:    raise SystemExit(error.message) for dom in data.items:    print(f"{dom.name} ({dom.status})")
Retorno esperado (data)data, error = client...
data.json
[  {    "id": "dom_7a8b9c",    "name": "seudominio.com.br",    "status": "verified"  }]
client.domainsPython 3.10+

.get()

Assinatura Síncrona:client.domains.get(domain_id: str) -> CoffeeMailResponse[DomainDetail]Assinatura Assíncrona:await async_client.domains.get(domain_id: str) -> CoffeeMailResponse[DomainDetail]

Obtém detalhes e status de propagação de um domínio.

Chamada do método
get.py
data, error = client.domains.get("dom_123456") if error:    raise SystemExit(error.message) print(f"Domínio: {data.name} | Status: {data.status}")
Retorno esperado (data)data, error = client...
data.json
{  "id": "dom_7a8b9c",  "name": "seudominio.com.br",  "status": "verified",  "spfValid": true,  "dkimValid": true,  "dmarcValid": true}
client.domainsPython 3.10+

.delete()

Assinatura Síncrona:client.domains.delete(domain_id: str) -> CoffeeMailResponse[dict[str, object]]Assinatura Assíncrona:await async_client.domains.delete(domain_id: str) -> CoffeeMailResponse[dict[str, object]]

Remove um domínio remetente cadastrado.

Chamada do método
delete.py
data, error = client.domains.delete("dom_123456") if error:    print("Erro ao excluir domínio:", error.message)else:    print("Domínio removido.")
Retorno esperado (data)data, error = client...
data.json
{  "id": "dom_7a8b9c",  "deleted": true}
client.templates

Modelos

client.templatesPython 3.10+

.create()

Assinatura Síncrona:client.templates.create(payload: CreateTemplatePayload | dict) -> CoffeeMailResponse[TemplateDetail]Assinatura Assíncrona:await async_client.templates.create(payload: CreateTemplatePayload | dict) -> CoffeeMailResponse[TemplateDetail]

Cadastra um modelo de e-mail com variáveis dinâmicas.

Chamada do método
create.py
data, error = client.templates.create({    "name": "Boas-vindas",    "subject": "Olá, {{name}}! Bem-vindo.",    "html": "<h1>Olá, {{name}}!</h1><p>Sua conta foi criada com sucesso.</p>",}) if error:    raise SystemExit(error.message) print("Template criado ID:", data.id)
Retorno esperado (data)data, error = client...
data.json
{  "id": "tpl_9f8e7d",  "name": "Boas-vindas",  "engine": "handlebars",  "version": 1}
client.templatesPython 3.10+

.preview()

Assinatura Síncrona:client.templates.preview(payload: PreviewTemplatePayload | dict) -> CoffeeMailResponse[PreviewTemplateResponse]Assinatura Assíncrona:await async_client.templates.preview(payload: PreviewTemplatePayload | dict) -> CoffeeMailResponse[PreviewTemplateResponse]

Gera pré-visualização HTML do template com dados simulados.

Chamada do método
preview.py
data, error = client.templates.preview({    "template_id": "tpl_123",    "variables": {"name": "Carlos Silva"},}) if error:    print("Erro ao renderizar:", error.message)else:    print("HTML renderizado:", data.html)
Retorno esperado (data)data, error = client...
data.json
{  "html": "<h1>Olá João!</h1><p>Seu código é: 123456</p>",  "text": "Olá João! Seu código é: 123456"}
client.templatesPython 3.10+

.list()

Assinatura Síncrona:client.templates.list() -> CoffeeMailResponse[list[TemplateDetail]]Assinatura Assíncrona:await async_client.templates.list() -> CoffeeMailResponse[list[TemplateDetail]]

Lista modelos de e-mail cadastrados na organização.

Chamada do método
list.py
data, error = client.templates.list() if error:    raise SystemExit(error.message) for tpl in data:    print(f"{tpl.id} - {tpl.name}")
Retorno esperado (data)data, error = client...
data.json
[  {    "id": "tpl_9f8e7d",    "name": "Boas-vindas",    "engine": "handlebars"  }]
client.audiences

Audiências

client.audiencesPython 3.10+

.create()

Assinatura Síncrona:client.audiences.create(payload: CreateAudiencePayload | dict) -> CoffeeMailResponse[AudienceDetail]Assinatura Assíncrona:await async_client.audiences.create(payload: CreateAudiencePayload | dict) -> CoffeeMailResponse[AudienceDetail]

Cria uma nova lista de audiência para contatos e campanhas.

Chamada do método
create.py
data, error = client.audiences.create({    "name": "Clientes VIP 2026",    "description": "Contatos com planos ativos Enterprise",}) if error:    raise SystemExit(error.message) print(f"Audiência criada: {data.id} ({data.name})")
Retorno esperado (data)data, error = client...
data.json
{  "id": "aud_1a2b3c",  "name": "Clientes VIP",  "totalContacts": 0}
client.audiencesPython 3.10+

.add_contact()

Assinatura Síncrona:client.audiences.add_contact(audience_id: str, payload: ContactPayload | dict) -> CoffeeMailResponse[ContactDetail]Assinatura Assíncrona:await async_client.audiences.add_contact(audience_id: str, payload: ContactPayload | dict) -> CoffeeMailResponse[ContactDetail]

Adiciona ou atualiza um contato em uma audiência.

Chamada do método
add_contact.py
data, error = client.audiences.add_contact(    audience_id="aud_123456",    payload={        "email": "lead@empresa.com",        "first_name": "Mariana",        "last_name": "Ribeiro",    },) if error:    raise SystemExit(error.message) print(f"Contato {data.email} adicionado.")
Retorno esperado (data)data, error = client...
data.json
{  "id": "cnt_4d5e6f",  "email": "lead@empresa.com",  "subscribed": true}
client.audiencesPython 3.10+

.bulk_add()

Assinatura Síncrona:client.audiences.bulk_add(audience_id: str, contacts: list[ContactPayload | dict]) -> CoffeeMailResponse[BulkContactResult]Assinatura Assíncrona:await async_client.audiences.bulk_add(audience_id: str, contacts: list[ContactPayload | dict]) -> CoffeeMailResponse[BulkContactResult]

Importa contatos em lote para uma audiência existente.

Chamada do método
bulk_add.py
data, error = client.audiences.bulk_add(    audience_id="aud_123456",    contacts=[        {"email": "joao@example.com", "first_name": "João"},        {"email": "maria@example.com", "first_name": "Maria"},    ],) if error:    raise SystemExit(error.message) print(f"Inseridos: {data.inserted_count} | Ignorados: {data.skipped_count}")
Retorno esperado (data)data, error = client...
data.json
{  "insertedCount": 150,  "failedCount": 0}
client.broadcasts

Broadcasts

client.broadcastsPython 3.10+

.create()

Assinatura Síncrona:client.broadcasts.create(payload: CreateBroadcastPayload | dict) -> CoffeeMailResponse[BroadcastDetail]Assinatura Assíncrona:await async_client.broadcasts.create(payload: CreateBroadcastPayload | dict) -> CoffeeMailResponse[BroadcastDetail]

Cria uma campanha de broadcast associada a uma audiência.

Chamada do método
create.py
data, error = client.broadcasts.create({    "name": "Lançamento Q4",    "audience_id": "aud_123456",    "from": "novidades@seudominio.com.br",    "subject": "Chegou a nova versão da plataforma!",    "html": "<h1>Novidades exclusivas</h1><p>Confira os detalhes.</p>",}) if error:    raise SystemExit(error.message) print("Broadcast rascunho criado ID:", data.id)
Retorno esperado (data)data, error = client...
data.json
{  "id": "brd_3c2b1a",  "name": "Campanha Outubro",  "status": "draft"}
client.broadcastsPython 3.10+

.send()

Assinatura Síncrona:client.broadcasts.send(broadcast_id: str) -> CoffeeMailResponse[BroadcastDetail]Assinatura Assíncrona:await async_client.broadcasts.send(broadcast_id: str) -> CoffeeMailResponse[BroadcastDetail]

Dispara o envio da campanha para todos os inscritos da audiência.

Chamada do método
send.py
data, error = client.broadcasts.send("bc_123456") if error:    raise SystemExit(error.message) print(f"Campanha disparada! Status: {data.status}")
Retorno esperado (data)data, error = client...
data.json
{  "id": "brd_3c2b1a",  "status": "sending",  "recipientsCount": 1250}
client.suppressions

Supressões

client.suppressionsPython 3.10+

.list()

Assinatura Síncrona:client.suppressions.list(limit: int | None = None, after: str | None = None) -> CoffeeMailResponse[ListSuppressionsResponse]Assinatura Assíncrona:await async_client.suppressions.list(limit: int | None = None, after: str | None = None) -> CoffeeMailResponse[ListSuppressionsResponse]

Lista e-mails na lista de supressão por bounce ou unsubscribe.

Chamada do método
list.py
data, error = client.suppressions.list() if error:    raise SystemExit(error.message) for item in data.items:    print(f"{item.email} (Motivo: {item.reason})")
Retorno esperado (data)data, error = client...
data.json
{  "data": [    {      "email": "bounced@invalid.com",      "reason": "hard_bounce",      "createdAt": "2026-09-30T10:00:00.000Z"    }  ],  "pagination": {    "page": 1,    "pageSize": 50,    "total": 1  }}
client.suppressionsPython 3.10+

.add()

Assinatura Síncrona:client.suppressions.add(email: str, reason: str = "manual") -> CoffeeMailResponse[SuppressionDetail]Assinatura Assíncrona:await async_client.suppressions.add(email: str, reason: str = "manual") -> CoffeeMailResponse[SuppressionDetail]

Adiciona manualmente um e-mail à lista de supressão.

Chamada do método
add.py
data, error = client.suppressions.add("optout@example.com", reason="unsubscribe") if error:    print("Erro ao suprimir:", error.message)else:    print("E-mail adicionado à supressão.")
Retorno esperado (data)data, error = client...
data.json
{  "email": "spam@complaint.com",  "reason": "complaint",  "created": true}
client.webhooks

Webhooks

client.webhooksPython 3.10+

.verify_signature()

Assinatura Síncrona:Webhooks.verify_signature(payload: str | bytes, signature: str, secret: str) -> bool

Valida criptograficamente o header de assinatura HMAC SHA-256 do webhook.

Chamada do método
verify_signature.py
from coffeemail import Webhooks # Em uma rota FastAPI ou Flask:is_valid = Webhooks.verify_signature(    payload=raw_request_body,    signature=request.headers.get("x-coffeemail-signature"),    secret="whsec_seu_segredo",) if not is_valid:    raise ValueError("Assinatura do webhook inválida ou adulterada.")
Retorno esperado (data)data, error = client...
data.json
true
client.stats

Estatísticas

client.statsPython 3.10+

.get()

Assinatura Síncrona:client.stats.get(from_date: str | None = None, to_date: str | None = None) -> CoffeeMailResponse[StatsOverview]Assinatura Assíncrona:await async_client.stats.get(from_date: str | None = None, to_date: str | None = None) -> CoffeeMailResponse[StatsOverview]

Consulta métricas agregadas de envio, entrega e rejeição por período.

Chamada do método
get.py
data, error = client.stats.get(from_date="2026-09-01", to_date="2026-09-30") if error:    raise SystemExit(error.message) print(f"Total enviados: {data.total_sent} | Entregues: {data.total_delivered} ({data.delivery_rate}%)")
Retorno esperado (data)data, error = client...
data.json
{  "delivered": 14200,  "bounced": 12,  "complaints": 2,  "deliveryRate": 99.91}