py-wecom-toolkit

企业微信 Python 扩展包 - 支持 Webhook 与 Server 端 API。

• 作者: 郭磊
• 邮箱: 174000902@qq.com
• 项目主页: https://gitee.com/guolei19850528/py_wecom_toolkit
• PyPI: https://pypi.org/project/py-wecom-toolkit/

项目结构

py-wecom-toolkit/
├── pyproject.toml              # 项目构建配置
├── README.md                   # 项目说明文档
├── LICENSE                     # MIT 许可证
├── src/
│   └── py_wecom_toolkit/       # 源代码包
│       ├── __init__.py         # 包入口，导出所有公共类
│       ├── webhook/            # Webhook 消息发送模块（原“群机器人”）
│       │   ├── __init__.py
│       │   ├── client.py       # WeComWebhookClient / AsyncWeComWebhookClient
│       │   ├── exceptions.py   # Webhook 异常定义
│       │   ├── models.py       # pydantic v2 消息模型（8种消息类型）
│       │   └── types.py        # 类型别名
│       └── server/             # Server 端 API 模块（企业微信服务端接口）
│           ├── __init__.py
│           ├── client.py       # WeComServerClient / AsyncWeComServerClient
│           ├── token.py        # WeComServerAccessTokenManager
│           ├── cache.py        # WeComServerMemoryCache / WeComServerDiskCache / WeComServerRedisCache
│           ├── models.py       # Server API 响应模型
│           ├── exceptions.py   # Server 异常定义
│           ├── message.py      # 应用消息发送（WeComServerMessageSendMixin）
│           └── media.py        # 素材上传/下载（WeComServerMediaMixin）
├── tests/                      # 测试用例（真实调用企业微信接口）
│   ├── webhook/                # Webhook 测试（按消息类型组织）
│   │   ├── __init__.py
│   │   ├── .env.example        # 环境变量模板（同时用作文件消息的发送素材）
│   │   ├── conftest.py         # 配置加载与公共 fixtures
│   │   ├── test_text.py        # 文本消息测试（同步/异步）
│   │   ├── test_markdown.py    # Markdown 消息测试（同步/异步）
│   │   ├── test_markdown_v2.py # Markdown V2 消息测试（同步/异步）
│   │   ├── test_image.py       # 图片消息测试（同步/异步）
│   │   ├── test_news.py        # 图文消息测试（同步/异步）
│   │   ├── test_file.py        # 文件消息测试（同步/异步）
│   │   ├── test_voice.py       # 语音消息测试（同步/异步）
│   │   └── test_template_card.py # 模版卡片消息测试（同步/异步）
│   └── server/                 # Server 端测试（按功能模块组织）
│       ├── __init__.py
│       ├── .env.example        # 环境变量模板（SERVER_ 前缀）
│       ├── conftest.py         # 配置加载与公共 fixtures
│       ├── test_token.py       # access_token 获取与缓存测试
│       ├── test_iplist.py      # 接口 IP 段 / 回调 IP 段测试
│       ├── test_text.py        # 文本应用消息发送测试（同步/异步）
│       └── test_media.py       # 素材上传下载测试（同步/异步）
├── examples/                   # 演示代码（文件名以 example_ 开头）
│   ├── webhook/                # Webhook 示例（与 tests/webhook 按消息类型对应）
│   │   ├── .env.example        # 环境变量模板（同时用作文件消息的发送素材）
│   │   ├── example_config.py   # 配置读取（WEBHOOK_KEY）
│   │   ├── example_text.py     # 文本消息示例
│   │   ├── example_markdown.py # Markdown 消息示例
│   │   ├── example_markdown_v2.py # Markdown V2 消息示例
│   │   ├── example_image.py    # 图片消息示例
│   │   ├── example_news.py     # 图文消息示例
│   │   ├── example_file.py     # 文件消息示例（上传 .env.example 后发送）
│   │   ├── example_voice.py    # 语音消息示例（生成最小 AMR 后发送）
│   │   └── example_template_card.py # 模版卡片消息示例
│   └── server/                 # Server 端示例（与 tests/server 按功能模块对应）
│       ├── .env.example        # 环境变量模板（SERVER_ 前缀）
│       ├── example_config.py   # 配置读取（SERVER_CORP_ID 等）
│       ├── example_token.py    # access_token 获取与缓存示例
│       ├── example_iplist.py   # IP 段查询示例
│       ├── example_text.py     # 发送文本应用消息示例
│       └── example_media.py    # 素材上传/下载示例

安装

基础安装（推荐使用 uv）

uv pip install py-wecom-toolkit

# 或使用 pip 安装
pip install py-wecom-toolkit

包含 Server 端支持（diskcache / redis 缓存）

uv pip install py-wecom-toolkit[server]

# 或使用 pip 安装
pip install py-wecom-toolkit[server]

开发环境（包含测试、代码检查、打包、上传工具）

uv pip install py-wecom-toolkit[dev]

