钉钉内部应用（也称为自建应用或微应用）发送单聊消息，主要通过钉钉开放平台的“发送工作通知消息”接口实现。该接口允许应用主动向指定的企业内部员工发送消息，即单聊。核心流程包括：获取企业内部应用的`access_token`，然后调用工作通知接口发送消息。以下是详细的实现步骤和代码示例。 ### 1. 准备工作与核心概念 在开始之前，你需要在[钉钉开发者后台](https://open-dev.dingtalk.com/)完成以下配置： 1. **创建企业内部应用**：在“应用开发”->“企业内部开发”中创建H5微应用或小程序。 2. **获取凭证**： * **AppKey** 与 **AppSecret**：在应用的基础信息页面获取，用于获取`access_token`。 * **AgentId**：在应用的功能信息页面获取，发送工作通知时需要。 3. **配置权限**：在应用的权限管理中，为应用添加“工作通知”的接口访问权限。 发送单聊消息的实质是**发送工作通知给单个用户**。钉钉提供了`topapi/message/corpconversation/asyncsend_v2`接口来完成此功能。 ### 2. 获取 Access Token 调用绝大多数钉钉服务端API都需要使用`access_token`，它是应用调用API的全局唯一凭证。获取方式是通过AppKey和AppSecret向`https://oapi.dingtalk.com/gettoken`接口发起请求。 ```python import requests import json def get_access_token(app_key, app_secret): """ 获取钉钉企业内部应用的access_token :param app_key: 应用的AppKey :param app_secret: 应用的AppSecret :return: access_token字符串 """ url = "https://oapi.dingtalk.com/gettoken" params = { "appkey": app_key, "appsecret": app_secret } try: response = requests.get(url, params=params, timeout=5) result = response.json() if result.get('errcode') == 0: return result.get('access_token') else: raise Exception(f"获取access_token失败: {result.get('errmsg')}") except Exception as e: raise Exception(f"请求token接口异常: {e}") # 使用示例 APP_KEY = "your_app_key" APP_SECRET = "your_app_secret" access_token = get_access_token(APP_KEY, APP_SECRET) print(f"获取到的access_token: {access_token}") ``` ### 3. 发送工作通知（单聊消息） 获取到`access_token`后，即可调用发送工作通知的接口。你需要指定接收者的用户ID（`userid_list`）、消息内容体（`msg`）以及应用的`agent_id`。钉钉支持多种消息类型，如文本（`text`）、链接（`link`）、Markdown（`markdown`）、OA消息（`oa`）、文件（`file`）等。 以下是一个发送**文本消息**和**Markdown消息**的示例： ```python def send_work_notification(access_token, agent_id, userid_list, msg_body, msg_type='text'): """ 发送工作通知（单聊消息） :param access_token: 接口调用凭证 :param agent_id: 微应用的AgentId :param userid_list: 接收者的用户userid列表，多个用逗号分隔，例如："user123,user456" :param msg_body: 消息内容体，字典格式，结构因消息类型而异 :param msg_type: 消息类型，如 'text', 'markdown', 'link', 'oa', 'file' 等 :return: 接口返回结果 """ url = "https://oapi.dingtalk.com/topapi/message/corpconversation/asyncsend_v2" headers = {'Content-Type': 'application/json'} # 构建请求体 data = { "agent_id": agent_id, "userid_list": userid_list, "msg": { msg_type: msg_body # 关键：消息类型作为键，消息体作为值 } } params = {'access_token': access_token} try: # 注意：这里需要使用json.dumps确保中文等字符正确编码 json_data = json.dumps(data, ensure_ascii=False).encode('utf-8') response = requests.post(url, params=params, data=json_data, headers=headers, timeout=10) result = response.json() return result except Exception as e: raise Exception(f"发送工作通知请求异常: {e}") # --- 示例1：发送文本消息 --- def send_text_message(access_token, agent_id, userid, content): """发送文本工作通知""" text_body = { "content": content # 文本消息内容 } return send_work_notification(access_token, agent_id, userid, text_body, 'text') # --- 示例2：发送Markdown消息 --- def send_markdown_message(access_token, agent_id, userid, title, text): """发送Markdown工作通知""" markdown_body = { "title": title, # 消息标题，会显示在通知摘要里 "text": text # Markdown格式的正文内容 } return send_work_notification(access_token, agent_id, userid, markdown_body, 'markdown') # 使用示例 AGENT_ID = 12345678 # 你的应用AgentId，是数字 TARGET_USER_ID = "manager123" # 接收消息的员工userid # 发送文本消息 text_result = send_text_message(access_token, AGENT_ID, TARGET_USER_ID, "您好，这是一条来自Python应用的测试通知。") print(f"文本消息发送结果: {text_result}") # 发送Markdown消息 markdown_text = """## 项目日报 **时间**: 2023-10-27 **状态**: 正常 - [x] 完成模块A开发 - [ ] 进行模块B测试 [查看详情](http://your-internal-site.com)""" md_result = send_markdown_message(access_token, AGENT_ID, TARGET_USER_ID, "项目状态更新", markdown_text) print(f"Markdown消息发送结果: {md_result}") ``` ### 4. 关键参数说明与消息类型对比 不同的消息类型需要构建不同的`msg_body`。下表对比了常用的几种消息类型及其数据结构： | 消息类型 | `msg_body` 结构示例 | 说明 | | :--- | :--- | :--- | | **文本 (text)** | `{"content": "消息内容"}` | 纯文本消息，最常用[ref_1]。 | | **Markdown (markdown)** | `{"title":"标题", "text":"# Markdown内容"}` | 支持Markdown语法，可展示更丰富的格式[ref_3]。 | | **链接 (link)** | `{"messageUrl":"链接", "picUrl":"图片", "title":"标题", "text":"摘要"}` | 发送一个图文链接卡片。 | | **OA消息 (oa)** | 结构复杂，包含`head`, `body`, `status_bar`等 | 仿钉钉OA工作流的消息，样式最丰富。 | | **文件 (file)** | `{"media_id": "文件media_id"}` | 需要先调用**媒体文件上传**接口获取`media_id`[ref_4]。 | **重要参数详解**： * `agent_id`：**必须**填写正确，否则收不到消息或报错。 * `userid_list`：接收消息的用户ID，多个用英文逗号分隔。用户ID需要在钉钉组织内唯一，可以通过钉钉API或管理员后台获取。 * `to_all_user`：如果设置为`True`，则发送给企业内所有用户。与`userid_list`、`dept_id_list`互斥。 * `dept_id_list`：接收消息的部门ID列表。与`userid_list`类似，用于群发。 ### 5. 错误处理与调试建议 在实际调用中，你可能会遇到以下常见问题及解决方案： 1. **`400014` 或 `invalid token`**：`access_token`无效或已过期。`access_token`有效期为7200秒（2小时），需要实现缓存和刷新机制。 2. **`400008` 或 `forbidden`**：应用没有调用“工作通知”接口的权限。请去开发者后台检查权限配置。 3. **`400004` 或 `invalid agent_id`**：`agent_id`填写错误。请确认应用基础信息里的`AgentId`。 4. **消息已发送成功但用户收不到**： * 检查`userid`是否正确。 * 确认接收用户是否已经安装了该微应用（工作通知的发送不强制要求安装，但某些企业设置可能影响）。 * 在开发者后台的“应用管理”->“相关权限”->“权限范围”中，检查该用户是否在应用的可使用范围之内。 一个健壮的生产环境代码应包含`access_token`的管理（如使用`cachetools`库进行缓存）和完整的异常捕获与重试逻辑。 ### 6. 完整流程示例：发送带本地文件的工作通知 发送文件消息的流程稍复杂，需要分两步：先上传文件获取`media_id`，再发送消息。 ```python import os def upload_media_and_send(access_token, agent_id, userid, file_path): """ 完整示例：上传本地文件并发送文件类型的工作通知 """ # 步骤1：上传媒体文件 upload_url = "https://oapi.dingtalk.com/media/upload" params_upload = {'access_token': access_token, 'type': 'file'} # type: file-普通文件, image-图片, voice-语音 with open(file_path, 'rb') as f: files = {'media': (os.path.basename(file_path), f)} upload_resp = requests.post(upload_url, params=params_upload, files=files) upload_result = upload_resp.json() if upload_result.get('errcode') != 0: print(f"文件上传失败: {upload_result}") return upload_result media_id = upload_result.get('media_id') print(f"文件上传成功，media_id: {media_id}") # 步骤2：发送文件消息 file_body = {"media_id": media_id} send_result = send_work_notification(access_token, agent_id, userid, file_body, 'file') print(f"文件消息发送结果: {send_result}") return send_result # 使用示例 # file_result = upload_media_and_send(access_token, AGENT_ID, TARGET_USER_ID, "/path/to/your/report.pdf") ``` 通过以上步骤，你可以基于Python和钉钉开放平台API，稳定地实现从企业内部应用向指定员工发送单聊（工作通知）消息的功能，并根据需要选择文本、富文本或文件等不同消息格式[ref_1][ref_4][ref_5]。