《​WorkBuddy 接入企业微信指南》

摘要：
企微助理集成支持通过企业微信远程操控电脑端的WorkBuddy，实现任务下发、进度跟进和团队协作。用户需确保WorkBuddy≥4.6.4版本、企业微信账号权限及网络畅通。配置分为管理员/成员两种角色操作：通过企业微信后台或客户端创建智能机器人，选择长连接模式（推荐）绑定BotID和Secret，或备选的URL回调模式（需回填Webhook）。绑定后，企业微信可远程触发电脑端任务执行，结果同步至聊天窗口，支持跨设备查看记录和持续交互。常见问题包括机器人无响应、URL验证失败等，需检查凭据、网络及服务状态。

企微助理集成让您可以通过企业微信远程控制电脑上的 WorkBuddy。在企业微信中向智能机器人发送任务指令，WorkBuddy 在电脑上自动执行，并将结果同步回企业微信聊天窗口。

企微助理能做什么​

绑定企微助理后，您可以在企业微信中完成以下操作：

• 在群聊中 @机器人下发团队协作任务，WorkBuddy 自动执行并返回结果
• 跟进任务进度：直接向机器人发送任务指令，随时跟进执行进度
• 创建专属任务群，长期绑定 WorkBuddy 处理团队日常工作
• 发布通知：任务完成后自动通知群聊相关人员
• 任务提醒：将待办、验收、协作事项同步到企微群
• 流程打通：连接 WorkBuddy 任务处理与团队现有工作流
• 确认操作：批准命令执行、文件修改等需要确认的操作

设置前的准备​

在开始配置之前，请确认满足以下条件：

要求	说明
WorkBuddy	已在电脑上安装 WorkBuddy >= 4.6.4
企业微信账号	拥有一个可创建智能机器人的企业微信账号
网络连接	电脑和手机均能正常访问互联网
WorkBuddy 运行中	电脑保持开机并运行 WorkBuddy

如果您还没有企业微信账号，可以前往 企业微信官网 免费注册一个企业。

连接企微助理​

企业微信集成需要先在平台完成机器人创建和配置。请根据您的账号角色选择对应的操作指南。

管理员操作​

第一步：进入管理后台创建机器人​

打开浏览器，访问 企业微信管理后台，使用管理员账号登录后，依次点击「安全与管理」→「管理工具」→「智能机器人」→「创建机器人」。

如果先进入 AI 自动生成页面，请点击左下角的「手动创建」：

在创建页面直接选择「API 模式创建」：

第二步：填写基本信息​

进入 API 模式页面后，请先完成以下公共配置：

配置项	说明
机器人名称	建议填写容易识别的名字，如「WorkBuddy 助手」
可见范围	选择哪些员工、部门或标签可以使用这个机器人

点击「可见范围」后的「添加」，选择需要使用机器人的成员、部门或标签。

完成以上配置后，请先点击页面底部的「保存」，再在右侧「API 配置」区域选择「使用长连接」。

第三步：获取凭证并完成绑定​

进入使用长连接方式接入完成后续步骤。

普通成员操作​

第一步：在企业微信客户端创建机器人​

打开企业微信客户端，进入「工作台」→「智能机器人应用」→「创建机器人」。

如果先进入 AI 自动生成页面，请点击左下角的「手动创建」，再选择「API 模式创建」：

第二步：填写基本信息​

进入 API 模式页面后，请先完成以下公共配置：

配置项	说明
机器人名称	建议填写容易识别的名字，如「WorkBuddy 助手」
可使用成员围	选择哪些员工、部门或标签可以使用这个机器人

点击「可使用成员」后的「修改」，选择需要使用机器人的成员、部门或标签。

完成以上配置后，再在下方「API 配置」区域选择「使用长连接」。

第三步：获取凭证并完成绑定​

进入使用长连接方式接入完成后续步骤。

使用长连接方式接入​

两种角色在此步骤的操作一致。推荐优先使用长连接模式，配置步骤更少，也不需要回填 Webhook URL。

在企业微信中获取 Bot ID 和 Secret：

