How to Build a Team Notification Priority System

Notification overload kills productivity. When everything is urgent, nothing is. I built a team notification priority system with smart alerts that classifies incoming notifications, routes urgent ones immediately, batches low-priority ones into digests, and respects quiet hours.

The right message at the right time through the right channel.

What You Need Before Starting

• Python 3.8+
• Slack bot token
• Team member preference configs
• A classification system for priority

Step 1: Define Priority Levels and Channels

PRIORITY_CONFIG = {
    "critical": {"channels": ["sms", "slack_dm", "email"], "delay": 0, "badge": True},
    "high": {"channels": ["slack_dm", "email"], "delay": 0, "badge": True},
    "medium": {"channels": ["slack_channel"], "delay": 0, "badge": False},
    "low": {"channels": ["digest"], "delay": 3600, "badge": False}
}

TEAM_PREFERENCES = {
    "alice": {"quiet_hours": "22:00-08:00", "timezone": "US/Eastern", "override_critical": True},
    "bob": {"quiet_hours": "21:00-07:00", "timezone": "US/Pacific", "override_critical": True},
}

Step 2: Classify Notification Priority

import anthropic
import json

client = anthropic.Anthropic()

def classify_priority(notification_text, source):
    keyword_rules = {
        "critical": ["server down", "data breach", "production error", "client emergency"],
        "high": ["deadline today", "approval needed", "blocker", "urgent"],
        "medium": ["update", "completed", "new comment", "assigned"],
        "low": ["newsletter", "reminder", "fyi", "digest"]
    }

    text_lower = notification_text.lower()
    for priority, keywords in keyword_rules.items():
        if any(kw in text_lower for kw in keywords):
            return priority

    return "medium"

Step 3: Apply Quiet Hours and Preferences

from datetime import datetime
import pytz

def should_deliver_now(recipient, priority):
    prefs = TEAM_PREFERENCES.get(recipient, {})
    quiet = prefs.get("quiet_hours", "")

    if not quiet:
        return True

    if priority == "critical" and prefs.get("override_critical", True):
        return True

    tz = pytz.timezone(prefs.get("timezone", "UTC"))
    now = datetime.now(tz)
    start_str, end_str = quiet.split("-")
    start_hour = int(start_str.split(":")[0])
    end_hour = int(end_str.split(":")[0])

    current_hour = now.hour
    if start_hour > end_hour:
        in_quiet = current_hour >= start_hour or current_hour < end_hour
    else:
        in_quiet = start_hour <= current_hour < end_hour

    return not in_quiet

Step 4: Route and Deliver

import sqlite3

def init_notification_queue(db_path="notifications.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS notification_queue (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            recipient TEXT,
            message TEXT,
            priority TEXT,
            channel TEXT,
            scheduled_for TEXT,
            delivered INTEGER DEFAULT 0,
            delivered_at TEXT
        )
    """)
    conn.commit()
    return conn

def route_notification(conn, recipient, message, priority):
    config = PRIORITY_CONFIG[priority]
    deliver_now = should_deliver_now(recipient, priority)

    if deliver_now:
        for channel in config["channels"]:
            if channel == "digest":
                queue_for_digest(conn, recipient, message)
            else:
                deliver_to_channel(channel, recipient, message)
    else:
        queue_for_later(conn, recipient, message, priority)

def deliver_to_channel(channel, recipient, message):
    print(f"[{channel}] -> {recipient}: {message[:100]}")

def queue_for_digest(conn, recipient, message):
    conn.execute(
        "INSERT INTO notification_queue (recipient, message, priority, channel, scheduled_for) VALUES (?,?,?,?,?)",
        (recipient, message, "low", "digest", "next_digest")
    )
    conn.commit()

def queue_for_later(conn, recipient, message, priority):
    conn.execute(
        "INSERT INTO notification_queue (recipient, message, priority, channel, scheduled_for) VALUES (?,?,?,?,?)",
        (recipient, message, priority, "deferred", "after_quiet_hours")
    )
    conn.commit()

Step 5: Send Daily Digests

def send_digest(conn, recipient):
    items = conn.execute(
        "SELECT message FROM notification_queue WHERE recipient=? AND channel='digest' AND delivered=0",
        (recipient,)
    ).fetchall()

    if not items:
        return

    digest = f"*Daily Notification Digest ({len(items)} items)*\n\n"
    for i, (msg,) in enumerate(items, 1):
        digest += f"{i}. {msg}\n"

    deliver_to_channel("slack_dm", recipient, digest)

    conn.execute(
        "UPDATE notification_queue SET delivered=1, delivered_at=? WHERE recipient=? AND channel='digest' AND delivered=0",
        (datetime.now().isoformat(), recipient)
    )
    conn.commit()

What to Build Next

Add machine learning that learns each person's actual urgency patterns. If Alice always opens messages from the CTO immediately, auto-upgrade those to high priority. Personalized priority based on behavior.

Related Reading

• Building Your First Automation: A Complete Guide - automation fundamentals
• The Feedback Loop That Powers Everything - notification feedback loops
• Identifying Your Biggest Bottleneck - notification overload as a bottleneck