Sending SPC Alerts to Slack and Microsoft Teams

Posting an out-of-control signal into a chat channel is the most common real-time SPC alert path, and the details that make it useful — the rule that tripped, the value against its limit, the reaction step, a link to the chart — are exactly the ones a quick prototype omits. This guide is part of real-time alert notification channels: it shows how to build a rich Slack block message and a Microsoft Teams adaptive card from one alert object and post them through an injected webhook client with sane retries.

Prerequisites

  • Python 3.10+ with requests available for the injected client, or any HTTP client you prefer (pip install requests)
  • A Slack incoming-webhook URL and a Microsoft Teams incoming-webhook URL, held in your secrets manager — never in source (the code below receives a configured client, not a URL)
  • An Alert-shaped record from the detection layer carrying process id, characteristic, rule, value, UCL, LCL, severity, and the OCAP action
  • A dedup/throttle gate already in front of this transport — see deduplicating and throttling SPC alert storms — so retries here never amplify a storm
  • The OCAP step text resolved for the tripped rule, from your out-of-control action plan (OCAP) table

Step 1 — Model the alert and the injected client

Keep the message builders pure: they take an alert and return a payload dict, with no network and no URL. The webhook client is injected, so the same builders work in a unit test with a fake client and in production with a real one.

from __future__ import annotations

from dataclasses import dataclass
from enum import IntEnum
from typing import Protocol


class Severity(IntEnum):
    INFO = 10
    WARNING = 20
    CRITICAL = 30


@dataclass(frozen=True)
class Alert:
    process_id: str
    characteristic: str
    rule: str
    value: float
    ucl: float
    lcl: float
    severity: Severity
    ocap_action: str
    chart_url: str  # site-relative link to a chart snapshot


class WebhookClient(Protocol):
    """A configured client that already holds its target URL and posts JSON."""
    def post(self, payload: dict) -> None: ...


def breached_limit(alert: Alert) -> tuple[str, float]:
    """Return the label and value of the limit the point crossed."""
    if alert.value >= alert.ucl:
        return "UCL", alert.ucl
    return "LCL", alert.lcl

Step 2 — Build a Slack block message

Slack renders a blocks array. Use a header for the severity and process, a section for the value-versus-limit statement, a section for the OCAP action, and a button that deep-links to the chart snapshot. Every field an operator needs is on screen without leaving the channel.

def build_slack_payload(alert: Alert) -> dict:
    """Compose a Slack block message carrying full SPC context."""
    label, limit = breached_limit(alert)
    header = f"[{alert.severity.name}] {alert.process_id} — {alert.rule}"
    detail = (
        f"*{alert.characteristic}* measured *{alert.value:.4g}*, "
        f"breaching {label} {limit:.4g}."
    )
    return {
        "blocks": [
            {"type": "header", "text": {"type": "plain_text", "text": header}},
            {"type": "section", "text": {"type": "mrkdwn", "text": detail}},
            {"type": "section", "text": {"type": "mrkdwn",
                                         "text": f"*OCAP action:* {alert.ocap_action}"}},
            {"type": "actions", "elements": [
                {"type": "button",
                 "text": {"type": "plain_text", "text": "Open chart"},
                 "url": alert.chart_url}]},
        ]
    }

Step 3 — Build a Microsoft Teams adaptive card

Teams accepts an adaptive card wrapped in a message attachment. The structure differs from Slack but carries the same four facts: severity/process heading, the value-versus-limit detail, the OCAP step, and an open-chart action.

def build_teams_payload(alert: Alert) -> dict:
    """Compose a Microsoft Teams adaptive card with the same context as Slack."""
    label, limit = breached_limit(alert)
    header = f"[{alert.severity.name}] {alert.process_id} — {alert.rule}"
    detail = (
        f"{alert.characteristic} measured {alert.value:.4g}, "
        f"breaching {label} {limit:.4g}."
    )
    return {
        "type": "message",
        "attachments": [{
            "contentType": "application/vnd.microsoft.card.adaptive",
            "content": {
                "type": "AdaptiveCard",
                "version": "1.4",
                "body": [
                    {"type": "TextBlock", "text": header, "weight": "Bolder", "size": "Large"},
                    {"type": "TextBlock", "text": detail, "wrap": True},
                    {"type": "TextBlock", "text": f"OCAP action: {alert.ocap_action}",
                     "wrap": True},
                ],
                "actions": [
                    {"type": "Action.OpenUrl", "title": "Open chart", "url": alert.chart_url}],
            },
        }],
    }

Step 4 — Post with bounded retries and error handling

