<!-- title: 造通知MCP：完活秒发钉钉飞书 -->

# Claude Code MCP 服务器开发实战：造一个多渠道通知推送工具

## 概述

本教程带你从零构建一个 **MCP（Model Context Protocol）通知服务器**，让 Claude Code 在执行任务过程中或任务完成后，能主动把消息推送到钉钉、飞书、企业微信等团队协作渠道。学完你将掌握：MCP 服务器的标准骨架、Python FastMCP 工具开发、主流 IM 的 webhook 加签算法、以及如何把 MCP 工具与 Claude Code 的 Stop 钩子组合成"完活即报"的自动化闭环。适合已经会用 Claude Code 基础命令、想给 AI 装上"对外喊话"能力的开发者与运维同学。

## 前置条件

- 已安装 Claude Code，并能跑通 `/mcp` 命令查看已注册服务器
- Python 3.10+，会用 `pip` 安装依赖
- 一个或多个可用的团队机器人 webhook：钉钉群机器人、飞书自定义机器人（没有的话，先用"模拟打印"模式跑通流程）
- 了解 JSON 与 HTTP 基本概念，看得懂 Python 函数

---

## 第一步：先想清楚——为什么通知这件事要做成 MCP

很多人第一反应是：发个 HTTP 请求不就行了，干嘛要包成 MCP？这里的关键区别在于 **谁来决定发、发什么**。

- **写死脚本**：你得提前想好触发条件、消息内容，写进 `notify.py`，再让 cron 或 CI 去跑。脚本是死的。
- **做成 MCP 工具**：Claude 在对话中 **自己决定** 何时调用、把当前任务的结论组织成消息发出去。它是活的、有上下文的。

举个真实场景：你让 Claude 跑一遍测试并修复失败的用例。它修完后，**自己判断** "修好了，该通知主人"，然后调用 `send_notify` 工具，把修复摘要推到钉钉。你不必盯着终端，去喝杯咖啡回来手机上已经收到结果。

MCP 在这里的角色，就是 **把"发消息"这个能力，以标准协议暴露给 Claude，让它像调用内置工具一样调用它**。理解了这一点，后面的代码就顺理成章了。

## 第二步：理解 MCP 通知服务器的整体架构

在动手前，先在脑子里把数据流走一遍。Claude Code（MCP 客户端）通过 stdio 与我们的 MCP 服务器进程通信，服务器进程再通过 HTTP webhook 与钉钉/飞书的服务器通信。

```mermaid
flowchart LR
    U["用户对 Claude 说:\n'跑测试，修完通知我'"]
    C["Claude Code\n(MCP 客户端)"]
    S["notify_mcp.py\n(MCP 服务器)"]

    subgraph IM["团队协作平台"]
        D["钉钉群机器人"]
        F["飞书自定义机器人"]
    end

    P["手机收到推送"]

    U --> C
    C -->|"stdio / JSON-RPC\n调用 send_notify 工具"| S
    S -->|"HTTPS + 签名\nwebhook"| D
    S -->|"HTTPS + 签名\nwebhook"| F
    D --> P
    F --> P
```

注意中间那段 `stdio / JSON-RPC`：Claude Code 和 MCP 服务器之间走的是标准协议，不是你自定义的接口。也就是说，只要你按 MCP 规范暴露工具，**任何支持 MCP 的客户端（Claude Code、Claude Desktop、其他 IDE 插件）都能用你的工具**。这正是 MCP 的价值——写一次，到处接。

## 第三步：搭建 FastMCP 服务器骨架

我们用官方 Python SDK 的 **FastMCP** 风格开发，它用装饰器把普通函数变成 MCP 工具，写法极简。

先建工作目录并安装 SDK：

```bash
mkdir ~/notify-mcp && cd ~/notify-mcp
python -m venv .venv && source .venv/bin/activate
pip install "mcp[cli]" httpx
```

> 说明：`mcp[cli]` 是官方 Python SDK（包名就叫 `mcp`），带 CLI 扩展；`httpx` 用来发 HTTP 请求。官方 stable v2 计划随规范一起发布，当前 v1.x 已稳定可用。

创建 `notify_mcp.py`，先写一个最小骨架验证 SDK 跑得通：

```python
# notify_mcp.py —— 最小骨架
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("notify-mcp")   # 服务器名，会显示在 Claude Code 的 /mcp 列表里

@mcp.tool()
def ping(message: str) -> str:
    """原样回显消息，用于验证连通性。"""
    return f"pong: {message}"

if __name__ == "__main__":
    mcp.run(transport="stdio")   # Claude Code 默认用 stdio 传输
```

