钉钉机器人 + 自部署 AI 服务 30 分钟复现

1. 登录 open-dev.dingtalk.com（钉钉开发者后台）→ 应用开发 → 企业内部应用 → 创建应用
2. 填应用名称/描述 → 创建后进入应用详情，左侧「凭证与基础信息」：记录 AppKey 和 AppSecret（AppSecret 只显示一次，关闭后需重置才能再看）
3. 左侧「机器人」→ 添加机器人：填名称、头像，消息接收模式选 Stream 模式（免公网 IP、免回调 URL，SDK 直接长连接）
4. 保存后，确认机器人出现在应用「版本管理与发布」里——企业内部应用要发布后，企业成员才搜得到这个机器人（见文末报错 2）

官方 Python SDK（dingtalk-stream）把 WebSocket 长连接、心跳、断线重连都封装好了。安装并保存下面代码为 bot.py：

pip install dingtalk-stream

import logging
import dingtalk_stream
from dingtalk_stream import AckMessage


def setup_logger():
    logger = logging.getLogger()
    handler = logging.StreamHandler()
    handler.setFormatter(
        logging.Formatter('%(asctime)s %(name)-8s %(levelname)-8s %(message)s'))
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    return logger


class EchoHandler(dingtalk_stream.ChatbotHandler):
    async def process(self, callback: dingtalk_stream.CallbackMessage):
        message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
        content = message.text.content.strip()
        self.logger.info('收到消息: %s' % content)
        self.reply_text('收到：' + content, message)
        return AckMessage.STATUS_OK, 'OK'


def main():
    logger = setup_logger()
    credential = dingtalk_stream.Credential('你的AppKey', '你的AppSecret')
    client = dingtalk_stream.DingTalkStreamClient(credential)
    client.register_callback_handler(
        dingtalk_stream.chatbot.ChatbotMessage.TOPIC, EchoHandler(logger))
    client.start_forever()


if __name__ == '__main__':
    main()

运行 python bot.py，日志出现 连接成功 后，到钉钉群里 @机器人 发一句「你好」，应该收到「收到：你好」。

把 EchoHandler 换成调用大模型 API。以 DeepSeek 为例（官网 platform.deepseek.com 充值后创建 API Key，价格在官网定价页实时可查，本文发布时 deepseek-chat 输出约 2 元/百万 tokens）：

import requests

API_KEY = '你的DeepSeek API Key'


def ask_ai(text):
    resp = requests.post(
        'https://api.deepseek.com/chat/completions',
        headers={'Authorization': 'Bearer ' + API_KEY},
        json={
            'model': 'deepseek-chat',
            'messages': [{'role': 'user', 'content': text}],
        },
        timeout=30)
    return resp.json()['choices'][0]['message']['content']


class AIHandler(dingtalk_stream.ChatbotHandler):
    async def process(self, callback: dingtalk_stream.CallbackMessage):
        message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
        content = message.text.content.strip()
        self.logger.info('收到: %s' % content)
        answer = ask_ai(content)
        self.reply_text(answer, message)
        return AckMessage.STATUS_OK, 'OK'

把第 2 步代码里的 EchoHandler 换成 AIHandler，重启脚本，再 @ 机器人提问，收到的就是模型回答。

1. 钉钉群 → 设置 → 机器人 → 添加机器人 → 选刚发布的应用机器人
2. 群里 @机器人 提问，收到回复即说明全流程可用
3. 想记多轮上下文：把历史消息拼进 messages 数组再发给模型（注意控制长度，超长就只保留最近几轮）
4. @ 机器人时消息文本会带上「@机器人 」前缀，content.strip() 后建议再 replace('@机器人','') 去掉前缀，避免把机器人名字喂给模型

开发机跑通后，把脚本放到服务器，用 systemd 托管（断线自动拉起，SDK 自身也会自动重连）：

sudo tee /etc/systemd/system/dingtalk-bot.service <<'EOF'
[Unit]
Description=DingTalk AI Bot
After=network-online.target

[Service]
WorkingDirectory=/opt/dingbot
ExecStart=/usr/bin/python3 /opt/dingbot/bot.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl daemon-reload
sudo systemctl enable --now dingtalk-bot
sudo journalctl -u dingtalk-bot -f

注意：AppKey/AppSecret 建议从环境变量或密钥文件读，别硬编码进 git 仓库。

1. 一个钉钉企业内部应用（AppKey/AppSecret 在手）
2. 一个 Stream 模式机器人，群里 @ 它就能对话
3. 脚本接了大模型 API，回答有实际内容
4. 服务在服务器上常驻，重启自动拉起

下一步可以做的：把机器人接进 知识库检索（先查内部文档再回答），或者接到 企业审批/工单系统（收到消息后调内部 API 干活）。这类「IM + AI + 内部系统」的串联，正是 Mule Agent 平台擅长的场景——多个 IM 统一接一个 AI 大脑，一个后台管理。