# Prefect 接入 Server酱：Flow 运行失败或完成时微信收到通知（2026 教程）

> 通过 Prefect 的 on_failure/on_completion 钩子或 Automations Webhook 调用 Server酱 API，实现 flow 运行状态变更时微信推送通知。

- 网页版：https://sct.ftqq.com/docs/integrations/prefect/
- 更新日期：2026-07-17

# Prefect + Server酱：Flow 运行状态变更微信通知

> 效果：Prefect flow 运行失败或完成时，微信收到状态通知

## 前置条件

- 获取 Server酱 SendKey（[30 秒获取教程](https://sct.ftqq.com/docs/getting-started/sendkey/)）
- 已安装 Prefect 及 `requests` 或 `httpx` 库的 Python 环境

## 配置步骤

接 Server酱最直接的方式是在 flow 钩子中直接调用 API，或通过 Prefect Automations 配置 Webhook。

**接法一：在 flow 钩子中调 API（推荐）**

直接在代码中定义 `on_failure` 或 `on_completion` 钩子，向 Server酱 发起 POST 请求。将 SendKey 存放于环境变量 `SERVERCHAN_SENDKEY` 中。

```python
import os
import requests
from prefect import flow

SENDKEY = os.environ.get("SERVERCHAN_SENDKEY")
SCT_API = f"https://sctapi.ftqq.com/{SENDKEY}.send"

def notify_serverchan(title: str, desp: str):
    if SENDKEY:
        requests.post(SCT_API, data={"title": title, "desp": desp})

@flow(on_failure=lambda flow, state: notify_serverchan("Flow 失败", f"详情:\n\n{state}"),
      on_completion=lambda flow, state: notify_serverchan("Flow 完成", f"详情:\n\n{state}"))
def my_flow():
    pass
```

如果你使用 Server酱³，请将 `SCT_API` 替换为 `https://<你的UID>.push.ft07.com/send/{SENDKEY}.send`（将 `<你的UID>` 替换为实际 UID，UID 取自 sctp 后的数字）。

**接法二：Prefect Automations 配 Webhook 动作**

在 Prefect Automations 中配置 Webhook 动作，URL 填写 `https://sctapi.ftqq.com/<你的SENDKEY>.send`（将 `<你的SENDKEY>` 替换为实际 SendKey），具体配置方式以 Prefect 官方文档为准。

## 完整示例

以下是一个复制即用的完整 ETL flow 示例，包含失败与完成时的微信推送：

```python
import os
import requests as req
from prefect import flow, task, get_run_logger

SENDKEY = os.environ.get("SERVERCHAN_SENDKEY")
SCT_API = f"https://sctapi.ftqq.com/{SENDKEY}.send"

def send_notification(title: str, desp: str):
    if not SENDKEY:
        return
    try:
        req.post(SCT_API, data={"title": title, "desp": desp})
    except Exception as e:
        logger = get_run_logger()
        logger.error(f"Server酱推送失败: {e}")

def on_flow_failure(flow, state):
    send_notification("【失败】Prefect Flow", f"Flow `{flow.name}` 运行失败。\n\n状态: {state}")

def on_flow_completion(flow, state):
    send_notification("【成功】Prefect Flow", f"Flow `{flow.name}` 运行完成。\n\n状态: {state}")

@task
def extract():
    return 42

@task
def transform(data):
    return data * 2

@flow(on_failure=on_flow_failure, on_completion=on_flow_completion)
def etl_flow():
    data = extract()
    return transform(data)

if __name__ == "__main__":
    etl_flow()
```

## 常见问题

**Server酱的发送额度是多少？**

免费版每日可发送 5 条，每分钟上限 50 条。订阅会员目前特价最低 3 元/月，一天最多可发 1000 条（[价格详情](https://sct.ftqq.com/subscribe)）。

**配置后没收到微信通知？**

请检查环境变量 `SERVERCHAN_SENDKEY` 是否正确设置，并参考[常见问题与排查](https://sct.ftqq.com/docs/getting-started/faq/)进行排查。

**触发频率过高被限流怎么办？**

Server酱 API 有每分钟 50 次的限制。如果 Prefect flow 短时间大量失败，请在代码中加入去重或限流逻辑，避免频繁请求。

## 相关集成

- [/integrations/python/](https://sct.ftqq.com/docs/integrations/python/)
- [/integrations/qinglong/](https://sct.ftqq.com/docs/integrations/qinglong/)
- [/integrations/uptime-kuma/](https://sct.ftqq.com/docs/integrations/uptime-kuma/)


## 懒人方案：把这段话复制给你的 AI 助手

把下面这段文字原样粘贴给 Claude Code、Cursor 或任何 AI 编程助手，它就会替你完成 Server酱 的接入：

```text
请帮我把 Server酱 微信通知接入到我当前的项目/工具中：
1. API：POST https://sctapi.ftqq.com/{SendKey}.send，参数 title（必填，不能含换行）、desp（正文，支持 Markdown，可选）；成功时返回 JSON 的 code 为 0。
2. 如果我的 SendKey 以 sctp 开头，则改用 https://{uid}.push.ft07.com/send/{SendKey}.send，uid 是 SendKey 中 sctp 与 t 之间的数字。
3. SendKey 从环境变量 SERVERCHAN_SENDKEY 读取；如果没有，提醒我到 https://sct.ftqq.com 免费获取（每天 5 条额度）。
4. 安全要求：不要把 SendKey 硬编码进代码、不要打印到日志、不要提交到 git。
5. 频率约定：任务结束发 1 条结果摘要即可，批量任务合并成一条，不要每步都推。
请根据我当前的项目和场景直接写好代码/配置并验证。
```