**预期结果**：这只是一个连通性测试。下一节接入后，在 Claude Code 里输入"调用 ping 工具，消息写 hi"，它应返回 `pong: hi`，证明你的 MCP 服务器已被识别。

**常见问题**：

- *报错 `ModuleNotFoundError: No module named 'mcp'`*：你大概率没激活虚拟环境，或装到了别的解释器里。Claude Code 启动 MCP 时用的是你配置里指定的那个 `python`，要确保那个解释器能 `import mcp`。
- *`from mcp.server.fastmcp import FastMCP` 报错*：旧版 SDK 路径不同。请用 `mcp>=1.x`。

## 第四步：实现钉钉渠道（含 HMAC 加签）

钉钉自定义机器人发消息的核心是 `POST` 一个 JSON 到 webhook URL。难点在于 **安全设置里的"加签"模式**：钉钉要求每个请求带上 `timestamp` 和 `sign` 两个参数，`sign` 由 `时间戳\n密钥` 经 HMAC-SHA256 后再 base64、再 URL 编码得到。

在 `notify_mcp.py` 里追加钉钉发送逻辑：

```python
import os, time, hmac, hashlib, base64, urllib.parse, json, urllib.request

# ---------- 钉钉 ----------
def _dingtalk_sign(secret: str, ts_ms: int) -> str:
    """钉钉加签：HMAC-SHA256(timestamp + '\\n' + secret) -> base64 -> URL 编码"""
    string_to_sign = f"{ts_ms}\n{secret}"
    digest = hmac.new(
        secret.encode("utf-8"),
        string_to_sign.encode("utf-8"),
        digestmod=hashlib.sha256,
    ).digest()
    return urllib.parse.quote_plus(base64.b64encode(digest))

def _dingtalk_send(webhook: str, secret: str, title: str, text: str) -> dict:
    ts_ms = int(round(time.time() * 1000))          # 钉钉要毫秒级时间戳
    if secret:
        sign = _dingtalk_sign(secret, ts_ms)
        webhook = f"{webhook}&timestamp={ts_ms}&sign={sign}"
    payload = {"msgtype": "markdown", "markdown": {"title": title, "text": text}}
    req = urllib.request.Request(
        webhook, data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=5) as resp:
        return json.loads(resp.read().decode("utf-8"))
```

这里有几个容易踩的坑，务必记住：

1. **时间戳是毫秒**（`* 1000`）。飞书用的是秒，两者别搞混。
2. **签名串里的换行符**是真实的 `\n`，不是字面量。`f"{ts}\n{secret}"` 这里的 `\n` 是转义，正确。
3. 钉钉机器人有 **频率限制：每个机器人每分钟最多 20 条**，超出会被限流 10 分钟。做监控告警类场景时，要在应用层做"合并摘要"，别把 MCP 当成高频日志通道。

如何拿到 webhook 和 secret：在钉钉群里「群设置 → 智能群助手 → 添加机器人 → 自定义」，安全设置勾选"加签"，创建后会给你一个 webhook URL（含 `access_token`）和一个 secret。两者都妥善保管。

## 第五步：实现飞书渠道

飞书自定义机器人的加签算法与钉钉 **形似而神不同**：同样是 HMAC-SHA256，但它的密钥和消息串拼法相反——飞书是 `"{timestamp}\n{secret}"` 作为 HMAC 的 **key**，消息体为空，digest 后直接 base64，不做 URL 编码。这是个反直觉的细节，写错就只会收到"签名不匹配"。

继续在文件里追加：

```python
# ---------- 飞书 ----------
def _feishu_sign(secret: str, ts_sec: int) -> str:
    """飞书加签：HMAC-SHA256(key=timestamp+'\\n'+secret, msg='') -> base64"""
    string_to_sign = f"{ts_sec}\n{secret}"
    digest = hmac.new(
        string_to_sign.encode("utf-8"),
        digestmod=hashlib.sha256,            # 注意：msg 为空，只传 key
    ).digest()
    return base64.b64encode(digest).decode("utf-8")

def _feishu_send(webhook: str, secret: str, title: str, text: str) -> dict:
    payload = {"msg_type": "text", "content": {"text": f"{title}\n{text}"}}
    if secret:
        ts_sec = int(time.time())            # 飞书要秒级时间戳
        payload["timestamp"] = str(ts_sec)
        payload["sign"] = _feishu_sign(secret, ts_sec)
    req = urllib.request.Request(
        webhook, data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=5) as resp:
        return json.loads(resp.read().decode("utf-8"))
```