1. 在「API 配置」区域选择「使用长连接」
2. 复制 Bot ID
3. 点击 点击获取 获取 Secret，并妥善保存

长连接模式不需要再填写 URL、Token 或 Encoding-AESKey。

在 WorkBuddy 中完成绑定​

1. 打开 WorkBuddy，点击左下角头像菜单中的「设置-助理设置」
2. 在「集成（BETA）」区域找到「企微助理集成」，点击「配置」
3. 在弹窗中选择「WebSocket 长连接」后，有两种绑定方式可选：
   • 手动填写 Bot ID 和 Secret
   • 点击快捷绑定接入，使用企微扫码完成绑定
   • 填入刚才复制的 Bot ID 和 Secret后点击「注册」完成绑定

关于 API 模式​

企业微信 API 模式支持两种接入方式：

• 长连接模式（推荐）：在 WorkBuddy 中填写 Bot ID 和 Secret 即可完成绑定，配置更简单，无需回填 Webhook URL。
• URL 回调模式：适用于需要 Webhook 回调的场景。需在 WorkBuddy 中填写 Token 和 Encoding-AESKey，生成 Webhook URL 后再回填到企业微信。

URL 回调模式配置指引见文末「使用 URL 回调接入（备选方案）」。

开始使用​

配置完成后，您就可以在企业微信中直接与机器人对话了。

找到机器人​

在企业微信通讯录的「企业创建的」分组下找到刚刚创建的机器人，点击「发消息」即可开始下发任务。

联通后对话效果​

发送一条简单消息（例如"你好"）进行联通测试。如果配置正确，WorkBuddy 会接收到消息并在企业微信中返回回复。

恭喜

如果机器人能够正常回复，说明您已成功完成 WorkBuddy 与企业微信的对接！

跨设备接续工作​

企微助理支持与 WorkBuddy 桌面端无缝衔接：

• 在企业微信中发起一个任务后，可以回到电脑上的 WorkBuddy 查看完整的执行过程和结果
• 在助理页面可查看远程任务的完整执行记录，包括思考过程、操作步骤、生成文件及最终结果
• 您可以在电脑上继续追问、细化任务，或在企业微信中随时查看进度

什么来自本地电脑​

当您通过企业微信向 WorkBuddy 发送任务时，处理过程发生在您的电脑上：

来源	说明
企业微信	仅发送指令、批准操作和后续消息
您的电脑	提供 WorkBuddy 执行任务所需的全部环境

电脑上的 WorkBuddy 使用您本地的项目文件、Shell 环境、凭证权限、插件和工具来执行任务。也就是说，您电脑上能做的事，通过企业微信也能远程完成。

远程任务与普通任务的区别​

对比项	普通任务	助理（远程任务）
工作目录	自由指定	固定使用助理专属文件夹
新建对话	支持多个并行任务	仅一个会话，所有远程指令集中处理
上下文管理	可清空历史重新开始	保留完整对话历史，不可清空

本地操作优先使用主界面的普通任务；需要远程触发或统一查看记录时，再使用助理。

使用 URL 回调接入（备选方案）​

如果您已经在使用 Webhook 配置，或因网络、部署限制需要通过 URL 回调接入，可按以下步骤操作。

在企业微信中生成回调凭据​

1. 在「API 配置」区域选择「使用 URL 回调」
2. 点击 Token 和 Encoding-AESKey 输入框右侧的「随机获取」
3. 保存这两个参数，后续需要在 WorkBuddy 中使用

重要

请务必保存好 Token 和 Encoding-AESKey，否则后续无法完成注册。

在 WorkBuddy 中生成 Webhook URL​

1. 打开 WorkBuddy，进入「助理设置」→「企微 AIBot 集成」→「配置」
2. 在注册弹窗中切换到「使用 URL 回调」
3. 填入刚才获取的 Token 和 Encoding-AESKey，点击「注册」
4. 注册成功后，复制生成的 Webhook URL

回填 URL 并保存​

回到企业微信的机器人创建页面：

1. 将刚才复制的 Webhook 地址粘贴到 URL 输入框
2. 点击「保存」完成配置

常见问题​

机器人没有响应怎么办？​

请按以下步骤排查：

