Skip to main content

钉钉机器人工具包,封装 HTTP 回调签名校验、回复体构造、单聊、群聊及固定 Webhook 主动推送。

Project description

zy-dingtalk-chat-bot (Python)

钉钉机器人工具包,封装 HTTP 回调签名校验、同步回复消息体、单聊发送、群聊发送,以及固定群机器人 Webhook 主动推送。

这是 npm 包 dingtalk-chat-bot 的 Python 版本,API 风格保持一致。

支持范围

这个包当前支持 HTTP 回调方式,不支持 Stream 长连接方式。

钉钉机器人常见有两类接入:

  • HTTP 回调:钉钉把用户消息 POST 到你的公网服务。
  • Stream 长连接:你的程序主动连接钉钉并保持长连接。

本包适合这些场景:

  • 用户私聊机器人后,自动回复或异步发送单聊消息。
  • 群里 @ 机器人后,回复当前群聊。
  • 监控告警、定时日报等任务,通过固定群机器人 Webhook 主动推送到群。

本包暂不包含:

  • Stream 长连接收消息。
  • 业务指令解析。
  • 数据库、任务队列、权限系统。

安装

pip install zy-dingtalk-chat-bot

如果你需要本地运行示例 Flask 服务:

pip install "zy-dingtalk-chat-bot[server]"

创建实例

import os
from dingtalk_chat_bot import create_dingtalk_bot

bot = create_dingtalk_bot(
    app_key=os.environ["DINGTALK_APP_KEY"],
    app_secret=os.environ["DINGTALK_APP_SECRET"],
    robot_code=os.environ.get("DINGTALK_ROBOT_CODE"),
)

配置说明:

  • app_key:钉钉应用的 AppKey。发送单聊消息时需要。
  • app_secret:钉钉应用的 AppSecret。校验 HTTP 回调签名、获取 accessToken 时需要。
  • robot_code:机器人编码。通常可以填 AppKey;如果钉钉后台单独展示 robotCode,就填 robotCode。
  • webhook_secret:可选,自定义机器人「加签」安全策略的密钥(SEC 开头)。设置后 send_webhook_message 会自动按官方算法拼 timestampsign。详见下方「自定义机器人加签」一节。
  • http_client:可选,自定义 requests.Session
  • signature_expires_in:可选,HTTP 回调签名时间窗口,单位毫秒,默认 1 小时。
  • request_timeout:可选,HTTP 请求超时时间,单位秒,默认 10。

如果你只使用固定群机器人 Webhook 推送,可以不传 app_keyapp_secret

bot = create_dingtalk_bot()

bot.send_webhook_message(os.environ["DINGTALK_WEBHOOK_URL"], "服务正常")

环境变量

仓库提供了 .env.example 作为模板。本地运行示例服务时,可以自己创建 .env

cp .env.example .env

示例:

PORT=3000
DINGTALK_APP_KEY=
DINGTALK_APP_SECRET=
DINGTALK_ROBOT_CODE=
DINGTALK_WEBHOOK_URL=

.env 不应该提交到 Git,也不会进入发布包。

HTTP 回调签名校验

当钉钉把消息 POST 到你的服务时,可以这样校验签名(以 Flask 为例):

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.post("/dingtalk/robot")
def dingtalk_robot():
    ok = bot.verify_signature(
        request.headers.get("timestamp"),
        request.headers.get("sign"),
    )
    if not ok:
        return jsonify({"error": "invalid dingtalk signature"}), 401

    return jsonify(bot.build_reply_payload("收到", request.get_json()))

同步回复

build_reply_payload 用来构造可直接返回给钉钉 HTTP 回调的消息体。

普通文本:

return jsonify(bot.build_reply_payload("你好,我收到了", request.get_json()))

Markdown:

return jsonify(bot.build_reply_payload({
    "msgtype": "markdown",
    "title": "处理结果",
    "content": "## 处理结果\n\n- 状态:成功\n- 耗时:120ms",
}, request.get_json()))

如果是群聊回调,并且消息体里有 senderStaffId,回复会默认在第一行真实 @ 提问人。

自动发送单聊或群聊

send_message 适合在收到钉钉回调后使用。它会根据回调消息体里的 conversationType 自动判断发送方式:

  • conversationType == "1":通过 OpenAPI 发送单聊消息。
  • conversationType == "2":通过回调里的 sessionWebhook 发送群聊消息。
bot.send_message(body, "这条消息会自动发到当前单聊或群聊")

群聊发送时默认会 @ 本次提问人。如果不想 @:

bot.send_message(body, "这条群消息不艾特任何人", {"atSender": False})

发送 Markdown:

bot.send_message(body, {
    "msgtype": "markdown",
    "title": "日报",
    "content": "## 今日日报\n\n- 完成机器人回复\n- 支持单聊和群聊",
})

主动发送单聊

bot.send_private_message("用户 staffId", "你好")

多个用户:

bot.send_private_message(["staffId1", "staffId2"], "批量单聊消息")

Markdown 单聊:

bot.send_private_message("用户 staffId", {
    "msgtype": "markdown",
    "title": "通知",
    "content": "## 通知\n\n请查看最新处理结果。",
})

通过 sessionWebhook 发群聊

sessionWebhook 来自钉钉 HTTP 回调消息体,适合在用户 @ 机器人之后回复当前群聊。

bot.send_group_message(body["sessionWebhook"], "群聊回复")

指定 @ 用户:

bot.send_group_message(body["sessionWebhook"], "请关注这条消息", {
    "atUserIds": ["staffId1"],
})

Markdown 群聊:

bot.send_group_message(body["sessionWebhook"], {
    "msgtype": "markdown",
    "title": "群聊通知",
    "content": "## 群聊通知\n\n- 已处理完成",
})

