Hermes Agent 微信对接部署指南

引言

Hermes Agent 是一款强大的 AI 自动化代理框架，支持技能系统、定时任务、文件操作和外部工具调用。但在实际生产环境中，我们往往需要一个即时触达的交互入口——微信。无论是企业微信机器人、个人微信助手还是微信公众号，将 Hermes Agent 接入微信生态，可以让你的智能体通过微信接收指令、执行任务并推送结果，真正做到”随时随地”。

本文将详细讲解如何将 Hermes Agent 与企业微信（WeCom）和微信公众平台对接，从环境准备到线上部署，完整覆盖核心步骤与常见坑点。

核心概念

在动手之前，先梳理几个关键概念：

整体架构如下：

用户微信 → 企业微信/公众号服务器 → Webhook → Nginx → Hermes Agent Webhook Skill → 处理逻辑 → 响应回微信

实战步骤

步骤一：创建企业微信自建应用

1. 登录 企业微信管理后台
2. 进入 应用管理 → 自建 → 创建应用
3. 填写应用名称（如 “Hermes 助手”）并上传 Logo
4. 创建成功后，记录 AgentId 和 Secret
5. 设置 企业ID（CorpId），在”我的企业”页底部查看
6. 配置 接收消息服务器 URL（稍后配置，先空着）

步骤二：创建 Hermes Agent 微信技能

在 Hermes 的技能目录下新建一个 wechat-bridge 技能：

# 进入 Hermes 的技能目录
cd ~/.hermes/profiles/default/skills/
mkdir -p wechat-bridge
cd wechat-bridge

创建技能定义文件 skill.json：

{
  "name": "wechat-bridge",
  "version": "1.0.0",
  "description": "企业微信消息接收与推送桥接技能",
  "author": "Hermes",
  "triggers": {
    "webhook": {
      "path": "/webhook/wechat",
      "methods": ["GET", "POST"],
      "enabled": true
    }
  },
  "config": {
    "corp_id": "your_corp_id",
    "agent_id": "your_agent_id",
    "secret": "your_app_secret",
    "token": "your_wechat_token",
    "encoding_aes_key": "your_encoding_aes_key"
  }
}

步骤三：编写 Webhook 处理逻辑

创建 handler.py 作为技能的核心处理脚本：

#!/usr/bin/env python3
"""
Hermes Agent - 企业微信 Webhook 消息处理
"""
import hashlib
import xml.etree.ElementTree as ET
import time
import json
import requests
import os

# 从环境变量读取配置（更安全）
CORP_ID = os.environ.get("WECHAT_CORP_ID", "")
AGENT_ID = os.environ.get("WECHAT_AGENT_ID", "")
SECRET = os.environ.get("WECHAT_SECRET", "")
TOKEN = os.environ.get("WECHAT_TOKEN", "")


def verify_url(msg_signature, timestamp, nonce, echo_str):
    """验证回调 URL 的签名（微信 GET 请求验证）"""
    arr = sorted([TOKEN, timestamp, nonce])
    sign_str = "".join(arr)
    sha1 = hashlib.sha1(sign_str.encode("utf-8")).hexdigest()
    if sha1 == msg_signature:
        return echo_str
    return "verification failed"


def parse_xml(xml_data):
    """解析微信 XML 消息"""
    root = ET.fromstring(xml_data)
    msg_type = root.find("MsgType").text
    content = root.find("Content").text if root.find("Content") is not None else ""
    from_user = root.find("FromUserName").text
    msg_id = root.find("MsgId").text if root.find("MsgId") is not None else ""
    return {
        "type": msg_type,
        "content": content,
        "from_user": from_user,
        "msg_id": msg_id,
    }


def get_access_token():
    """获取企业微信 Access Token"""
    url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken"
    params = {"corpid": CORP_ID, "corpsecret": SECRET}
    resp = requests.get(url, params=params)
    data = resp.json()
    if data.get("errcode") == 0:
        return data["access_token"]
    raise Exception(f"获取 token 失败: {data}")


def send_text_message(user_id, content):
    """向指定用户发送文本消息"""
    token = get_access_token()
    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
    payload = {
        "touser": user_id,
        "msgtype": "text",
        "agentid": AGENT_ID,
        "text": {"content": content},
        "safe": 0,
    }
    resp = requests.post(url, json=payload)
    return resp.json()


def handle_message(xml_data):
    """主消息处理入口"""
    msg = parse_xml(xml_data)
    user_msg = msg["content"].strip()

    # 将消息转发给 Hermes Agent 处理
    # 这里可以调用 Hermes 的 Skill API 或直接执行命令
    reply = process_with_hermes(user_msg)

    # 回复用户
    result = send_text_message(msg["from_user"], reply)
    return result


