企业微信机器人 API 3.0 接入实战5步配置智能回复支持 Markdown 消息企业微信作为企业级沟通协作平台其机器人 API 3.0 版本为开发者提供了更强大的自动化能力。本文将带你从零开始通过五个关键步骤实现智能回复功能并支持丰富的 Markdown 消息格式。1. 准备工作与环境配置在开始接入企业微信机器人 API 3.0 之前需要完成以下基础准备工作企业微信管理员权限确保你有权限在企业微信后台创建和配置应用可公网访问的服务器用于接收企业微信的回调消息基础开发环境Python 3.7 或 Node.js 14 环境1.1 创建企业微信应用登录企业微信管理后台https://work.weixin.qq.com/进入「应用管理」→「自建应用」→「创建应用」填写应用名称如智能客服机器人、上传应用图标记录下生成的AgentId和CorpId后续会用到1.2 获取 API 调用凭证企业微信 API 调用需要以下关键参数参数名称获取位置说明CorpId我的企业 → 企业信息企业唯一标识AgentId应用详情页应用唯一IDSecret应用详情页 → 查看Secret应用凭证密钥注意Secret 是敏感信息请妥善保管不要直接暴露在客户端代码中。2. 配置回调服务器企业微信机器人采用回调机制实现双向通信。当用户发送消息时企业微信会将消息推送到你配置的服务器地址。2.1 服务器端基础代码Python示例from flask import Flask, request, jsonify import hashlib import xml.etree.ElementTree as ET app Flask(__name__) # 配置参数 TOKEN your_token # 自定义Token需与企业微信后台一致 EncodingAESKey your_encoding_aes_key # 43位随机字符串 CORP_ID your_corp_id # 企业ID app.route(/callback, methods[GET, POST]) def wechat_callback(): if request.method GET: # 验证回调URL return verify_url(request.args) else: # 处理消息回调 return handle_message(request.data) def verify_url(args): # URL验证逻辑实现 signature args.get(msg_signature, ) timestamp args.get(timestamp, ) nonce args.get(nonce, ) echostr args.get(echostr, ) # 验证签名逻辑省略具体实现 if check_signature(signature, timestamp, nonce, TOKEN): return echostr return Verification failed, 403 def handle_message(xml_data): # 消息处理逻辑 xml_tree ET.fromstring(xml_data) msg_type xml_tree.find(MsgType).text # 根据不同类型处理消息 return success2.2 企业微信后台配置进入应用详情页 → 接收消息 → 设置API接收填写以下信息URL你的服务器回调地址如https://yourdomain.com/callbackToken与代码中TOKEN一致EncodingAESKey43位随机字符串点击保存完成配置3. 实现消息接收与解析企业微信支持多种消息类型我们需要正确处理这些消息格式。3.1 支持的消息类型企业微信机器人 API 3.0 支持以下常见消息类型文本消息图片消息语音消息视频消息位置消息链接消息事件消息如成员加入聊天3.2 消息解析示例代码def parse_message(xml_data): 解析企业微信回调的XML消息 xml_tree ET.fromstring(xml_data) msg { ToUserName: xml_tree.find(ToUserName).text, FromUserName: xml_tree.find(FromUserName).text, CreateTime: xml_tree.find(CreateTime).text, MsgType: xml_tree.find(MsgType).text, MsgId: xml_tree.find(MsgId).text, AgentID: xml_tree.find(AgentID).text, } # 根据不同类型提取额外字段 if msg[MsgType] text: msg[Content] xml_tree.find(Content).text elif msg[MsgType] image: msg[PicUrl] xml_tree.find(PicUrl).text msg[MediaId] xml_tree.find(MediaId).text # 其他消息类型处理... return msg def handle_message(xml_data): msg parse_message(xml_data) response_url request.args.get(response_url) # 获取response_url # 根据消息类型生成回复 if msg[MsgType] text: reply generate_text_reply(msg[Content]) send_reply(response_url, reply) return success4. 实现主动回复与 Markdown 支持企业微信 3.0 API 提供了response_url机制允许开发者在处理完业务逻辑后主动回复消息。4.1 主动回复机制详解用户发送消息到企业微信企业微信将消息推送到你的服务器并在回调中附带response_url你可以在处理完业务逻辑后向该response_url发送回复每个response_url只能使用一次有效期为1小时4.2 Markdown 消息格式示例def generate_markdown_reply(content): 生成Markdown格式回复 markdown_content f # 智能回复 {content} ## 常用功能 - **查询订单**输入订单号查询状态 - **联系客服**输入人工转接 - **帮助**查看所有功能 服务时间工作日 9:00-18:00 return { msgtype: markdown, markdown: { content: markdown_content, feedback: { id: feedback_123 # 可选用于用户反馈追踪 } } } def send_reply(response_url, reply_content): 发送回复到企业微信 headers {Content-Type: application/json} response requests.post( response_url, headersheaders, jsonreply_content ) return response.json()4.3 Markdown 支持语法企业微信 Markdown 支持丰富的格式# 一级标题 ## 二级标题 ### 三级标题 *斜体文本* **加粗文本** - 无序列表项 1. 有序列表项 引用文本 [链接文本](https://example.com)  行内代码代码块| 表格 | 对齐 | |:----|:----:| | 左对齐 | 居中 |5. 完整示例与部署指南5.1 Python 完整示例代码import requests from flask import Flask, request import xml.etree.ElementTree as ET import hashlib import time app Flask(__name__) # 配置参数 CONFIG { TOKEN: your_token, EncodingAESKey: your_aes_key, CORP_ID: your_corp_id, SECRET: your_app_secret } app.route(/callback, methods[GET, POST]) def callback(): if request.method GET: return verify_url(request.args) else: return handle_message(request.data) def verify_url(args): # 实现URL验证逻辑 signature args.get(msg_signature, ) timestamp args.get(timestamp, ) nonce args.get(nonce, ) echostr args.get(echostr, ) if check_signature(signature, timestamp, nonce, CONFIG[TOKEN]): return echostr return Verification failed, 403 def handle_message(xml_data): msg parse_message(xml_data) response_url request.args.get(response_url) if msg[MsgType] text: reply process_text_message(msg[Content]) send_reply(response_url, reply) return success def process_text_message(content): # 简单的关键词回复逻辑 if 订单 in content: return generate_markdown_reply( ### 订单查询 请输入完整的订单编号我们将为您查询最新状态。 **示例** 订单 20230615001 ) else: return generate_markdown_reply(f 已收到您的消息{content} 输入帮助查看可用命令。 ) # 其他辅助函数...5.2 Node.js 实现关键部分const express require(express); const crypto require(crypto); const axios require(axios); const xml2js require(xml2js); const app express(); app.use(express.text({ type: application/xml })); // 配置参数 const config { token: your_token, encodingAESKey: your_aes_key, corpId: your_corp_id, secret: your_app_secret }; // 验证URL app.get(/callback, (req, res) { const { msg_signature, timestamp, nonce, echostr } req.query; if (verifySignature(msg_signature, timestamp, nonce, config.token)) { res.send(echostr); } else { res.status(403).send(Verification failed); } }); // 处理消息 app.post(/callback, async (req, res) { const { response_url } req.query; const xmlData req.body; try { const result await xml2js.parseStringPromise(xmlData); const message result.xml; if (message.MsgType[0] text) { const reply processTextMessage(message.Content[0]); await sendReply(response_url, reply); } res.send(success); } catch (error) { console.error(Error processing message:, error); res.status(500).send(Internal Server Error); } }); // 处理文本消息 function processTextMessage(content) { if (content.includes(帮助)) { return { msgtype: markdown, markdown: { content: # 帮助菜单\n\n- **订单查询**输入订单订单号\n- **人工客服**输入人工\n- **营业时间**工作日9:00-18:00 } }; } // 其他处理逻辑... } // 发送回复 async function sendReply(responseUrl, reply) { try { const response await axios.post(responseUrl, reply); return response.data; } catch (error) { console.error(Error sending reply:, error); throw error; } }5.3 部署与测试建议服务器部署推荐使用云服务如阿里云、腾讯云等确保服务器有公网IP或域名配置HTTPS企业微信要求回调地址必须为HTTPS测试流程使用企业微信开发者调试工具发送测试消息检查服务器日志确认收到回调验证回复消息是否正确显示性能优化使用消息队列处理高并发场景实现消息去重利用MsgId添加超时和重试机制高级功能与最佳实践消息加解密实现企业微信要求敏感数据需要加密传输以下是Python实现示例from Crypto.Cipher import AES import base64 import random import string import socket import struct import time import hashlib from xml.etree import ElementTree class WXBizMsgCrypt: def __init__(self, sToken, sEncodingAESKey, sCorpId): self.key base64.b64decode(sEncodingAESKey ) self.token sToken self.corp_id sCorpId def decrypt(self, sEncryptMsg): cipher AES.new(self.key, AES.MODE_CBC, self.key[:16]) msg cipher.decrypt(base64.b64decode(sEncryptMsg)) # 移除补位字符 pad ord(msg[-1:]) content msg[:-pad] # 解析XML xml_content content[16:] xml_len socket.ntohl(struct.unpack(I, xml_content[:4])[0]) xml_content xml_content[4:xml_len4] return xml_content def encrypt(self, sReplyMsg): # 加密逻辑实现 pass def verify_url(self, sMsgSignature, sTimeStamp, sNonce, sEchoStr): # URL验证逻辑 pass消息处理性能优化对于高并发场景建议采用以下优化策略异步处理使用消息队列如RabbitMQ、Kafka解耦消息接收和处理缓存机制缓存常用回复和用户会话状态连接池为HTTP请求维护连接池超时设置合理设置请求超时避免阻塞# 使用Redis缓存用户会话状态示例 import redis redis_client redis.Redis(hostlocalhost, port6379, db0) def get_user_session(user_id): session redis_client.get(fsession:{user_id}) return json.loads(session) if session else {} def update_user_session(user_id, session_data): redis_client.setex( fsession:{user_id}, 3600, # 1小时过期 json.dumps(session_data) )安全最佳实践敏感信息保护不要将Secret硬编码在代码中使用环境变量或配置中心管理敏感信息实现定期密钥轮换输入验证验证所有传入消息的签名过滤特殊字符防止注入攻击日志记录记录关键操作日志避免记录敏感信息实现日志审计功能# 安全日志记录示例 import logging from logging.handlers import RotatingFileHandler # 配置日志 logger logging.getLogger(wechat_bot) logger.setLevel(logging.INFO) handler RotatingFileHandler( bot.log, maxBytes5*1024*1024, backupCount3 ) formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) handler.setFormatter(formatter) logger.addHandler(handler) # 记录安全事件 def log_security_event(event_type, user_id, ip): logger.info( fSecurity Event - {event_type} - User: {user_id} - IP: {ip}, extra{type: security} )
企业微信机器人 API 3.0 接入实战：5步配置智能回复，支持 Markdown 消息
企业微信机器人 API 3.0 接入实战5步配置智能回复支持 Markdown 消息企业微信作为企业级沟通协作平台其机器人 API 3.0 版本为开发者提供了更强大的自动化能力。本文将带你从零开始通过五个关键步骤实现智能回复功能并支持丰富的 Markdown 消息格式。1. 准备工作与环境配置在开始接入企业微信机器人 API 3.0 之前需要完成以下基础准备工作企业微信管理员权限确保你有权限在企业微信后台创建和配置应用可公网访问的服务器用于接收企业微信的回调消息基础开发环境Python 3.7 或 Node.js 14 环境1.1 创建企业微信应用登录企业微信管理后台https://work.weixin.qq.com/进入「应用管理」→「自建应用」→「创建应用」填写应用名称如智能客服机器人、上传应用图标记录下生成的AgentId和CorpId后续会用到1.2 获取 API 调用凭证企业微信 API 调用需要以下关键参数参数名称获取位置说明CorpId我的企业 → 企业信息企业唯一标识AgentId应用详情页应用唯一IDSecret应用详情页 → 查看Secret应用凭证密钥注意Secret 是敏感信息请妥善保管不要直接暴露在客户端代码中。2. 配置回调服务器企业微信机器人采用回调机制实现双向通信。当用户发送消息时企业微信会将消息推送到你配置的服务器地址。2.1 服务器端基础代码Python示例from flask import Flask, request, jsonify import hashlib import xml.etree.ElementTree as ET app Flask(__name__) # 配置参数 TOKEN your_token # 自定义Token需与企业微信后台一致 EncodingAESKey your_encoding_aes_key # 43位随机字符串 CORP_ID your_corp_id # 企业ID app.route(/callback, methods[GET, POST]) def wechat_callback(): if request.method GET: # 验证回调URL return verify_url(request.args) else: # 处理消息回调 return handle_message(request.data) def verify_url(args): # URL验证逻辑实现 signature args.get(msg_signature, ) timestamp args.get(timestamp, ) nonce args.get(nonce, ) echostr args.get(echostr, ) # 验证签名逻辑省略具体实现 if check_signature(signature, timestamp, nonce, TOKEN): return echostr return Verification failed, 403 def handle_message(xml_data): # 消息处理逻辑 xml_tree ET.fromstring(xml_data) msg_type xml_tree.find(MsgType).text # 根据不同类型处理消息 return success2.2 企业微信后台配置进入应用详情页 → 接收消息 → 设置API接收填写以下信息URL你的服务器回调地址如https://yourdomain.com/callbackToken与代码中TOKEN一致EncodingAESKey43位随机字符串点击保存完成配置3. 实现消息接收与解析企业微信支持多种消息类型我们需要正确处理这些消息格式。3.1 支持的消息类型企业微信机器人 API 3.0 支持以下常见消息类型文本消息图片消息语音消息视频消息位置消息链接消息事件消息如成员加入聊天3.2 消息解析示例代码def parse_message(xml_data): 解析企业微信回调的XML消息 xml_tree ET.fromstring(xml_data) msg { ToUserName: xml_tree.find(ToUserName).text, FromUserName: xml_tree.find(FromUserName).text, CreateTime: xml_tree.find(CreateTime).text, MsgType: xml_tree.find(MsgType).text, MsgId: xml_tree.find(MsgId).text, AgentID: xml_tree.find(AgentID).text, } # 根据不同类型提取额外字段 if msg[MsgType] text: msg[Content] xml_tree.find(Content).text elif msg[MsgType] image: msg[PicUrl] xml_tree.find(PicUrl).text msg[MediaId] xml_tree.find(MediaId).text # 其他消息类型处理... return msg def handle_message(xml_data): msg parse_message(xml_data) response_url request.args.get(response_url) # 获取response_url # 根据消息类型生成回复 if msg[MsgType] text: reply generate_text_reply(msg[Content]) send_reply(response_url, reply) return success4. 实现主动回复与 Markdown 支持企业微信 3.0 API 提供了response_url机制允许开发者在处理完业务逻辑后主动回复消息。4.1 主动回复机制详解用户发送消息到企业微信企业微信将消息推送到你的服务器并在回调中附带response_url你可以在处理完业务逻辑后向该response_url发送回复每个response_url只能使用一次有效期为1小时4.2 Markdown 消息格式示例def generate_markdown_reply(content): 生成Markdown格式回复 markdown_content f # 智能回复 {content} ## 常用功能 - **查询订单**输入订单号查询状态 - **联系客服**输入人工转接 - **帮助**查看所有功能 服务时间工作日 9:00-18:00 return { msgtype: markdown, markdown: { content: markdown_content, feedback: { id: feedback_123 # 可选用于用户反馈追踪 } } } def send_reply(response_url, reply_content): 发送回复到企业微信 headers {Content-Type: application/json} response requests.post( response_url, headersheaders, jsonreply_content ) return response.json()4.3 Markdown 支持语法企业微信 Markdown 支持丰富的格式# 一级标题 ## 二级标题 ### 三级标题 *斜体文本* **加粗文本** - 无序列表项 1. 有序列表项 引用文本 [链接文本](https://example.com)  行内代码代码块| 表格 | 对齐 | |:----|:----:| | 左对齐 | 居中 |5. 完整示例与部署指南5.1 Python 完整示例代码import requests from flask import Flask, request import xml.etree.ElementTree as ET import hashlib import time app Flask(__name__) # 配置参数 CONFIG { TOKEN: your_token, EncodingAESKey: your_aes_key, CORP_ID: your_corp_id, SECRET: your_app_secret } app.route(/callback, methods[GET, POST]) def callback(): if request.method GET: return verify_url(request.args) else: return handle_message(request.data) def verify_url(args): # 实现URL验证逻辑 signature args.get(msg_signature, ) timestamp args.get(timestamp, ) nonce args.get(nonce, ) echostr args.get(echostr, ) if check_signature(signature, timestamp, nonce, CONFIG[TOKEN]): return echostr return Verification failed, 403 def handle_message(xml_data): msg parse_message(xml_data) response_url request.args.get(response_url) if msg[MsgType] text: reply process_text_message(msg[Content]) send_reply(response_url, reply) return success def process_text_message(content): # 简单的关键词回复逻辑 if 订单 in content: return generate_markdown_reply( ### 订单查询 请输入完整的订单编号我们将为您查询最新状态。 **示例** 订单 20230615001 ) else: return generate_markdown_reply(f 已收到您的消息{content} 输入帮助查看可用命令。 ) # 其他辅助函数...5.2 Node.js 实现关键部分const express require(express); const crypto require(crypto); const axios require(axios); const xml2js require(xml2js); const app express(); app.use(express.text({ type: application/xml })); // 配置参数 const config { token: your_token, encodingAESKey: your_aes_key, corpId: your_corp_id, secret: your_app_secret }; // 验证URL app.get(/callback, (req, res) { const { msg_signature, timestamp, nonce, echostr } req.query; if (verifySignature(msg_signature, timestamp, nonce, config.token)) { res.send(echostr); } else { res.status(403).send(Verification failed); } }); // 处理消息 app.post(/callback, async (req, res) { const { response_url } req.query; const xmlData req.body; try { const result await xml2js.parseStringPromise(xmlData); const message result.xml; if (message.MsgType[0] text) { const reply processTextMessage(message.Content[0]); await sendReply(response_url, reply); } res.send(success); } catch (error) { console.error(Error processing message:, error); res.status(500).send(Internal Server Error); } }); // 处理文本消息 function processTextMessage(content) { if (content.includes(帮助)) { return { msgtype: markdown, markdown: { content: # 帮助菜单\n\n- **订单查询**输入订单订单号\n- **人工客服**输入人工\n- **营业时间**工作日9:00-18:00 } }; } // 其他处理逻辑... } // 发送回复 async function sendReply(responseUrl, reply) { try { const response await axios.post(responseUrl, reply); return response.data; } catch (error) { console.error(Error sending reply:, error); throw error; } }5.3 部署与测试建议服务器部署推荐使用云服务如阿里云、腾讯云等确保服务器有公网IP或域名配置HTTPS企业微信要求回调地址必须为HTTPS测试流程使用企业微信开发者调试工具发送测试消息检查服务器日志确认收到回调验证回复消息是否正确显示性能优化使用消息队列处理高并发场景实现消息去重利用MsgId添加超时和重试机制高级功能与最佳实践消息加解密实现企业微信要求敏感数据需要加密传输以下是Python实现示例from Crypto.Cipher import AES import base64 import random import string import socket import struct import time import hashlib from xml.etree import ElementTree class WXBizMsgCrypt: def __init__(self, sToken, sEncodingAESKey, sCorpId): self.key base64.b64decode(sEncodingAESKey ) self.token sToken self.corp_id sCorpId def decrypt(self, sEncryptMsg): cipher AES.new(self.key, AES.MODE_CBC, self.key[:16]) msg cipher.decrypt(base64.b64decode(sEncryptMsg)) # 移除补位字符 pad ord(msg[-1:]) content msg[:-pad] # 解析XML xml_content content[16:] xml_len socket.ntohl(struct.unpack(I, xml_content[:4])[0]) xml_content xml_content[4:xml_len4] return xml_content def encrypt(self, sReplyMsg): # 加密逻辑实现 pass def verify_url(self, sMsgSignature, sTimeStamp, sNonce, sEchoStr): # URL验证逻辑 pass消息处理性能优化对于高并发场景建议采用以下优化策略异步处理使用消息队列如RabbitMQ、Kafka解耦消息接收和处理缓存机制缓存常用回复和用户会话状态连接池为HTTP请求维护连接池超时设置合理设置请求超时避免阻塞# 使用Redis缓存用户会话状态示例 import redis redis_client redis.Redis(hostlocalhost, port6379, db0) def get_user_session(user_id): session redis_client.get(fsession:{user_id}) return json.loads(session) if session else {} def update_user_session(user_id, session_data): redis_client.setex( fsession:{user_id}, 3600, # 1小时过期 json.dumps(session_data) )安全最佳实践敏感信息保护不要将Secret硬编码在代码中使用环境变量或配置中心管理敏感信息实现定期密钥轮换输入验证验证所有传入消息的签名过滤特殊字符防止注入攻击日志记录记录关键操作日志避免记录敏感信息实现日志审计功能# 安全日志记录示例 import logging from logging.handlers import RotatingFileHandler # 配置日志 logger logging.getLogger(wechat_bot) logger.setLevel(logging.INFO) handler RotatingFileHandler( bot.log, maxBytes5*1024*1024, backupCount3 ) formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) handler.setFormatter(formatter) logger.addHandler(handler) # 记录安全事件 def log_security_event(event_type, user_id, ip): logger.info( fSecurity Event - {event_type} - User: {user_id} - IP: {ip}, extra{type: security} )
