bavimail-python

Python SDK for the Bavimail email service API.

Version 1.0 targets the current Bavimail API. Upgrading from 0.x? Read MIGRATION.md.

Installation

pip install bavimail

Quick Start

Synchronous

from bavimail import Bavimail

client = Bavimail(api_key="YOUR_API_KEY")

# Send an email from an inbox on a verified domain
email = client.emails.send(
    inbox_id="your-inbox-id",
    to_emails=["recipient@example.com"],
    subject="Hello!",
    body="<p>Welcome!</p>",
)
print(f"Accepted: {email.id} (status: {email.status})")

client.close()

The API key travels in the x-api-key header. Sends are accepted as queued (or scheduled) and delivered by Bavimail; follow them with client.emails.get(email.id) or webhooks.

Asynchronous

Every method has an _async variant. Use async with for automatic cleanup:

import asyncio
from bavimail import Bavimail

async def main():
    async with Bavimail(api_key="YOUR_API_KEY") as client:
        domains = await client.domains.list_async()
        email = await client.emails.send_async(
            inbox_id="your-inbox-id",
            to_emails=["recipient@example.com"],
            subject="Hello!",
            body="<p>Welcome!</p>",
        )

asyncio.run(main())

Resources

Domains

# 1. Publish the account's ownership TXT record (it must stay published)
ownership = client.domains.get_ownership("example.com")
print(f"{ownership.record.type} {ownership.record.name} -> {ownership.record.value}")

# 2. Add the domain and publish its setup records
domain = client.domains.create("example.com")
setup = client.domains.get_setup(domain.id)
for record in setup.dns_records:
    print(f"{record.type} {record.name} -> {record.value}")

# Or let Bavimail publish them through your DNS provider's API
options = client.domains.get_setup_options(domain.id)
if options.detected_provider.supports_api_setup:
    client.domains.auto_setup(domain.id, options.detected_provider.id, {"api_token": "..."})

# 3. Bavimail verifies the domain on its own; watch the progress
status = client.domains.get_setup_status(domain.id)
print(f"Verified checks: {status.progress.verified}/{status.progress.total}, ready={status.ready}")

domain = client.domains.update(domain.id, inbound_enabled=False)
client.domains.delete(domain.id)

Inboxes

inbox = client.inboxes.create(domain.id, "support")  # support@example.com
inboxes = client.inboxes.list(domain_id=domain.id)
client.inboxes.update(inbox.id, signature_html="<p>The team</p>")
client.inboxes.delete(inbox.id)

Emails (Outbound)

email = client.emails.send(
    inbox_id=inbox.id,
    to_emails=["user@example.com"],
    subject="Welcome",
    body="<h1>Hello!</h1>",
    track_opens=True,
    headers={"X-Campaign": "welcome"},
    attachments=[
        {"filename": "doc.pdf", "content": "<base64>", "content_type": "application/pdf"},
        {"attachment_id": "uploaded-attachment-id"},
    ],
)

# Schedule (local time in an IANA timezone) and cancel
later = client.emails.send(
    inbox.id, ["user@example.com"], "Tomorrow", "<p>Hi</p>",
    scheduled_at="2026-10-20T09:00:00", scheduled_timezone="Europe/Rome",
)
client.emails.cancel(later.id)

page = client.emails.list(inbox_id=inbox.id, status="delivered", limit=50)
print(page.total, [e.subject for e in page.data])

events = client.emails.list_events(email.id)
tracking = client.emails.get_tracking(email.id)
result = client.emails.batch_send([
    {"inbox_id": inbox.id, "to_emails": ["a@example.com"], "subject": "Hi", "body": "Hello"},
])
check = client.emails.validate("someone@example.com")

Inbound Emails and Threads

page = client.inbound_emails.list(inbox_id=inbox.id, limit=25)
detail = client.inbound_emails.get(page.data[0].id)
data = client.inbound_emails.download_attachment(detail.id, detail.attachments[0].id)

threads = client.threads.list(unread=True)
thread = client.threads.get(threads.data[0].id)
for message in thread.messages:
    print(message.type, message.subject)

client.threads.update(thread.id, unread=False, add_label_ids=["label-id"])
client.threads.update(thread.id, folder_id="trash")
client.threads.delete(thread.id)  # permanent, only from trash

Labels and Mailing Lists

label = client.labels.create("Important", color="#ff0000")
folder = client.labels.create("Projects", kind="folder")
labels = client.labels.list()

lists = client.mailing_lists.list(status="active")
client.mailing_lists.unsubscribe(lists.data[0].id, trash_threads=True)

Attachments

