安装 OpenClaw

openclaw openclaw官方 2

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

安装 OpenClaw-第1张图片-OpenClaw开源下载|官方OpenClaw下载

前期准备

钉钉开发者账号

  • 访问 钉钉开放平台
  • 使用企业管理员账号登录
  • 创建企业内部应用或第三方企业应用

OpenClaw 环境

# 安装钉钉适配器
pip install openclaw-adapter-dingtalk

钉钉应用配置

创建应用

  1. 登录钉钉开放平台
  2. 进入「应用开发」→「企业内部开发」
  3. 点击「创建应用」选择「H5微应用」
  4. 填写应用信息:
    • 应用名称: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 配置

设置回调地址

在钉钉开放平台:

  1. 进入应用详情
  2. 选择「事件与回调」
  3. 点击「设置回调地址」
  4. 填写:
    请求地址: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: "不合法的媒体文件类型",
    # ... 更多错误码
}

权限不足

  • 检查应用权限是否已开通
  • 确认管理员已审批
  • 验证用户是否在可见范围

安全建议

  1. Token 管理:使用环境变量存储敏感信息
  2. IP 白名单:配置钉钉服务器 IP 白名单
  3. 消息加密:启用消息加密功能
  4. 频率限制:实现消息频率限制
  5. 日志审计:记录所有交互日志

示例项目结构

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

需要我详细说明某个具体部分吗?比如回调验证的详细步骤、消息加解密的实现,或者特定业务场景的集成方案?

标签: 安装 OpenClaw

抱歉,评论功能暂时关闭!