Documentation

SDK Python

PyPI: coffeemail v0.1.0

Referencia completa del paquete oficial coffeemail para Python (PyPI). Soporte síncrono y asíncrono (FastAPI/asyncio), modelos Pydantic v2, retorno seguro data, error y método .unwrap().

¿Buscas la documentación para Node.js o TypeScript? Consulta la referencia oficial del paquete @coffeemail/node.

SDK Node.js

Instalación

pypi.org/project/coffeemail ↗

pip install coffeemail

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

Django • Flask • Celery

Soporte para clientes síncronos (CoffeeMail) y asíncronos (AsyncCoffeeMail). La API Key se carga automáticamente de la variable de entorno COFFEEMAIL_API_KEY si se omite.

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 Asíncrono (FastAPI, asyncio, Temporal)

FastAPI • asyncio • Temporal

Soporte para clientes síncronos (CoffeeMail) y asíncronos (AsyncCoffeeMail). La API Key se carga automáticamente de la variable de entorno COFFEEMAIL_API_KEY si se omite.

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())

Desempaquetado de Tupla (Recomendado)

Padrão Idiomático Seguro

Todas las operaciones retornan una instancia de CoffeeMailResponse[T] permitiendo desempaquetar data, error directamente o invocar .unwrap() para lanzar excepciones capturables.

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 Excepción)

Try / Except

Todas las operaciones retornan una instancia de CoffeeMailResponse[T] permitiendo desempaquetar data, error directamente o invocar .unwrap() para lanzar excepciones capturables.

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}")

Validación Criptográfica de Webhooks

La utilidad Webhooks.verify_signature valida la firma HMAC SHA-256 en el encabezado x-coffeemail-signature contra la manipulación de cargas útiles.

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()

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

Firma Asíncrona:await async_client.emails.send(payload: SendEmailPayload | dict) -> CoffeeMailResponse[SendEmailResponse]

Envía un correo transaccional de forma síncrona o asíncrona.

Llamada del 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()

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

Firma Asíncrona:await async_client.emails.send_batch(emails: list[SendEmailPayload | dict]) -> CoffeeMailResponse[BatchSendEmailResult]

Envía múltiples correos en un solo lote optimizado.

Llamada del 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()

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

Consulta detalles, estado de entrega y marcas de tiempo de un correo.

Llamada del 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()

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

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

Lista correos recientes con soporte para paginación y filtros de estado.

Llamada del 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()

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

Cancela un correo programado antes de su envío.

Llamada del 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

Dominios

client.domainsPython 3.10+

.create()

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

Registra un nuevo dominio remitente y devuelve los registros DNS requeridos.

Llamada del 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()

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

Ejecuta la verificación activa de registros SPF, DKIM, DMARC y MX.

Llamada del 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()

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

Lista los dominios configurados para la organización.

Llamada del 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()

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

Obtiene detalles y estado de propagación DNS de un dominio.

Llamada del 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()

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

Elimina un dominio remitente registrado.

Llamada del 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

Plantillas

client.templatesPython 3.10+

.create()

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

Firma Asíncrona:await async_client.templates.create(payload: CreateTemplatePayload | dict) -> CoffeeMailResponse[TemplateDetail]

Registra una plantilla de correo con variables dinámicas.

Llamada del 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()

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

Firma Asíncrona:await async_client.templates.preview(payload: PreviewTemplatePayload | dict) -> CoffeeMailResponse[PreviewTemplateResponse]

Genera una vista previa HTML de la plantilla con datos simulados.

Llamada del 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()

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

Lista plantillas de correo registradas en la organización.

Llamada del 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

Audiencias

client.audiencesPython 3.10+

.create()

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

Firma Asíncrona:await async_client.audiences.create(payload: CreateAudiencePayload | dict) -> CoffeeMailResponse[AudienceDetail]

Crea una nueva lista de audiencia para contactos y campañas.

Llamada del 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()

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

Firma Asíncrona:await async_client.audiences.add_contact(audience_id: str, payload: ContactPayload | dict) -> CoffeeMailResponse[ContactDetail]

Agrega o actualiza un contacto en una audiencia.

Llamada del 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()

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

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

Importa contactos en lote a una audiencia existente.

Llamada del 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()

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

Firma Asíncrona:await async_client.broadcasts.create(payload: CreateBroadcastPayload | dict) -> CoffeeMailResponse[BroadcastDetail]

Crea una campaña de broadcast asociada a una audiencia.

Llamada del 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()

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

Inicia el envío de la campaña a todos los suscriptores activos.

Llamada del 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

Supresiones

client.suppressionsPython 3.10+

.list()

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

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

Lista direcciones suprimidas por rebote o cancelación de suscripción.

Llamada del 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()

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

Firma Asíncrona:await async_client.suppressions.add(email: str, reason: str = "manual") -> CoffeeMailResponse[SuppressionDetail]

Agrega manualmente un correo a la lista de supresión.

Llamada del 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()

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

Valida criptográficamente el encabezado de firma HMAC SHA-256 del webhook.

Llamada del 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

Estadísticas

client.statsPython 3.10+

.get()

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

Firma Asíncrona:await async_client.stats.get(from_date: str | None = None, to_date: str | None = None) -> CoffeeMailResponse[StatsOverview]

Consulta métricas agregadas de envío, entrega y rechazo por período.

Llamada del 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}