☰
RealPython 中文系列教程(二十八):用 Flask 与 Slack API 搭建可验证的告警机器人
2026/10/28 3:11:43 网站建设 项目流程

1. 从一条“告警发不出去”的深夜说起:Flask 与 Slack API 集成到底解决什么问题

监控告警发不出去,是很多团队都踩过的坑。服务器 CPU 打满、定时任务失败、接口 5xx 飙升,脚本明明检测到了异常,可消息就是躺在日志里没人看见。你需要的不是再装一套重型告警平台,而是一条足够短、足够可控的链路:Python 脚本发现异常,Flask 暴露一个接收端点,Slack API 把消息推到指定频道,人一眼就能看到。

这篇内容聚焦 RealPython 教程里 Flask 与 Slack API 的集成实践,面向需要把监控告警推送到 Slack 的开发者。核心检索词是 Flask 与 Slack API 集成告警机器人,它能做什么:把本地或服务器上的监控事件,通过 Flask 路由接收、校验、转发到 Slack 频道;适合谁:手里有 Python 脚本、想快速搭一条可验证告警链路的后端或运维开发。整条链路我会拆成可复制的 Flask 路由、Slack Webhook 配置、本地 curl 验证三步,并说明如何用 GitHub Copilot 辅助补全重复代码,最后在本地跑通一条告警消息的端到端链路。

先明确一个概念区分,避免后面混淆。Slack 侧有两种常见入口:一种是 Incoming Webhook,你拿到一个固定 URL,往它 POST JSON 就能发消息,适合“只发不收”的告警推送;另一种是 Events API / Outgoing Webhook,Slack 把频道里的消息 POST 给你的 Flask 端点,适合“接收并处理”。RealPython 原文用的是 slackclient 库加 Outgoing Webhook 接收消息,而告警场景更常用 Incoming Webhook 主动推送。本文两条都覆盖:§3 用 Incoming Webhook 做告警推送,§4 用 Flask 路由接收并回显,形成闭环验证。

为什么不用现成的告警平台?因为链路越长,排障越难。Flask 加 Slack Webhook 的组合,代码量小、依赖少、每一步都能用 curl 单独验证。你出问题时能立刻定位是“Flask 没收到”还是“Slack 没发出”,而不是在一堆中间件里猜。这也是我倾向自己搭一条最小链路的原因:可控性比功能丰富更重要。

还有一个现实问题:很多教程只讲“怎么发”,不讲“怎么验证发成功了”。告警系统最怕的是静默失败——你以为发出去了,其实 401 了、代理挂了、JSON 格式错了。所以本文会把验证步骤写得很细,每个环节都给出预期输出和失败时的排查方向。你照着做,能在半小时内看到一条真实消息出现在 Slack 频道里。

2. 前置准备:TaoToken 接入与 Slack Webhook 获取的完整路径

在写 Flask 代码之前,先把两样东西准备好:一个能调用模型的 API 入口(用于后续让模型帮你生成或润色告警文案、补全代码),以及 Slack 的 Webhook URL。前者走 TaoToken,后者在 Slack 后台生成。

TaoToken 的定位是统一的模型 API 接入层,官网入口是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 基址是 https://taotoken.net/api 。你需要先去控制台创建 API Key,路径是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ,Key 管理页在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。拿到 Key 后不要硬编码进代码,用环境变量注入,这一点和 Slack Token 的处理方式一致。

模型选择上,如果你只是让模型帮忙补全 Flask 路由、生成告警文案模板,用模型对话页 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=chat&utm_campaign=rewrite 先试效果最省事;如果后续要做长期的编码辅助或 Agent 流程,可以看 Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,Claude Code 相关配置参考 https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode&utm_campaign=rewrite 。这些链接先存着,§3 会用到其中的 Base URL 和 Key。