对比两个加签函数，体会差异：钉钉 `hmac.new(key=secret, msg=串)`，飞书 `hmac.new(key=串, msg=None)`。这种"长得像但参数位置相反"的设计，是踩坑重灾区，建议把两个函数的注释写得足够显眼。

## 第六步：封装统一的 MCP 工具 send_notify

现在把两个渠道整合成一个对外工具。设计要点：**渠道用参数区分，密钥走环境变量**——绝不能把 secret 硬编码进代码或写进会被提交的配置文件。

```python
@mcp.tool()
def send_notify(message: str, channel: str = "dingtalk", title: str = "Claude 通知") -> str:
    """向团队通知渠道推送一条消息。

    Args:
        message: 消息正文（钉钉支持 markdown，飞书按纯文本处理）。
        channel: 渠道名，dingtalk 或 feishu，默认 dingtalk。
        title: 消息标题，仅对支持标题的渠道生效。

    Returns:
        发送结果摘要，便于 Claude 判断是否成功。
    """
    channel = (channel or "dingtalk").lower()

    if channel == "dingtalk":
        webhook = os.environ.get("DINGTALK_WEBHOOK", "")
        secret = os.environ.get("DINGTALK_SECRET", "")
        if not webhook:
            return "❌ 未配置环境变量 DINGTALK_WEBHOOK"
        res = _dingtalk_send(webhook, secret, title, message)

    elif channel == "feishu":
        webhook = os.environ.get("FEISHU_WEBHOOK", "")
        secret = os.environ.get("FEISHU_SECRET", "")
        if not webhook:
            return "❌ 未配置环境变量 FEISHU_WEBHOOK"
        res = _feishu_send(webhook, secret, title, message)

    else:
        return f"❌ 未知渠道：{channel}（支持 dingtalk / feishu）"

    # 钉钉成功返回 errcode=0，飞书返回 code=0
    ok = res.get("errcode", res.get("code", -1)) == 0
    return f"{'✅' if ok else '⚠️'} {channel} 推送结果：{res}"
```

**为什么把 docstring 写这么细**：MCP 工具的 docstring 会被原样喂给 Claude，作为它判断"什么时候该调用这个工具、参数怎么填"的依据。docstring 写得越清楚（含参数含义、默认值、返回值含义），Claude 调用得越准。这是 MCP 工具开发的一条黄金法则：**docstring 就是给 AI 看的 API 文档**。

## 第七步：接入 Claude Code

FastMCP 的 `mcp.run(transport="stdio")` 已经让服务器通过标准输入输出与客户端对话。现在把它注册进 Claude Code。推荐用命令行方式（会自动写入项目级 `.mcp.json`）：

```bash
# 注意 venv 的 python 绝对路径，避免 Claude Code 找不到依赖
claude mcp add notify-mcp \
  --env DINGTALK_WEBHOOK='https://oapi.dingtalk.com/robot/send?access_token=你的token' \
  --env DINGTALK_SECRET='SEC你的密钥' \
  --env FEISHU_WEBHOOK='https://open.feishu.cn/open-apis/bot/v2/hook/你的hook' \
  --env FEISHU_SECRET='你的飞书密钥' \
  -- ~/notify-mcp/.venv/bin/python ~/notify-mcp/notify_mcp.py
```

也可以直接手写 `.mcp.json`（效果等价，便于版本管理与团队共享）：

```json
{
  "mcpServers": {
    "notify-mcp": {
      "command": "/home/you/notify-mcp/.venv/bin/python",
      "args": ["/home/you/notify-mcp/notify_mcp.py"],
      "env": {
        "DINGTALK_WEBHOOK": "https://oapi.dingtalk.com/robot/send?access_token=xxx",
        "DINGTALK_SECRET": "SECxxx",
        "FEISHU_WEBHOOK": "https://open.feishu.cn/open-apis/bot/v2/hook/xxx",
        "FEISHU_SECRET": "xxx"
      }
    }
  }
}
```

