微信服务通知

借助微信公众号的 Webhook 机制，服务器既可以被动回复用户消息，也可以主动推送模板通知（告警、部署结果等）。微信测试号免费、无需认证，适合个人项目快速接入。

基本概念

概念	说明
微信测试号	公众平台提供的免费沙箱账号，支持大多数公众号接口，无需认证
OpenID	用户关注公众号后生成的唯一标识，向用户发消息必须持有
Token	你自定义的任意字符串，用于微信服务器调用你的 Webhook 时做签名验证，与 access_token 无关
Access Token	调用微信主动 API（客服消息、模板消息等）时的凭证，有效期 7200 秒，需缓存复用
被动回复	用户发消息 → 服务器在 5 秒内返回 XML 响应；逾期微信会重试最多 3 次
客服消息	主动推送接口，不受 5s 限制，需要 access_token；用户须在 48h 内发过消息才能收到
模板消息	固定格式的结构化通知（告警/订单等），需 access_token，适合服务器主动推送

申请测试号

访问微信公众平台测试号，使用微信扫码登录
记录页面顶部「测试号信息」中的 appID 和 appsecret，后续调用微信 API 时作为身份凭证使用
在「测试号二维码」处扫码关注（关注后才能在「用户列表」看到自己的 OpenID）
点击「接口配置信息」右侧的修改，填写以下两项并提交：
- URL：你的服务器 Webhook 地址（必须是 HTTPS 443 端口，如 https://your-domain.com/wechat）；该地址同时处理两类请求：GET 用于接入验证（填写后点提交时触发一次），POST 用于接收用户消息（用户每次发消息时触发）
- Token：自定义任意字符串（如 MyToken123）；微信每次推送时会带上用 Token 生成的签名，服务器用同一个 Token 验证签名，确认请求确实来自微信服务器而非伪造；填写的值需同步设置到服务器的 WECHAT_TOKEN 环境变量，见下方「接入验证」章节的签名校验代码
点击提交后，微信会立即向 URL 发送一次 GET 请求做接入验证，通过后配置才生效。本地开发时可用 ngrok / cloudflared 内网穿透生成临时 HTTPS 地址，详见下方「接入验证」章节。
按需在「模板消息接口」区域新增测试模板（仅主动推送时需要）

接入 Webhook

工作原理

微信要求 Webhook URL 只能使用 80/443 端口，不支持自定义端口。本地开发可使用 ngrok / cloudflared 内网穿透暴露本地服务。

接入验证（GET 请求）

在测试号管理页「接口配置信息」填写 URL 和 Token 后点击「提交」，微信会向 URL 发 GET 请求验证签名。通过后才算接入成功，后续消息才会推送进来。

签名验证逻辑：将 token、timestamp、nonce 三个值排序后拼接，计算 SHA1，与微信传来的 signature 比较。

import crypto from 'node:crypto'

const TOKEN = process.env.WECHAT_TOKEN // 与测试号后台填写的 Token 完全一致

// GET /wechat  — 接入验证
app.get('/wechat', (req, res) => {
  const { signature, timestamp, nonce, echostr } = req.query as Record<string, string>

  if (!signature || !timestamp || !nonce) {
    res.status(403).send('Forbidden')
    return
  }

  // timestamp 5 分钟时效校验，防止截获后重放
  const now = Math.floor(Date.now() / 1000)
  if (Math.abs(now - Number(timestamp)) > 300) {
    res.status(403).send('Timestamp expired')
    return
  }

  const hash = crypto
    .createHash('sha1')
    .update([TOKEN, timestamp, nonce].sort().join(''))
    .digest('hex')

  // 常数时间比较，防止时序攻击
  const a = Buffer.from(hash)
  const b = Buffer.from(signature)
  if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
    res.status(403).send('Invalid signature')
    return
  }

  res.send(echostr) // 验证通过，原样返回 echostr
})

接收消息（POST 请求）