Slack 侧的准备分两步。第一步,创建一个 Slack App 或直接用 Incoming Webhook。进入 Slack API 应用管理页,新建 App,选择 “From scratch”,填 App 名称和工作区。然后在左侧菜单找到 “Incoming Webhooks”,打开开关,点击 “Add New Webhook to Workspace”,选择要接收告警的频道(建议单独建一个 #alerts 频道,别混在闲聊频道里),授权后会得到一个形如https://hooks.slack.com/services/T000/B000/XXXX的 URL。这个 URL 就是你的告警出口,泄露了别人就能往你频道发消息,所以同样走环境变量。

第二步,如果你还想验证“接收”方向(Slack 消息进 Flask),需要在 App 里配置 Event Subscriptions 或 Outgoing Webhook,并填一个公网可达的 URL。本地开发时这个 URL 不好搞,所以本文的接收验证用 curl 模拟 Slack 的 POST 请求,不依赖公网隧道,这样你在纯本地环境也能跑通。

环境依赖清单:Python 3.9+、Flask、requests。用虚拟环境隔离,命令如下:

mkdir slack-alert-bot && cd slack-alert-bot python3 -m venv venv source venv/bin/activate pip install flask requests

Windows 下激活命令是venv\Scripts\activate。装完后用pip list确认 flask 和 requests 都在。这里不装 slackclient,因为告警推送用 requests 直接 POST Webhook 更轻,也更容易看清请求体结构;slackclient 适合做频道列表、消息检索这类复杂操作,告警场景用不上。

环境变量准备两个:SLACK_WEBHOOK_URL和TAOTOKEN_API_KEY。Linux/macOS 用 export,Windows PowerShell 用$env:语法。注意 Webhook URL 里有斜杠和特殊字符,export 时用单引号包住,避免 shell 解析出错。这一步做完,前置就齐了,可以进代码。

3. 可复制配置:Flask 路由、Slack Webhook 与模型调用的 settings 片段

这一节是全文的核心,给出可直接复制的代码和配置。先看 Flask 应用的整体结构:一个app.py,包含告警接收路由/alert、Slack 消息构造与发送函数、以及一个健康检查路由/health。目录结构建议这样:

slack-alert-bot/ ├── app.py ├── config.py ├── .env └── requirements.txt

config.py负责读取环境变量,避免在业务代码里散落os.environ:

import os SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL", "") TAOTOKEN_API_KEY = os.environ.get("TAOTOKEN_API_KEY", "") TAOTOKEN_BASE_URL = "https://taotoken.net/api" ALERT_CHANNEL = os.environ.get("ALERT_CHANNEL", "#alerts")

注意TAOTOKEN_BASE_URL用的是 API 基址,不带 UTM 参数,这是调用时的规范写法。Key 从环境变量读,.env文件只放本地开发用,记得加进.gitignore。

app.py的主体如下,先看告警接收和 Slack 发送部分:

import json import logging import requests from flask import Flask, request, jsonify from config import SLACK_WEBHOOK_URL, ALERT_CHANNEL app = Flask(__name__) logging.basicConfig(level=logging.INFO) def build_slack_payload(title, detail, level="warning"): color_map = {"info": "#36a64f", "warning": "#ffcc00", "critical": "#ff0000"} return { "channel": ALERT_CHANNEL, "attachments": [ { "color": color_map.get(level, "#cccccc"), "title": title, "text": detail, "footer": "flask-slack-alert-bot", } ], } def send_to_slack(payload): if not SLACK_WEBHOOK_URL: raise RuntimeError("SLACK_WEBHOOK_URL is not set") resp = requests.post( SLACK_WEBHOOK_URL, data=json.dumps(payload), headers={"Content-Type": "application/json"}, timeout=5, ) resp.raise_for_status() return resp.status_code @app.route("/health", methods=["GET"]) def health(): return jsonify({"status": "ok"}), 200 @app.route("/alert", methods=["POST"]) def alert(): data = request.get_json(silent=True) or {} title = data.get("title", "Untitled alert") detail = data.get("detail", "") level = data.get("level", "warning") payload = build_slack_payload(title, detail, level) try: status = send_to_slack(payload) return jsonify({"sent": True, "slack_status": status}), 200 except Exception as exc: app.logger.exception("send to slack failed") return jsonify({"sent": False, "error": str(exc)}), 502 if __name__ == "__main__": app.run(host="127.0.0.1", port=5000, debug=True)

这段代码的关键点:build_slack_payload用 attachments 结构,颜色按告警级别区分,比纯文本更醒目;send_to_slack用raise_for_status()把 HTTP 错误暴露出来,而不是静默吞掉;/alert路由返回明确的sent字段,方便调用方判断。timeout=5是必须的,否则 Slack 侧慢响应会拖住 Flask 工作线程。

如果你想让模型帮忙生成告警文案或补全这段代码,TaoToken 的调用配置可以写成这样(以 OpenAI 兼容格式为例,具体模型 ID 以文档为准):

import requests from config import TAOTOKEN_API_KEY, TAOTOKEN_BASE_URL def ask_model(prompt, model="gpt-4o-mini"): resp = requests.post( f"{TAOTOKEN_BASE_URL}/v1/chat/completions", headers={ "Authorization": f"Bearer {TAOTOKEN_API_KEY}", "Content-Type": "application/json", }, json={ "model": model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.3, }, timeout=30, ) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"]

这里 Base URL 是https://taotoken.net/api,Key 走TAOTOKEN_API_KEY,Model ID 按你控制台里可用的填。三件套齐了:Base URL、Key、Model ID。如果你用 Claude Code 做编码辅助,配置参考文档页,同样是这三项对齐。

GitHub Copilot 在这里的角色是补全重复代码。比如你写完build_slack_payload的函数签名和第一行,Copilot 往往能根据上下文补出 attachments 结构;写send_to_slack时,它会提示requests.post加raise_for_status的常见模式。但要注意,Copilot 补出的 Webhook 请求可能漏掉Content-Type头,或者把data=写成json=导致 Slack 返回invalid_payload。所以补全后要对照本文的代码检查请求体格式,别直接 Tab 到底。

4. 验证请求与成功结果:用 curl 跑通一条告警的端到端链路

代码写完,先启动 Flask:

python app.py

预期输出里有Running on http://127.0.0.1:5000。先打健康检查:

curl -s http://127.0.0.1:5000/health

预期返回{"status":"ok"}。这一步确认 Flask 本身活着,排除端口占用、依赖缺失这类基础问题。

然后发一条告警:

curl -s -X POST http://127.0.0.1:5000/alert \ -H "Content-Type: application/json" \ -d '{"title":"CPU 使用率过高","detail":"web-01 CPU 92% 持续 5 分钟","level":"critical"}'

预期返回类似{"sent":true,"slack_status":200}。同时去 Slack 的 #alerts 频道看,应该出现一条带红色边条的卡片,标题是“CPU 使用率过高”,正文是详情,footer 是flask-slack-alert-bot。看到这条消息,说明 Flask 接收、payload 构造、Slack 发送三段全部打通。

如果 Slack 频道没消息但 curl 返回了 200,先检查ALERT_CHANNEL是否和 Webhook 绑定的频道一致。Incoming Webhook 的频道在创建时就固定了,payload 里的channel字段可能被忽略,这是常见误解。以 Webhook 实际绑定的频道为准。

再验证接收方向,模拟 Slack 往 Flask 发消息。虽然本文的/alert是给内部脚本调的,但你可以加一个/slack/events路由来接收 Slack 的 POST,验证请求解析逻辑:

@app.route("/slack/events", methods=["POST"]) def slack_events(): data = request.get_json(silent=True) or {} if data.get("type") == "url_verification": return jsonify({"challenge": data.get("challenge", "")}), 200 app.logger.info("slack event: %s", json.dumps(data)[:500]) return jsonify({"ok": True}), 200

用 curl 模拟验证请求:

curl -s -X POST http://127.0.0.1:5000/slack/events \ -H "Content-Type: application/json" \ -d '{"type":"url_verification","challenge":"abc123"}'

预期返回{"challenge":"abc123"}。这一步是 Slack Events API 的握手验证,能过说明你的解析和返回格式正确。真实接入时还需要校验签名,本文聚焦本地链路,签名校验留到上线前补。

模型调用也可以单独验证。写个小脚本调ask_model("用一句话描述 CPU 告警"),预期返回一段中文文案。如果返回 401,检查 Key 是否带Bearer前缀、是否有多余空格;如果返回 404,检查 Base URL 是否多了或少了/v1,以文档为准。

整条链路验证顺序建议:health → alert → Slack 频道确认 → slack/events 握手 → 模型调用。每步都有明确预期输出,哪步断了就停在哪步排查,不要跳步。我试过跳过 health 直接发 alert,结果端口被旧进程占着,curl 连到了另一个服务,白折腾十分钟。

5. 本篇常见错排查:401、local proxy failed、reading choices 与 OAuth 报错对照

告警链路跑不通时,报错信息往往很含糊。这一节按真实报错逐条对照,给出定位方向。

401 Unauthorized(Slack 侧):curl 返回{"sent":false,"error":"401 Client Error..."}。原因通常是 Webhook URL 失效或被撤销。去 Slack App 的 Incoming Webhooks 页面确认 URL 还在,重新生成一个替换环境变量。注意 URL 里的T000/B000/XXXX三段任何一段错都会 401,复制时别漏字符。

401 Unauthorized(TaoToken 侧):模型调用返回 401。检查TAOTOKEN_API_KEY是否设置成功,echo $TAOTOKEN_API_KEY看有没有值;检查请求头是否是Authorization: Bearer <key>,少了Bearer或多了引号都会失败。Key 管理页在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite ,可以对照确认 Key 状态。

local proxy failed:这个报错通常出现在 requests 走了系统代理但代理不可用。检查环境变量HTTP_PROXY、HTTPS_PROXY是否指向了一个已失效的地址。告警脚本运行环境应该直连,把这两个变量清掉再试:unset HTTP_PROXY HTTPS_PROXY。注意不要在任何配置里写代理相关的绕过方案,直连是唯一正确姿势。

reading 'choices' / KeyError: 'choices':模型返回体里没有choices字段,说明请求没成功但代码没先检查状态码。典型原因是 Base URL 拼错(比如写成https://taotoken.net/api/v1/chat/completions但实际路径不同),或者模型 ID 不存在导致返回错误 JSON。修复方式:在resp.json()之前先resp.raise_for_status(),并把原始返回打印出来看。对照接入文档确认路径和模型 ID。

invalid_payload(Slack 侧):curl 返回 400,Slack 说invalid_payload。最常见是Content-Type没设成application/json,或者用了json=但 Slack 期望的是data=json.dumps(...)。本文代码用后者,照抄即可。另一个原因是 payload 里attachments结构写错,比如color值不是合法十六进制。

OAuth / invalid_auth:如果你走的是 OAuth 流程拿 token,报invalid_auth说明 token 过期或 scope 不足。告警场景用 Incoming Webhook 就绕开了 OAuth,不需要处理 token 刷新。如果你确实要用 Bot Token 调chat.postMessage,确认 scope 里有chat:write,并且 token 以xoxb-开头。

Flask 端口占用:Address already in use。用lsof -i :5000找到进程 kill 掉,或者换端口app.run(port=5001)。debug 模式下 Flask 会起 reloader,有时两个进程抢端口,加use_reloader=False可避免。

Slack 频道收不到但返回 200:前面提过,Webhook 绑定频道优先。另外检查频道是否是私有频道,私有频道需要把 App 邀请进去。在频道里输入/invite @你的App名即可。

排查原则:先看 HTTP 状态码,再看返回体原文,最后对照本文的请求格式。不要凭报错关键词猜,把resp.text打出来,90% 的问题一眼就能看出来。

6. 语义一致的后续路径:按你的下一步选模型对话、接入文档还是 Coding Plan

链路跑通后,下一步取决于你要往哪走。如果你只是想验证模型能不能帮你写告警文案、补全 Flask 代码,直接去模型对话页 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=chat&utm_campaign=rewrite 试几条 prompt,成本最低。如果你在接入过程中卡在鉴权、路径、参数上,接入文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 里有完整的请求示例和错误码说明,配合 API Keys 页 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 对照 Key 状态,基本能自助解决。

如果你打算把这条告警链路做成长期运行的编码辅助或 Agent 流程,比如让模型自动分析告警日志、生成修复建议、甚至触发回滚脚本,那 Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 更适合,它面向的是持续性的编码与 Agent 场景,不是一次性对话。Claude Code 用户看 https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode&utm_campaign=rewrite ,配置时同样对齐 Base URL、Key、Model ID 三件套。

最后给一个实用技巧:把本文的/alert路由挂到你的监控脚本里,用requests.post调本地 Flask,而不是让监控脚本直接调 Slack Webhook。多这一层的好处是,你可以在 Flask 里加去重、限流、告警聚合逻辑,而不用改监控脚本。比如同一告警 5 分钟内只发一次,在alert()里加个内存字典记录时间戳就行。这条链路的价值不在于“能发消息”,而在于“发消息之前你能控制什么”。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询