侧边栏壁纸
博主头像
码农小札

行动起来,活在当下

  • 累计撰写 6 篇文章
  • 累计创建 3 个标签
  • 累计收到 0 条评论

目 录CONTENT

文章目录

OpenClaw 接入主流通讯工具(微信/企业微信/钉钉)完整教程

OpenClaw 作为轻量级的自动化框架,接入主流通讯工具能极大扩展其应用场景(如消息推送、告警通知、业务审批自动提醒等)。本文将手把手教你实现 OpenClaw 对接**微信公众号**、**企业微信**、**钉钉**三大主流通讯工具,代码可直接复用,适合自动化开发新手和运维人员。

## 前置准备

1. **环境要求**:Python 3.7+(OpenClaw 核心依赖)、pip 包管理工具

2. **安装核心依赖**:

```bash

pip install openclaw requests pycryptodome # requests用于接口调用,pycryptodome用于企业微信/钉钉加解密

```

3. **账号准备**:

- 微信公众号:需注册公众号(测试号即可,地址:https://mp.weixin.qq.com/debug/cgi-bin/sandbox?t=sandbox/login)

- 企业微信:拥有企业微信账号,创建应用并获取 CorpID、Secret

- 钉钉:创建钉钉自定义机器人,获取 Webhook 地址

## 一、OpenClaw 对接微信公众号(消息推送)

### 1.1 获取微信公众号接口凭证

1. 登录微信公众号测试号,获取 `APPID` 和 `APPSECRET`;

2. 新增测试用户(绑定个人微信),获取用户 `openid`(测试号页面可直接查看)。

### 1.2 编写 OpenClaw 对接代码

创建 `openclaw_wechat.py`,核心逻辑是通过 OpenClaw 封装微信公众号接口调用:

```python

import openclaw

import requests

import time

# 微信公众号配置

WECHAT_CONFIG = {

"appid": "你的测试号APPID",

"appsecret": "你的测试号APPSECRET",

"openid": "接收消息的用户openid"

}

class WeChatNotifier(openclaw.BaseTask):

"""OpenClaw 微信公众号消息推送任务"""

def get_access_token(self):

"""获取微信接口访问凭证"""

url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={WECHAT_CONFIG['appid']}&secret={WECHAT_CONFIG['appsecret']}"

try:

res = requests.get(url, timeout=10)

res.raise_for_status() # 抛出HTTP异常

token_data = res.json()

if "access_token" in token_data:

return token_data["access_token"]

else:

self.logger.error(f"获取凭证失败:{token_data}")

return None

except Exception as e:

self.logger.error(f"获取access_token异常:{str(e)}")

return None

def send_text_msg(self, content):

"""发送文本消息"""

token = self.get_access_token()

if not token:

return False

url = f"https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token={token}"

data = {

"touser": WECHAT_CONFIG["openid"],

"msgtype": "text",

"text": {

"content": content

}

}

try:

res = requests.post(url, json=data, timeout=10)

res_data = res.json()

if res_data["errcode"] == 0:

self.logger.info("微信消息发送成功")

return True

else:

self.logger.error(f"消息发送失败:{res_data}")

return False

except Exception as e:

self.logger.error(f"发送消息异常:{str(e)}")

return False

def run(self):

"""OpenClaw 任务执行入口"""

# 业务逻辑:这里可替换为你的自动化场景(如监控告警、任务完成通知)

self.send_text_msg(f"【OpenClaw 通知】{time.strftime('%Y-%m-%d %H:%M:%S')} 自动化任务执行完成!")

# 运行 OpenClaw 任务

if __name__ == "__main__":

task = WeChatNotifier()

openclaw.run_task(task)

```

### 1.3 测试验证

1. 替换代码中的 `APPID`、`APPSECRET`、`openid`;

2. 运行脚本,个人微信会收到公众号推送的消息,说明对接成功。

## 二、OpenClaw 对接企业微信(应用消息推送)

企业微信更适合内部办公场景,支持向员工批量推送消息,步骤如下:

### 2.1 获取企业微信配置信息

1. 登录企业微信管理后台(https://work.weixin.qq.com/);

2. 进入「应用管理」→ 「创建应用」,获取 `CorpID`、`Secret`(应用详情页);

3. 记录接收消息的员工 `userid`(通讯录→成员详情)。

### 2.2 对接代码实现

创建 `openclaw_wework.py`:

```python

import openclaw

import requests

import time

# 企业微信配置

WEWORK_CONFIG = {

"corpid": "你的企业CorpID",

"corpsecret": "你的应用Secret",

"agentid": 1000002, # 应用ID(数字)

"userid": "接收消息的员工userid"

}

class WeWorkNotifier(openclaw.BaseTask):

"""OpenClaw 企业微信消息推送任务"""

def get_access_token(self):

"""获取企业微信接口凭证"""

url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={WEWORK_CONFIG['corpid']}&corpsecret={WEWORK_CONFIG['corpsecret']}"

try:

res = requests.get(url, timeout=10)

res_data = res.json()

if res_data["errcode"] == 0:

return res_data["access_token"]

else:

self.logger.error(f"获取凭证失败:{res_data}")

return None

except Exception as e:

self.logger.error(f"获取access_token异常:{str(e)}")

return None

def send_text_msg(self, content):

"""发送企业微信文本消息"""

token = self.get_access_token()

if not token:

return False

url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"

data = {

"touser": WEWORK_CONFIG["userid"],

"msgtype": "text",

"agentid": WEWORK_CONFIG["agentid"],

"text": {

"content": content

},

"safe": 0 # 0表示普通消息,1表示加密消息

}

try:

res = requests.post(url, json=data, timeout=10)

res_data = res.json()

if res_data["errcode"] == 0:

self.logger.info("企业微信消息发送成功")

return True

else:

self.logger.error(f"消息发送失败:{res_data}")

return False

except Exception as e:

self.logger.error(f"发送消息异常:{str(e)}")

return False

def run(self):

"""任务执行入口"""

self.send_text_msg(f"【OpenClaw 企业微信通知】{time.strftime('%Y-%m-%d %H:%M:%S')} 自动化任务执行完成!")

if __name__ == "__main__":

task = WeWorkNotifier()

openclaw.run_task(task)

```

## 三、OpenClaw 对接钉钉(自定义机器人)

钉钉机器人是最简单的对接方式,无需复杂鉴权,适合快速集成:

### 3.1 创建钉钉自定义机器人

1. 打开钉钉群聊 → 群设置 → 智能群助手 → 添加机器人 → 自定义机器人;

2. 输入机器人名称,勾选「自定义关键词」(如「OpenClaw」),复制 Webhook 地址;

3. (可选)开启加签,记录密钥(提高安全性)。

### 3.2 对接代码实现

创建 `openclaw_dingtalk.py`:

```python

import openclaw

import requests

import time

import hmac

import hashlib

import base64

import urllib.parse

# 钉钉配置

DINGTALK_CONFIG = {

"webhook": "你的钉钉机器人Webhook地址",

"secret": "加签密钥(无则留空)"

}

class DingTalkNotifier(openclaw.BaseTask):

"""OpenClaw 钉钉消息推送任务"""

def get_sign(self):

"""生成钉钉加签(可选)"""

if not DINGTALK_CONFIG["secret"]:

return ""

timestamp = str(round(time.time() * 1000))

secret_enc = DINGTALK_CONFIG["secret"].encode('utf-8')

string_to_sign = f"{timestamp}\n{DINGTALK_CONFIG['secret']}"

string_to_sign_enc = string_to_sign.encode('utf-8')

hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()

sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))

return f"&timestamp={timestamp}&sign={sign}"

def send_text_msg(self, content):

"""发送钉钉文本消息"""

sign = self.get_sign()

url = f"{DINGTALK_CONFIG['webhook']}{sign}"

data = {

"msgtype": "text",

"text": {

"content": content

}

}

try:

res = requests.post(url, json=data, timeout=10)

res_data = res.json()

if res_data["errcode"] == 0:

self.logger.info("钉钉消息发送成功")

return True

else:

self.logger.error(f"消息发送失败:{res_data}")

return False

except Exception as e:

self.logger.error(f"发送消息异常:{str(e)}")

return False

def run(self):

"""任务执行入口"""

self.send_text_msg(f"【OpenClaw 钉钉通知】{time.strftime('%Y-%m-%d %H:%M:%S')} 自动化任务执行完成!")

if __name__ == "__main__":

task = DingTalkNotifier()

openclaw.run_task(task)

```

## 四、进阶优化建议

1. **凭证缓存**:微信/企业微信的 `access_token` 有效期为2小时,可缓存到本地文件/Redis,避免频繁调用接口;

2. **异常重试**:添加重试机制(如 `tenacity` 库),处理网络波动导致的接口调用失败;

3. **消息类型扩展**:支持图片、markdown、卡片消息(参考各平台官方文档);

4. **集成到 OpenClaw 工作流**:将消息推送作为任务节点,嵌入到复杂的自动化流程中。

## 五、常见问题排查

1. **接口调用失败**:检查配置信息(CorpID/Secret/Webhook)是否正确,网络是否能访问公网;

2. **微信公众号消息发送失败**:测试号的接口权限是否开启,用户是否已关注测试号;

3. **钉钉消息无推送**:确保消息包含自定义关键词,加签参数是否正确。

---

### 总结

1. OpenClaw 对接主流通讯工具的核心是**封装各平台的 HTTP 接口**,通过 `openclaw.BaseTask` 实现任务化调用;

2. 微信公众号/企业微信需先获取 `access_token`,钉钉直接通过 Webhook 推送更简单;

3. 关键注意点:配置信息准确、接口权限开启、异常处理完善,可适配不同自动化场景的消息通知需求。

0

评论区