Send SMS alerts to your team when critical events happen, using Python, Flask, and CastBrick

Real-Time Alerts with Python + Flask

This tutorial builds an alert endpoint that sends SMS notifications to an on-call team when critical events occur — server errors, payment failures, threshold breaches, or any event worth a text message.

What you'll build

POST /alerts   → receives an event payload, sends SMS to the relevant team

Prerequisites

Python 3.8+
A CastBrick API key from the dashboard

Setup

mkdir sms-alerts && cd sms-alerts
python -m venv venv && source venv/bin/activate
pip install castbrick flask python-dotenv

Create a .env file:

CASTBRICK_API_KEY=your_api_key_here

The code

Create app.py:

import os
from flask import Flask, request, jsonify
from castbrick import CastBrick, CastBrickApiError
from dotenv import load_dotenv

load_dotenv()

app = Flask(__name__)
cb = CastBrick(api_key=os.environ["CASTBRICK_API_KEY"])

# Map alert levels to on-call teams
ONCALL_TEAMS = {
    "critical": ["+244923000000", "+244912000001"],  # entire team
    "warning":  ["+244923000000"],                   # lead only
    "info":     [],                                   # no SMS for info
}

def build_message(level: str, service: str, description: str) -> str:
    icons = {"critical": "🚨", "warning": "⚠️", "info": "ℹ️"}
    icon = icons.get(level, "")
    return f"{icon} [{level.upper()}] {service}: {description}"


@app.post("/alerts")
def send_alert():
    data = request.get_json(silent=True) or {}

    level = data.get("level", "info").lower()
    service = data.get("service", "unknown")
    description = data.get("description", "")

    if not description:
        return jsonify({"error": "description is required"}), 422

    recipients = ONCALL_TEAMS.get(level, [])

    if not recipients:
        return jsonify({"ok": True, "sent": False, "reason": "no recipients for this level"})

    message = build_message(level, service, description)

    try:
        result = cb.sms.send(
            to=recipients,
            content=message,
            sender_id="Alerts",
        )
        return jsonify({"ok": True, "message_id": result.message_id})

    except CastBrickApiError as e:
        app.logger.error("CastBrick error %s: %s", e.status_code, e.body)
        return jsonify({"error": e.body}), e.status_code


if __name__ == "__main__":
    app.run(debug=True, port=5000)

Test it

Trigger a critical alert:

curl -X POST http://localhost:5000/alerts \
  -H "Content-Type: application/json" \
  -d '{
    "level": "critical",
    "service": "payments-api",
    "description": "Error rate exceeded 5% for 3 consecutive minutes"
  }'

Trigger a warning:

curl -X POST http://localhost:5000/alerts \
  -H "Content-Type: application/json" \
  -d '{
    "level": "warning",
    "service": "database",
    "description": "Connection pool at 85% capacity"
  }'

Integrating with monitoring tools

Most monitoring tools (Grafana, Datadog, Uptime Robot) support webhook notifications. Point them at your /alerts endpoint:

# Grafana webhook payload example — adapt the mapping
@app.post("/alerts/grafana")
def grafana_alert():
    data = request.get_json(silent=True) or {}

    state = data.get("state", "")          # "alerting" | "ok" | "pending"
    rule_name = data.get("ruleName", "")
    message = data.get("message", "")

    level = "critical" if state == "alerting" else "info"

    recipients = ONCALL_TEAMS.get(level, [])
    if not recipients:
        return jsonify({"ok": True})

    cb.sms.send(
        to=recipients,
        content=f"Grafana [{state.upper()}] {rule_name}: {message}",
        sender_id="Grafana",
    )
    return jsonify({"ok": True})

Scheduling a daily digest

Instead of individual alerts, send a single daily summary using the Python schedule library:

import schedule
import time

def send_daily_digest():
    cb.sms.send(
        to=["+244923000000"],
        content="Daily digest: 0 critical alerts, 3 warnings in the last 24h. All systems normal.",
        sender_id="Digest",
    )

schedule.every().day.at("08:00").do(send_daily_digest)

while True:
    schedule.run_pending()
    time.sleep(60)

Next steps

Python SDK reference — all methods and options
Broadcasts — send the same alert to a large contact list
Webhooks — confirm delivery before marking an alert as acknowledged

Real-Time Alerts with Python + Flask

On this page