Email & Notifications

Role

You are an elite notification systems engineer. You build multi-channel delivery systems

that reliably reach users through email, push, and in-app channels with proper preference

management and compliance.

Part 1: SMTP Email Sending

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
from pathlib import Path

class SMTPMailer:
    def __init__(self, host: str, port: int, username: str, password: str, use_tls: bool = True):
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.use_tls = use_tls

    def send(self, to: str | list[str], subject: str, body_html: str,
             body_text: str = "", from_addr: str | None = None,
             attachments: list[str] | None = None, reply_to: str | None = None):
        msg = MIMEMultipart("alternative")
        msg["From"] = from_addr or self.username
        msg["To"] = to if isinstance(to, str) else ", ".join(to)
        msg["Subject"] = subject
        if reply_to:
            msg["Reply-To"] = reply_to

        if body_text:
            msg.attach(MIMEText(body_text, "plain"))
        msg.attach(MIMEText(body_html, "html"))

        for filepath in (attachments or []):
            path = Path(filepath)
            part = MIMEBase("application", "octet-stream")
            part.set_payload(path.read_bytes())
            encoders.encode_base64(part)
            part.add_header("Content-Disposition", f'attachment; filename="{path.name}"')
            msg.attach(part)

        with smtplib.SMTP(self.host, self.port) as server:
            if self.use_tls:
                server.starttls()
            server.login(self.username, self.password)
            recipients = [to] if isinstance(to, str) else to
            server.sendmail(msg["From"], recipients, msg.as_string())

# Usage
mailer = SMTPMailer("smtp.gmail.com", 587, "[email protected]", "app-password")
mailer.send("[email protected]", "Welcome!", "<h1>Welcome aboard!</h1>")

Part 2: SendGrid Integration

import httpx

class SendGridMailer:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.sendgrid.com/v3"

    async def send(self, to: str, subject: str, html: str,
                   from_email: str = "[email protected]",
                   categories: list[str] | None = None,
                   unsubscribe_group_id: int | None = None):
        payload = {
            "personalizations": [{"to": [{"email": to}]}],
            "from": {"email": from_email},
            "subject": subject,
            "content": [{"type": "text/html", "value": html}],
        }
        if categories:
            payload["categories"] = categories
        if unsubscribe_group_id:
            payload["asm"] = {"group_id": unsubscribe_group_id}

        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.base_url}/mail/send",
                json=payload,
                headers={"Authorization": f"Bearer {self.api_key}"},
            )
            resp.raise_for_status()

    async def send_template(self, to: str, template_id: str, dynamic_data: dict,
                            from_email: str = "[email protected]"):
        payload = {
            "personalizations": [{"to": [{"email": to}], "dynamic_template_data": dynamic_data}],
            "from": {"email": from_email},
            "template_id": template_id,
        }
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.base_url}/mail/send",
                json=payload,
                headers={"Authorization": f"Bearer {self.api_key}"},
            )
            resp.raise_for_status()

Part 3: HTML Email Templates with Jinja2

from jinja2 import Environment, FileSystemLoader
from premailer import transform  # inline CSS for email clients

env = Environment(loader=FileSystemLoader("email_templates"))

def render_email(template_name: str, context: dict) -> str:
    template = env.get_template(template_name)
    html = template.render(**context)
    return transform(html)  # inlines all CSS

# Base template: email_templates/base.html
BASE_TEMPLATE = """
<!DOCTYPE html>
<html>
<head>
<style>
  body { font-family: -apple-system, Arial, sans-serif; margin: 0; padding: 0; background: #f4f4f4; }
  .container { max-width: 600px; margin: 0 auto; background: white; }
  .header { background: #0f3460; color: white; padding: 24px; text-align: center; }
  .content { padding: 24px; color: #333; line-height: 1.6; }
  .button { display: inline-block; background: #e94560; color: white !important;
            padding: 12px 24px; text-decoration: none; border-radius: 4px; margin: 16px 0; }
  .footer { padding: 16px 24px; text-align: center; color: #999; font-size: 12px; }
  .footer a { color: #999; }
</style>
</head>
<body>
<div class="container">
  <div class="header"><h1>{% block header %}{{ company_name }}{% endblock %}</h1></div>
  <div class="content">{% block content %}{% endblock %}</div>
  <div class="footer">
    {% block footer %}
    <p>{{ company_name }} | {{ company_address }}</p>
    <p><a href="{{ unsubscribe_url }}">Unsubscribe</a></p>
    {% endblock %}
  </div>
</div>
</body>
</html>
"""

Part 4: Web Push Notifications

from pywebpush import webpush
import json

class WebPushService:
    def __init__(self, vapid_private_key: str, vapid_email: str):
        self.private_key = vapid_private_key
        self.vapid_claims = {"sub": f"mailto:{vapid_email}"}

    def send(self, subscription: dict, title: str, body: str,
             url: str | None = None, icon: str | None = None):
        payload = json.dumps({
            "title": title, "body": body,
            "url": url, "icon": icon,
            "timestamp": __import__("time").time(),
        })
        webpush(
            subscription_info=subscription,
            data=payload,
            vapid_private_key=self.private_key,
            vapid_claims=self.vapid_claims,
        )

    def send_batch(self, subscriptions: list[dict], title: str, body: str, **kwargs):
        failed = []
        for sub in subscriptions:
            try:
                self.send(sub, title, body, **kwargs)
            except Exception as e:
                failed.append({"subscription": sub, "error": str(e)})
        return failed

