Dispatching Multi-Channel Notifications: SMS, Email, and Web

When a Tier 1 event fires, a water system does not get to pick one convenient way to reach its customers — the primacy agency prescribes a set of delivery methods, and the utility must prove that each prescribed method was actually attempted and completed. This page sits inside the Public Notification Workflows for SDWA Violations section and covers the delivery layer specifically: the code that takes one approved notice and fans it out across SMS and voice, email, the utility website and social feeds, and broadcast media, then assembles a certification of completion that names every channel used and the receipt that proves it. The trigger logic that decides a notice is owed lives in automating Tier 1 public notification triggers; the clock that says how fast delivery must happen lives in tracking state primacy notification deadlines. What follows is the dispatcher that runs between them.

Prerequisites & Environment Setup

The design targets Python 3.10+ and leans on a small, boring dependency set on purpose: notice dispatch is a regulated record-keeping activity, so every moving part should be inspectable. Email uses the standard-library smtplib and email.message, which means no third-party mail SDK sits between you and the audit trail. SMS and voice go through an HTTP gateway behind an abstract adapter, so the concrete carrier (a commercial aggregator, a state emergency-alert bridge, or a municipal SMS gateway) is a swappable detail rather than a hard dependency. pydantic validates the notice and the receipts, and requests handles the gateway calls.

Give the process outbound network reach to exactly two places — your relay host and the SMS gateway endpoint — and nothing else. Credentials come from the environment or a secrets manager, never from source. Under 40 CFR Part 141 (Subpart Q) the delivery record is part of the compliance file, so the same append-only storage discipline the rest of the compliance platform uses applies to the receipts this module emits.

python3 -m venv .venv && source .venv/bin/activate
pip install "pydantic==2.7.*" "requests==2.32.*"
# smtplib and email are standard library; nothing to install for the mail path

Step-by-Step Implementation

Step 1 — Model the notice, the channels, and the receipt

Start from the data. A PublicNotice carries the approved message text and the set of required delivery methods; a Receipt records what happened on one channel for one attempt. Keeping the receipt immutable and explicit is what later lets you certify completion rather than merely assert it.

from __future__ import annotations

import time
import uuid
from datetime import datetime, timezone
from enum import Enum
from typing import Protocol

from pydantic import BaseModel, Field


class Method(str, Enum):
    SMS = "sms"
    VOICE = "voice"
    EMAIL = "email"
    WEB = "web"
    BROADCAST = "broadcast"


class DeliveryState(str, Enum):
    PENDING = "PENDING"
    DISPATCHED = "DISPATCHED"
    CONFIRMED = "CONFIRMED"
    FAILED = "FAILED"


class PublicNotice(BaseModel):
    notice_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    violation_id: str
    subject: str
    body: str
    required_methods: list[Method]
    audience: dict[Method, list[str]]  # method -> recipients/targets


class Receipt(BaseModel):
    notice_id: str
    method: Method
    state: DeliveryState
    attempts: int
    gateway_reference: str | None = None
    detail: str | None = None
    completed_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

Step 2 — Define the abstract Channel interface

Every delivery mechanism reduces to the same contract: given a notice, try to deliver it and return a receipt whose state is either CONFIRMED or FAILED. A channel never raises on a delivery failure — a failure is data the certification needs, not an exception that aborts the run. Using a Protocol keeps adapters decoupled from any base class import.

class Channel(Protocol):
    method: Method

    def deliver(self, notice: PublicNotice) -> Receipt:
        """Attempt one delivery. Must return a Receipt, never raise for
        an ordinary delivery failure (raise only for programmer error)."""
        ...

Step 3 — Implement the email adapter with smtplib

The email adapter builds a proper MIME message and hands it to an SMTP relay. It treats any recipient the relay refuses as a partial failure and records the refusal detail so the certification can show exactly who was and was not reached.

import smtplib
from email.message import EmailMessage


class EmailChannel:
    method = Method.EMAIL

    def __init__(self, host: str, port: int, sender: str,
                 username: str, password: str) -> None:
        self._host, self._port = host, port
        self._sender = sender
        self._username, self._password = username, password

    def deliver(self, notice: PublicNotice) -> Receipt:
        recipients = notice.audience.get(Method.EMAIL, [])
        if not recipients:
            return Receipt(notice_id=notice.notice_id, method=self.method,
                           state=DeliveryState.FAILED, attempts=1,
                           detail="no email recipients configured")
        msg = EmailMessage()
        msg["Subject"] = notice.subject
        msg["From"] = self._sender
        msg["To"] = ", ".join(recipients)
        msg.set_content(notice.body)
        try:
            with smtplib.SMTP(self._host, self._port, timeout=30) as smtp:
                smtp.starttls()
                smtp.login(self._username, self._password)
                refused = smtp.send_message(msg)  # {addr: (code, reason)}
        except (smtplib.SMTPException, OSError) as exc:
            return Receipt(notice_id=notice.notice_id, method=self.method,
                           state=DeliveryState.FAILED, attempts=1,
                           detail=f"relay error: {exc}")
        if refused:
            return Receipt(notice_id=notice.notice_id, method=self.method,
                           state=DeliveryState.FAILED, attempts=1,
                           detail=f"refused: {sorted(refused)}")
        return Receipt(notice_id=notice.notice_id, method=self.method,
                       state=DeliveryState.CONFIRMED, attempts=1,
                       gateway_reference=msg["To"], detail="accepted by relay")

