灵墨笔记
返回

企业微信消息转发：从零到一的完整实现指南

更新于:
编辑本页

前言

在企业数字化转型过程中，企业微信已成为许多企业的首选沟通工具。为了更好地整合内部系统和外部应用，实现企业微信消息的接收与转发变得尤为重要。本文将详细介绍如何从零开始搭建企业微信消息转发系统，涵盖注册、配置、开发、部署等全流程，帮助开发者快速实现企业微信消息的转发功能。

一、注册企业微信应用

首先，你需要在企业微信管理后台注册一个应用，这是实现消息转发的第一步。

  1. 登录企业微信管理后台 https://work.weixin.qq.com/wework_admin/
  2. 如果目前还没有注册过企业，点击注册企业，如下图。

http://work.weixin.qq.com/wework\_admin/qrlogin/mp/scan?op\_from=register&key=563476099956534 (二维码自动识别)

  1. 企业注册之后，进入”应用管理”页面。
  2. 创建一个新的应用，如下图：

二、获取关键配置信息

在实现消息转发前，需要获取以下三个关键信息：

企业ID

企业ID是企业微信中企业的唯一标识符，用于区分不同的企业。

  1. 在企业微信管理页面，点击我的企业。
  2. 滚动到页面最下方，找到企业ID，并记下。

应用ID (Agent ID)和Secret

应用ID是企业内特定应用的标识符，用于区分同一企业内的不同应用。Secret是应用的密钥，用于获取访问令牌和消息加解密。

  1. 在企业微信管理页面，点击应用管理，找到刚刚创建的应用，点击打开。
  2. 在应用名称下方可以看到AgentID 和 Secet。
  3. 在Secret 旁边点击查看，secret 将发送到企业微信上。

到此三个必要信息：企业ID, AgentID, Secret 都已经具备。但是，此时应用系统还无法使用这三个信息进行企业微信消息发送。常见的问题是企业应用系统对外发送消息的IP地址是不授信的IP。因此需要进行下面的步骤。

三、设置接收消息服务器URL

为什么要设置接收消息服务器URL？

设置接收消息服务器URL是企业微信消息转发系统搭建的关键步骤，其核心目的是：

  1. 验证服务器可达性：确认企业微信能够成功访问我们搭建的回调服务器。
  2. 为后续配置做准备：识别到的IP地址将用于配置企业可信IP白名单。
  3. 设置接收消息服务器URL比设置可信域名更简单。

要设置接收消息服务器，需要搭建一个能够处理企业微信回调的服务搭建一个能够处理企业微信回调的服务，且此服务需要能在公网上被访问到。

为了简化搭建步骤，我们直接使用github上的开源库weworkapi_python。 配置步骤如下：

下载企业微信API库

从GitHub下载weworkapi_python源代码：

# 克隆企业微信官方Python API库
git clone https://github.com/sbzhu/weworkapi_python.git

该库包含以下关键组件：

  • api/ - API接口封装
  • callback_python3/ - Python3版本的加解密库（本文使用）
  • callback_json_python3/ - JSON格式加解密库（适用于企业机器人/智能机器人）

编写回调服务的主程序

以下是使用Python Flask实现的示例代码：

#-*- encoding:utf-8 -*-
from flask import abort, request
from flask import Flask
from xml.dom.minidom import parseString
import _thread
import time
import os
import sys
sys.path.append("weworkapi_python/callback_python3")  # 正确的模块导入路径
from WXBizMsgCrypt import WXBizMsgCrypt   # https://github.com/sbzhu/weworkapi_python 项目地址
app = Flask(__name__)

# 对应步骤4中接受消息回调模式中的URL，如域名是'www.example.com' 那么在步骤4中填入的url就为"http://www.example.com/hook_path"
@app.route('/hook_path', methods=['GET','POST'])
def douban():
    if request.method == 'GET':
        echo_str = signature(request, 0)
        return(echo_str)
    elif request.method == 'POST':
        echo_str = signature2(request, 0)
        return(echo_str)