接入后，用户的所有消息以 XML 格式 POST 到同一 URL。用 xml2js 解析后按 MsgType 路由处理。

微信在 5s 超时后最多重试 3 次，需要用 MsgId 做去重，避免同一条消息被处理多次：

import { parseStringPromise } from 'xml2js'

// 内存去重缓存，TTL 60s（覆盖微信全部重试窗口）
const msgCache = new Map<string, number>()

function isDuplicate(msgId: string): boolean {
  const now = Date.now()
  const expiresAt = msgCache.get(msgId)
  if (expiresAt && expiresAt > now) return true // 重复，跳过
  msgCache.set(msgId, now + 60_000)
  return false
}

// POST /wechat  — 接收用户消息
app.post('/wechat', async (req, res) => {
  const rawXml: string = req.body // 需配置 bodyParser 以 text 格式接收

  let parsed: { xml: Record<string, string[]> }
  try {
    parsed = await parseStringPromise(rawXml)
  } catch {
    res.send('success') // XML 解析失败，返回 success 避免微信无限重试
    return
  }

  const msg = parsed.xml
  const msgId = msg.MsgId?.[0]

  // 事件消息无 MsgId，用 OpenID + Event 构造唯一键
  const dedupeKey = msgId ?? `${msg.FromUserName?.[0]}_${msg.Event?.[0]}`

  if (isDuplicate(dedupeKey)) {
    res.send('success')
    return
  }

  const replyXml = resolveReply(msg)
  if (!replyXml) {
    res.send('success')
    return
  }

  res.set('Content-Type', 'text/xml')
  res.send(replyXml)
})

消息解析与被动回复

消息结构

xml2js 默认把每个字段解析为 string[]，取第一个元素即为实际值：

const msgType = msg.MsgType?.[0] // 'text' | 'image' | 'event' | ...
const fromUser = msg.FromUserName?.[0] // 用户 OpenID
const toUser = msg.ToUserName?.[0] // 公众号 gh_xxx
const content = msg.Content?.[0] // 文本内容（仅 text 消息）
const event = msg.Event?.[0] // 'subscribe' | 'unsubscribe'（仅 event）
const mediaId = msg.MediaId?.[0] // 媒体 ID（image / voice / video）

构建回复 XML

被动回复必须以 XML 格式同步返回，字段顺序严格按微信官方文档：ToUserName / FromUserName 互换（公众号 → 用户），内容用 CDATA 包裹，CreateTime 为 Unix 秒级时间戳。

安全提示：若用户输入含有 ]]>，直接拼入 CDATA 会提前终止标签破坏 XML 结构，必须先转义：

// 转义 CDATA 终止序列，防止用户内容注入 XML
function escapeCdata(str: string): string {
  return str.replace(/]]>/g, ']]]]><![CDATA[>')
}

// 公共 XML 包装（字段顺序遵照官方文档）
function wrapXml(from: string, to: string, msgType: string, body: string): string {
  return `<xml>
  <ToUserName><![CDATA[${escapeCdata(to)}]]></ToUserName>
  <FromUserName><![CDATA[${escapeCdata(from)}]]></FromUserName>
  <CreateTime>${Math.floor(Date.now() / 1000)}</CreateTime>
  <MsgType><![CDATA[${msgType}]]></MsgType>
  ${body}
</xml>`
}

// 文本消息
function buildTextReply(from: string, to: string, content: string): string {
  return wrapXml(from, to, 'text', `<Content><![CDATA[${escapeCdata(content)}]]></Content>`)
}

// 图片消息（回传同一张图）
function buildImageReply(from: string, to: string, mediaId: string): string {
  return wrapXml(from, to, 'image', `<Image><MediaId><![CDATA[${mediaId}]]></MediaId></Image>`)
}