1. 检查 WorkBuddy 状态：确保电脑上的 WorkBuddy 正在运行，且助理服务已开启
2. 核对接入方式：确认企业微信与 WorkBuddy 中选择的是同一种接入方式
3. 检查凭据：长连接模式请核对 Bot ID 和 Secret；URL 回调模式请核对 Token 和 Encoding-AESKey
4. 检查网络连接：确保电脑能够正常访问网络

URL 验证失败怎么办？​

1. 确保 WorkBuddy 处于运行状态
2. 检查助理服务是否已正常启动
3. 重新复制 Webhook URL，确保没有遗漏或多余字符
4. 确认 Token 和 Encoding-AESKey 与 WorkBuddy 中配置的完全一致

长连接注册失败怎么办？​

1. 确认企业微信侧已选择「使用长连接」
2. 重新复制 Bot ID 和 Secret，避免带入多余空格
3. 如 Secret 已失效，可在企业微信后台重新获取后再次注册

实战代码案例

以下提供 3 个可直接运行的代码案例，覆盖企微机器人消息推送、长连接消息接收与自动回复、以及 WorkBuddy 自动化任务配置三大核心场景。

---

案例一：通过 Webhook 向企微群发送通知（Python）

场景：任务完成后，自动向企微群发送文本或 Markdown 通知，与 WorkBuddy 的「发布通知」功能配合使用。

import requests
import json
import os

class WeComBotNotifier:
    """
    企业微信机器人消息推送类
    支持文本、Markdown、图片、文件等多种消息类型
    """
    
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
        self.headers = {"Content-Type": "application/json"}
    
    def send_text(self, content: str, mentioned_list: list = None) -> dict:
        """
        发送文本消息
        :param content: 消息内容（最长2048字节）
        :param mentioned_list: 需要@的成员手机号列表，如 ["13800138000"]
        """
        payload = {
            "msgtype": "text",
            "text": {
                "content": content,
                "mentioned_list": mentioned_list or []
            }
        }
        return self._send(payload)
    
    def send_markdown(self, content: str) -> dict:
        """
        发送 Markdown 格式消息
        支持语法：标题、加粗、链接、引用、颜色高亮等
        """
        payload = {
            "msgtype": "markdown",
            "markdown": {
                "content": content
            }
        }
        return self._send(payload)
    
    def send_image(self, image_path: str) -> dict:
        """
        发送图片消息（需先转为 Base64）
        :param image_path: 本地图片路径（不超过2MB）
        """
        import base64
        import hashlib
        
        with open(image_path, "rb") as f:
            img_data = f.read()
        
        base64_data = base64.b64encode(img_data).decode("utf-8")
        md5_value = hashlib.md5(img_data).hexdigest()
        
        payload = {
            "msgtype": "image",
            "image": {
                "base64": base64_data,
                "md5": md5_value
            }
        }
        return self._send(payload)
    
    def _send(self, payload: dict) -> dict:
        """统一发送请求并处理响应"""
        try:
            resp = requests.post(
                self.webhook_url,
                headers=self.headers,
                data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
                timeout=10
            )
            resp.raise_for_status()
            result = resp.json()
            if result.get("errcode") == 0:
                print(f"✅ 消息发送成功")
            else:
                print(f"❌ 发送失败: {result}")
            return result
        except Exception as e:
            print(f"❌ 请求异常: {e}")
            return {"errcode": -1, "errmsg": str(e)}


# ========== 使用示例 ==========
if __name__ == "__main__":
    # 从环境变量读取 Webhook URL，避免硬编码
    WEBHOOK_URL = os.getenv("WECOM_BOT_WEBHOOK", 
        "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY_HERE")
    
    bot = WeComBotNotifier(WEBHOOK_URL)
    
    # 1. 发送普通文本通知
    bot.send_text(
        content="🎉 WorkBuddy 任务执行完毕！\n项目构建成功，已部署到测试环境。",
        mentioned_list=["@all"]  # @所有人
    )
    
    # 2. 发送 Markdown 富文本通知
    bot.send_markdown(
        content="""## 📊 每日构建报告
> **项目**: WorkBuddy-Assistant  
> **分支**: main  
> **状态**: <font color='green'>✅ 成功</font>  
> **耗时**: 3m 42s  
> **提交**: `feat: 优化企微消息推送逻辑`  
> [查看详细日志](https://your-ci-server.com/build/12345)
"""
    )

