Documentation

Python SDK

PyPI: coffeemail v0.1.0

Complete reference for the official coffeemail package for Python (PyPI). Synchronous and asynchronous (FastAPI/asyncio) support, Pydantic v2 models, safe return data, error pattern and .unwrap() method.

Looking for Node.js or TypeScript docs? Check the official reference for the @coffeemail/node package.

Node.js SDK

Installation

pypi.org/project/coffeemail ↗

pip install coffeemail

1. Synchronous Client (Django, Flask, Celery)

Django • Flask • Celery

Support for synchronous (CoffeeMail) and asynchronous (AsyncCoffeeMail) clients. The API Key is automatically loaded from COFFEEMAIL_API_KEY if omitted.

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. Asynchronous Client (FastAPI, asyncio, Temporal)

FastAPI • asyncio • Temporal

Support for synchronous (CoffeeMail) and asynchronous (AsyncCoffeeMail) clients. The API Key is automatically loaded from COFFEEMAIL_API_KEY if omitted.

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

Tuple Unpacking (Recommended)

Padrão Idiomático Seguro

All operations return an instance of CoffeeMailResponse[T] allowing direct unpacking of data, error or calling .unwrap() to raise catchable exceptions.

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

.unwrap() Method (Exception Style)

Try / Except

All operations return an instance of CoffeeMailResponse[T] allowing direct unpacking of data, error or calling .unwrap() to raise catchable exceptions.

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

Cryptographic Webhook Validation

The Webhooks.verify_signature utility validates the HMAC SHA-256 signature in the x-coffeemail-signature header against payload tampering.

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

Synchronous Signature:client.emails.send(payload: SendEmailPayload | dict) -> CoffeeMailResponse[SendEmailResponse]

Asynchronous Signature:await async_client.emails.send(payload: SendEmailPayload | dict) -> CoffeeMailResponse[SendEmailResponse]

Sends a transactional email synchronously or asynchronously.

Method invocation

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)

Expected response (data)data, error = client...

data.json

{  "id": "eml_8f3a2c1b",  "status": "queued",  "queuedAt": "2026-10-02T00:00:00.000Z"}

client.emailsPython 3.10+

.send_batch()

Synchronous Signature:client.emails.send_batch(emails: list[SendEmailPayload | dict]) -> CoffeeMailResponse[BatchSendEmailResult]

Asynchronous Signature:await async_client.emails.send_batch(emails: list[SendEmailPayload | dict]) -> CoffeeMailResponse[BatchSendEmailResult]

Sends multiple emails in a single optimized batch.

Method invocation

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

Synchronous Signature:client.emails.get(email_id: str) -> CoffeeMailResponse[EmailDetail]Asynchronous Signature:await async_client.emails.get(email_id: str) -> CoffeeMailResponse[EmailDetail]

Retrieves email details, delivery status, and timestamps.

Method invocation

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

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

Synchronous Signature:client.emails.list(limit: int | None = None, after: str | None = None, status: str | None = None) -> CoffeeMailResponse[ListEmailsResponse]

Asynchronous Signature:await async_client.emails.list(limit: int | None = None, after: str | None = None, status: str | None = None) -> CoffeeMailResponse[ListEmailsResponse]

Lists recent emails with pagination and status filters.

Method invocation

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

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

Synchronous Signature:client.emails.cancel(email_id: str) -> CoffeeMailResponse[dict[str, object]]Asynchronous Signature:await async_client.emails.cancel(email_id: str) -> CoffeeMailResponse[dict[str, object]]

Cancels a scheduled email before delivery.

Method invocation

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

Expected response (data)data, error = client...

data.json

{  "id": "eml_agendado_123",  "cancelled": true}

client.domains

Domains

client.domainsPython 3.10+

.create()

Synchronous Signature:client.domains.create(name: str) -> CoffeeMailResponse[CreateDomainResponse]Asynchronous Signature:await async_client.domains.create(name: str) -> CoffeeMailResponse[CreateDomainResponse]

Registers a new sending domain and returns required DNS records.

Method invocation

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

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

Synchronous Signature:client.domains.verify(domain_id: str) -> CoffeeMailResponse[DomainDetail]Asynchronous Signature:await async_client.domains.verify(domain_id: str) -> CoffeeMailResponse[DomainDetail]