// 图文消息（最多 8 条）
function buildNewsReply(
  from: string,
  to: string,
  articles: Array<{ title: string; description: string; picUrl: string; url: string }>,
): string {
  const items = articles
    .map(
      a => `<item>
    <Title><![CDATA[${escapeCdata(a.title)}]]></Title>
    <Description><![CDATA[${escapeCdata(a.description)}]]></Description>
    <PicUrl><![CDATA[${a.picUrl}]]></PicUrl>
    <Url><![CDATA[${a.url}]]></Url>
  </item>`,
    )
    .join('\n  ')
  return wrapXml(
    from,
    to,
    'news',
    `<ArticleCount>${articles.length}</ArticleCount>\n  <Articles>\n  ${items}\n  </Articles>`,
  )
}

处理逻辑示例

resolveReply 接收解析后的 msg 对象，返回回复 XML 字符串或 null（不回复）：

function resolveReply(msg: Record<string, string[]>): string | null {
  const from = msg.FromUserName?.[0] ?? ''
  const to = msg.ToUserName?.[0] ?? ''
  const msgType = msg.MsgType?.[0] ?? ''

  switch (msgType) {
    // ── 文本消息 ────────────────────────────────────────────────
    case 'text': {
      const content = msg.Content?.[0] ?? ''

      if (content === '帮助') {
        return buildTextReply(from, to, '发送任意文字即可对话')
      }

      return buildTextReply(from, to, content) // 默认原文回显
    }

    // ── 事件消息 ────────────────────────────────────────────────
    case 'event': {
      const event = msg.Event?.[0] ?? ''

      if (event === 'subscribe') {
        return buildTextReply(from, to, '欢迎关注！发送任意消息开始体验。')
      }

      return null // 其他事件不回复
    }

    // ── 图片消息：回传同一张图 ──────────────────────────────────
    case 'image': {
      const mediaId = msg.MediaId?.[0] ?? ''
      return buildImageReply(from, to, mediaId)
    }

    default:
      return null
  }
}

转发消息到其他接口

收到微信消息后，将其 POST 给业务服务，由业务服务决定回复内容，Webhook 服务只做透传：

微信服务器 → Webhook 服务（POST /wechat） → 业务服务（POST /your-api）
                                                    ↓ 返回回复内容
           微信服务器 ← Webhook 服务（返回 XML） ←

5s 内能响应的情况（业务服务处理快）：

case 'text': {
  const openid  = msg.FromUserName?.[0] ?? ''
  const content = msg.Content?.[0]      ?? ''

  // 带超时控制，留出余量给微信 5s TTL
  const controller = new AbortController()
  setTimeout(() => controller.abort(), 4000)

  try {
    const res = await fetch('https://your-api.com/handle', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ openid, content }),
      signal: controller.signal,
    })
    const { reply } = await res.json() as { reply: string }
    return buildTextReply(from, to, reply)
  } catch {
    // 超时或出错：不回复，微信会重试（配合去重缓存避免重复处理）
    return null
  }
}

5s 内无法响应的情况（调用 AI、数据库等耗时操作）：先立即回复占位文字，业务服务处理完成后通过 api.sendText() 异步推送结果：

case 'text': {
  const openid  = msg.FromUserName?.[0] ?? ''
  const content = msg.Content?.[0]      ?? ''

  // 立即返回占位，避免超时
  void fetch('https://your-api.com/handle', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ openid, content }),
  }).then(async (res) => {
    const { reply } = await res.json() as { reply: string }
    api.sendText(openid, reply) // 业务服务处理完成后通过客服接口推送
  })

  return buildTextReply(from, to, '正在处理，请稍候...')
}

业务服务也可以反过来主动调回 Webhook 服务，让 Webhook 服务统一调用 api.sendText() 推送——这样业务服务不需要持有微信凭证，权限更清晰。这正是 msg-bridge 项目中 POST /wechat/send 端点的用途。

主动推送通知

主动发送消息（模板通知、客服消息）统一使用 co-wechat-api，Access Token 自动管理，无需手动维护缓存和刷新逻辑。

