用 Python FastAPI + Webhook 实现 Alertmanager 邮件告警（绕过 WorkMail STARTTLS 限制）

在使用 AWS WorkMail 作为邮件发送服务时，我发现一个坑：
Alertmanager 默认的 SMTP 发送方式无法直接对接 WorkMail，因为 WorkMail 不支持 STARTTLS，而是仅支持 465 端口 + SSL 直连方式。

这就导致你在 Alertmanager smtp_smarthost 里设置：

smtp_smarthost: 'smtp.mail.us-west-2.awsapps.com:465'
smtp_require_tls: true

依然会发送失败。

为了让 Alertmanager 能正常通过 WorkMail 发出邮件，我的解决思路是：

不直接让 Alertmanager 发邮件，而是让它调用一个 Webhook API。
这个 API 接收告警信息，渲染 HTML 邮件模板，然后通过 SMTP_SSL 方式发送邮件。

解决方案架构

整体流程如下：

Alertmanager → Webhook API（FastAPI） → WorkMail（SMTP_SSL 465端口） → 收件人

Python FastAPI Webhook 实现

核心功能

接收 Alertmanager POST 请求
解析 UTC 时间并转为北京时间
Jinja2 渲染 HTML 模板
通过 WorkMail SMTP_SSL 465 发送邮件

完整代码如下：

from fastapi import FastAPI, Request
from jinja2 import Environment, FileSystemLoader, select_autoescape
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import smtplib
import asyncio
import concurrent.futures
import logging
from datetime import datetime
import pytz

app = FastAPI()
executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("email_sender")

# 加载模板
env = Environment(
    loader=FileSystemLoader('.'),
    autoescape=select_autoescape(['html', 'xml'])
)
template = env.get_template('email_template.html')

def parse_time(time_str):
    dt = datetime.strptime(time_str, "%Y-%m-%dT%H:%M:%S.%fZ")
    dt = dt.replace(tzinfo=pytz.utc).astimezone(pytz.timezone("Asia/Shanghai"))
    return dt

def render_email(data):
    alerts_firing = []
    alerts_resolved = []
    alerts = data.get("alerts", [])
    for alert in alerts:
        alert_copy = alert.copy()
        if "startsAt" in alert:
            alert_copy["startsAt"] = parse_time(alert["startsAt"])
        if alert.get("status") == "firing":
            alerts_firing.append(alert_copy)
        elif alert.get("status") == "resolved":
            alerts_resolved.append(alert_copy)

    return template.render(
        alerts_firing=alerts_firing,
        alerts_resolved=alerts_resolved
    )

def send_email_sync(to_emails, subject, html_content):
    try:
        msg = MIMEMultipart()
        msg["From"] = "devnotebot@pfdev2025.awsapps.com"
        msg["To"] = ", ".join(to_emails)
        msg["Subject"] = subject
        msg.attach(MIMEText(html_content, "html"))

        with smtplib.SMTP_SSL("smtp.mail.us-west-2.awsapps.com", 465) as server:
            server.login("workmail邮箱", "workmail密码")
            server.sendmail("workmail邮箱", to_emails, msg.as_string())
        logger.info(f"Sent email to {to_emails}")
    except Exception as e:
        logger.error(f"Failed to send email: {e}")

async def send_email_async(to_emails, subject, html_content):
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(executor, send_email_sync, to_emails, subject, html_content)

@app.post("/alert")
async def alert(request: Request):
    data = await request.json()
    subject = "Alertmanager通知"
    html_content = render_email(data)
    to_emails = [
        "收件人",
        "收件人"
    ]
    await send_email_async(to_emails, subject, html_content)
    return {"status": "ok"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("alert:app", host="0.0.0.0", port=880, reload=True)

在脚本内填入workmail的账号密码，发件人邮箱和收件人邮箱
smtp.mail.us-west-2.awsapps.com服务器地址根据自己的workmail地区修改

然后在同目录下用了一个email_template.html文件作为发建模板

{# email.to.html 模板 #}
<style>
.alert {
    display: inline-block;
    background-color: red;
    color: white;
    font-weight: bold;
    padding: 5px;
    margin-bottom: 10px;
}
.recovery {
    display: inline-block;
    background-color: green;
    color: white;
    font-weight: bold;
    padding: 5px;
    margin-bottom: 10px;
}
</style>

{% if alerts_firing|length > 0 %}
    {% for alert in alerts_firing %}
<div class="alert">@告警:</div> <br>
告警程序: prometheus_alert <br>
告警类型: {{ alert.labels.alertname }} <br>
故障主机: {{ alert.labels.instance_name }} ({{ alert.labels.instance }}) <br>
告警主题: {{ alert.annotations.summary }} <br>
告警详情: {{ alert.annotations.description }} <br>
触发时间: {{ alert.startsAt.strftime('%Y-%m-%d %H:%M:%S') }} <br>
    {% endfor %}
{% endif %}

{% if alerts_resolved|length > 0 %}
    {% for alert in alerts_resolved %}
<div class="recovery">@已恢复:</div> <br>
告警主机：{{ alert.labels.instance_name }} ({{ alert.labels.instance }}) <br>
告警主题：{{ alert.annotations.summary }} <br>
故障时间: {{ alert.startsAt.strftime('%Y-%m-%d %H:%M:%S') }} <br>
    {% endfor %}
{% endif %}

运行脚本后持续监听880端口

Alertmanager 配置 Webhook

在 Alertmanager 配置文件中添加 Webhook 接收地址：

receivers:
  - name: 'webhook-receiver'
    webhook_configs:
      - url: 'http://your-server-ip:880/alert'

测试后可以正常发送邮件