釘釘機器人 + 自部署 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 大腦，一個後臺管理。