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"×tamp={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. 关键注意点:配置信息准确、接口权限开启、异常处理完善,可适配不同自动化场景的消息通知需求。
评论区