Runs active verification for SPF, DKIM, DMARC, and MX records.

Method invocation

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

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

Synchronous Signature:client.domains.list() -> CoffeeMailResponse[ListDomainsResponse]Asynchronous Signature:await async_client.domains.list() -> CoffeeMailResponse[ListDomainsResponse]

Lists configured domains for the organization.

Method invocation

list.py

data, error = client.domains.list() if error:    raise SystemExit(error.message) for dom in data.items:    print(f"{dom.name} ({dom.status})")

Expected response (data)data, error = client...

data.json

[  {    "id": "dom_7a8b9c",    "name": "seudominio.com.br",    "status": "verified"  }]

client.domainsPython 3.10+

.get()

Synchronous Signature:client.domains.get(domain_id: str) -> CoffeeMailResponse[DomainDetail]Asynchronous Signature:await async_client.domains.get(domain_id: str) -> CoffeeMailResponse[DomainDetail]

Gets details and DNS propagation status for a domain.

Method invocation

get.py

data, error = client.domains.get("dom_123456") if error:    raise SystemExit(error.message) print(f"Domínio: {data.name} | Status: {data.status}")

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

Synchronous Signature:client.domains.delete(domain_id: str) -> CoffeeMailResponse[dict[str, object]]Asynchronous Signature:await async_client.domains.delete(domain_id: str) -> CoffeeMailResponse[dict[str, object]]

Removes a registered sending domain.

Method invocation

delete.py

data, error = client.domains.delete("dom_123456") if error:    print("Erro ao excluir domínio:", error.message)else:    print("Domínio removido.")

Expected response (data)data, error = client...

data.json

{  "id": "dom_7a8b9c",  "deleted": true}

client.templates

Templates

client.templatesPython 3.10+

.create()

Synchronous Signature:client.templates.create(payload: CreateTemplatePayload | dict) -> CoffeeMailResponse[TemplateDetail]

Asynchronous Signature:await async_client.templates.create(payload: CreateTemplatePayload | dict) -> CoffeeMailResponse[TemplateDetail]

Creates an email template with dynamic variables.

Method invocation

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)

Expected response (data)data, error = client...

data.json

{  "id": "tpl_9f8e7d",  "name": "Boas-vindas",  "engine": "handlebars",  "version": 1}

client.templatesPython 3.10+

.preview()

Synchronous Signature:client.templates.preview(payload: PreviewTemplatePayload | dict) -> CoffeeMailResponse[PreviewTemplateResponse]

Asynchronous Signature:await async_client.templates.preview(payload: PreviewTemplatePayload | dict) -> CoffeeMailResponse[PreviewTemplateResponse]

Renders an HTML preview of the template with simulated variables.

Method invocation

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)

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

Synchronous Signature:client.templates.list() -> CoffeeMailResponse[list[TemplateDetail]]Asynchronous Signature:await async_client.templates.list() -> CoffeeMailResponse[list[TemplateDetail]]

Lists email templates registered in the organization.

Method invocation

list.py

data, error = client.templates.list() if error:    raise SystemExit(error.message) for tpl in data:    print(f"{tpl.id} - {tpl.name}")

Expected response (data)data, error = client...

data.json

[  {    "id": "tpl_9f8e7d",    "name": "Boas-vindas",    "engine": "handlebars"  }]

client.audiences

Audiences

client.audiencesPython 3.10+

.create()

Synchronous Signature:client.audiences.create(payload: CreateAudiencePayload | dict) -> CoffeeMailResponse[AudienceDetail]

Asynchronous Signature:await async_client.audiences.create(payload: CreateAudiencePayload | dict) -> CoffeeMailResponse[AudienceDetail]

Creates a new audience list for contacts and campaigns.

Method invocation

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

Expected response (data)data, error = client...

data.json

{  "id": "aud_1a2b3c",  "name": "Clientes VIP",  "totalContacts": 0}

client.audiencesPython 3.10+

.add_contact()

Synchronous Signature:client.audiences.add_contact(audience_id: str, payload: ContactPayload | dict) -> CoffeeMailResponse[ContactDetail]

