企业微信二次开发做机器人：如何让企业微信自动执行重复任务

E云管家

4人浏览 · 2026-09-18 14:07:55

E云管家 · 2026-09-18 14:07:55 发布

销售每天要干的重复活不少：新客户加好友发欢迎语、跟进提醒、群发消息、同步客户资料、整理群聊记录。这些动作占了工作时间的 30-40%，但价值不高。用企微 API 做个机器人，把这些重复任务自动化，销售就能把时间花在真正有价值的跟进上。这篇讲重复任务自动化的设计。

一、什么任务适合自动化

不是所有任务都该自动化。适合自动化的任务有几个特征：

重复性高：每天/每周都做，步骤固定
判断简单：不需要复杂决策，按规则执行
出错代价低：即使执行错了，影响不大，可人工修正

不适合自动化的：需要复杂判断的（客户谈判策略）、高风险的（退款审核）、需要情感沟通的（安抚投诉客户）。

适合自动化的典型任务：

任务	触发条件	执行动作
新客户欢迎	好友通过	发欢迎语+打标签
跟进提醒	客户 N 天未互动	提醒销售
消息归档	收到消息	存进数据库
群公告更新	每周一	发本周重点到部门群
客户资料同步	资料变更回调	更新 CRM
离职清理	员工离职	客户转交、退群

二、任务的结构化定义

自动化任务要结构化定义，别写成硬编码的 if-else。一个任务包含：

class AutomationTask:
    name: str           # 任务名称
    trigger: Trigger    # 触发条件
    actions: list[Action]  # 执行动作列表
    condition: Condition  # 前置条件（可选）

触发条件分两类：

事件触发：企微侧回调事件驱动。比如好友通过触发欢迎语、资料变更触发同步
时间触发：定时任务驱动。比如每周一更新群公告、每天 9 点发跟进提醒

执行动作是对企微 API 的调用序列。一个任务可以有多个动作，按顺序执行。

TASKS = [
    {
        "name": "新客户欢迎",
        "trigger": {"type": "event", "event": "friend.added"},
        "actions": [
            {"api": "contact/updateLabel", "params": {"labels": ["新客户"]}},
            {"api": "message/sendText", "params": {"template": "tpl_welcome"}},
        ],
    },
    {
        "name": "周一群公告",
        "trigger": {"type": "cron", "schedule": "0 9 * * 1"},
        "actions": [
            {"api": "room/setNotice", "params": {"room_id": "dept_group", "template": "tpl_weekly"}},
        ],
    },
]

结构化定义的好处是：任务可以在后台配置，不用改代码。运营或管理员自己就能加任务、改任务。

三、事件触发型任务

事件触发型任务靠回调驱动。收到事件后，匹配对应的任务执行：

def on_callback_event(event, body):
    # 匹配该事件的所有任务
    matched = [t for t in TASKS if t.trigger.type == "event" and t.trigger.event == event]
    for task in matched:
        execute_task(task, body)

def execute_task(task, context):
    # 检查前置条件
    if not task.condition.check(context):
        return
    # 顺序执行动作
    for action in task.actions:
        result = api.call(action.api, action.params, context)
        if not result.success:
            log_failure(task, action, result)
            break  # 一个动作失败，后续不执行

事件触发的关键是幂等。同一个事件可能推送多次（回调重推），任务执行必须幂等：打标签重复打没影响、发消息要防重发。防重发靠消息内容指纹——同一个任务对同一个客户发过相同内容，不再重发。

四、时间触发型任务

时间触发型任务靠定时调度器。推荐用 APScheduler 或 Celery Beat 这类成熟的调度库，别自己写 while True 轮询。

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

def register_cron_tasks():
    for task in TASKS:
        if task.trigger.type == "cron":
            scheduler.add_job(
                execute_task,
                trigger="cron",
                **task.trigger.schedule,
                args=[task, {}],
                id=task.name,
            )

时间触发的任务要注意执行窗口。比如"每天 9 点提醒跟进"，如果调度服务在 9 点正好重启，任务可能漏执行。解决方案：任务有"最迟执行时间"，服务启动时检查有没有错过的任务，补执行。

五、任务编排和依赖

有些任务不是独立的，有依赖关系。比如"客户回访"任务依赖"客户标签更新"任务先跑完。

依赖管理的做法：任务有 depends_on 字段，调度时先执行被依赖的任务。或者用事件驱动——任务 A 执行完发一个内部事件，任务 B 监听这个事件再执行。

TASKS = [
    {"name": "标签更新", "trigger": {"type": "cron", "schedule": "0 8 * * *"}},
    {"name": "客户回访", "trigger": {"type": "event", "event": "task.label_update.done"}},
]

事件驱动比显式依赖灵活，加新任务不用改老任务的依赖关系。