Access Token 有效期 7200s，每日最多 2000 次，必须缓存复用；不要将 appsecret 硬编码在代码中，通过环境变量注入。

安装

bash

npm install co-wechat-api

初始化

import WechatAPI from 'co-wechat-api'

// 单实例：token 存内存，重启后自动重新获取
const api = new WechatAPI(process.env.WX_APP_ID, process.env.WX_APP_SECRET)

多实例部署时，传入 getToken / saveToken 把 token 持久化到 Redis，避免多个进程各自刷新超出每日限额：

import WechatAPI from 'co-wechat-api'
import { redisClient } from './redis.js'

const TOKEN_KEY = 'wx:access_token'

const api = new WechatAPI(
  process.env.WX_APP_ID,
  process.env.WX_APP_SECRET,
  async () => {
    const raw = await redisClient.get(TOKEN_KEY)
    return raw ? JSON.parse(raw) : null // 返回 { accessToken, expireTime }
  },
  async token => {
    const ttl = Math.floor((token.expireTime - Date.now()) / 1000)
    await redisClient.set(TOKEN_KEY, JSON.stringify(token), { EX: ttl })
  },
)

发送模板消息

模板消息适合服务器主动推送结构化通知，用户无需先发消息。收到消息有两个前提：

接收者已关注该测试号（扫码关注后 OpenID 才会出现在「用户列表」）
sendTemplate 传入的 data 字段名必须与模板正文中的变量名完全一致，否则消息发出但字段为空

测试号限制：测试号发出的模板消息不会触发手机通知栏推送，只会静默出现在会话列表中，需要手动打开微信才能看到。如需在开发阶段验证通知效果，可改用客服消息（api.sendText()）——客服消息在测试号上会正常触发通知。正式上线后申请服务号，模板消息同样会触发推送。

第一步：配置模板内容

在测试号管理页「模板消息接口」点击「新增测试模板」，填写标题和内容。动态字段用 {{变量名.DATA}} 占位，变量名自定义，传参时 key 与之对应。

以下是一个通用的告警模板，适合服务器状态、部署结果等场景：

标题：📢 服务通知

内容：
事件：{{event.DATA}}
状态：{{status.DATA}}
时间：{{time.DATA}}
─────────────────
{{detail.DATA}}

渲染效果如下：

📢 服务通知

事件：GitHub Actions 部署
状态：✅ 成功
时间：2026/4/16 14:30:00
─────────────────
modux-summarize main · commit a1b2c3d

保存后，页面会生成一个模板 ID（字符串），复制备用，传参时作为 templateId 传入。

字段名 .DATA 是固定后缀，微信要求，不可省略。字段顺序决定消息展示顺序。

第二步：调用发送接口

data 对象的 key 与模板正文中 {{变量名.DATA}} 的变量名一一对应；value 是实际展示的字符串，color 可选，控制该字段文字颜色：

await api.sendTemplate(
  openid, // 接收用户的 OpenID
  templateId, // 第一步获取的模板 ID
  null, // 点击消息跳转的 URL，不需要传 null
  {
    event: { value: 'GitHub Actions 部署' },
    status: { value: '✅ 成功', color: '#07c160' }, // color 可选
    time: { value: new Date().toLocaleString('zh-CN') },
    detail: { value: `modux-summarize main · commit ${process.env.GITHUB_SHA?.slice(0, 7)}` },
  },
)

字段缺失或命名不匹配时，微信不会报错，但对应位置会显示为空。发送前建议对照模板逐一核对 key 名称。

发送客服消息

客服消息不受 5s 时限，适合耗时操作完成后异步推结果。用户必须在 48h 内主动向公众号发过消息，否则接口拒绝。

与模板消息不同，客服消息在测试号上会正常触发手机通知，开发阶段可用来验证通知效果。