def process_with_hermes(user_input):
    """调用 Hermes Agent 处理用户输入"""
    import subprocess
    try:
        # 调用 Hermes CLI 执行
        result = subprocess.run(
            ["hermes", "run", f"用户消息: {user_input}"],
            capture_output=True,
            text=True,
            timeout=60,
        )
        return result.stdout[:2000] or "处理完成，但无输出。"
    except subprocess.TimeoutExpired:
        return "请求超时，请稍后重试。"
    except Exception as e:
        return f"处理出错: {str(e)}"


# Flask 应用入口 (Hermes 内置 Web 服务器)
from flask import Flask, request

app = Flask(__name__)


@app.route("/webhook/wechat", methods=["GET", "POST"])
def wechat_webhook():
    if request.method == "GET":
        # 验证回调 URL
        msg_signature = request.args.get("msg_signature", "")
        timestamp = request.args.get("timestamp", "")
        nonce = request.args.get("nonce", "")
        echo_str = request.args.get("echostr", "")
        return verify_url(msg_signature, timestamp, nonce, echo_str)

    # POST: 接收消息
    xml_data = request.data.decode("utf-8")
    try:
        result = handle_message(xml_data)
        return "success"
    except Exception as e:
        print(f"处理消息失败: {e}")
        return "fail", 500


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=9090, debug=False)

步骤四：配置 Nginx 反向代理

由于微信回调要求公网可访问，且通常使用 80 或 443 端口，使用 Nginx 反向代理到 Hermes Webhook 服务：

