py-wecom-toolkit

企业微信 SDK 工具包，提供企业微信 Webhook 和 Server API 的 Python 封装，支持同步和异步调用方式。

项目信息

• 项目地址: https://gitee.com/guolei19850528/py_wecom_toolkit
• 作者: Guo Lei
• 邮箱: 174000902@qq.com
• 许可证: MIT License

功能特性

• Webhook 模块: 企业微信机器人消息发送和媒体上传
• Server 模块: 企业微信服务端 API，包含 access_token 管理、消息发送、素材管理
• 支持同步/异步: 所有 API 均提供同步和异步两种调用方式
• 缓存支持: access_token 支持 diskcache 和 Redis 缓存
• 类型安全: 使用 Pydantic 进行数据验证和类型提示

项目结构

src/py_wecom_toolkit/
├── __init__.py              # 主模块入口，包含版本信息
├── server/                  # Server API 模块
│   ├── __init__.py          # Server 基础类（access_token管理、IP查询）
│   ├── models/              # 响应模型
│   │   ├── __init__.py
│   │   └── response.py      # Base、Success 响应模型
│   ├── materials/           # 素材管理
│   │   ├── __init__.py
│   │   └── media.py         # Media 类（临时素材上传、永久图片上传）
│   └── messages/            # 消息发送
│       ├── __init__.py      # Message 类（消息发送）
│       └── model.py         # 消息类型模型（Text、Markdown、Image 等）
└── webhook/                 # Webhook 模块
    ├── __init__.py          # Webhook 客户端类
    ├── models/              # 消息和响应模型
    │   ├── __init__.py
    │   ├── msgtype.py       # 消息类型模型
    │   └── response.py      # 响应模型
    └── util.py              # 工具函数（图片转 Base64 和 MD5）

安装

pip install py-wecom-toolkit

UV 安装

uv add py-wecom-toolkit

快速开始

Webhook 使用示例

from py_wecom_toolkit.webhook import Webhook
from py_wecom_toolkit.webhook.models import msgtype

# 初始化 Webhook
webhook = Webhook(key="your_webhook_key")

# 发送文本消息
webhook.send(msgtype.Text(text={"content": "Hello, World!"}))

# 发送 Markdown 消息
webhook.send(msgtype.Markdown(markdown={"content": "# 标题\n\n**粗体文本**"}))

# 发送图片消息（需先转换为 Base64 和 MD5）
from py_wecom_toolkit.webhook.util import image_to_base64_and_md5
base64_str, md5_str = image_to_base64_and_md5("image.jpg")
webhook.send(msgtype.Image(image={"base64": base64_str, "md5": md5_str}))

# 异步发送消息
import asyncio
asyncio.run(webhook.async_send(msgtype.Text(text={"content": "Async Message"})))

# 上传媒体文件
response = webhook.upload_media(ft="file", files={"media": open("test.pdf", "rb")})
media_id = response.json().get("media_id")

# 发送文件消息
webhook.send(msgtype.File(file={"media_id": media_id}))

Server 使用示例

from py_wecom_toolkit.server import Server
from py_wecom_toolkit.server.messages import Message
from py_wecom_toolkit.server.messages.model import Text
from py_wecom_toolkit.server.materials.media import Media

# 初始化 Server（基础类）
server = Server(
    corpid="your_corpid",
    corpsecret="your_corpsecret",
    agentid="your_agentid"
)

# 刷新 access_token（带缓存）
server.refresh_access_token()

# 使用 Message 类发送消息
message = Message(
    corpid="your_corpid",
    corpsecret="your_corpsecret",
    agentid="your_agentid"
)
message.refresh_access_token()
response = message.send(Text(text={"content": "Hello from Server API"}))

# 使用 Media 类上传素材
media = Media(
    corpid="your_corpid",
    corpsecret="your_corpsecret",
    agentid="your_agentid"
)
media.refresh_access_token()

# 上传临时素材
response = media.upload(ft="image", files={"media": open("test.jpg", "rb")})
media_id = response.json().get("media_id")

# 上传永久图片
response = media.uploadimg(files={"media": open("test.jpg", "rb")})
image_url = response.json().get("url")

# 异步调用示例
import asyncio
async def main():
    await message.async_refresh_access_token()
    await message.async_send(Text(text={"content": "Async Server Message"}))
asyncio.run(main())

官方文档

核心模块说明

Webhook 模块

Webhook 类 (webhook/__init__.py)

支持的消息类型:

• Text: 文本消息（支持 @ 指定用户）
• Markdown: Markdown 消息
• MarkdownV2: MarkdownV2 消息
• Image: 图片消息
• News: 图文消息
• File: 文件消息
• Voice: 语音消息
• TemplateCard: 模板卡片消息

Server 模块

Server 类 (server/__init__.py)

Message 类 (server/messages/__init__.py)

Media 类 (server/materials/media.py)

工具函数

Webhook 工具 (webhook/util.py)

响应模型

Server 响应 (server/models/response.py)

Webhook 响应 (webhook/models/response.py)

响应模型使用示例：

from py_wecom_toolkit.server.models.response import build_base_instance, Base

# 发送请求并解析响应
response = server.gettoken()
base_instance = build_base_instance(response)

# 检查响应是否成功
if base_instance.is_success():
    print("请求成功")
    print(f"access_token: {base_instance.access_token}")
else:
    print(f"请求失败: {base_instance.errcode} - {base_instance.errmsg}")

# 直接从字典构建
response_dict = {"errcode": 0, "errmsg": "ok", "access_token": "xxx"}
base_instance = Base(**response_dict)
if base_instance.is_success():
    print("响应成功")

缓存配置

Server 模块支持 access_token 缓存，减少 API 调用次数。

# 使用 diskcache
import diskcache
cache = diskcache.Cache("./cache")
server = Server(
    corpid="xxx",
    corpsecret="xxx",
    agentid="xxx",
    cache_config={"instance": cache}
)

# 使用 Redis
import redis
r = redis.Redis(host="localhost", port=6379)
server = Server(
    corpid="xxx",
    corpsecret="xxx",
    agentid="xxx",
    cache_config={"instance": r}
)

客户端配置

支持自定义 HTTP 客户端配置：

webhook = Webhook(
    key="your_key",
    client_kwargs={
        "timeout": 30,
        "verify": True,
        "proxies": {"http": "http://proxy:8080"}
    }
)

server = Server(
    corpid="xxx",
    corpsecret="xxx",
    client_kwargs={
        "timeout": 60,
        "verify": False
    }
)

依赖

• httpx >= 0.27.0
• pydantic >= 2.0
• py-httpx-toolkit >= 1.0.0
• diskcache >= 5.0（可选，用于缓存）
• redis >= 5.0（可选，用于缓存）

贡献

欢迎提交 Issue 和 Pull Request！

许可证

MIT License