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。
  • 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 上;如果开启了关键词安全设置,消息内容必须包含对应关键词。

示例服务

仓库里的 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.0.tar.gz (11.3 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.0-py3-none-any.whl (9.5 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: zy_dingtalk_chat_bot-1.1.0.tar.gz
  • Upload date:
  • Size: 11.3 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.0.tar.gz
Algorithm Hash digest
SHA256 a28bed11823f82dde9e12fbba2a0bc1fd1a5ca1241bf9f968c02f9604ae4866b
MD5 8897102d0d1af47570da85a9780fc092
BLAKE2b-256 9f7eb17f2fd4a422b6ce0075802042a8718399ae177923b00a1093c640cae323

See more details on using hashes here.

File details

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

File metadata

File hashes

Hashes for zy_dingtalk_chat_bot-1.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 33269f4c5fabd63dec12c2c63a5fa16dda72e8742907a730b927d678a6b09522
MD5 5b5f6396988aaadd7f72fd7be2620f19
BLAKE2b-256 0026614ed0c67cb5c5eb1545345463139b2bd4db0615f4500732f1384aa8ac70

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