# /etc/nginx/conf.d/wechat-hermes.conf
server {
    listen 80;
    server_name your-domain.com;

    # 微信回调专用路径
    location /webhook/wechat {
        proxy_pass http://127.0.0.1:9090;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_read_timeout 90s;
        proxy_buffering off;
    }

    # 其他 Hermes 接口可按需配置
    location / {
        proxy_pass http://127.0.0.1:9090;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}

配置完成后重载 Nginx：

# 测试配置语法
nginx -t

# 重载配置
nginx -s reload

步骤五：使用 Docker Compose 一键部署

为方便部署，准备一个 docker-compose.yml：

version: "3.8"

services:
  hermes:
    image: nousresearch/hermes-agent:latest
    container_name: hermes-agent
    restart: unless-stopped
    ports:
      - "9090:9090"
    volumes:
      - ./profiles:/root/.hermes/profiles
      - ./skills:/root/.hermes/skills
    environment:
      # 微信配置
      - WECHAT_CORP_ID=${WECHAT_CORP_ID}
      - WECHAT_AGENT_ID=${WECHAT_AGENT_ID}
      - WECHAT_SECRET=${WECHAT_SECRET}
      - WECHAT_TOKEN=${WECHAT_TOKEN}
      - WECHAT_ENCODING_AES_KEY=${WECHAT_ENCODING_AES_KEY}
      # Hermes 配置
      - HERMES_PROFILE=default
      - HERMES_LOG_LEVEL=info
    networks:
      - hermes_net

  nginx:
    image: nginx:alpine
    container_name: wechat-nginx
    restart: unless-stopped
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/conf.d:/etc/nginx/conf.d
      - ./nginx/ssl:/etc/nginx/ssl
      - ./nginx/html:/usr/share/nginx/html
    depends_on:
      - hermes
    networks:
      - hermes_net

networks:
  hermes_net:
    driver: bridge

创建 .env 环境变量文件：

# .env 文件 —— 请勿提交到 Git！
WECHAT_CORP_ID=ww1234567890abcdef
WECHAT_AGENT_ID=1000001
WECHAT_SECRET=your_app_secret_here
WECHAT_TOKEN=your_random_token_here
WECHAT_ENCODING_AES_KEY=your_43_char_encoding_aes_key_here

启动服务：

# 启动所有服务
docker compose up -d

# 查看日志
docker compose logs -f hermes

# 检查 Webhook 是否正常运行
curl -s "http://localhost:9090/webhook/wechat"

步骤六：配置微信回调 URL

回到企业微信后台的自建应用配置页：

1. 填写接收消息服务器 URL：
   http://your-domain.com/webhook/wechat
2. 填写 Token（与 skill.json 中一致）
3. 填写 EncodingAESKey（可随机生成）
4. 点击保存，微信会发送 GET 请求验证 URL——我们的 handler 会自动响应验证

验证通过后，在应用管理的”配置接收消息”中开启相应消息类型（文字、图片等）。

步骤七：配置定时推送（Cron 消息）

Hermes Agent 支持 Cron 定时任务，可用于定期向微信群或用户推送消息。在技能目录创建 cron.yaml：

# ~/.hermes/profiles/default/skills/wechat-bridge/cron.yaml
tasks:
  - name: "daily-report"
    schedule: "0 9 * * *"
    command: "python3 /root/hermes-skills/wechat-bridge/cron_report.py"
    description: "每天早上 9 点推送日报"

  - name: "reminder"
    schedule: "*/30 * * * *"
    command: "python3 /root/hermes-skills/wechat-bridge/cron_reminder.py"
    description: "每 30 分钟检查待办并提醒"

定时推送脚本示例 cron_report.py：

#!/usr/bin/env python3
"""每日工作报告推送"""
import os
import requests
from datetime import datetime

WECHAT_CORP_ID = os.environ["WECHAT_CORP_ID"]
WECHAT_AGENT_ID = os.environ["WECHAT_AGENT_ID"]
WECHAT_SECRET = os.environ["WECHAT_SECRET"]

# 目标用户（"@all" 表示群发或发送到全部成员）
TARGET_USER = "@all"  # 或指定 UserID: "zhangsan|lisi"


def get_token():
    url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
    params = {"corpid": WECHAT_CORP_ID, "corpsecret": WECHAT_SECRET}
    return requests.get(url, params=params).json()["access_token"]


def send_markdown(token):
    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
    today = datetime.now().strftime("%Y-%m-%d %A")
    content = (
        f"## 📊 每日工作日报n"
        f"**日期**：{today}nn"
        f"### 今日要点n"
        f"- 🔹 Hermes Agent 运行状态：✅ 正常n"
        f"- 🔹 待处理任务：3 项n"
        f"- 🔹 已完成任务：12 项nn"
        f"### 系统状态n"
        f"> CPU 使用率：23%n"
        f"> 内存使用率：45%n"
        f"> 磁盘使用率：67%n"
    )
    payload = {
        "touser": TARGET_USER,
        "msgtype": "markdown",
        "agentid": WECHAT_AGENT_ID,
        "markdown": {"content": content},
    }
    return requests.post(url, json=payload).json()


if __name__ == "__main__":
    token = get_token()
    result = send_markdown(token)
    print(f"[{datetime.now()}] 推送结果: {result}")

常见问题

Q1：微信回调 URL 验证失败

原因：Token 不匹配或签名计算方式有误
解决：
– 确认 skill.json 中 token 与微信后台填写的完全一致
– 确保 Webhook 服务在微信发起 GET 请求前已启动
– 检查 Nginx 日志：tail -f /var/log/nginx/access.log

Q2：消息能收到但无法回复

原因：Access Token 获取失败或用户未关注应用
解决：

# 手动测试 Token 获取
curl "https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=YOUR_CORPID&corpsecret=YOUR_SECRET"

# 发送测试消息
curl -X POST "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=TOKEN" 
  -H "Content-Type: application/json" 
  -d '{"touser":"USERID","msgtype":"text","agentid":1000001,"text":{"content":"test"}}'

Q3：消息处理超时

原因：微信要求 5 秒内返回响应，而 Hermes Agent 处理复杂任务可能超过此时间
解决：
– 采用异步处理模式：先回复 “正在处理…”，后台执行完后再主动推送结果
– 或使用企业微信的”被动回复+主动推送”双通道策略

Q4：如何对接微信公众号而非企业微信？

如果需要对接微信公众号（订阅号/服务号），差异如下：

只需将 handler 中的 API 地址和认证方式替换为公众号的即可，Webhook 校验逻辑类似。

总结

通过本文的完整步骤，你已经成功将 Hermes Agent 接入到企业微信生态中：

• ✅ 创建 Hermes Agent 微信桥接技能（Skill）
• ✅ 编写 Webhook 消息处理逻辑（Python + Flask）
• ✅ 配置 Nginx 反向代理与 Docker Compose 部署
• ✅ 设置 Cron 定时消息推送
• ✅ 完成微信回调 URL 验证与消息收发

这套方案将 Hermes Agent 强大的 AI 自动化能力与微信的即时通讯优势结合起来，无论是日常信息推送、智能问答助手，还是自动化运维通知，都能高效完成。

你可以在此基础上进一步扩展：接入多模态消息（图片、语音、文件）、集成 LLM 对话能力、对接多个群聊机器人等。Hermes Agent 的技能系统使得扩展模块化、可维护，是构建生产级 AI 助手的理想基座。