py-wecom-toolkit

文档语言

• 中文 (README.md)
• English (README.en.md)

介绍

py-wecom-toolkit 是一个企业微信 Python 扩展包，支持 Webhook（群机器人） 与 Server 端 API。提供同步/异步双模客户端、基于 Pydantic v2 的消息模型，以及自动缓存的 access_token 管理。

py-wecom-toolkit 让你极其简单地与企业微信 API 交互。无需手动拼接 URL，无需手动构建 JSON 负载，自动缓存 access_token，支持同步和异步客户端。

核心特性

• 双模客户端：同时提供同步（WeComWebhookClient / WeComServerClient）与异步（AsyncWeComWebhookClient / AsyncWeComServerClient）客户端，方法名完全一致，支持 with / async with 上下文管理器。
• Webhook 群机器人：支持文本、Markdown、Markdown V2、图片、图文、文件、语音、模版卡片消息发送，以及媒体文件上传。
• Server 端 API：支持 access_token 自动获取与缓存、接口/回调 IP 段查询、应用消息发送（文本/Markdown/图片/语音/视频/文件/文本卡片/图文/mpnews/markdown_card）、素材上传/下载。
• 自动缓存 access_token：支持内存（默认）、diskcache、Redis 三种缓存后端，自动处理 token 过期与刷新（提前 5 分钟视为过期）。
• Pydantic v2 消息模型：所有消息载荷基于 Pydantic v2 构建，字段自动校验，类型安全。
• 基于 httpx：底层 HTTP 客户端使用 httpx，支持同步与异步请求。

参考文档

• 企业微信 Webhook 接口文档
• 企业微信 Server 端 API 文档
• access_token 接口文档
• 应用消息发送接口文档
• 素材管理接口文档

软件架构

py-wecom-toolkit/
├── pyproject.toml              # 项目构建配置（PEP 621 / setuptools）
├── README.md                   # 项目说明文档（中文）
├── README.en.md                # 项目说明文档（英文）
├── LICENSE                     # MIT 许可证
├── uv.lock                     # 依赖锁定文件
├── 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 示例（文本/Markdown/图片/图文/文件/语音/模版卡片，同步+异步）
    └── server/                 # Server 端示例（文本/token/素材/IP 段，同步+异步）

安装教程

推荐使用 uv 进行安装：

1. 基础安装（推荐 uv）

uv pip install py-wecom-toolkit

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

uv pip install py-wecom-toolkit[cache]

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

uv pip install py-wecom-toolkit[dev]

4. 从源码安装

git clone https://gitee.com/guolei19850528/py_wecom_toolkit.git
cd py_wecom_toolkit
uv pip install -e .

使用说明

一、Webhook 群机器人（Webhook）

Webhook 适用于企业微信群机器人场景，只需配置 Webhook Key 即可发送消息到群聊。

1. 创建客户端

from py_wecom_toolkit import WeComWebhookClient, AsyncWeComWebhookClient

# 同步客户端
client = WeComWebhookClient(key="你的webhook_key")

# 异步客户端
async_client = AsyncWeComWebhookClient(key="你的webhook_key")

推荐使用上下文管理器，自动管理 HTTP 连接池：

with WeComWebhookClient(key="你的webhook_key") as client:
    # 使用 client 发送消息
    pass

async with AsyncWeComWebhookClient(key="你的webhook_key") as client:
    # 使用 client 发送消息
    pass

2. 发送文本消息（同步 / 异步）

同步：

from py_wecom_toolkit import WeComWebhookClient

with WeComWebhookClient(key="你的webhook_key") as client:
    resp = client.send_text("Hello, World!", mentioned_list=["@all"])
    print(f"发送结果: {resp.errmsg}")

异步：

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!", mentioned_list=["@all"])
        print(f"发送结果: {resp.errmsg}")

asyncio.run(main())

3. 发送 Markdown 消息

with WeComWebhookClient(key="你的webhook_key") as client:
    content = """### 标题示例

- 列表项一
- 列表项二
**加粗文本**"""
    resp = client.send_markdown(content)