固定群机器人 Webhook 推送

如果你在群机器人设置里复制到了固定 Webhook:

https://oapi.dingtalk.com/robot/send?access_token=xxx

可以使用 send_webhook_message 主动推送消息,适合监控告警、定时日报、定时巡检等场景,不需要等用户先私聊或 @ 机器人。

普通文本:

bot.send_webhook_message(
    os.environ["DINGTALK_WEBHOOK_URL"],
    "监控告警:订单 API 响应超时",
)

Markdown:

bot.send_webhook_message(os.environ["DINGTALK_WEBHOOK_URL"], {
    "msgtype": "markdown",
    "title": "监控告警",
    "content": "## 监控告警\n\n- 服务:订单 API\n- 状态:响应超时\n- 请及时处理",
})

@ 指定用户:

bot.send_webhook_message(
    os.environ["DINGTALK_WEBHOOK_URL"],
    "请关注这条告警",
    {"atUserIds": ["staffId1"]},
)

@ 所有人:

bot.send_webhook_message(
    os.environ["DINGTALK_WEBHOOK_URL"],
    "重要告警,请所有人关注",
    {"isAtAll": True},
)

如果开启了关键词安全设置,消息内容必须包含对应关键词,否则会被钉钉拒绝。

自定义机器人加签

如果群机器人安全设置勾选了「加签」,调用 webhook 时必须在 URL 上拼 timestampsign,本包已内置处理。

构造时传入 webhook_secret,后续所有 send_webhook_message 调用都会自动加签:

bot = create_dingtalk_bot(webhook_secret=os.environ["DINGTALK_ROBOT_SECRET"])
bot.send_webhook_message(os.environ["DINGTALK_WEBHOOK_URL"], "已加签")

也可以在单次调用时通过 secret 参数覆盖(优先级高于构造时的 webhook_secret):

bot.send_webhook_message(webhook_url, "临时加签", secret="SECxxxxxx")

如果只想拿到签好的 URL 自己用,可以调用静态方法:

from dingtalk_chat_bot import DingTalkBot

signed = DingTalkBot.sign_webhook_url(webhook_url, "SECxxxxxx")

webhook_secret / secret 留空时不加签,方便同一份代码同时兼容启用 / 未启用加签的机器人。

send_group_message 走的是钉钉 HTTP 回调里的 sessionWebhook(临时地址,自带 token),不需要也不应该加签。

示例消息与文本路由 (dingtalk_chat_bot.examples)

开箱即用的菜单 / Markdown / 时间回复 / 默认路由,免去自己再写一份样板代码。

from dingtalk_chat_bot.examples import (
    build_menu,            # → markdown 菜单
    build_markdown_demo,   # → markdown 示例
    build_text_demo,       # → 文本示例
    build_time_reply,      # → 当前服务器时间
    handle_user_message,   # → 默认文本路由
)

bot.send_webhook_message(WEBHOOK_URL, build_menu())

handle_user_message(text) 把用户文本映射到合适的示例回复,可直接接到 HTTP 回调里:

用户发送 回复内容
菜单 / help / /help / 帮助 菜单 markdown
文本 / text 文本示例
markdown / md / 示例markdown Markdown 示例
时间 / time 当前服务器时间
空字符串 提示重新发送
其他 回声 + 菜单引导

接入示例:

from dingtalk_chat_bot.examples import handle_user_message

@app.post("/dingtalk/robot")
def robot():
    body = request.get_json(silent=True) or {}
    text = (body.get("text") or {}).get("content", "").strip()
    return jsonify(bot.build_reply_payload(handle_user_message(text), body))

示例服务

仓库里的 server.py 是一个 Flask 示例,不会进入发布包。你可以本地运行它测试 HTTP 回调:

pip install "zy-dingtalk-chat-bot[server]"
cp .env.example .env
python server.py

默认接口:

  • POST /dingtalk/robot:同步回复示例。
  • POST /dingtalk/robot-async:先同步确认,再异步发送消息示例。
  • GET /health:健康检查。

发布说明

pyproject.toml[tool.setuptools] 字段只显式包含 dingtalk_chat_bot 包目录。因此 .envserver.pyrequirements-dev.txt.venv 等都不会进入发布包。

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

zy_dingtalk_chat_bot-1.1.1.tar.gz (12.7 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

zy_dingtalk_chat_bot-1.1.1-py3-none-any.whl (10.5 kB view details)

Uploaded Python 3

File details

Details for the file zy_dingtalk_chat_bot-1.1.1.tar.gz.

File metadata

  • Download URL: zy_dingtalk_chat_bot-1.1.1.tar.gz
  • Upload date:
  • Size: 12.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.10.11

File hashes

Hashes for zy_dingtalk_chat_bot-1.1.1.tar.gz
Algorithm Hash digest
SHA256 0437240f7596d613b038dd34774d194a40e1e8b374ed09c9024aaebe568b0777
MD5 d74672d98eb2c4976793ed9b6c5986a7
BLAKE2b-256 dd805be4b9735213c3a02fd29fd48480e57bf259ac3c3b5de16a9e3b22ee39b6

See more details on using hashes here.

File details

Details for the file zy_dingtalk_chat_bot-1.1.1-py3-none-any.whl.

File metadata

File hashes

Hashes for zy_dingtalk_chat_bot-1.1.1-py3-none-any.whl
Algorithm Hash digest
SHA256 587116e8afa9d72c927aa21481dd5d8fb9250e36bbba8979a3060d0273bb0fc3
MD5 e6ea971c1fe822f8a825d74c75c9c37b
BLAKE2b-256 af3b52f79c94104c6dd07c0e8e835ff6f9ac1c95c09de342aded87031f03a7b3

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page