---

案例二：长连接模式接收消息并自动回复（Python）

场景：基于企微智能机器人长连接协议，实现一个轻量级消息网关，接收用户 @机器人 的指令并自动回复，与 WorkBuddy 的「在群聊中 @机器人下发任务」功能对应。

import json
import websocket
import threading
import time
from typing import Callable

class WeComAIBotGateway:
    """
    企业微信智能机器人长连接网关
    基于 WebSocket 接收消息回调，并主动回复消息
    """
    
    def __init__(self, bot_id: str, secret: str, 
                 on_message_handler: Callable[[dict], str] = None):
        self.bot_id = bot_id
        self.secret = secret
        self.ws_url = "wss://aibot-work.weixin.qq.com/ws/open/aibot"
        self.ws = None
        self.on_message_handler = on_message_handler or self._default_handler
        self._connected = False
    
    def _default_handler(self, msg_body: dict) -> str:
        """默认消息处理逻辑"""
        msg_type = msg_body.get("msgtype", "text")
        if msg_type == "text":
            content = msg_body.get("text", {}).get("content", "")
            # 去除 @机器人 前缀
            clean_content = content.split(" ", 1)[-1] if " " in content else content
            return f"收到您的指令：「{clean_content}」，WorkBuddy 正在处理中..."
        return "暂不支持该消息类型的自动回复"
    
    def _build_auth_request(self) -> dict:
        """构建长连接鉴权请求"""
        return {
            "cmd": "aibot_auth",
            "headers": {
                "req_id": f"auth_{int(time.time() * 1000)}"
            },
            "body": {
                "aibotid": self.bot_id,
                "secret": self.secret
            }
        }
    
    def _build_reply(self, req_id: str, chat_id: str, content: str) -> dict:
        """构建回复消息请求"""
        return {
            "cmd": "aibot_send_msg",
            "headers": {
                "req_id": req_id
            },
            "body": {
                "chatid": chat_id,
                "chat_type": 2,  # 2=群聊，1=单聊
                "msgtype": "markdown",
                "markdown": {
                    "content": content
                }
            }
        }
    
    def _on_open(self, ws):
        """连接建立后发送鉴权"""
        print("🔌 WebSocket 连接已建立，正在鉴权...")
        auth_req = self._build_auth_request()
        ws.send(json.dumps(auth_req))
    
    def _on_message(self, ws, message):
        """接收消息并处理"""
        try:
            data = json.loads(message)
            cmd = data.get("cmd", "")
            body = data.get("body", {})
            headers = data.get("headers", {})
            req_id = headers.get("req_id", "")
            
            # 鉴权响应
            if cmd == "aibot_auth_resp":
                errcode = body.get("errcode", -1)
                if errcode == 0:
                    print("✅ 鉴权成功，长连接已就绪")
                    self._connected = True
                else:
                    print(f"❌ 鉴权失败: {body}")
                return
            
            # 消息回调（用户发送消息）
            if cmd == "aibot_msg_callback":
                chat_id = body.get("chatid", "")
                from_user = body.get("from", {}).get("userid", "")
                msg_type = body.get("msgtype", "")
                
                print(f"📩 收到消息 | 来自: {from_user} | 类型: {msg_type}")
                
                # 调用业务处理函数生成回复内容
                reply_content = self.on_message_handler(body)
                
                # 发送回复
                reply_msg = self._build_reply(req_id, chat_id, reply_content)
                ws.send(json.dumps(reply_msg))
                print(f"📤 已回复消息到群聊 {chat_id}")
            
            # 事件回调（如进入会话）
            elif cmd == "aibot_event_callback":
                event_type = body.get("event", {}).get("eventtype", "")
                if event_type == "enter_chat":
                    # 回复欢迎语
                    welcome = {
                        "cmd": "aibot_respond_welcome_msg",
                        "headers": {"req_id": req_id},
                        "body": {
                            "msgtype": "text",
                            "text": {
                                "content": "👋 您好！我是 WorkBuddy 智能助理，\n请直接 @我 下发任务，我会调用本地 WorkBuddy 为您执行。"
                            }
                        }
                    }
                    ws.send(json.dumps(welcome))
                    print("👋 已发送欢迎语")
                    
        except Exception as e:
            print(f"❌ 消息处理异常: {e}")
    
    def _on_error(self, ws, error):
        print(f"⚠️ WebSocket 错误: {error}")
    
    def _on_close(self, ws, close_status_code, close_msg):
        print(f"🔌 连接已关闭 [{close_status_code}]: {close_msg}")
        self._connected = False
    
    def start(self):
        """启动长连接"""
        print(f"🚀 正在连接企微智能机器人长连接网关...")
        self.ws = websocket.WebSocketApp(
            self.ws_url,
            on_open=self._on_open,
            on_message=self._on_message,
            on_error=self._on_error,
            on_close=self._on_close
        )
        
        # 使用独立线程运行 WebSocket
        self.ws_thread = threading.Thread(target=self.ws.run_forever)
        self.ws_thread.daemon = True
        self.ws_thread.start()
    
    def stop(self):
        """关闭连接"""
        if self.ws:
            self.ws.close()
        print("👋 网关已停止")


