dingtalk-openclaw — dingtalk-OpenClaw

钉钉连接 OpenClaw

本 技能 指导你将钉钉机器人连接到 OpenClaw，实现通过钉钉与 AI 对话。

架构 钉钉消息 → dingtalk_流 → OpenClaw API → AI回复 → 钉钉网页hook → 用户

前置条件 钉钉企业应用（创建应用并开通 流 模式） OpenClaw 已运行并启用 HTTP API Python 3.8+ 配置步骤

• 安装依赖

• 创建配置文件 config.py

# 钉钉应用凭证（从钉钉开放平台获取） 代理_ID = "你的代理Id" 应用_KEY = "你的应用Key" 应用_SECRET = "你的应用Secret"

# 机器人消息回调Topic 机器人_MESSAGE_TOPIC = "/v1.0/im/机器人/messages/获取"

# 日志配置 记录_LEVEL = "信息" 记录_格式化 = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" 记录_FILE = "记录s/dingtalk_流.记录"

• 创建主程序 mAIn.py

导入 sys 导入 记录ging 导入 请求s from pathlib 导入 Path

try: 导入 dingtalk_流 from dingtalk_流 导入 AckMessage except 导入Error: print("请先安装: pip 安装 dingtalk-流") sys.exit(1)

导入 config

# ============ 配置区域（根据你的环境修改）============ # OpenClaw API 地址 OpenClaw_URL = "http://127.0.0.1:18789/v1/响应s" # OpenClaw Gateway 令牌（在 OpenClaw.json 中配置） OpenClaw_令牌 = "你的OpenClaw_令牌"

# 钉钉 网页hook（用于发送回复） # 创建机器人后获取：群设置 → 智能群助手 → 添加机器人 → 自定义机器人 网页HOOK_URL = "你的钉钉网页hook_URL" # ==================================================

PROCESSED_FILE = Path("C:/Users/MyPC/.OpenClaw/workspace/dingtalk_processed.txt") PROCESSED_IDS = 设置()

def load_processed(): global PROCESSED_IDS if PROCESSED_FILE.exists(): with open(PROCESSED_FILE, "r", encoding="utf-8") as f: PROCESSED_IDS = 设置(line.strip() for line in f if line.strip())

def save_processed(msg_id): with open(PROCESSED_FILE, "a", encoding="utf-8") as f: f.write(msg_id + "\n") PROCESSED_IDS.添加(msg_id)

# 调用 OpenClaw AI def 获取_OpenClaw_响应(user_msg: str) -> str: try: headers = { "Authorization": f"Bearer {OpenClaw_令牌}", "Content-Type": "应用/json" } data = { "模型": "OpenClaw", "输入": user_msg } r = 请求s.post(OpenClaw_URL, headers=headers, json=data, timeout=60) if r.状态_code == 200: 结果 = r.json() 输出 = 结果.获取("输出", []) if 输出: content = 输出[0].获取("content", []) if content: return content[0].获取("text", "抱歉，我无法回答这个问题。") 记录ging.error(f"OpenClaw API 错误: {r.状态_code} {r.text}") return "抱歉，我现在有点忙，请稍后再试。" except 异常 as e: 记录ging.error(f"调用 OpenClaw 失败: {e}") return "抱歉，连接出现问题，请稍后再试。"

# 发送回复到钉钉 def 发送_reply(content): data = { "msgtype": "text", "text": {"content": content}, "at": {"atUserIds": [], "isAtAll": False} } try: r = 请求s.post(网页HOOK_URL, json=data, timeout=10) 结果 = r.json() 记录ging.信息(f"回复结果: {结果}") return 结果.获取("errcode", 1) == 0 except 异常 as e: 记录ging.error(f"发送失败: {e}") return False

# 消息处理器 class 处理器(dingtalk_流.Chat机器人处理器): def __init__(self): super().__init__() a同步 def process(self, callback): try: msg = dingtalk_流.Chat机器人Message.from_dict(callback.data) content = msg.text.content if msg.text else "" if not content: return AckMessage.状态_OK, 'OK' msg_id = f"{msg.conversation_id}_{msg.发送er_id}_{len(content)}" if msg_id in PROCESSED_IDS: 记录ging.信息(f"重复消息，跳过: {content[:20]}") return AckMessage.状态_OK, 'OK' save_processed(msg_id) 记录ging.信息(f"收到: {content}") # 获取 AI 回复 响应 = 获取_OpenClaw_响应(content) 记录ging.信息(f"回复: {响应}") # 发送到钉钉 发送_reply(响应) return AckMessage.状态_OK, 'OK' except 异常 as e: 记录ging.error(f"处理失败: {e}") return AckMessage.状态_ERROR, str(e)

def mAIn(): print("=" 50) print(" 钉钉 + OpenClaw AI 对话机器人") print("=" 50) load_processed() 记录ging.信息(f"已加载 {len(PROCESSED_IDS)} 条已处理消息") 凭证 = dingtalk_流.凭证(config.应用_KEY, config.应用_SECRET) 命令行工具ent = dingtalk_流.DingTalk流命令行工具ent(凭证) 命令行工具ent.register_callback_处理器(dingtalk_流.Chat机器人Message.TOPIC, 处理器()) 记录ging.信息("正在连接钉钉...") 命令行工具ent.启动_forever()

if __name__ == "__mAIn__": mAIn()

• 启用 OpenClaw HTTP API

编辑 C:\Users\MyPC\.OpenClaw\OpenClaw.json，添加：

{ "gateway": { "http": { "端点s": { "响应s": { "enabled": true } } } } }

重启 OpenClaw Gateway 后生效。

• 运行

注意事项 应用_KEY 和 应用_SECRET 在钉钉开放平台的企业应用详情页获取 网页HOOK_URL 创建自定义机器人后获取，注意安全设置（加签） OpenClaw_令牌 在 OpenClaw.js