py-wecom-toolkit 1.0.5

企业微信 SDK，提供企业微信 API 的 Python 封装，支持同步和异步调用。

py-wecom-toolkit

企业微信 API 的 Python SDK，提供完整的企业微信接口封装，支持同步和异步调用方式。

项目主页

• 项目地址: https://gitee.com/guolei19850528/py_wecom_toolkit

功能特性

Webhook 模块 (py_wecom_toolkit.webhook)

• 企业微信 Webhook 消息发送
• 支持文本、Markdown、MarkdownV2、图片、图文、文件、语音、模板卡片等消息类型
• 支持同步和异步两种调用方式
• 媒体文件上传功能

Server 模块 (py_wecom_toolkit.server)

• 企业微信 Server API 集成
• 获取 access_token（支持 diskcache 和 Redis 缓存）
• 获取 API 域名 IP 列表
• 获取回调 IP 列表
• 消息发送和素材管理
• 支持同步和异步两种调用方式

依赖包

httpx>=0.27.0
pydantic>=2.0
jsonpath-ng>=1.5.3
jsonschema>=4.21.0
diskcache>=5.6.3
redis>=4.6.0
py-httpx-toolkit>=1.0.1

安装方式

pip install py-wecom-toolkit

UV 安装

uv add py-wecom-toolkit

项目结构

src/py_wecom_toolkit/
├── __init__.py              # 主模块入口
├── server/                  # Server API 模块
│   ├── __init__.py          # Base 类（access_token管理、IP查询）
│   ├── materials.py         # 素材管理
│   ├── messages.py          # 消息模型
│   ├── responses.py         # 响应模型
│   └── utils.py             # 工具函数
└── webhook/                 # Webhook 模块
    ├── __init__.py          # Webhook 客户端类
    ├── messages.py          # 消息模型
    ├── responses.py         # 响应模型
    └── utils.py             # 工具函数

使用示例

Webhook 使用示例

同步发送文本消息：

from py_wecom_toolkit.webhook import Webhook
from py_wecom_toolkit.webhook.messages import Text, TextContent

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

# 发送文本消息
response = webhook.send(Text(text=TextContent(content="Hello World")))
print(response.json())

异步发送消息：

import asyncio
from py_wecom_toolkit.webhook import Webhook
from py_wecom_toolkit.webhook.messages import Markdown, MarkdownContent

async def main():
    webhook = Webhook(key="your_webhook_key")
    response = await webhook.async_send(
        Markdown(markdown=MarkdownContent(content="**Hello** from async"))
    )
    print(response.json())

asyncio.run(main())

上传媒体文件：

from py_wecom_toolkit.webhook import Webhook

webhook = Webhook(key="your_webhook_key")

with open("test.txt", "rb") as f:
    response = webhook.upload_media(files={"file": ("test.txt", f)})
print(response.json())  # {"errcode": 0, "media_id": "xxx", ...}

Server 使用示例

获取 access_token：

from py_wecom_toolkit.server import Base
from diskcache import Cache

# 创建缓存实例
cache = Cache("./cache")

# 初始化 Server 客户端
server = Base(
    corpid="your_corpid",
    corpsecret="your_corpsecret",
    cache_config={"instance": cache}
)

# 刷新 access_token
server.refresh_access_token()
print(server.access_token)

异步获取 access_token：

import asyncio
import redis
from py_wecom_toolkit.server import Base

async def main():
    r = redis.Redis(host='localhost', port=6379, db=0)
    server = Base(
        corpid="your_corpid",
        corpsecret="your_corpsecret",
        cache_config={"instance": r}
    )
    await server.async_refresh_access_token()
    print(server.access_token)

asyncio.run(main())

获取回调 IP：

from py_wecom_toolkit.server import Base

server = Base(corpid="your_corpid", corpsecret="your_corpsecret")
server.refresh_access_token()
response = server.getcallbackip()
print(response.json())

API 说明

Webhook 模块 API

Webhook 类

初始化参数：

参数	类型	必填	默认值	说明
key	str	是	-	Webhook密钥
base_url	str	否	https://qyapi.weixin.qq.com/	企业微信API基础URL
mentioned_list	list/tuple	否	[]	默认@用户ID列表
mentioned_mobile_list	list/tuple	否	[]	默认@用户手机号列表
client_kwargs	dict	否	{}	HTTP客户端配置参数

