← Home Blog 中文版

DingTalk Bot + Self-hosted AI Service in 30 Minutes

From creating an internal app to getting your first AI reply in a DingTalk group (Stream Mode, no public IP needed, 4 error fixes included)

2026-08-16 · Leo · Tutorial · 15 min read
TL;DR — A 30-minute checklist: DingTalk developer console → create an internal app → add a bot with Stream Mode → use the official dingtalk-stream SDK to receive and reply to messages → plug in an LLM API → @ the bot in a group and get an AI answer. No public IP, no callback URL. It is a full order of magnitude simpler than WeCom's AES callback setup. 4 common errors and fixes at the end.
⚠️ The biggest trap first: a DingTalk "custom robot" webhook can only send messages — it cannot receive what employees send. For a conversational AI assistant you need an internal app + bot, and the bot's message-receiving mode must be Stream Mode. Everything below follows that path.

Step 1: Create an internal app + bot

📌 Get AppKey / AppSecret, add a bot with Stream Mode 5 min
  1. Log in to open-dev.dingtalk.com (DingTalk developer console) → App Development → Enterprise Internal App → Create App
  2. Fill in name/description → open the app detail page → "Credentials & Basic Info" on the left: note down AppKey and AppSecret (AppSecret is shown only once; you must reset it to view it again)
  3. Left menu "Bot" → Add Bot: set name and avatar, and set Message Receiving Mode to Stream Mode (no public IP, no callback URL — the SDK keeps a long-lived connection)
  4. After saving, make sure the bot is in the app's "Version Management & Release" — an internal app must be released before employees can find the bot (see error 2 below)

Step 2: Get the minimal receiving script running

🐍 dingtalk-stream SDK: echo back any message 8 min

The official Python SDK (dingtalk-stream) wraps the WebSocket connection, heartbeat and auto-reconnect. Install it and save the code below as 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('Received: %s' % content)
        self.reply_text('Echo: ' + content, message)
        return AckMessage.STATUS_OK, 'OK'


def main():
    logger = setup_logger()
    credential = dingtalk_stream.Credential('YOUR_APP_KEY', 'YOUR_APP_SECRET')
    client = dingtalk_stream.DingTalkStreamClient(credential)
    client.register_callback_handler(
        dingtalk_stream.chatbot.ChatbotMessage.TOPIC, EchoHandler(logger))
    client.start_forever()


if __name__ == '__main__':
    main()

Run python bot.py. Once the log shows the connection is up, @ the bot in a DingTalk group with "hi" — you should get "Echo: hi".

Step 3: Forward messages to an LLM

🤖 Plug in DeepSeek API (OpenAI-compatible) 10 min

Replace EchoHandler with a handler that calls an LLM API. Using DeepSeek as an example (create an API key at platform.deepseek.com after topping up; live prices are on their pricing page — at the time of writing, deepseek-chat output is about ¥2 per million tokens):

import requests

API_KEY = 'YOUR_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('Received: %s' % content)
        answer = ask_ai(content)
        self.reply_text(answer, message)
        return AckMessage.STATUS_OK, 'OK'

Swap EchoHandler for AIHandler in the Step 2 code, restart the script, and @ the bot again — the reply is now the model's answer.

Step 4: Test the conversation in a group

💬 @ the bot: from single turns to multi-turn 3 min
  1. DingTalk group → Settings → Bots → Add Bot → pick the released app bot
  2. @ the bot in the group and ask a question — a reply means the pipeline works end to end
  3. For multi-turn context: append the history to the messages array before sending to the model (keep it short — if too long, keep only the most recent turns)
  4. When you @ the bot, the message text carries an "@bot " prefix. After content.strip(), also do replace('@bot','') so the bot's name is not fed to the model

Step 5: Deploy it on a server

🚀 Run it with nohup / systemd 2 min

Once it works on your dev machine, move the script to a server and let systemd manage it (auto-restart on crash; the SDK also reconnects by itself):

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

Note: read AppKey/AppSecret from environment variables or a secrets file — do not hardcode them into a git repo.

30 minutes later, you should have

  1. A DingTalk internal app (AppKey/AppSecret in hand)
  2. A Stream Mode bot you can talk to in a group
  3. The script wired to an LLM API, with real answers
  4. The service running on a server with auto-restart

Next steps: connect the bot to a knowledge base (retrieve internal docs before answering), or to approval/ticket systems (call internal APIs when a message arrives). Connecting "IM + AI + internal systems" like this is exactly what the Mule Agent platform does — multiple IMs wired to one AI brain, managed from one console.

4 common errors and fixes

Error 1: Cannot connect, log says Invalid client_id or client_secret

Wrong AppKey/AppSecret, or extra spaces/newlines were copied along.

✅ Fix: copy again from "Credentials & Basic Info"; make sure it is the internal app's AppKey, not a custom bot's webhook access_token. If you reset AppSecret, update the code too.
Error 2: The bot is not findable / cannot be @'d in the group

The app was never released, or the bot was not added to the current group.

✅ Fix: developer console → App Release → Version Management & Release → create a version and release it to the org; in the group: Settings → Bots → Add Bot. Both steps are required.
Error 3: Connection is up but no messages arrive

The bot's message-receiving mode was set to HTTP callback (which needs a public URL), or receiving was never configured.

✅ Fix: on the bot config page switch Message Receiving Mode to Stream Mode and save; confirm the registered callback topic is ChatbotMessage.TOPIC (note the 'o' in Chatbot).
Error 4: Reply fails, log reports a reply exception

reply_text must be called inside the process callback — DingTalk's reply context only exists within that callback. An LLM API timeout also prevents a reply.

✅ Fix: set the ask_ai timeout to 30+ seconds; if the model takes long, reply "processing…" first and send the real answer asynchronously afterwards (Stream Mode supports proactive messaging).
💡 Skip the self-hosting step? The Mule Agent platform natively supports DingTalk, WeCom, Feishu and 7 major IM platforms: one AI brain wired to all of them, messages routed to knowledge bases/workflows, configured from a console without code. Leave your email on the site for a demo.
📮 Questions? Email 278946228@qq.com or leave a message on the blog guestbook.