await api.sendText(openid, '你好，消息已处理完成。')

其他消息类型：

方法	用途
`api.sendText(openid, text)`	文本消息
`api.sendImage(openid, mediaId)`	图片消息
`api.sendVoice(openid, mediaId)`	语音消息
`api.sendNews(openid, articles)`	图文消息

上传临时素材

微信支持将图片、语音、视频等文件先上传为临时素材，获取一个有效期 3 天的 media_id，再用于客服消息回复，避免每次发送都重复传输文件内容。

类型	格式	大小上限	时长上限	有效期	典型用途
`image`	jpg / png / bmp / gif	≤2 MB	—	3 天	图片客服消息
`voice`	amr / mp3	≤2 MB	≤60 秒	3 天	语音客服消息
`video`	mp4	≤10 MB	—	3 天	视频客服消息
`thumb`	jpg	≤64 KB	—	3 天	视频消息缩略图

所有临时素材的 media_id 有效期均为 3 天，超期微信接口拒绝使用，需重新上传获取新 ID；有效期内同一个 media_id 可多次复用。如需长期保存，应改用永久素材接口（见「待补充」）。

接收上传文件（multer）

服务端用 multer 接收 multipart/form-data 格式的文件，先写入系统临时目录，再交给微信 API 上传：

bash

npm install multer

import os from 'node:os'
import multer from 'multer'

const upload = multer({
  dest: os.tmpdir(), // 写入系统临时目录，避免污染项目目录
  limits: { fileSize: 10 * 1024 * 1024 }, // 10 MB 兜底限制（各类型上限以微信为准）
  fileFilter: (_req, file, cb) => {
    const allowed = /^(image|audio|video)\// // 过滤非媒体 MIME 类型，拒绝文档/压缩包等
    if (allowed.test(file.mimetype)) {
      cb(null, true)
    } else {
      cb(new Error(`不支持的文件类型：${file.mimetype}`))
    }
  },
})

在路由中调用 upload.single('file')，文件信息挂载到 req.file（字段名 file 为约定，客户端 FormData 同名对应）：

// POST /upload  — 接收文件并上传到微信
app.post('/upload', upload.single('file'), async (req, res) => {
  if (!req.file || req.file.size === 0) {
    res.status(400).json({ error: '请上传文件，字段名：file' })
    return
  }
  // ...
})

上传到微信

api.uploadMedia(filepath, type) 接受本地文件路径和素材类型，返回微信分配的 media_id。上传完成后无论成功与否都要删除本地临时文件，否则 tmpdir 会持续积压：

import fs from 'node:fs/promises'

async function uploadTempMedia(filepath: string, type: string): Promise<string> {
  try {
    const result = await api.uploadMedia(filepath, type)
    return result.media_id
  } finally {
    await fs.unlink(filepath).catch(() => {}) // finally 确保临时文件必定被清理
  }
}

完整路由示例（含参数校验与临时文件清理）：

import fs from 'node:fs/promises'

const ALLOWED_TYPES = ['image', 'voice', 'video', 'thumb']

app.post('/upload', upload.single('file'), async (req, res) => {
  if (!req.file || req.file.size === 0) {
    if (req.file) await fs.unlink(req.file.path).catch(() => {}) // 参数校验失败也要清理
    res.status(400).json({ error: '请上传文件，字段名：file' })
    return
  }

  const type = (req.body.type as string) ?? 'image' // 默认 image
  if (!ALLOWED_TYPES.includes(type)) {
    await fs.unlink(req.file.path).catch(() => {})
    res.status(400).json({ error: `type 必须为 ${ALLOWED_TYPES.join(' / ')}` })
    return
  }

  const mediaId = await uploadTempMedia(req.file.path, type)
  res.json({ mediaId, type }) // media_id 有效期 3 天
})

在消息中使用 mediaId

拿到 mediaId 后，直接传给对应客服接口发送媒体消息：