Markdown V2：

resp = client.send_markdown_v2("# 标题\n\n正文内容")

4. 发送图片消息

with WeComWebhookClient(key="你的webhook_key") as client:
    resp = client.send_image("path/to/image.png")

5. 发送图文消息（最多 8 条）

with WeComWebhookClient(key="你的webhook_key") as client:
    articles = [
        {
            "title": "标题 1",
            "description": "描述 1",
            "url": "https://example.com",
        },
        {
            "title": "标题 2",
            "description": "描述 2",
            "url": "https://example.com",
            "picurl": "https://example.com/image.png",
        },
    ]
    resp = client.send_news(articles)

6. 发送文件消息（先上传获取 media_id）

with WeComWebhookClient(key="你的webhook_key") as client:
    media_id = client.upload_media("path/to/file.pdf", media_type="file")
    resp = client.send_file(media_id)

7. 发送语音消息（先上传获取 media_id）

with WeComWebhookClient(key="你的webhook_key") as client:
    media_id = client.upload_media("path/to/voice.amr", media_type="voice")
    resp = client.send_voice(media_id)

8. 发送模版卡片消息

文本通知卡片（text_notice）：

with WeComWebhookClient(key="你的webhook_key") as client:
    card = {
        "card_type": "text_notice",
        "source": {"desc": "py_wecom_toolkit"},
        "main_title": {"title": "卡片标题", "desc": "标题辅助信息"},
        "emphasis_content": {"title": "关键数据", "desc": "数据描述"},
        "sub_title_text": "二级普通文本",
        "card_action": {"type": 1, "url": "https://example.com"},
    }
    resp = client.send_template_card(card)

图文展示卡片（news_notice）：

card = {
    "card_type": "news_notice",
    "source": {"desc": "py_wecom_toolkit"},
    "main_title": {"title": "卡片标题"},
    "card_image": {"url": "https://example.com/image.png"},
    "image_text_area": {
        "type": 1,
        "url": "https://example.com",
        "title": "左图右文标题",
        "desc": "左图右文描述",
        "image_url": "https://example.com/image.png",
    },
    "card_action": {"type": 1, "url": "https://example.com"},
}
resp = client.send_template_card(card)

9. 默认 @ 提醒（可选）

在初始化客户端时可设置默认的 mentioned_list 和 mentioned_mobile_list，每次发送文本消息时自动合并：

client = WeComWebhookClient(
    key="你的webhook_key",
    mentioned_list=["@all"],
    mentioned_mobile_list=["13800000000"],
)
client.send_text("这条消息会自动 @ 所有人")

二、Server 端 API（企业微信服务端接口）

Server 端适用于企业微信应用开发，需要配置 CorpID、CorpSecret 和 AgentID。

1. 创建客户端

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

# 同步客户端（使用磁盘缓存）
client = WeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
    agentid="你的AgentID",
    cache=WeComServerDiskCache(directory=".runtime.cache.diskcache"),
)

# 异步客户端（使用内存缓存，默认）
from py_wecom_toolkit.server.cache import WeComServerMemoryCache
async_client = AsyncWeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
    agentid="你的AgentID",
    cache=WeComServerMemoryCache(),
)

2. 获取 access_token（自动缓存）

with WeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
) as client:
    token = client.get_access_token()
    print(f"access_token: {token}")

    # 二次获取将命中缓存
    token_again = client.get_access_token()
    print(f"缓存命中: {token == token_again}")

    # 清除缓存
    client.clear_access_token_cache()

异步：

async with AsyncWeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
) as client:
    token = await client.get_access_token()
    print(f"access_token: {token}")

3. 查询 IP 段

with WeComServerClient(corpid="你的CorpID", corpsecret="你的CorpSecret") as client:
    # 接口 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}")

4. 发送应用消息

发送文本消息：

with WeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
    agentid="你的AgentID",
) as client:
    resp = client.send_text(
        content="Hello, WeCom!",
        touser="@all",
    )
    print(f"发送结果: {resp.errmsg}")