from bavimail import AttachmentUploadFile

files = client.attachments.upload([AttachmentUploadFile("doc.pdf", b"...", "application/pdf")])
content = client.attachments.download(files[0].id)
client.attachments.delete(files[0].id)

Downloads use the signed URL Bavimail returns; the API key is never sent to storage.

Webhooks

# Create (returns the secret, shown once!)
wh = client.webhooks.create(
    url="https://hooks.example.com/bavimail",
    event_types=["email.inbound.received", "domain.verified"],
)
print(f"Secret: {wh.secret}")  # Store this!

# Bavimail posts a signed {"type": "webhook.verification", "verification_code": ...}
# challenge to the URL; activate the webhook with that code
wh = client.webhooks.verify(wh.id, "verification-code-from-endpoint")

webhooks = client.webhooks.list()
client.webhooks.update(wh.id, status="disabled")
client.webhooks.test(wh.id)
secret = client.webhooks.rotate_secret(wh.id)
client.webhooks.delete(wh.id)

Suppressions, Warmup, Tracking Domains, Reputation

client.suppressions.create("blocked@example.com", note="asked to stop")
page = client.suppressions.list(reason="bounce")

warmup = client.warmup.enable(inbox.id, "bavimail", daily_volume=20)
client.warmup.update(warmup.id, daily_volume=40)
client.warmup.pause(warmup.id)

tracking = client.tracking_domains.create("track.example.com")
client.tracking_domains.verify(tracking.id)

reputation = client.reputation.get()

Agent self-registration

from bavimail import Bavimail

result = Bavimail.register(name="my-agent")  # no credentials needed
client = Bavimail(api_key=result.api_key.key)
print("Ask a person to claim the account:", result.claim.claim_url)

Pagination

Offset lists return a Page (data, total); threads and the event feed return a CursorPage (data, next_cursor). Iterate everything with the helpers:

from bavimail import iter_cursor_pages, iter_pages, iter_pages_async

for email in iter_pages(client.emails.list, inbox_id="...", page_size=50):
    print(email.subject)

for thread in iter_cursor_pages(client.threads.list, unread=True):
    print(thread.subject)

async for email in iter_pages_async(client.inbound_emails.list_async, page_size=25):
    print(email.subject)

Receiving Webhooks

from bavimail import Bavimail, EventType, WebhookEvent
from bavimail.events import InboundReceivedData

client = Bavimail(api_key="YOUR_API_KEY")

@client.on(EventType.INBOUND_RECEIVED)
def handle(event: WebhookEvent):
    data = InboundReceivedData.from_dict(event.data)
    print(data.from_email, data.subject)

# Standalone server: creates and verifies the webhook, deletes it on exit
client.listen(port=8080, webhook_url="https://my-server.example.com/webhooks")

# Or mount into Flask / FastAPI with an existing secret
blueprint = client.webhook_blueprint("/webhooks", secret="your-hex-secret")
router = client.webhook_router(
    "/webhooks",
    secret="your-hex-secret",
    on_verification=lambda code: client.webhooks.verify("webhook-id", code),
)

Verification challenges and test deliveries are answered with 200 and are not dispatched to event handlers.

Signature verification

from bavimail import verify_webhook_signature, WebhookVerificationError

try:
    verify_webhook_signature(
        payload=request.body,
        signature=request.headers["x-webhook-signature"],
        timestamp=request.headers["x-webhook-timestamp"],
        secret="your-hex-secret",
    )
except WebhookVerificationError as e:
    print(f"Invalid webhook: {e}")

Error Handling

All API errors raise typed exceptions carrying the API's machine-readable code:

from bavimail import (
    APIError,
    NotFoundError,
    QuotaExceededError,
    ValidationError,
)

try:
    client.emails.send(inbox_id, ["user@example.com"], "Hi", "Hello")
except ValidationError as e:
    print(e.code, e.message)       # e.g. "recipient_suppressed"
except QuotaExceededError as e:
    print("Plan limit reached:", e.code)
except APIError as e:
    print(f"API error ({e.status_code}): {e.message}")

Custom HTTP Client

Pass a custom httpx.Client for advanced configuration (proxies, custom TLS, retries):

import httpx

custom = httpx.Client(
    base_url="https://api.bavimail.com",
    headers={"x-api-key": "YOUR_API_KEY"},
    transport=httpx.HTTPTransport(retries=3),
    timeout=60.0,
)

client = Bavimail(api_key="YOUR_API_KEY", http_client=custom)

Development

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest tests/ -v

# Type check
mypy src/bavimail --strict

# Lint
ruff check src/ tests/

License

MIT