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/

安装

基础安装（推荐使用 uv）

uv pip install py-wecom-toolkit

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

# 或从本地源码安装（开发模式）
uv pip install -e .

带缓存依赖安装（diskcache / redis）

uv pip install py-wecom-toolkit[cache]

# 或从本地源码安装（开发模式，带缓存依赖）
uv pip install -e ".[cache]"

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

uv pip install py-wecom-toolkit[dev]

# 或从本地源码安装（开发模式）
uv pip install -e ".[dev]"

快速开始

Webhook 消息发送（同步）

from py_wecom_toolkit.webhook.client 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.webhook.client 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.server.client 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.server.client 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"),
    ) as client:

        # 获取 access_token（自动缓存，方法名与同步客户端一致）
        token = await client.get_access_token()
        print(f"access_token: {token}")

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

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

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

asyncio.run(main())

API 参考

Webhook 客户端

WeComWebhookClient（同步）

AsyncWeComWebhookClient（异步）

异步方法名与同步客户端完全一致（不带 async_ 前缀），支持 async with 上下文管理器。

Server 端客户端

WeComServerClient（同步）

AsyncWeComServerClient（异步）

异步方法名与同步客户端完全一致（不带 async_ 前缀），支持 async with 上下文管理器。

缓存后端

异常

测试

测试用例位于 tests/ 目录，基于 pytest + pytest-asyncio，统一从 .env 读取配置。

# 安装测试依赖
uv pip install -e ".[dev]"

# 运行全部测试
pytest tests/ -q

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

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

未配置相关环境变量时，测试会自动跳过。

项目结构

py-wecom-toolkit/
├── pyproject.toml              # 项目构建配置
├── README.md                   # 项目说明文档
├── LICENSE                     # MIT 许可证
├── src/
│   └── py_wecom_toolkit/       # 源代码包
│       ├── __init__.py         # 包入口
│       ├── py.typed            # PEP 561 类型标记
│       ├── webhook/            # Webhook 消息发送模块（原“群机器人”）
│       │   ├── __init__.py
│       │   ├── client.py       # WeComWebhookClient / AsyncWeComWebhookClient / WeComWebhookClientConfig
│       │   ├── models.py       # pydantic v2 消息模型与异常定义
│       │   └── types.py        # 类型别名
│       └── server/             # Server 端 API 模块（企业微信服务端接口）
│           ├── __init__.py
│           ├── client.py       # WeComServerClient / AsyncWeComServerClient / WeComServerClientConfig
│           ├── token.py        # WeComServerAccessTokenManager / AsyncWeComServerAccessTokenManager
│           ├── cache.py        # WeComServerMemoryCache / WeComServerDiskCache / WeComServerRedisCache
│           ├── models.py       # Server API 响应模型与异常定义
│           ├── message.py      # 应用消息发送（WeComServerMessageSendMixin / AsyncWeComServerMessageSendMixin）
│           └── media.py        # 素材上传/下载（WeComServerMediaMixin / AsyncWeComServerMediaMixin）
├── tests/                      # 测试用例
│   ├── webhook/                # Webhook 测试（按消息类型组织）
│   └── server/                 # Server 端测试（按功能模块组织）
└── examples/                   # 演示代码
    ├── webhook/                # Webhook 示例
    └── server/                 # Server 端示例

消息类型参考

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

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

License

MIT