Asynchronous Signature:await async_client.audiences.add_contact(audience_id: str, payload: ContactPayload | dict) -> CoffeeMailResponse[ContactDetail]

Adds or updates a contact in an audience.

Method invocation

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

Expected response (data)data, error = client...

data.json

{  "id": "cnt_4d5e6f",  "email": "lead@empresa.com",  "subscribed": true}

client.audiencesPython 3.10+

.bulk_add()

Synchronous Signature:client.audiences.bulk_add(audience_id: str, contacts: list[ContactPayload | dict]) -> CoffeeMailResponse[BulkContactResult]

Asynchronous Signature:await async_client.audiences.bulk_add(audience_id: str, contacts: list[ContactPayload | dict]) -> CoffeeMailResponse[BulkContactResult]

Imports contacts in bulk into an existing audience.

Method invocation

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

Expected response (data)data, error = client...

data.json

{  "insertedCount": 150,  "failedCount": 0}

client.broadcasts

Broadcasts

client.broadcastsPython 3.10+

.create()

Synchronous Signature:client.broadcasts.create(payload: CreateBroadcastPayload | dict) -> CoffeeMailResponse[BroadcastDetail]

Asynchronous Signature:await async_client.broadcasts.create(payload: CreateBroadcastPayload | dict) -> CoffeeMailResponse[BroadcastDetail]

Creates a broadcast campaign associated with an audience.

Method invocation

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)

Expected response (data)data, error = client...

data.json

{  "id": "brd_3c2b1a",  "name": "Campanha Outubro",  "status": "draft"}

client.broadcastsPython 3.10+

.send()

Synchronous Signature:client.broadcasts.send(broadcast_id: str) -> CoffeeMailResponse[BroadcastDetail]Asynchronous Signature:await async_client.broadcasts.send(broadcast_id: str) -> CoffeeMailResponse[BroadcastDetail]

Triggers broadcast delivery to all active subscribers in the audience.

Method invocation

send.py

data, error = client.broadcasts.send("bc_123456") if error:    raise SystemExit(error.message) print(f"Campanha disparada! Status: {data.status}")

Expected response (data)data, error = client...

data.json

{  "id": "brd_3c2b1a",  "status": "sending",  "recipientsCount": 1250}

client.suppressions

Suppressions

client.suppressionsPython 3.10+

.list()

Synchronous Signature:client.suppressions.list(limit: int | None = None, after: str | None = None) -> CoffeeMailResponse[ListSuppressionsResponse]

Asynchronous Signature:await async_client.suppressions.list(limit: int | None = None, after: str | None = None) -> CoffeeMailResponse[ListSuppressionsResponse]

Lists suppressed email addresses from bounces or unsubscribes.

Method invocation

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

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

Synchronous Signature:client.suppressions.add(email: str, reason: str = "manual") -> CoffeeMailResponse[SuppressionDetail]

Asynchronous Signature:await async_client.suppressions.add(email: str, reason: str = "manual") -> CoffeeMailResponse[SuppressionDetail]

Manually adds an email address to the suppression list.

Method invocation

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

Expected response (data)data, error = client...

data.json

{  "email": "spam@complaint.com",  "reason": "complaint",  "created": true}

client.webhooks

Webhooks

client.webhooksPython 3.10+

.verify_signature()

Synchronous Signature:Webhooks.verify_signature(payload: str | bytes, signature: str, secret: str) -> bool

Cryptographically verifies the HMAC SHA-256 signature in the webhook header.

Method invocation

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

Expected response (data)data, error = client...

data.json

true

client.stats

Statistics

client.statsPython 3.10+

.get()

Synchronous Signature:client.stats.get(from_date: str | None = None, to_date: str | None = None) -> CoffeeMailResponse[StatsOverview]

Asynchronous Signature:await async_client.stats.get(from_date: str | None = None, to_date: str | None = None) -> CoffeeMailResponse[StatsOverview]

Queries aggregate sending, delivery, and rejection statistics by period.

Method invocation

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

Expected response (data)data, error = client...

data.json

{  "delivered": 14200,  "bounced": 12,  "complaints": 2,  "deliveryRate": 99.91}