Lesson 03 · WhatsApp Business API for your app

Tenant storage & token lifecycle

Where "how do I store that credential" gets a concrete answer. One encrypted row per org, a thin provider interface so BSP and direct-Meta are swappable, and a token lifecycle that reacts to revocation instead of chasing timers. Python / FastAPI.

1. The token you're storing (why it changes the design)

After Embedded Signup (Lesson 04), your server exchanges a code for a Business Integration System User (BISU) access token — customer-scoped and long-lived. This is the opposite of a plain user access token, which expires in hours and must be regenerated constantly. [Meta: access tokens]

2. The multi-tenant table

One row per connected org. The token is the only sensitive field — it's stored encrypted, never plaintext.

# models.py
import enum, datetime as dt
from sqlalchemy import String, LargeBinary, Enum, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase

class Base(DeclarativeBase):
    pass

class WaStatus(enum.StrEnum):
    connected     = "connected"       # good to send
    token_revoked = "token_revoked"   # org must re-connect
    disconnected  = "disconnected"    # never connected / removed

class OrgWhatsApp(Base):
    __tablename__ = "org_whatsapp"

    org_id:          Mapped[str]   = mapped_column(String, primary_key=True)  # YOUR tenant id
    provider:        Mapped[str]   = mapped_column(String, default="meta")    # "meta" | "360dialog" | ...
    waba_id:         Mapped[str]   = mapped_column(String)
    phone_number_id: Mapped[str]   = mapped_column(String)                    # the "from" address
    access_token_enc:Mapped[bytes] = mapped_column(LargeBinary)               # ENCRYPTED bytes, never plaintext
    status:          Mapped[WaStatus] = mapped_column(Enum(WaStatus), default=WaStatus.connected)
    connected_at:    Mapped[dt.datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

Portable DDL (same shape, any DB) so this survives a stack change:

CREATE TABLE org_whatsapp (
  org_id           TEXT PRIMARY KEY,
  provider         TEXT NOT NULL DEFAULT 'meta',
  waba_id          TEXT NOT NULL,
  phone_number_id  TEXT NOT NULL,
  access_token_enc BYTEA NOT NULL,          -- encrypted at rest
  status           TEXT NOT NULL DEFAULT 'connected',
  connected_at     TIMESTAMPTZ NOT NULL DEFAULT now()
);

3. Encrypt the token at rest

The rule (from Lesson 01): the token can send messages billed to that org and read their messages — treat it like a password. Encrypt before it hits the DB, decrypt only in-process at send time.

# settings.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    wa_token_key: str          # base64 32-byte Fernet key, from a secrets manager (NOT git)
    meta_app_id: str
    meta_app_secret: str
    meta_graph_version: str = "v21.0"

settings = Settings()  # reads env / secrets

# crypto.py
from cryptography.fernet import Fernet
from .settings import settings

_fernet = Fernet(settings.wa_token_key.encode())

def encrypt(plaintext: str) -> bytes:
    return _fernet.encrypt(plaintext.encode())

def decrypt(ciphertext: bytes) -> str:
    return _fernet.decrypt(ciphertext).decode()

4. The provider boundary (Lesson 02's portability, made real)

This is the interface that makes "BSP now, Meta later" a one-file swap. Your app code only ever calls send_template — it never knows who's behind it.

# provider.py  — the portability boundary
from typing import Protocol
from dataclasses import dataclass

@dataclass
class SendResult:
    message_id: str

class ProviderAuthError(Exception):
    """Token revoked/expired — the org must re-connect."""

class MessagingProvider(Protocol):
    def send_template(
        self, *, phone_number_id: str, to: str,
        template: str, lang: str, variables: list[str],
    ) -> SendResult: ...

The direct-Meta implementation. Note how it maps a revoked token (Graph error code 190) onto your own ProviderAuthError — that's the lifecycle hook.

# meta_provider.py
import httpx
from .settings import settings
from .provider import SendResult, ProviderAuthError

class MetaCloudProvider:
    def __init__(self, access_token: str):
        self._token = access_token

    def send_template(self, *, phone_number_id, to, template, lang, variables) -> SendResult:
        url = f"https://graph.facebook.com/{settings.meta_graph_version}/{phone_number_id}/messages"
        body = {
            "messaging_product": "whatsapp",
            "to": to,
            "type": "template",
            "template": {
                "name": template,
                "language": {"code": lang},
                "components": [{
                    "type": "body",
                    "parameters": [{"type": "text", "text": v} for v in variables],
                }],
            },
        }
        r = httpx.post(url, json=body,
                       headers={"Authorization": f"Bearer {self._token}"}, timeout=15)
        if r.status_code in (400, 401):
            err = r.json().get("error", {})
            if err.get("code") == 190:                 # invalid/expired/revoked token
                raise ProviderAuthError(err.get("message", "token revoked"))
        r.raise_for_status()
        return SendResult(message_id=r.json()["messages"][0]["id"])

A BSP implementation would satisfy the exact same Protocol — different URL, different auth header, same method signature and same ProviderAuthError contract. Swapping providers never touches your app code. [Meta: send messages]

5. Wiring it together: load tenant → decrypt → send

A factory turns an org_id into a ready provider, and the use-case handles the one lifecycle event that matters — revocation.

# factory.py
from .models import OrgWhatsApp, WaStatus
from .crypto import decrypt
from .meta_provider import MetaCloudProvider
from .provider import MessagingProvider

def provider_for(org_id: str, session) -> tuple[MessagingProvider, OrgWhatsApp]:
    row = session.get(OrgWhatsApp, org_id)
    if row is None or row.status != WaStatus.connected:
        raise RuntimeError(f"org {org_id} has no active WhatsApp connection")
    token = decrypt(row.access_token_enc)          # decrypt only here, in-process
    if row.provider == "meta":
        return MetaCloudProvider(token), row
    # if row.provider == "360dialog": return BspProvider(token), row
    raise RuntimeError(f"unknown provider {row.provider!r}")

# notifications.py  — your product use-case
from .factory import provider_for
from .provider import ProviderAuthError, SendResult
from .models import WaStatus

def send_billing_reminder(org_id, parent_phone, child_name, amount, session) -> SendResult:
    provider, row = provider_for(org_id, session)
    try:
        return provider.send_template(
            phone_number_id=row.phone_number_id,
            to=parent_phone,
            template="billing_reminder_v1",        # a UTILITY template (Lesson 02)
            lang="en",
            variables=[child_name, amount],
        )
    except ProviderAuthError:
        row.status = WaStatus.token_revoked        # stop retrying; prompt re-connect in UI
        session.commit()
        raise

6. Storing the connection (forward hook to Lesson 04)

After the "Connect" flow returns and your server exchanges the code, you persist it here. Use an upsert so re-connecting an org (e.g. after revocation) overwrites cleanly.

# onboarding.py
from .models import OrgWhatsApp, WaStatus
from .crypto import encrypt

def save_connection(org_id, waba_id, phone_number_id, access_token, session, provider="meta"):
    session.merge(OrgWhatsApp(                      # merge = upsert on org_id
        org_id=org_id,
        provider=provider,
        waba_id=waba_id,
        phone_number_id=phone_number_id,
        access_token_enc=encrypt(access_token),     # encrypted before it touches the DB
        status=WaStatus.connected,
    ))
    session.commit()

Check yourself