flowbox

A vendor-neutral email connector: read inboxes and send mail through the adapter you choose. Standard library only.

../project is a small flowlab app that shows how to wire it into an application: settings, one factory module, a queued send job, polling and a webhook route.

LogAdapter sends nothing; it logs each message on the flowbox.log logger, like Laravel's log mailer. Use it for local work, not production.

FlowOutbox takes an OutboxAdapterABC. FlowInbox takes an InboxAdapterABC (a mailbox you read from), a WebhookInboxAdapterABC (a provider that pushes mail to you), or an adapter that is both. Using an adapter for something it cannot do — FlowInbox(SMTPAdapter(...)), or parse_webhook() on an IMAP inbox — is a type error, so mypy or your editor flags it before it runs.

Settings from the environment

pip install flowbox[settings] adds FlowboxSettings, a pydantic-settings class that picks the adapters from environment variables and builds the flows. Subclass it for your app's own settings:

from pydantic_settings import SettingsConfigDict

from flowbox.settings import FlowboxSettings


class Settings(FlowboxSettings):
    model_config = SettingsConfigDict(env_file=".env")


settings = Settings()

with settings.outbox() as outbox:
    outbox.send_mail(message)

with settings.inbox() as inbox:
    messages = inbox.retrieve_mails("INBOX", since)

Each outbox() / inbox() call builds a new flow: open one per job, command or request, since IMAP and SMTP adapters hold a socket and are not thread-safe. Construction fails when a selected driver is missing its credentials, and the error names the variables to set, never their values.

Webhooks stay per vendor: settings.lettermint_webhooks() returns a FlowInbox[LettermintAdapter] for that vendor's webhook route. Your own adapter goes in by overriding outbox_adapter() or inbox_adapter():

class Settings(FlowboxSettings):
    crm_api_key: str | None = None

    def outbox_adapter(self) -> OutboxAdapterABC:
        if self.crm_api_key:
            return CRMAdapter(self.crm_api_key)
        return super().outbox_adapter()

flowbox.settings is the only module that reads the environment, and the core never imports it.

Sending

from flowbox import Address, Attachment, FlowOutbox, Message

with FlowOutbox.connect_smtp("smtp.example.com", "user", "secret", sender=Address("noreply@acme.com", "Acme")) as outbox:
    message_id = outbox.send_mail(
        Message(
            subject="Your invoice",
            to=[Address("customer@example.com")],
            text="Your invoice is attached.",
            attachments=[Attachment.from_path("invoice.pdf")],
        )
    )

Sending options

A few Message fields are for providers that support them; an adapter whose provider has no equivalent ignores them.

Errors

Every provider failure is an AdapterError. status_code is the provider's HTTP status when it answered with one, and transient says whether trying again later may succeed (timeouts, dropped connections, rate limits, server errors, SMTP 4xx replies). A transient failure while sending can mean the message went out before the connection broke: retry with the same idempotency_key.

try:
    outbox.send_mail(message)
except AdapterError as exc:
    if not exc.transient:
        raise
    ...  # retry later, same message, same idempotency_key

Reading

from datetime import datetime, timedelta

from flowbox import FlowInbox

with FlowInbox.connect_imap("imap.example.com", "user", "secret") as inbox:
    print(inbox.get_mailboxes())

    for message in inbox.retrieve_mails("INBOX", datetime.now() - timedelta(days=1)):
        print(message.received_at, message.sender, message.subject)

retrieve_mails() returns everything received at or after from_datetime, oldest first, and never marks mail as read.

Webhooks

Providers that push received mail to you sign and shape their webhooks differently; the adapter verifies the signature and translates the payload. Hand it the raw request body, before any JSON or form parsing, and the headers:

from fastapi import FastAPI, HTTPException, Request

from flowbox import FlowInbox, SignatureError

app = FastAPI()
inbox = FlowInbox.connect_lettermint(webhook_secret="whsec_...")


@app.post("/webhooks/inbound")
async def inbound(request: Request) -> None:
    try:
        message = inbox.parse_webhook(await request.body(), request.headers)
    except SignatureError:
        raise HTTPException(status_code=401)
    ...

Responding is up to your app. Providers retry deliveries that get no 2xx, so the same message can arrive twice; message.id stays the same.

A correctly signed webhook for another event (a delivery report sent to the same URL) raises UnsupportedWebhookError, a subclass of AdapterError with the event name in event; answer it with a 2xx so the provider stops retrying. Fields the provider sends that Message has no place for — Lettermint's route, envelope recipient, is_spam, authentication_results and the event id — are in message.provider_data, as sent, without the body and attachment contents.

A Lettermint route can deliver large attachments as signed download links instead of inline. parse_webhook() downloads them by default; FlowInbox.connect_lettermint(webhook_secret=..., download_attachments=False) leaves their content empty and their url in provider_data, so a queued job can fetch them after the webhook has been answered.

Lettermint

One adapter covers everything; each part needs only its own credential. Sending uses a project token (lm_...), retrieve_mails() a team token (lm_team_...) with the team's inbound routes as the mailboxes, and parse_webhook() the inbound route webhook's signing secret (whsec_...).

from flowbox import Address, FlowInbox, FlowOutbox
from flowbox.adapters import LettermintAdapter

lettermint = LettermintAdapter(project_token="lm_...", team_token="lm_team_...", webhook_secret="whsec_...")

inbox = FlowInbox(lettermint)
outbox = FlowOutbox(lettermint, sender=Address("noreply@acme.com"))