# ========== 使用示例 ==========
if __name__ == "__main__":
    # 替换为从企微后台获取的实际凭证
    BOT_ID = "YOUR_BOT_ID_HERE"
    SECRET = "YOUR_SECRET_HERE"
    
    # 自定义业务处理：将用户指令转发给 WorkBuddy 本地服务
    def custom_handler(msg_body: dict) -> str:
        content = msg_body.get("text", {}).get("content", "")
        # 这里可以接入 WorkBuddy 本地 API 或 Shell 调用
        # 例如：requests.post("http://localhost:8080/execute", json={"command": content})
        return f"🤖 WorkBuddy 已收到指令：\n```\n{content}\n```\n\n⏳ 正在本地执行，完成后将推送结果..."
    
    gateway = WeComAIBotGateway(
        bot_id=BOT_ID,
        secret=SECRET,
        on_message_handler=custom_handler
    )
    
    gateway.start()
    
    # 保持主线程运行
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        gateway.stop()

---

案例三：WorkBuddy 自动化任务配置（YAML + Python）

场景：配置一个每日定时任务，自动抓取数据、生成报告并通过企微群推送，与 WorkBuddy 的「任务提醒」「发布通知」功能配合使用。

# ========== workbuddy_task.py ==========
"""
WorkBuddy 自动化任务脚本
功能：每日自动生成本地项目构建报告，并推送到企微群
"""

import subprocess
import os
import json
from datetime import datetime
from pathlib import Path

# 引入案例一的推送类（需将案例一代码保存为 wecom_bot.py）
from wecom_bot import WeComBotNotifier


def run_shell_command(cmd: str, cwd: str = None) -> tuple:
    """
    执行本地 Shell 命令并捕获输出
    对应 WorkBuddy 使用本地 Shell 环境执行任务的能力
    """
    try:
        result = subprocess.run(
            cmd,
            shell=True,
            cwd=cwd,
            capture_output=True,
            text=True,
            timeout=300
        )
        return result.returncode, result.stdout, result.stderr
    except subprocess.TimeoutExpired:
        return -1, "", "命令执行超时"


def generate_build_report(project_path: str) -> dict:
    """
    生成项目构建报告
    模拟 WorkBuddy 在本地执行代码分析任务
    """
    report = {
        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "project": Path(project_path).name,
        "git_branch": "unknown",
        "git_commit": "unknown",
        "file_count": 0,
        "test_result": "未执行"
    }
    
    # 获取 Git 信息
    rc, stdout, _ = run_shell_command("git branch --show-current", cwd=project_path)
    if rc == 0:
        report["git_branch"] = stdout.strip()
    
    rc, stdout, _ = run_shell_command("git log -1 --oneline", cwd=project_path)
    if rc == 0:
        report["git_commit"] = stdout.strip()
    
    # 统计代码文件数量
    rc, stdout, _ = run_shell_command(
        "find . -name '*.py' -o -name '*.js' -o -name '*.ts' | wc -l",
        cwd=project_path
    )
    if rc == 0:
        report["file_count"] = int(stdout.strip())
    
    # 运行测试（示例）
    rc, stdout, stderr = run_shell_command("python -m pytest --tb=short", cwd=project_path)
    report["test_result"] = "✅ 通过" if rc == 0 else f"❌ 失败 (exit {rc})"
    report["test_output"] = stdout[-500:] if stdout else stderr[-500:]  # 截取最后500字符
    
    return report