> 安全提醒：`.mcp.json` 里含 secret，**务必加入 `.gitignore`**，或把 `env` 中的密钥改为从系统环境变量读取（`os.environ` 仍可用，只要 Claude Code 进程能继承到这些变量）。绝对不要把带密钥的 `.mcp.json` 提交到公开仓库。

## 第八步：验证——让 Claude 主动发消息

重启 Claude Code 让配置生效，然后在对话里直接说需求，不要教它怎么调工具：

```
> 跑一下 pytest，把通过/失败的用例数发到钉钉群通知我。
```

Claude 会自动：读测试 → 跑 pytest → 解析结果 → 调用 `send_notify(message="pytest 结果：12 通过，0 失败", channel="dingtalk")` → 你的手机"叮"地收到一条群消息。

用 `/mcp` 命令可以查看 `notify-mcp` 是否在线、工具是否注册成功（应显示 `send_notify`）。如果显示连接失败，常见原因是 `command` 路径写错、venv 未激活、或端口/stdio 被占用——回到第七步逐项核对。

## 第九步：升级为"完活即报"——配合 Stop 钩子

手动让 Claude 发消息还不够爽。真正的自动化是：**每次 Claude 跑完一轮任务，自动静默地通知你**。这要用到 Claude Code 的 **Stop 钩子**——在 Claude 结束响应时触发的脚本。

Stop 钩子与 MCP 的分工很清晰：

- **MCP 工具**：Claude **主动**调用，消息内容由 Claude 的上下文决定，智能化、有针对性。
- **Stop 钩子**：系统在 Claude 停下时 **自动**触发，适合"无脑提醒一下人回来"。

两者可以并存：复杂结论走 MCP 工具（Claude 组织内容），简单"我干完了"走 Stop 钩子（脚本固定文案）。

先写一个独立的通知脚本 `notify_on_stop.py`（复用钉钉加签，但不走 MCP——因为它是脚本，不需要协议）：

```python
import sys, json, os, time, hmac, hashlib, base64, urllib.parse, urllib.request

def dingtalk_send(text: str) -> str:
    webhook = os.environ["DINGTALK_WEBHOOK"]
    secret = os.environ.get("DINGTALK_SECRET", "")
    ts = int(round(time.time() * 1000))
    if secret:
        sign = urllib.parse.quote_plus(base64.b64encode(
            hmac.new(secret.encode(), f"{ts}\n{secret}".encode(), hashlib.sha256).digest()))
        webhook = f"{webhook}&timestamp={ts}&sign={sign}"
    data = json.dumps({"msgtype": "markdown",
        "markdown": {"title": "Claude 完工", "text": text}}).encode()
    req = urllib.request.Request(webhook, data=data,
        headers={"Content-Type": "application/json"})
    return urllib.request.urlopen(req, timeout=5).read().decode()

def main():
    payload = json.load(sys.stdin)        # Claude Code 把会话信息以 JSON 传入 stdin
    if payload.get("stop_hook_active"):   # 关键：避免与 MCP 通知互相触发导致循环
        return
    dingtalk_send("✅ Claude Code 已结束本轮任务，回终端看看吧。")

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

然后在 Claude Code 的 `settings.json` 里注册这个钩子（项目级 `.claude/settings.json` 或用户级 `~/.claude/settings.json`）：

```json
{
  "hooks": {
    "Stop": [
      {
        "matcher": "",
        "hooks": [
          { "type": "command",
            "command": "/home/you/notify-mcp/.venv/bin/python /home/you/notify-mcp/notify_on_stop.py" }
        ]
      }
    ]
  }
}
```

注意钩子脚本的运行环境要能读到 `DINGTALK_WEBHOOK`/`DINGTALK_SECRET` 环境变量。如果它们只写在 `.mcp.json` 的 `env` 里，钩子进程是读不到的——需要把这些变量也 export 到 shell 环境里（例如写进 `~/.bashrc` 或一个专用的 `envrc`），或在脚本里直接 `os.environ` 读。

`stop_hook_active` 那一行检查很重要：如果 Claude 自己调了 `send_notify` 发消息，发完它停下来又触发 Stop 钩子发一遍，就可能陷入提醒风暴。加这层防护可以避免重复通知。

## 实战案例：CI 构建结果通知机器人

把前面学到的组合成一个真实可用的场景：在 Claude Code 帮你排查 CI 失败、改完代码后，自动把"改了什么、CI 现在状态"推送到飞书。完整工作流如下：

```mermaid
sequenceDiagram
    participant U as 你
    participant C as Claude Code
    participant M as notify-mcp(MCP)
    participant CI as GitHub Actions
    participant F as 飞书群

    U->>C: "CI 挂了，看下日志改一下，改完通知飞书"
    C->>CI: gh run view / 读日志
    CI-->>C: 失败堆栈
    C->>C: 定位 bug 并修复代码
    C->>CI: 推送修复 / 触发重跑
    CI-->>C: 通过 ✅
    C->>M: send_notify(message="修复 Lint 报错,CI 已绿", channel="feishu")
    M->>F: POST webhook + 签名
    F-->>U: 群消息: ✅ CI 已修复通过
    Note over C: 同时 Stop 钩子兜底: "本轮任务结束"