Step 4 — Implement the SMS/voice adapter over an HTTP gateway

The SMS adapter posts to a gateway and reads back a provider reference. Because SMS and voice share the same request shape at most aggregators, one adapter serves both by parameterizing the channel type. The gateway client is injected so tests never touch the network.

import requests


class HttpSmsChannel:
    def __init__(self, base_url: str, api_key: str,
                 method: Method = Method.SMS,
                 session: requests.Session | None = None) -> None:
        self.method = method
        self._url = base_url.rstrip("/") + "/messages"
        self._key = api_key
        self._session = session or requests.Session()

    def deliver(self, notice: PublicNotice) -> Receipt:
        targets = notice.audience.get(self.method, [])
        if not targets:
            return Receipt(notice_id=notice.notice_id, method=self.method,
                           state=DeliveryState.FAILED, attempts=1,
                           detail="no targets configured")
        payload = {"channel": self.method.value, "to": targets,
                   "text": notice.body, "reference": notice.notice_id}
        try:
            resp = self._session.post(
                self._url, json=payload,
                headers={"Authorization": f"Bearer {self._key}"}, timeout=20)
            resp.raise_for_status()
        except requests.RequestException as exc:
            return Receipt(notice_id=notice.notice_id, method=self.method,
                           state=DeliveryState.FAILED, attempts=1,
                           detail=f"gateway error: {exc}")
        ref = resp.json().get("id")
        return Receipt(notice_id=notice.notice_id, method=self.method,
                       state=DeliveryState.CONFIRMED, attempts=1,
                       gateway_reference=ref, detail="queued at gateway")

A web/social adapter follows the same shape — it POSTs the notice to the utility’s content endpoint — and a broadcast adapter can wrap the EAS/IPAWS bridge behind the identical deliver contract, so nothing downstream cares which concrete class it is.

Step 5 — Dispatch with retry and exponential backoff

The Dispatcher walks every required method, retries transient failures, and collects one authoritative receipt per channel. Retries use exponential backoff so a briefly unavailable gateway is not hammered: the delay before attempt nn is

tn=b⋅2 n−1,n=1,2,…,Nt_n = b \cdot 2^{\,n-1}, \qquad n = 1, 2, \dots, N

where bb is the base delay and NN the attempt ceiling. Each notice moves through the states PENDING → DISPATCHED → CONFIRMED or FAILED, and the aggregate becomes CERTIFIED only when every required method reports CONFIRMED.

class Dispatcher:
    def __init__(self, channels: list[Channel],
                 max_attempts: int = 3, base_delay: float = 2.0,
                 sleep=time.sleep) -> None:
        self._channels = {c.method: c for c in channels}
        self._max_attempts = max_attempts
        self._base_delay = base_delay
        self._sleep = sleep

    def _deliver_with_retry(self, channel: Channel,
                            notice: PublicNotice) -> Receipt:
        last: Receipt | None = None
        for attempt in range(1, self._max_attempts + 1):
            receipt = channel.deliver(notice)
            receipt.attempts = attempt
            if receipt.state == DeliveryState.CONFIRMED:
                return receipt
            last = receipt
            if attempt < self._max_attempts:
                self._sleep(self._base_delay * (2 ** (attempt - 1)))
        return last  # exhausted: the final FAILED receipt

    def dispatch(self, notice: PublicNotice) -> "CertificationRecord":
        receipts: list[Receipt] = []
        for method in notice.required_methods:
            channel = self._channels.get(method)
            if channel is None:
                receipts.append(Receipt(
                    notice_id=notice.notice_id, method=method,
                    state=DeliveryState.FAILED, attempts=0,
                    detail="no adapter registered for required method"))
                continue
            receipts.append(self._deliver_with_retry(channel, notice))
        return CertificationRecord.build(notice, receipts)

Step 6 — Assemble the certification of completion

The certification is the artifact the compliance file needs. It is complete only when the set of confirmed methods covers the set of required methods, and it carries the individual receipts so an auditor can trace each one back to a gateway reference.