方法：

方法名	说明	参数	返回值
send(message)	同步发送消息	message: 消息对象或字典	httpx.Response
async_send(message)	异步发送消息	message: 消息对象或字典	await httpx.Response
upload_media(file_type, **kwargs)	同步上传媒体文件	file_type: "file"或"voice"，files: 文件参数	httpx.Response
async_upload_media(file_type, **kwargs)	异步上传媒体文件	file_type: "file"或"voice"，files: 文件参数	await httpx.Response

Webhook 消息类型

消息类型	类名	说明
文本	Text	支持@用户，最长2048字节
Markdown	Markdown	支持标准Markdown语法
MarkdownV2	MarkdownV2	支持字体颜色等扩展样式
图片	Image	需要base64编码和md5值
图文	News	支持1-8篇文章
文件	File	需要先上传获取media_id
语音	Voice	需要先上传获取media_id
模板卡片	TemplateCard	支持多种卡片类型

文本消息示例：

from py_wecom_toolkit.webhook.messages import Text, TextContent

message = Text(
    text=TextContent(
        content="Hello World",
        mentioned_list=["user1", "@all"],
        mentioned_mobile_list=["13800138000"]
    )
)

图文消息示例：

from py_wecom_toolkit.webhook.messages import News, NewsContent, NewsArticle

message = News(
    news=NewsContent(
        articles=[
            NewsArticle(
                title="文章标题",
                description="文章描述",
                url="https://example.com",
                picurl="https://example.com/image.jpg"
            )
        ]
    )
)

Server 模块 API

Base 类

初始化参数：

参数	类型	必填	默认值	说明
corpid	str	是	-	企业ID
corpsecret	str	是	-	应用密钥
agentid	str/int	否	-	企业应用ID
base_url	str	否	https://qyapi.weixin.qq.com/	企业微信API基础URL
cache_config	dict	否	{}	缓存配置
client_kwargs	dict	否	{}	HTTP客户端配置参数

cache_config 参数：

参数	类型	必填	默认值	说明
instance	Cache/Redis	否	None	缓存实例（diskcache.Cache或redis.Redis）
key	str	否	pywecom_server_{corpid}_{agentid}	缓存键名
expire	int	否	7100	缓存过期时间（秒）

方法：

方法名	说明	参数	返回值
gettoken()	同步获取access_token	-	httpx.Response
async_gettoken()	异步获取access_token	-	await httpx.Response
get_api_domain_ip()	同步获取API域名IP	-	httpx.Response
async_get_api_domain_ip()	异步获取API域名IP	-	await httpx.Response
getcallbackip()	同步获取回调IP	-	httpx.Response
async_getcallbackip()	异步获取回调IP	-	await httpx.Response
refresh_access_token()	同步刷新access_token	cache_config	self
async_refresh_access_token()	异步刷新access_token	cache_config	await self
request_with_access_token()	同步请求，自动添加access_token	client, client_kwargs, **kwargs	httpx.Response
async_request_with_access_token()	异步请求，自动添加access_token	client, client_kwargs, **kwargs	await httpx.Response

Sender 类

继承自 Base 类，新增消息发送方法：

方法名	说明	参数	返回值
send(message)	同步发送消息	message: 消息对象或字典	httpx.Response
async_send(message)	异步发送消息	message: 消息对象或字典	await httpx.Response

Server 消息类型

消息类型	类名	说明
文本	Text	支持@用户
图片	Image	需要media_id
语音	Voice	需要media_id
视频	Video	需要media_id
文件	File	需要media_id
文本卡片	TextCard	带标题和链接的卡片
图文	News	支持1-8篇文章
图文(mpnews)	MpNews	素材库文章
Markdown	Markdown	Markdown格式消息
小程序通知	MiniprogramNotice	小程序模板消息
模板卡片	TemplateCard	支持多种卡片类型

发送文本消息示例：

from py_wecom_toolkit.server.messages import Sender, Text, TextContent

sender = Sender(corpid="your_corpid", corpsecret="your_corpsecret", agentid=1000001)
sender.refresh_access_token()
response = sender.send(Text(text=TextContent(content="Hello")))

发送图文消息示例：