qy_api = [
    WXBizMsgCrypt("<token>", "<EncodingAESKey>", "<企业ID>"),
] #对应接受消息回调模式中的token，EncodingAESKey 和 企业信息中的企业id   # 只改这里即可

# 开启消息接受模式时验证接口连通性
def signature(request, i):
    msg_signature = request.args.get('msg_signature', '')
    timestamp = request.args.get('timestamp', '')
    nonce = request.args.get('nonce', '')
    echo_str = request.args.get('echostr', '')
    ret,sEchoStr=qy_api[i].VerifyURL(msg_signature, timestamp,nonce,echo_str)
    if (ret != 0):
        print("ERR: VerifyURL ret: " + str(ret))
        return("failed")
    else:
        return(sEchoStr)

# 实际接受消息
def signature2(request, i):
    msg_signature = request.args.get('msg_signature', '')
    timestamp = request.args.get('timestamp', '')
    nonce = request.args.get('nonce', '')
    data = request.data.decode('utf-8')
    ret,sMsg=qy_api[i].DecryptMsg(data,msg_signature, timestamp,nonce)
    if (ret != 0):
        print("ERR: DecryptMsg ret: " + str(ret))
        return("failed")
    else:
        with open ("/var/log/qywx.log", 'a+') as f: # 消息接收日志
            doc = parseString(sMsg)
            collection = doc.documentElement
            name_xml = collection.getElementsByTagName("FromUserName")
            msg_xml = collection.getElementsByTagName("Content")
            type_xml = collection.getElementsByTagName("MsgType")
            pic_xml = collection.getElementsByTagName("PicUrl")
            msg = ""
            name = ""
            msg_type = type_xml[0].childNodes[0].data
            if msg_type == "text": #文本消息
                name = name_xml[0].childNodes[0].data        #发送者id
                msg = msg_xml[0].childNodes[0].data          #发送的消息内容
                f.write(time.strftime('[%Y-%m-%d %H:%M:%S]') + "[ch%d] %s:%s\n" % (i, name, msg))
                _thread.start_new_thread(os.system, ("python3 command.py '%s' '%s' '%d' '%d'" % (name, msg, i, 0), )) #此处将消息进行外部业务处理

            elif msg_type == "image": #图片消息
                name = name_xml[0].childNodes[0].data
                pic_url = pic_xml[0].childNodes[0].data
                f.write(time.strftime('[%Y-%m-%d %H:%M:%S]') + "[ch%d] %s:图片消息\n" % (i, name))
                _thread.start_new_thread(os.system, ("python3 command.py '%s' '%s' '%d' '%d'" % (name, pic_url, i, 1), ))  #此处将消息进行外部业务处理

            f.close()

        return("ok")

if __name__=='__main__':
    app.run("0.0.0.0", 10086)  #本地监听端口,可自定义

接收消息服务器配置

在企业微信管理页面，点击新创建的应用，配置接收消息服务器，如下图。

  1. URL 为上一步搭建起来URL, 注意下URL最后加上 hook_path
  2. Token: 点击随机获取
  3. EncodingAESKey：点击随机获取

将Token、EncodingAESKey、企业ID 填入准备好的主程序，并运行主程序。

运行回调主程序

服务器运行成功示例：

(myenv) root@ubuntu:/home/admin# python wework.py
 * Serving Flask app 'wework'
 * Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on all addresses (0.0.0.0)
 * Running on http://127.0.0.1:10086
 * Running on http://192.168.10.253:10086
Press CTRL+C to quit

常见问题解决

在运行上述代码时，可能会遇到 PY_SSIZE_T_CLEAN macro must be defined for '#' formats 错误，解决方案如下：

  • 卸载现有的pycryptodome：
# 激活虚拟环境
source myenv/bin/activate
# 卸载旧版本
pip uninstall -y pycryptodome pycrypto  # 确保清理干净
  • 定义宏并重新安装pycryptodome：
# Linux/macOS 下定义宏并安装
export CFLAGS="-DPY_SSIZE_T_CLEAN"
pip install pycryptodome==3.20.0  # 安装稳定版，避免最新版兼容问题

查看调用日志