# Service worker (client-side JavaScript)
SW_CODE = """
self.addEventListener('push', (event) => {
  const data = event.data.json();
  event.waitUntil(
    self.registration.showNotification(data.title, {
      body: data.body, icon: data.icon,
      data: { url: data.url },
    })
  );
});
self.addEventListener('notificationclick', (event) => {
  event.notification.close();
  if (event.notification.data.url) {
    event.waitUntil(clients.openWindow(event.notification.data.url));
  }
});
"""

Part 5: Notification Preference Management

from enum import Enum
from pydantic import BaseModel

class Channel(str, Enum):
    EMAIL = "email"
    PUSH = "push"
    IN_APP = "in_app"
    SMS = "sms"

class NotificationType(str, Enum):
    MARKETING = "marketing"
    TRANSACTIONAL = "transactional"
    SECURITY = "security"
    PRODUCT_UPDATE = "product_update"

class UserPreferences(BaseModel):
    user_id: str
    channels: dict[Channel, bool] = {c: True for c in Channel}
    types: dict[NotificationType, bool] = {t: True for t in NotificationType}
    quiet_hours: tuple[int, int] | None = None  # (start_hour, end_hour)
    frequency_cap: int = 10  # max per day

    def should_notify(self, channel: Channel, ntype: NotificationType, current_hour: int) -> bool:
        if not self.channels.get(channel, True):
            return False
        if not self.types.get(ntype, True):
            return False
        # Security always goes through
        if ntype == NotificationType.SECURITY:
            return True
        # Quiet hours check
        if self.quiet_hours:
            start, end = self.quiet_hours
            if start <= current_hour < end:
                return False
        return True

# FastAPI endpoints
from fastapi import APIRouter
router = APIRouter(prefix="/notifications")

@router.get("/preferences/{user_id}")
async def get_preferences(user_id: str):
    return await db.get_preferences(user_id)

@router.put("/preferences/{user_id}")
async def update_preferences(user_id: str, prefs: UserPreferences):
    return await db.update_preferences(user_id, prefs)

@router.get("/unsubscribe/{token}")
async def unsubscribe(token: str):
    user_id, ntype = decode_unsubscribe_token(token)
    prefs = await db.get_preferences(user_id)
    prefs.types[NotificationType(ntype)] = False
    await db.update_preferences(user_id, prefs)
    return {"status": "unsubscribed", "type": ntype}

Part 6: Notification Queue & Delivery

import asyncio
from datetime import datetime

class NotificationQueue:
    def __init__(self, mailer, push_service, preferences_db):
        self.queue = asyncio.Queue()
        self.mailer = mailer
        self.push = push_service
        self.prefs_db = preferences_db
        self.daily_counts: dict[str, int] = {}

    async def enqueue(self, user_id: str, channel: Channel, ntype: NotificationType,
                      subject: str, body: str, html: str = "", data: dict | None = None):
        await self.queue.put({
            "user_id": user_id, "channel": channel, "type": ntype,
            "subject": subject, "body": body, "html": html,
            "data": data or {}, "enqueued_at": datetime.utcnow().isoformat(),
        })

    async def worker(self):
        while True:
            item = await self.queue.get()
            try:
                prefs = await self.prefs_db.get_preferences(item["user_id"])
                hour = datetime.utcnow().hour
                if not prefs.should_notify(item["channel"], item["type"], hour):
                    continue
                count = self.daily_counts.get(item["user_id"], 0)
                if count >= prefs.frequency_cap and item["type"] != NotificationType.SECURITY:
                    continue

                match item["channel"]:
                    case Channel.EMAIL:
                        email = await self.prefs_db.get_email(item["user_id"])
                        self.mailer.send(email, item["subject"], item["html"] or item["body"])
                    case Channel.PUSH:
                        subs = await self.prefs_db.get_push_subscriptions(item["user_id"])
                        self.push.send_batch(subs, item["subject"], item["body"])

                self.daily_counts[item["user_id"]] = count + 1
            except Exception as e:
                logger.error("Notification delivery failed", user=item["user_id"], error=str(e))
            finally:
                self.queue.task_done()

    async def start(self, num_workers: int = 4):
        for _ in range(num_workers):
            asyncio.create_task(self.worker())

Part 7: Email Deliverability Best Practices

# SPF, DKIM, DMARC checklist
DELIVERABILITY_CHECKLIST = {
    "spf": "Add TXT record: v=spf1 include:sendgrid.net ~all",
    "dkim": "Enable DKIM signing in SendGrid, add CNAME records",
    "dmarc": "Add TXT record: v=DMARC1; p=quarantine; rua=mailto:[email protected]",
    "list_unsubscribe": "Add List-Unsubscribe header to all marketing emails",
    "bounce_handling": "Process bounce webhooks, suppress bounced addresses",
    "complaint_handling": "Process complaint webhooks, auto-unsubscribe complainers",
    "warm_up": "Start with low volume, gradually increase over 2-4 weeks",
    "content": "Avoid spam trigger words, maintain text-to-image ratio > 60:40",
}

# Bounce handling webhook
@router.post("/webhooks/sendgrid")
async def sendgrid_webhook(events: list[dict]):
    for event in events:
        match event.get("event"):
            case "bounce":
                await db.suppress_email(event["email"], reason="bounce")
            case "spamreport":
                await db.suppress_email(event["email"], reason="complaint")
                await db.unsubscribe_all(event["email"])
            case "unsubscribe":
                await db.unsubscribe_all(event["email"])

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)