from py_wecom_toolkit.server.messages import Sender, News, NewsArticlesContent, NewsArticleContent

sender = Sender(corpid="your_corpid", corpsecret="your_corpsecret", agentid=1000001)
sender.refresh_access_token()

message = News(
    news=NewsArticlesContent(
        articles=[
            NewsArticleContent(
                title="文章标题",
                description="描述",
                url="https://example.com",
                picurl="https://example.com/image.jpg"
            )
        ]
    )
)
response = sender.send(message)

响应模型

响应结果可通过 response.json() 获取，企业微信API响应结构：

{
    "errcode": 0,          # 错误码，0表示成功
    "errmsg": "ok",        # 错误信息
    "access_token": "...", # access_token（仅gettoken接口）
    "media_id": "...",     # 媒体文件ID（仅上传接口）
    "ip_list": [...]       # IP列表（仅IP查询接口）
}

错误码说明：

错误码	说明
0	成功
40001	获取access_token时AppSecret错误
40002	不合法的凭证类型
40003	不合法的UserID
40014	不合法的access_token
40033	不合法的字符
40056	不合法的agentid
40060	不合法的media_id
50000	企业不存在
60001	应用不存在
60002	应用已停用

工具函数说明

Webhook 工具函数 (py_wecom_toolkit.webhook.utils)

函数名	说明	参数	返回值
json_find_first(expression, data)	使用JSONPath查找第一个匹配项	expression: JSONPath表达式，data: JSON数据	匹配值或None
json_is_valid(schema, data)	JSON Schema校验	schema: Schema字典，data: 待校验数据	bool
image_to_base64_and_md5(image_path)	图片转Base64和MD5	image_path: 图片文件路径	Tuple[str, str]
build_success_instance(response)	转换为Success响应模型	response: Response对象或dict	Success
build_send_instance(response)	转换为Send响应模型	response: Response对象或dict	Send
build_upload_media_instance(response)	转换为UploadMedia响应模型	response: Response对象或dict	UploadMedia
success_validator(response, schema)	校验errcode是否为0	response: Response对象或dict	bool

图片转换示例：

from py_wecom_toolkit.webhook.utils import image_to_base64_and_md5
from py_wecom_toolkit.webhook.messages import Image, ImageContent

base64_str, md5_str = image_to_base64_and_md5("path/to/image.jpg")
message = Image(image=ImageContent(base64=base64_str, md5=md5_str))

响应校验示例：

from py_wecom_toolkit.webhook import Webhook
from py_wecom_toolkit.webhook.utils import build_send_instance, success_validator

webhook = Webhook(key="your_key")
response = webhook.send(message)

# 校验响应
if success_validator(response):
    result = build_send_instance(response)
    print(f"发送成功: {result.errmsg}")
else:
    print("发送失败")

Server 工具函数 (py_wecom_toolkit.server.utils)

函数名	说明	参数	返回值
json_find_first(expression, data)	使用JSONPath查找第一个匹配项	expression: JSONPath表达式，data: JSON数据	匹配值或None
json_is_valid(schema, data)	JSON Schema校验	schema: Schema字典，data: 待校验数据	bool
build_success_instance(response)	转换为Success响应模型	response: Response对象或dict	Success
success_is_valid(response, schema)	校验errcode是否为0	response: Response对象或dict	bool

响应校验示例：

from py_wecom_toolkit.server.messages import Sender, Text, TextContent
from py_wecom_toolkit.server.utils import build_success_instance, success_is_valid

sender = Sender(corpid="your_corpid", corpsecret="your_corpsecret", agentid=1000001)
sender.refresh_access_token()
response = sender.send(Text(text=TextContent(content="Hello")))

# 校验响应
if success_is_valid(response):
    result = build_success_instance(response)
    print(f"发送成功: {result.errmsg}")
else:
    print("发送失败")

JSONPath查找示例：

from py_wecom_toolkit.server.utils import json_find_first

data = {
    "result": {
        "items": [
            {"id": 1, "name": "item1"},
            {"id": 2, "name": "item2"}
        ]
    }
}

# 使用JSONPath查找第一个item的name
name = json_find_first("$.result.items[0].name", data)
print(name)  # item1

参考文档

• 企业微信 Webhook API
• 企业微信 Server API

许可证

MIT License