192.168.10.252 - - [31/Dec/2025 13:59:10] "GET /hook_path?msg_signature=9af0675b6f473514562e68110306db00fbc642a0&timestamp=1767160750&nonce=1767381859&echostr=OhuRuTo30/wXcFq%2BGUKcwIUE9PIWdLExPJlzMuxbx1FYJ8tAcGtgg3aMhzEJJ9TDbjJ70Vp6mfOydwBbaVh12g%3D%3D HTTP/1.1" 200 -
192.168.10.252 - - [31/Dec/2025 14:16:41] "POST /hook_path?msg_signature=cfeef9dc6cbc0586fe25a00158e556253da50b44&timestamp=1767161801&nonce=1767809672 HTTP/1.1" 200 -

URL认证成功后，企业微信会自动识别服务器的公网IP地址。

重要提示：

  • 如果服务器在内网，需要确保端口映射正确配置

四、配置企业可信IP

为什么要配置企业可信IP？

在完成URL认证后，企业微信已经识别并记录了服务器的公网IP地址。配置企业可信IP白名单是为了：

  1. 安全性保障：限制只有指定IP地址的服务器才能调用企业微信API，防止未授权访问
  2. API调用权限：只有配置在可信IP列表中的服务器才能成功调用企业微信的消息发送接口

配置步骤：

  1. 进入应用管理 → 打开新建的应用→ 滑到下方找到企业可信IP。
  2. 添加服务器的公网IP地址（该IP在URL认证时已被企业微信识别）。

五、测试消息转发功能

完成配置后，可以进行消息转发测试。以下是测试脚本的示例输出：

# -*- coding: utf-8 -*-
import urllib
import urllib2
import json
import sys

# 解决Python 2.7中文编码问题
reload(sys)
sys.setdefaultencoding('utf-8')

def get_access_token(corpid, corpsecret):
    """
    获取企业微信access_token（接口调用凭证）
    :param corpid: 企业ID（企业微信后台获取）
    :param corpsecret: 应用凭证密钥（对应应用的secret）
    :return: 成功返回access_token，失败返回None
    """
    token_url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
    # 拼接请求参数
    params = {
        "corpid": corpid,
        "corpsecret": corpsecret
    }
    full_url = token_url + "?" + urllib.urlencode(params)

    try:
        # 发送GET请求
        response = urllib2.urlopen(full_url, timeout=10)
        result = json.loads(response.read())

        if result.get("errcode") == 0:
            return result.get("access_token")
        else:
            print u"获取access_token失败：%s（错误码：%s）" % (result.get("errmsg"), result.get("errcode"))
            return None
    except urllib2.URLError as e:
        print u"获取access_token网络异常：%s" % e
        return None

def send_wechat_msg(access_token, agentid, touser, content):
    """
    发送企业微信文本消息
    :param access_token: 接口调用凭证
    :param agentid: 应用ID（整数类型）
    :param touser: 接收人（多个用|分隔，全员发送填@all）
    :param content: 消息内容（支持中文）
    :return: 发送结果字典，失败返回None
    """
    send_url = "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=%s" % access_token

    # 构造消息体
    msg_data = {
        "touser": touser,
        "msgtype": "text",
        "agentid": agentid,
        "text": {
            "content": content
        },
        "safe": 0  # 0=普通消息，1=保密消息
    }

    try:
        # 转换为JSON字符串（处理中文编码）
        json_data = json.dumps(msg_data, ensure_ascii=False).encode('utf-8')
        # 构造POST请求
        req = urllib2.Request(
            url=send_url,
            data=json_data,
            headers={"Content-Type": "application/json"}
        )
        # 发送请求
        response = urllib2.urlopen(req, timeout=10)
        result = json.loads(response.read())

        if result.get("errcode") == 0:
            print u"消息发送成功！"
            return result
        else:
            print u"消息发送失败：%s（错误码：%s）" % (result.get("errmsg"), result.get("errcode"))
            return result
    except urllib2.URLError as e:
        print u"发送消息网络异常：%s" % e
        return None