// 发送图片消息（type 为 image 时上传）
await api.sendImage(openid, mediaId)

// 发送语音消息（type 为 voice 时上传）
await api.sendVoice(openid, mediaId)

// 发送视频消息（需额外上传 thumb 类型缩略图）
await api.sendVideo(openid, videoMediaId, thumbMediaId)

sendImage / sendVoice 要求 media_id 类型与消息类型一致；类型不匹配时微信接口会返回错误，不会静默忽略。

最佳实践

GitHub Actions 部署成功通知

在 CI 流水线末尾调用通知函数，自动推送部署结果到微信。

yaml

# .github/workflows/deploy.yml 末尾追加
- name: 发送微信通知
  run: node scripts/notify.js
  env:
    WX_APP_ID: ${{ secrets.WX_APP_ID }}
    WX_APP_SECRET: ${{ secrets.WX_APP_SECRET }}
    WX_OPEN_ID: ${{ secrets.WX_OPEN_ID }}
    WX_TEMPLATE_ID: ${{ secrets.WX_TEMPLATE_ID }}

// scripts/notify.js
import WechatAPI from 'co-wechat-api'

const api = new WechatAPI(process.env.WX_APP_ID, process.env.WX_APP_SECRET)

await api.sendTemplate(process.env.WX_OPEN_ID, process.env.WX_TEMPLATE_ID, null, {
  event: { value: `${process.env.GITHUB_REPOSITORY} 部署` },
  status: { value: '✅ 成功', color: '#07c160' },
  time: { value: new Date().toLocaleString('zh-CN') },
  detail: { value: `commit: ${process.env.GITHUB_SHA?.slice(0, 7)}` },
})

服务器健康巡检告警

配合系统 crontab 定时检测服务状态，异常时推送告警。

// scripts/monitor.js
import WechatAPI from 'co-wechat-api'

const api = new WechatAPI(process.env.WX_APP_ID, process.env.WX_APP_SECRET)

async function checkHealth(url) {
  try {
    const res = await fetch(url, { signal: AbortSignal.timeout(5000) })
    return res.ok
  } catch {
    return false // 超时或连接失败均视为异常
  }
}

const ok = await checkHealth('https://your-site.com')
if (!ok) {
  await api.sendTemplate(process.env.WX_OPEN_ID, process.env.WX_TEMPLATE_ID, null, {
    event: { value: '健康巡检' },
    status: { value: '❌ 服务不可用', color: '#ff4d4f' },
    time: { value: new Date().toLocaleString('zh-CN') },
    detail: { value: 'your-site.com 无法访问，请立即检查' },
  })
}

crontab 配置（每 5 分钟执行一次）：

bash

*/5 * * * * cd /path/to/project && node scripts/monitor.js >> /var/log/monitor.log 2>&1

微信服务通知 ​

基本概念 ​

申请测试号 ​

接入 Webhook ​

工作原理 ​

接入验证（GET 请求） ​

接收消息（POST 请求） ​

消息解析与被动回复 ​

消息结构 ​

构建回复 XML ​

处理逻辑示例 ​

转发消息到其他接口 ​

主动推送通知 ​

安装 ​

初始化 ​

发送模板消息 ​

发送客服消息 ​

上传临时素材 ​

接收上传文件（multer） ​

上传到微信 ​

在消息中使用 mediaId ​

最佳实践 ​

GitHub Actions 部署成功通知 ​

服务器健康巡检告警 ​

微信服务通知

基本概念

申请测试号

接入 Webhook

工作原理

接入验证（GET 请求）

接收消息（POST 请求）

消息解析与被动回复

消息结构

构建回复 XML

处理逻辑示例

转发消息到其他接口

主动推送通知

安装

初始化

发送模板消息

发送客服消息

上传临时素材

接收上传文件（multer）

上传到微信

在消息中使用 mediaId

最佳实践

GitHub Actions 部署成功通知

服务器健康巡检告警