异步：

async with AsyncWeComServerClient(
    corpid="你的CorpID",
    corpsecret="你的CorpSecret",
    agentid="你的AgentID",
) as client:
    resp = await client.send_text(
        content="Hello, WeCom!",
        touser="@all",
    )
    print(f"发送结果: {resp.errmsg}")

发送 Markdown 消息：

resp = client.send_markdown(
    content="### Markdown 消息\n\n**加粗文本**",
    touser="@all",
)

发送图片消息（需先上传素材获取 media_id）：

# 上传素材
media_resp = client.upload_media("path/to/image.png", media_type="image")
media_id = media_resp.media_id

# 发送图片消息
resp = client.send_image(
    media_id=media_id,
    touser="@all",
)

发送文件消息：

media_resp = client.upload_media("path/to/file.pdf", media_type="file")
resp = client.send_file(
    media_id=media_resp.media_id,
    touser="@all",
)

发送语音消息：

media_resp = client.upload_media("path/to/voice.amr", media_type="voice")
resp = client.send_voice(
    media_id=media_resp.media_id,
    touser="@all",
)

发送视频消息：

media_resp = client.upload_media("path/to/video.mp4", media_type="video")
resp = client.send_video(
    media_id=media_resp.media_id,
    title="视频标题",
    description="视频描述",
    touser="@all",
)

发送文本卡片消息：

resp = client.send_textcard(
    title="卡片标题",
    description="卡片描述",
    url="https://example.com",
    btntxt="详情",
    touser="@all",
)

发送图文消息（news）：

resp = client.send_news(
    articles=[
        {
            "title": "文章标题 1",
            "description": "文章描述 1",
            "url": "https://example.com/1",
            "picurl": "https://example.com/1.png",
        },
    ],
    touser="@all",
)

发送 mpnews 消息：

resp = client.send_mpnews(
    articles=[
        {
            "title": "mpnews 标题",
            "thumb_media_id": "你的缩略图 media_id",
            "author": "作者",
            "content_source_url": "https://example.com",
            "content": "<h1>正文内容</h1>",
            "digest": "摘要",
        },
    ],
    touser="@all",
)

发送 markdown_card 消息：

resp = client.send_markdown_card(
    title="卡片标题",
    content="### Markdown 内容\n\n**加粗**",
    touser="@all",
)

5. 素材上传 / 下载

上传临时素材：

resp = client.upload_media(
    file_path="path/to/image.png",
    media_type="image",  # image / voice / video / file
)
print(f"media_id: {resp.media_id}")

下载临时素材：

data = client.download_media(media_id="你的media_id")
with open("downloaded_file", "wb") as f:
    f.write(data)

6. 指定 access_token（跳过自动获取）

如果你已有 access_token，可直接传入，跳过自动获取：

client = WeComServerClient(
    corpid="你的CorpID",
    access_token="你的access_token",  # 直接指定，不再自动获取
    agentid="你的AgentID",
)
resp = client.send_text(content="Hello", touser="@all")

三、异常处理

所有 API 调用失败时会抛出相应异常：

from py_wecom_toolkit import (
    WeComWebhookAPIError,
    WeComWebhookRateLimitError,
    WeComWebhookValidationError,
    WeComWebhookNetworkError,
)
from py_wecom_toolkit.server.models import (
    WeComServerAPIError,
    WeComServerNetworkError,
    WeComServerConfigError,
)

try:
    resp = client.send_text("Hello")
except WeComWebhookRateLimitError as e:
    print(f"触发频率限制: {e}")
except WeComWebhookAPIError as e:
    print(f"API 错误: [{e.errcode}] {e.errmsg}")
except WeComWebhookNetworkError as e:
    print(f"网络错误: {e}")

四、贡献指南

1. Fork 本仓库
2. 新建 Feat_xxx 分支
3. 提交代码
4. 新建 Pull Request

五、许可证

本项目采用 MIT 许可证。