# 或使用 pip 安装
pip install py-wecom-toolkit[dev]

功能特性

Webhook 消息发送模块

• 支持全部 8 种 Webhook 消息类型: text, markdown, markdown_v2, image, news, file, voice, template_card
• 基于 pydantic v2 的消息模型验证
• 基于 httpx 的异步 HTTP 客户端
• 同步与异步客户端双模式支持 (WeComWebhookClient / AsyncWeComWebhookClient)
• 消息发送支持 @ 成员列表合并与去重
• 文件上传接口 (base64 编码 + md5 校验)

Server 端 API 模块

• access_token 管理: 自动获取与缓存（支持 memory/diskcache/redis 三种后端）
• 接口 IP 段: 获取企业微信服务器 IP 段
• 回调 IP 段: 获取企业微信回调服务器 IP 段
• 应用消息发送: 文本、markdown、图片、文件、语音、图文、模版卡片等
• 素材管理: 上传临时素材、下载临时素材
• 同步与异步客户端双模式支持 (WeComServerClient / AsyncWeComServerClient)
• agentid 支持：可在客户端初始化时指定，发送消息时未传则自动使用客户端 agentid

快速开始

Webhook 消息发送（同步）

from py_wecom_toolkit import WeComWebhookClient

# 创建同步客户端，支持上下文管理器自动管理连接池
with WeComWebhookClient(key="你的webhook_key") as client:

    # 发送文本消息
    resp = client.send_text("Hello, World!")
    print(f"发送结果: {resp.errmsg}")

    # 发送文本消息并 @某人
    resp = client.send_text(
        "你好, 请查看",
        mentioned_list=["zhangsan"],
        mentioned_mobile_list=["13800001111"],
    )
    print(f"发送结果: {resp.errmsg}")

    # 发送 Markdown 消息
    resp = client.send_markdown(
        '实时新增用户反馈<font color="warning">132例</font>'
    )
    print(f"发送结果: {resp.errmsg}")

    # 发送图文消息
    resp = client.send_news([{
        "title": "中秋节礼品领取",
        "description": "今年中秋节公司有豪礼相送",
        "url": "https://work.weixin.qq.com",
        "picurl": "https://example.com/pic.png",
    }])
    print(f"发送结果: {resp.errmsg}")

    # 发送图片
    resp = client.send_image("/path/to/image.png")
    print(f"发送结果: {resp.errmsg}")

    # 上传文件并发送
    media_id = client.upload_media("/path/to/file.pdf")
    resp = client.send_file(media_id)
    print(f"发送结果: {resp.errmsg}")

Webhook 消息发送（异步）

import asyncio
from py_wecom_toolkit import AsyncWeComWebhookClient

async def main():
    async with AsyncWeComWebhookClient(key="你的webhook_key") as client:

        # 发送文本消息
        resp = await client.send_text("Hello, World!")
        print(f"发送结果: {resp.errmsg}")

        # 发送 Markdown 消息
        resp = await client.send_markdown(
            '实时新增用户反馈<font color="warning">132例</font>'
        )
        print(f"发送结果: {resp.errmsg}")

        # 上传文件并发送
        media_id = await client.upload_media("/path/to/file.pdf")
        resp = await client.send_file(media_id)
        print(f"发送结果: {resp.errmsg}")

asyncio.run(main())

Server 端 API（同步）

from py_wecom_toolkit import WeComServerClient
from py_wecom_toolkit.server.cache import WeComServerDiskCache

# 关键配置请通过环境变量或配置文件读取，不要硬编码在代码中
client = WeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
    agentid="你的AgentID",
    cache=WeComServerDiskCache(directory=".runtime.cache.diskcache"),  # 默认使用 diskcache 缓存 access_token
)

# 获取 access_token（自动缓存）
token = client.get_access_token()
print(f"access_token: {token}")

# 获取企业微信接口 IP 段
iplist = client.get_iplist()
print(f"接口 IP 段: {iplist.ip_list}")

# 获取企业微信回调 IP 段
callback_ip = client.get_callback_ip()
print(f"回调 IP 段: {callback_ip.ip_list}")

# 发送文本应用消息
resp = client.send_text(
    content="Hello, WeCom!",
    touser="Guolei",
)
print(f"发送结果: {resp.errmsg}")

# 上传素材并发送文件消息
media = client.upload_media("/path/to/README.md", media_type="file")
resp = client.send_file(media.media_id, touser="Guolei")
print(f"发送文件结果: {resp.errmsg}")

client.close()

Server 端 API（异步）

import asyncio
from py_wecom_toolkit import AsyncWeComServerClient
from py_wecom_toolkit.server.cache import WeComServerDiskCache