if __name__ == "__main__":
    # ===================== 请替换为自己的企业微信配置 =====================
    CORPID = "<企业ID>"  # 企业ID（我的企业 -> 企业信息 -> 企业ID）
    CORPSECRET = "<Secret>"  # 应用Secret（应用管理 -> 对应应用 -> 应用信息 -> 凭证密钥）
    AGENTID = <AgentID>  # 应用ID（同上，整数类型）
    TOUSER = "@all"  # 接收人：单个用户填账号，多个用|分隔，全员填@all
    MSG_CONTENT = u"【Python 2.7测试】这是企业微信消息发送测试内容！"
    # ====================================================================

    # 1. 获取access_token
    access_token = get_access_token(CORPID, CORPSECRET)
    if not access_token:
        sys.exit(1)  # 获取失败则退出
    print u"\n成功获取access_token：%s" % access_token

    # 2. 发送消息
    print u"\n开始发送消息..."
    send_result = send_wechat_msg(access_token, AGENTID, TOUSER, MSG_CONTENT)
    if send_result:
        print u"发送结果详情：%s" % json.dumps(send_result, ensure_ascii=False, indent=2)

运行测试发送消息程序

[root@172-17-3-2 ~]# python test_wecom.py

成功获取access_token：wK_nSJVAK65KYR5aaIrXA_sOzuhswx_DF-CONfZDoklphL4qTD7THcVd6rzunFaQEfKnXV9DrjceBQ3oXRYhl-gnwIUsewh4a2hM36RlB-owl_HX582C1mg-ZM_0K34AkjJ6vZEgW5a_EnrCcz5aRjzaSv_N4o0gWmPhYFzSPtM9HnPBnivIuvlaPhjCYnl-DycVHflpdsDosJwupp_J5w

开始发送消息...
消息发送成功！
发送结果详情：{
  "msgid": "oykM8gZThPh2qFctYhHlfINo-3aRNqE5gvdoy60pT5Y-fxdOC4IxBkGr4MQR7PoR5MmDH8TjbtsYv51K_j67Te5_b1AXcf91WEfYjoAjvJs",
  "errcode": 0,
  "errmsg": "ok"
}

六、验证消息接收

测试完成后，验证应用系统的消息转发是否成功：

  1. 在企业应用系统中配置 企业ID、AgentID、Secret
  2. 发送消息
  3. 在企业微信中验证消息接收

企微中接收消息

微信中接收消息

如果需要在微信中查看企微发送的消息，此时需要在微信中关注企业微信的消息。

  1. 在我的企业中 打开微信插件
  2. 找到邀请关注的二维码
  3. 用微信扫描二维码。

七、常见问题排查

1. URL认证失败

  • 检查Token、EncodingAESKey和企业ID是否正确
  • 确认服务器能够正常访问互联网
  • 验证防火墙设置，确保端口开放

2. 消息加密解密错误

  • 检查WXBizMsgCrypt库是否正确安装
  • 确认加密密钥设置是否正确
  • 验证消息体格式是否符合企业微信规范

3. 访问令牌获取失败

  • 检查Secret是否正确
  • 确认企业ID和应用ID是否匹配
  • 验证网络连接是否正常

总结

本文详细介绍了企业微信消息转发的完整实现流程，从注册应用到最终测试验证。通过以上步骤，您可以成功搭建一个企业微信消息转发系统，实现企业微信与其他系统的消息互通。

在实际部署过程中，还需要考虑以下几点：

  1. 安全性：确保API密钥等敏感信息的安全存储
  2. 可靠性：实现错误处理和重试机制
  3. 监控：添加日志记录和监控告警功能
  4. 性能：优化代码逻辑，提高处理企业微信消息转发功能的实现为企业数字化转型提供了重要支持，通过本文的指导，您可以快速搭建属于自己的消息转发系统。
编辑本页
分享这篇文章：
在 微信 分享Sina Weibo在 微博 分享通过邮件分享
扫描二维码分享
打开应用扫一扫，即可分享
上一篇
Hermes Agent 与OpenClaw有什么不同？
下一篇
火速实测：Graphify项目将Karpathy 知识库方法从理论推到生产