我来为你详细介绍 OpenClaw 与钉钉的对接教程。

前期准备
钉钉开发者账号
- 访问 钉钉开放平台
- 使用企业管理员账号登录
- 创建企业内部应用或第三方企业应用
OpenClaw 环境
# 安装钉钉适配器 pip install openclaw-adapter-dingtalk
钉钉应用配置
创建应用
- 登录钉钉开放平台
- 进入「应用开发」→「企业内部开发」
- 点击「创建应用」选择「H5微应用」
- 填写应用信息:
- 应用名称:OpenClaw机器人
- 应用描述:智能对话机器人
- 应用图标:上传图标
获取关键凭证
# 记录以下信息 AppKey: your_app_key AppSecret: your_app_secret AgentId: your_agent_id CorpId: your_corp_id
配置权限
在应用权限管理中添加:
- 机器人权限
- 消息通知权限
- 通讯录权限(如需获取用户信息)
OpenClaw 配置
配置文件
# config/dingtalk.yaml
dingtalk:
app_key: "your_app_key"
app_secret: "your_app_secret"
agent_id: "your_agent_id"
corp_id: "your_corp_id"
# 消息类型配置
message:
text_enabled: true
markdown_enabled: true
image_enabled: true
file_enabled: true
# 回调配置
callback:
token: "your_callback_token"
aes_key: "your_encrypt_aes_key"
url: "https://your-domain.com/callback"
# 功能配置
features:
group_chat: true # 支持群聊
single_chat: true # 支持单聊
at_all: false # 是否允许@所有人
初始化代码
# dingtalk_bot.py
from openclaw import OpenClaw
from openclaw.adapters.dingtalk import DingTalkAdapter
# 创建 OpenClaw 实例
claw = OpenClaw()
# 配置钉钉适配器
dingtalk_config = {
"app_key": "your_app_key",
"app_secret": "your_app_secret",
"agent_id": "your_agent_id",
"corp_id": "your_corp_id",
"callback_url": "https://your-domain.com/callback"
}
# 添加钉钉适配器
dingtalk_adapter = DingTalkAdapter(config=dingtalk_config)
claw.add_adapter(dingtalk_adapter)
# 启动机器人
claw.start()
Webhook 配置
设置回调地址
在钉钉开放平台:
- 进入应用详情
- 选择「事件与回调」
- 点击「设置回调地址」
- 填写:
请求地址:https://your-domain.com/callback 加签/Token:your_callback_token 加密AES密钥:your_encrypt_aes_key
验证回调
# callback_handler.py
from flask import Flask, request
import hashlib
import time
app = Flask(__name__)
@app.route('/callback', methods=['POST'])
def dingtalk_callback():
# 获取签名
timestamp = request.args.get('timestamp')
nonce = request.args.get('nonce')
signature = request.args.get('signature')
# 验证签名
token = "your_callback_token"
list = [token, timestamp, nonce]
list.sort()
sha1 = hashlib.sha1()
sha1.update("".join(list).encode('utf-8'))
hashcode = sha1.hexdigest()
if hashcode == signature:
# 处理回调消息
return handle_message(request.json)
else:
return "验证失败", 403
消息处理
接收消息
def handle_message(data):
msg_type = data.get('msgtype')
sender_id = data.get('senderId')
content = data.get('text', {}).get('content', '')
# 消息类型判断
if msg_type == 'text':
return process_text_message(sender_id, content)
elif msg_type == 'image':
return process_image_message(sender_id, data.get('image'))
# ... 其他消息类型
return {"msgtype": "text", "text": {"content": "收到消息"}}
发送消息
from openclaw.adapters.dingtalk import DingTalkMessageSender
sender = DingTalkMessageSender(
app_key="your_app_key",
app_secret="your_app_secret"
)
# 发送文本消息
def send_text_message(user_id, content):
message = {
"agent_id": "your_agent_id",
"userid_list": user_id,
"msg": {
"msgtype": "text",
"text": {"content": content}
}
}
return sender.send(message)
# 发送 Markdown 消息
def send_markdown_message(user_id, title, text):
message = {
"agent_id": "your_agent_id",
"userid_list": user_id,
"msg": {
"msgtype": "markdown",
"markdown": {
"title": title,
"text": text
}
}
}
return sender.send(message)
高级功能
消息卡片
def send_action_card(user_id, title, text, buttons):
message = {
"agent_id": "your_agent_id",
"userid_list": user_id,
"msg": {
"msgtype": "action_card",
"action_card": {
"title": title,
"markdown": text,
"btn_orientation": "0",
"btn_json_list": buttons
}
}
}
return sender.send(message)
工作通知
def send_work_notification(user_ids, msg_content):
"""
发送工作通知(无需用户关注应用)
"""
url = "https://oapi.dingtalk.com/topapi/message/corpconversation/asyncsend_v2"
params = {
"access_token": get_access_token(),
"agent_id": agent_id,
"userid_list": ",".join(user_ids),
"msg": json.dumps(msg_content)
}
response = requests.post(url, json=params)
return response.json()
获取用户信息
def get_user_info(user_id):
"""
获取钉钉用户详细信息
"""
url = f"https://oapi.dingtalk.com/user/get?access_token={access_token}&userid={user_id}"
response = requests.get(url)
return response.json()
部署与运维
服务器部署
# 使用 Gunicorn 部署 gunicorn -w 4 -b 0.0.0.0:8000 dingtalk_bot:app # 或使用 supervisor 管理 [program:dingtalk-bot] command=/path/to/venv/bin/gunicorn -w 4 -b 0.0.0.0:8000 dingtalk_bot:app directory=/path/to/your/project user=www-data autostart=true autorestart=true
HTTPS 配置
# Nginx 配置
server {
listen 443 ssl;
server_name your-domain.com;
ssl_certificate /path/to/cert.pem;
ssl_certificate_key /path/to/key.pem;
location /callback {
proxy_pass http://127.0.0.1:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
}
常见问题解决
签名验证失败
- 检查时间戳是否同步
- 确认 Token 配置一致
- 验证加密 AES 密钥
消息发送失败
# 错误代码参考
ERROR_CODES = {
40001: "获取access_token时AppSecret错误,或者access_token无效",
40002: "不合法的凭证类型",
40003: "不合法的UserID",
40004: "不合法的媒体文件类型",
# ... 更多错误码
}
权限不足
- 检查应用权限是否已开通
- 确认管理员已审批
- 验证用户是否在可见范围
安全建议
- Token 管理:使用环境变量存储敏感信息
- IP 白名单:配置钉钉服务器 IP 白名单
- 消息加密:启用消息加密功能
- 频率限制:实现消息频率限制
- 日志审计:记录所有交互日志
示例项目结构
openclaw-dingtalk/
├── config/
│ ├── dingtalk.yaml
│ └── openclaw_config.yaml
├── src/
│ ├── __init__.py
│ ├── main.py
│ ├── callback_handler.py
│ └── message_processor.py
├── requirements.txt
├── docker-compose.yml
└── README.md
需要我详细说明某个具体部分吗?比如回调验证的详细步骤、消息加解密的实现,或者特定业务场景的集成方案?
版权声明:除非特别标注,否则均为本站原创文章,转载时请以链接形式注明文章出处。