```

这个案例的价值在于：**它把 AI 从"被动回答"变成了"主动闭环交付"**。你不盯着屏幕，CI 的修复结论会自己飞到群里。把它套到任何"耗时任务 + 需要告知他人"的场景上（数据清洗跑完通知、批量重构完通知、长时间 Agent 任务完通知）都成立。

## 进阶：消息格式与可靠性

- **富文本/卡片**：钉钉的 `markdown` 类型支持标题、加粗、链接、图片；飞书用 `msg_type=interactive` 可发交互卡片。想要更好看的报告，把 `msgtype` 换掉、payload 结构按官方文档改即可。
- **失败重试**：当前实现是"发一次就完"。生产环境建议加 1～2 次指数退避重试，并对钉钉的 `errcode=130101`（频率限制）做识别与降级（合并多条消息为一条摘要）。
- **多群路由**：把 `channel` 参数从枚举扩展为"群别名→webhook 映射表"配置文件，就能用一个工具推到不同项目群，由 Claude 根据任务上下文选群。

## 训练营适用性评估

| 维度 | 说明 |
|------|------|
| 适合训练营类型 | 后端/MCP 开发营、AI 工程化营、运维自动化营 |
| 学员基础 | 会 Python 函数、懂 HTTP，能看懂 HMAC 概念即可 |
| 建议教学时长 | 1.5～2 小时（含动手实现钉钉+飞书+钩子） |
| 可裁剪版本 | 只做钉钉单一渠道 + 骨架验证，45 分钟可完成 |
| 拓展作业 | 加企业微信/Slack/Discord 渠道；用 Channels 方向做反向推送；加失败重试与限流合并 |

本文核心是一条可复用的 MCP 交付路径：**把任何"对外动作"封装成标准工具 → 注册进 Claude Code → 用钩子补齐自动化闭环**。掌握这套范式，下一个你想让 AI 控制的东西（发邮件、建工单、调 K8s）都能照此办理。

## 延伸阅读

- 官方 Python SDK 与 FastMCP 指南：`py.sdk.modelcontextprotocol.io/server/`
- Claude Code Channels reference（反向：把 webhook 推进 Claude 会话）：`code.claude.com/docs/en/channels-reference`
- 钉钉自定义机器人发消息文档：`open.dingtalk.com/document/orgapp/enterprise-internal-robots-send-markdown-messages`
- 飞书自定义机器人使用指南：`open.feishu.cn/document/client-docs/bot-v3/add-custom-bot`

## 参考来源

1. [Building Servers - MCP Python SDK](https://py.sdk.modelcontextprotocol.io/server/) — 官方 FastMCP 服务器开发指南，本文 FastMCP 骨架与 `@mcp.tool()` 用法来源
2. [modelcontextprotocol/python-sdk](https://github.com/modelcontextprotocol/python-sdk) — 官方 Python SDK 仓库，stable v2 随规范发布的进展与版本信息
3. [Channels reference - Claude Code Docs](https://code.claude.com/docs/en/channels-reference) — Claude Code 关于"构建推送类 MCP / webhook 回灌会话"的官方参考
4. [企业内部机器人实现群聊发送 Markdown 消息](https://open.dingtalk.com/document/orgapp/enterprise-internal-robots-send-markdown-messages) — 钉钉 webhook 消息格式与加签（HMAC-SHA256）算法来源
5. [自定义机器人使用指南 - Feishu Open Platform](https://open.feishu.cn/document/client-docs/bot-v3/add-custom-bot) — 飞书自定义机器人加签算法与 msg_type 字段定义来源

> [!tip] 51学通信知识星球
> **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。
> **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。
> **当前规模**：5200 篇主题 / 990 个回答，198 元/年。
> **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。
> **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

![51学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)