async def main():
    async with AsyncWeComServerClient(
        corpid="你的CorpID",
        corpsecret="你的CorpSecret",
        agentid="你的AgentID",
        cache=WeComServerDiskCache(directory=".runtime.cache.diskcache"),  # 默认使用 diskcache 缓存 access_token
    ) as client:

        # 获取 access_token（自动缓存）
        token = await client.async_get_access_token()
        print(f"access_token: {token}")

        # 获取接口 IP 段
        iplist = await client.async_get_iplist()
        print(f"接口 IP 段: {iplist.ip_list}")

        # 发送文本应用消息
        resp = await client.async_send_text(
            content="Hello, WeCom Async!",
            touser="Guolei",
        )
        print(f"发送结果: {resp.errmsg}")

        # 上传素材并发送文件消息
        media = await client.async_upload_media("/path/to/README.md", media_type="file")
        resp = await client.async_send_file(media.media_id, touser="Guolei")
        print(f"发送文件结果: {resp.errmsg}")

asyncio.run(main())

高级用法

import httpx
from py_wecom_toolkit import WeComWebhookClient, AsyncWeComWebhookClient
from py_wecom_toolkit.server.cache import WeComServerDiskCache, WeComServerRedisCache

# 使用自定义 httpx 客户端（Webhook 异步客户端）
async with httpx.AsyncClient(timeout=30.0) as http_client:
    client = AsyncWeComWebhookClient(key="xxx", client=http_client)
    await client.send_text("使用自定义客户端")

# 使用上下文管理器（Webhook 同步客户端）
with WeComWebhookClient(key="xxx") as client:
    client.send_text("自动管理连接池")

# 使用 diskcache 缓存 access_token
client = WeComServerClient(
    corpid="xxx",
    corpsecret="yyy",
    agentid="zzz",
    cache=WeComServerDiskCache(directory=".runtime.cache.diskcache"),
)

# 使用 redis 缓存 access_token
client = WeComServerClient(
    corpid="xxx",
    corpsecret="yyy",
    agentid="zzz",
    cache=WeComServerRedisCache(host="localhost", port=6379),
)

API 参考

WeComWebhookClient / AsyncWeComWebhookClient

WeComServerClient / AsyncWeComServerClient

测试用例

测试按消息类型组织，会 真实调用企业微信 Webhook 接口 发送消息：

• 每条测试消息以 【同步】/【异步】 前缀标记客户端类型，末尾附带当前时间；
• 每次发送完成后，自动补发一条文本消息，报告刚才的发送是否成功；
• 文件消息使用 tests/webhook/.env.example 作为上传/发送素材；
• 企业微信机器人限制 20 条/分钟，测试会自动按固定间隔发送，全套约需 2 分钟。

运行前请在 tests/webhook/ 下准备 .env（直接设置 WEBHOOK_KEY 环境变量也可）：

cp tests/webhook/.env.example tests/webhook/.env

# 运行全部 Webhook 实测测试
pytest tests/webhook/

# 运行单个消息类型的测试
pytest tests/webhook/test_text.py

测试覆盖（同步客户端 + 异步客户端各一组）：

• tests/webhook/test_text.py - 文本消息
• tests/webhook/test_markdown.py - Markdown 消息
• tests/webhook/test_markdown_v2.py - Markdown V2 消息
• tests/webhook/test_image.py - 图片消息
• tests/webhook/test_news.py - 图文消息
• tests/webhook/test_file.py - 文件消息
• tests/webhook/test_voice.py - 语音消息
• tests/webhook/test_template_card.py - 模版卡片消息

Server 端测试 (tests/server/)

按功能模块组织，会 真实调用企业微信 Server API，配置未填写或仍为占位值时自动跳过：

# 运行全部 Server 端实测测试
pytest tests/server/

# 运行单个模块
pytest tests/server/test_token.py

运行前请在 tests/server/ 下准备 .env（直接设置 SERVER_ 前缀环境变量也可）：

cp tests/server/.env.example tests/server/.env

环境变量（前缀 SERVER_）：

测试覆盖（同步客户端 + 异步客户端各一组）：

• tests/server/test_token.py - access_token 获取与缓存（获取 / 缓存命中 / 清空重取）
• tests/server/test_iplist.py - 接口 IP 段 / 回调 IP 段查询
• tests/server/test_text.py - 发送文本应用消息
• tests/server/test_media.py - 素材上传 / 下载（文件使用当前目录 .env.example）

演示代码

示例与测试一一对应，文件名以 example_ 开头，每个示例同时包含同步与异步用法，运行后会打印发送是否成功：

# Webhook 示例（examples/webhook/.env 需配置 WEBHOOK_KEY）
cd examples/webhook
python example_text.py

# Server 端示例（examples/server/.env 需配置 SERVER_CORP_ID 等）
cd examples/server
python example_token.py

未配置时会提示缺少哪个环境变量并自动退出。

构建与发布

# 安装开发依赖
pip install py-wecom-toolkit[dev]

# 构建分发包
python -m build

# 上传到 PyPI
python -m twine upload dist/*

消息类型参考

参考 企业微信文档 获取 Webhook 消息格式说明。

参考 企业微信服务端 API 获取 Server 端 API 说明。

License

MIT