py-wecom-toolkit

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

• 版本: 3.0.0
• 作者: 郭磊
• 邮箱: 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       # WebhookClient / AsyncWebhookClient
│       │   ├── exceptions.py   # Webhook 异常定义
│       │   ├── models.py       # pydantic v2 消息模型（8种消息类型）
│       │   └── types.py        # 类型别名
│       └── server/             # Server 端 API 模块（企业微信服务端接口）
│           ├── __init__.py
│           ├── client.py       # WeComServerClient / AsyncWeComServerClient
│           ├── token.py        # AccessToken 管理（支持缓存）
│           ├── cache.py        # MemoryCache / DiskCache / RedisCache
│           ├── models.py       # Server API 响应模型
│           ├── exceptions.py   # Server 异常定义
│           ├── message.py      # 应用消息发送（MessageSendMixin）
│           └── media.py        # 素材上传/下载（MediaMixin）
├── tests/                      # 测试用例
│   ├── __init__.py
│   ├── webhook/                # Webhook 测试
│   │   ├── __init__.py
│   │   ├── test_client.py      # 同步客户端测试
│   │   ├── test_async_client.py # 异步客户端测试
│   │   ├── test_models.py      # 消息模型测试
│   │   └── test_integration.py # 真实集成测试
│   └── server/                 # Server 端测试
│       ├── __init__.py
│       ├── test_client.py      # 同步客户端测试
│       ├── test_async_client.py # 异步客户端测试
│       ├── test_token.py       # AccessToken 管理器测试
│       ├── test_cache.py       # 缓存测试
│       ├── test_integration_real.py # 真实集成测试
│       └── test_upload_readme_to_guolei.py # 上传README并发送测试
└── examples/                   # 演示代码
    ├── webhook/                # Webhook 示例
    │   ├── README.md
    │   ├── .env.example        # 环境变量模板
    │   ├── config.py           # 配置读取
    │   ├── basic.py            # 基础用法（上下文管理器、手动连接、自定义配置、原始dict）
    │   ├── text.py             # 文本消息示例
    │   ├── markdown.py         # Markdown 消息示例
    │   ├── markdown_v2.py      # Markdown V2 消息示例
    │   ├── image.py            # 图片消息示例
    │   ├── news.py             # 图文消息示例
    │   ├── file.py             # 文件消息示例
    │   ├── voice.py            # 语音消息示例
    │   ├── template_card.py    # 模版卡片消息示例
    │   ├── async_basic.py      # 异步基础示例
    │   ├── async_text.py       # 异步文本消息示例
    │   └── async_file.py       # 异步文件消息示例
    └── server/                 # Server 端示例
        ├── README.md
        ├── .env.example        # 环境变量模板
        ├── config.py           # 配置读取
        ├── basic_sync.py       # 同步基础示例
        ├── basic_async.py      # 异步基础示例
        ├── send_message.py     # 发送应用消息示例
        └── upload_file.py      # 上传文件并发送示例

安装

pip install py-wecom-toolkit

安装包含 Server 端支持 (diskcache / redis 缓存):

pip install py-wecom-toolkit[server]

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

pip install py-wecom-toolkit[dev]

功能特性

Webhook 模块

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

Server 端模块

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

快速开始

Webhook 消息发送

import asyncio
from py_wecom_toolkit import WebhookClient

async def main():
    client = WebhookClient(key="你的webhook_key")

    # 发送文本消息
    await client.send_text("Hello, World!")

    # 发送文本消息并 @某人
    await client.send_text(
        "你好, 请查看",
        mentioned_list=["zhangsan"],
        mentioned_mobile_list=["13800001111"],
    )

    # 发送 Markdown 消息
    await client.send_markdown(
        '实时新增用户反馈<font color="warning">132例</font>'
    )

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

    # 发送图片
    await client.send_image("/path/to/image.png")

    # 上传文件并发送
    media_id = await client.upload_media("/path/to/file.pdf")
    await client.send_file(media_id)

    await client.close()

asyncio.run(main())

Webhook 消息发送 (同步)

from py_wecom_toolkit import WebhookClient

# 创建同步客户端，支持上下文管理器自动管理连接池
with WebhookClient(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}")

Server 端 API (同步)

from py_wecom_toolkit import WeComServerClient

# 关键配置请通过环境变量或配置文件读取，不要硬编码在代码中
client = WeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
    agentid="你的AgentID",
)

# 获取 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

async def main():
    client = AsyncWeComServerClient(
        corpid="你的CorpID",
        corpsecret="你的CorpSecret",
        agentid="你的AgentID",
    )

    # 获取 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}")

    await client.close()

asyncio.run(main())

高级用法

import httpx
from py_wecom_toolkit import WebhookClient

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

# 使用上下文管理器
async with WebhookClient(key="xxx") as client:
    await client.send_text("自动管理连接池")

# 使用 diskcache 缓存 access_token
from py_wecom_toolkit.server.cache import DiskCache
client = WeComServerClient(
    corpid="xxx",
    corpsecret="yyy",
    agentid="zzz",
    cache=DiskCache(directory=".wecom_cache"),
)

# 使用 redis 缓存 access_token
from py_wecom_toolkit.server.cache import RedisCache
client = WeComServerClient(
    corpid="xxx",
    corpsecret="yyy",
    agentid="zzz",
    cache=RedisCache(host="localhost", port=6379),
)

API 参考

WebhookClient / AsyncWebhookClient

WeComServerClient / AsyncWeComServerClient

测试用例

项目包含完整的单元测试和集成测试：

# 运行所有测试
pytest tests/

# 仅运行 Webhook 测试
pytest tests/webhook/

# 仅运行 Server 端测试
pytest tests/server/

# 运行真实集成测试（需要配置有效的 Webhook Key / CorpID/CorpSecret/AgentID）
pytest tests/webhook/test_integration.py
pytest tests/server/test_integration_real.py

测试覆盖：

• tests/webhook/test_client.py - 同步 Webhook 客户端测试
• tests/webhook/test_async_client.py - 异步 Webhook 客户端测试
• tests/webhook/test_models.py - Webhook 消息模型测试
• tests/webhook/test_integration.py - Webhook 真实集成测试
• tests/server/test_client.py - 同步 Server 客户端测试
• tests/server/test_async_client.py - 异步 Server 客户端测试
• tests/server/test_token.py - AccessToken 管理器测试
• tests/server/test_cache.py - 缓存测试 (memory/diskcache/redis)
• tests/server/test_integration_real.py - Server 端真实集成测试
• tests/server/test_upload_readme_to_guolei.py - 上传 README 并发送测试

演示代码

Webhook 示例 (examples/webhook/)

Server 端示例 (examples/server/)

使用前请先配置环境变量或 .env 文件：

cp examples/webhook/.env.example examples/webhook/.env
# 或
cp examples/server/.env.example examples/server/.env

构建与发布

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

# 构建分发包
python -m build

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

消息类型参考

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

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

License

MIT