class CertificationRecord(BaseModel):
    notice_id: str
    violation_id: str
    required: list[Method]
    confirmed: list[Method]
    receipts: list[Receipt]
    complete: bool
    issued_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    @classmethod
    def build(cls, notice: PublicNotice,
              receipts: list[Receipt]) -> "CertificationRecord":
        confirmed = [r.method for r in receipts
                     if r.state == DeliveryState.CONFIRMED]
        required = set(notice.required_methods)
        return cls(
            notice_id=notice.notice_id, violation_id=notice.violation_id,
            required=notice.required_methods, confirmed=confirmed,
            receipts=receipts, complete=required.issubset(set(confirmed)))

For a high-fan-out notice — thousands of SMS targets across many treatment zones — run the dispatcher as a task rather than inline, using the same worker pattern described in async batch processing setup so a slow gateway never blocks the request that raised the violation.

Per-channel delivery states rolling up into a certified notice A notice channel starts in PENDING, transitions to DISPATCHED when an attempt is sent, and then to CONFIRMED on a delivery receipt or FAILED when the retry budget is exhausted. A retry can move a FAILED channel back to CONFIRMED. Once all required channels reach CONFIRMED the aggregate notice becomes CERTIFIED. PENDING DISPATCHED CONFIRMED FAILED CERTIFIED attempt sent receipt held retries exhausted all methods done send receipt retry ok timeout

Configuration Reference

Treat every value below as versioned configuration, not an inline constant. The retry parameters govern the backoff series above; the channel rows map a required method to the concrete adapter that satisfies it.

Parameter Type Default Meaning
max_attempts int 3 Attempt ceiling NN per channel before FAILED
base_delay float 2.0 Base backoff bb in seconds; delay is b⋅2 n−1b\cdot2^{\,n-1}
required_methods list — Methods the primacy agency mandates for this notice
audience[method] list — Recipients or targets keyed by method
smtp host/port str/int — Relay for the email adapter; TLS via starttls
gateway base_url str — HTTP endpoint for the SMS/voice adapter
CONFIRMED state — Channel returned a delivery receipt
FAILED state — Channel exhausted retries or refused all targets
CERTIFIED flag false Confirmed methods cover all required methods

Verification & Testing

Test the dispatcher with fake channels so no mail relay or gateway is touched. The decisive property is that a channel which fails on early attempts but succeeds within the retry budget still certifies, while a permanently failing required channel leaves the record incomplete.

class FlakyChannel:
    def __init__(self, method, succeed_on):
        self.method, self._succeed_on, self._calls = method, succeed_on, 0

    def deliver(self, notice):
        self._calls += 1
        state = (DeliveryState.CONFIRMED if self._calls >= self._succeed_on
                 else DeliveryState.FAILED)
        return Receipt(notice_id=notice.notice_id, method=self.method,
                       state=state, attempts=self._calls)


def _notice(methods):
    return PublicNotice(violation_id="V-1", subject="Boil Water",
                        body="Boil water before use.",
                        required_methods=methods,
                        audience={m: ["target"] for m in methods})


def test_retry_reaches_certification():
    disp = Dispatcher([FlakyChannel(Method.SMS, succeed_on=2)],
                      max_attempts=3, base_delay=0, sleep=lambda _: None)
    record = disp.dispatch(_notice([Method.SMS]))
    assert record.complete is True
    assert record.receipts[0].attempts == 2


def test_missing_required_method_blocks_certification():
    disp = Dispatcher([FlakyChannel(Method.SMS, succeed_on=1)],
                      base_delay=0, sleep=lambda _: None)
    record = disp.dispatch(_notice([Method.SMS, Method.EMAIL]))
    assert record.complete is False
    assert Method.EMAIL not in record.confirmed

Acceptance criteria before this dispatcher carries a real Tier 1 notice:

Troubleshooting & Gotchas

  • A notice reads CERTIFIED but a channel was never really required. The certification compares confirmed methods against required_methods, so an under-specified requirement set produces a false green. Derive required_methods from the tier decision made in automating Tier 1 public notification triggers, never hand-populate it per dispatch.
  • SMTP reports success but recipients see nothing. send_message returning no refusals means the relay accepted the envelope, not that a mailbox received it. Treat relay acceptance as CONFIRMED only if your relay is configured to bounce hard failures back into the receipt pipeline; otherwise reconcile with a delivery-status callback before certifying.
  • Retries silently blow the deadline. Three attempts with a two-second base backoff is fast, but a large max_attempts on a timing-out gateway can consume the whole regulatory window. Cap total elapsed retry time against the clock from tracking state primacy notification deadlines, and fail over to an alternate channel rather than retrying a dead one.
  • The gateway reference is lost on failure. A FAILED SMS receipt with no reference cannot be reconciled later against the carrier’s logs. Capture whatever correlation id the gateway returns even on non-2xx responses, and store it in the receipt detail.
  • One noisy channel starves the others. Because channels dispatch in sequence, a channel that burns its full backoff budget delays the rest. For urgent multi-zone notices, run each channel as an independent task using the async batch processing setup pattern and join on the receipts.

Related pages