def format_markdown_report(report: dict) -> str:
    """将报告格式化为 Markdown 消息"""
    return f"""## 📊 WorkBuddy 自动化日报
> **生成时间**: {report['timestamp']}
> **项目**: `{report['project']}`
> **分支**: `{report['git_branch']}`
> **最新提交**: {report['git_commit']}

---

**📁 代码统计**
- 源码文件数: **{report['file_count']}**

**🧪 测试结果**
- 状态: {report['test_result']}

**📋 输出摘要**

{report.get('test_output', '无')}

plain

---
<font color='info'>💡 本报告由 WorkBuddy 企微助理自动生成</font>
"""


def main():
    """主执行流程"""
    # 配置项（实际使用建议从环境变量或配置文件读取）
    PROJECT_PATH = os.getenv("PROJECT_PATH", "/path/to/your/project")
    WEBHOOK_URL = os.getenv("WECOM_BOT_WEBHOOK", "YOUR_WEBHOOK_URL")
    
    print(f"🚀 [{datetime.now()}] 开始执行 WorkBuddy 自动化任务...")
    
    # Step 1: 生成本地报告（使用本地环境）
    print("📊 正在生成本地项目报告...")
    report = generate_build_report(PROJECT_PATH)
    
    # Step 2: 格式化消息
    markdown_msg = format_markdown_report(report)
    
    # Step 3: 推送到企微群（远程通知）
    print("📤 正在推送报告到企微群...")
    notifier = WeComBotNotifier(WEBHOOK_URL)
    result = notifier.send_markdown(markdown_msg)
    
    if result.get("errcode") == 0:
        print("✅ 任务执行完毕，报告已推送")
    else:
        print(f"❌ 推送失败: {result}")
    
    # Step 4: 将报告保存到本地（助理专属文件夹）
    report_dir = Path.home() / "WorkBuddy" / "assistant_reports"
    report_dir.mkdir(parents=True, exist_ok=True)
    report_file = report_dir / f"report_{datetime.now().strftime('%Y%m%d')}.json"
    with open(report_file, "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    print(f"💾 报告已保存到: {report_file}")


if __name__ == "__main__":
    main()

配套定时任务配置（WorkBuddy 自动化面板 YAML）：

yaml

# workbuddy_schedule.yaml
# 在 WorkBuddy 自动化任务面板中导入此配置

task_name: "每日项目报告推送"
description: "每天 9:00 自动生成本地项目构建报告并推送到企微群"

schedule:
  type: "cron"
  expression: "0 9 * * *"  # 每天上午 9:00 执行
  timezone: "Asia/Shanghai"

runtime:
  type: "python_script"
  script_path: "./workbuddy_task.py"
  working_dir: "~/WorkBuddy/assistant_tasks"  # 助理专属文件夹
  timeout: 300  # 5分钟超时

environment:
  PROJECT_PATH: "/Users/yourname/projects/my-project"
  WECOM_BOT_WEBHOOK: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"

notify:
  on_success: true
  on_failure: true
  failure_alert_channel: "wecom_bot"  # 失败时通过企微机器人告警

---

相关文档

企业微信 + 豆包新模型：API 接口驱动的智能自动回复体系构建_豆包智能体结合企微-CSDN博客

企业微信与DeepSeek大模型结合实现智能消息推送_实现微信消息推送-CSDN博客

• WorkBuddy 助理功能说明

• 企业微信智能机器人长连接官方文档

• 企业微信机器人发送消息 API