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")],
        )
    )

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.

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