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

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

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

# 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 传输

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"))

# ---------- 飞书 ----------
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"))

@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}"

# 注意 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

{
  "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"
      }
    }
  }
}

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

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()

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

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 钩子兜底: "本轮任务结束"