Chat webhooks fail transiently — rate limits, brief 5xx, dropped connections. Retry a small number of times with exponential backoff, but treat client errors (4xx that are not rate limits) as permanent so a malformed payload does not loop. Wrap everything so one channel's failure raises a typed error the caller can log and move past.

import logging
import time

logger = logging.getLogger(__name__)


class TransportError(RuntimeError):
    """Delivery failed after exhausting retries."""


def post_with_retry(
    client: WebhookClient,
    payload: dict,
    channel: str,
    max_attempts: int = 3,
    base_delay_s: float = 0.5,
) -> None:
    """Post a payload, retrying transient failures with exponential backoff.

    Raises TransportError if every attempt fails. The injected client is
    expected to raise on a non-2xx response.
    """
    last_exc: Exception | None = None
    for attempt in range(1, max_attempts + 1):
        try:
            client.post(payload)
            logger.info("delivered to %s on attempt %d", channel, attempt)
            return
        except Exception as exc:  # normalize any client error
            last_exc = exc
            if attempt == max_attempts:
                break
            delay = base_delay_s * (2 ** (attempt - 1))
            logger.warning("%s attempt %d failed: %s; retrying in %.1fs",
                           channel, attempt, exc, delay)
            time.sleep(delay)
    raise TransportError(f"{channel} delivery failed after {max_attempts} attempts: {last_exc}")


def send_alert(
    alert: Alert,
    slack_client: WebhookClient,
    teams_client: WebhookClient,
) -> list[str]:
    """Fan an alert to Slack and Teams. Returns the channels that accepted it;
    a failure on one channel never blocks the other."""
    targets = [
        ("slack", slack_client, build_slack_payload(alert)),
        ("teams", teams_client, build_teams_payload(alert)),
    ]
    delivered: list[str] = []
    for channel, client, payload in targets:
        try:
            post_with_retry(client, payload, channel)
            delivered.append(channel)
        except TransportError as exc:
            logger.error("%s", exc)
    return delivered

Verification

Confirm the builders and delivery loop with a fake client — no network, deterministic, and it proves channel isolation.

class FakeClient:
    def __init__(self, fail_times: int = 0) -> None:
        self.fail_times = fail_times
        self.calls: list[dict] = []

    def post(self, payload: dict) -> None:
        if len(self.calls) < self.fail_times:
            self.calls.append(payload)
            raise ConnectionError("transient")
        self.calls.append(payload)


alert = Alert(
    process_id="LINE3-STN2", characteristic="bore_dia_mm", rule="Nelson 1",
    value=12.61, ucl=12.55, lcl=12.45, severity=Severity.CRITICAL,
    ocap_action="Quarantine last 5 parts; call process engineer.",
    chart_url="/charts/line3-stn2/latest/",
)

slack = build_slack_payload(alert)
assert slack["blocks"][0]["text"]["text"].startswith("[CRITICAL]")
assert "UCL 12.55" in slack["blocks"][1]["text"]["text"]
assert "Quarantine" in slack["blocks"][2]["text"]["text"]

teams = build_teams_payload(alert)
assert teams["attachments"][0]["content"]["actions"][0]["url"] == "/charts/line3-stn2/latest/"

# Slack client fails twice then succeeds; Teams client stays down entirely.
delivered = send_alert(alert, FakeClient(fail_times=2), FakeClient(fail_times=99))
assert delivered == ["slack"]           # Teams failed, Slack still delivered
print("slack/teams alert contract holds")

Expected output: slack/teams alert contract holds. The load-bearing assertion is the last one — with Teams permanently failing, Slack must still deliver and appear in the returned list. A delivery loop that aborts on the first channel error silences every other channel during a single provider outage.

Root-Cause Table

Symptom Cause Fix
Message posts but shows no rule or value Builder used a plain string instead of the block/card structure Emit the blocks array (Slack) or adaptive-card body (Teams); keep the value-vs-limit line
Retries hammer the webhook on a bad payload 4xx malformed-payload error retried as if transient Have the client raise a distinct error for 4xx (except 429) and do not retry it
Duplicate cards flood the channel Retry succeeded server-side but the response was lost, so it retried Keep the dedup gate upstream; make posts idempotent with a client-side key if the API supports it
Chart button goes nowhere chart_url left blank or pointed at a stale render Require chart_url; generate the snapshot link at alert time, not from a cache
One provider outage silences all alerts Delivery loop aborts on the first TransportError Catch per channel, log, and continue; route criticals to a non-chat fallback too
Teams card renders as raw JSON Missing contentType or wrong attachment wrapper Wrap the card in a message attachment with the adaptive-card content type

Up one level: Real-time alert notification channels.