Async WeCom (Enterprise WeChat) SDK built with httpx.
Project description
wecomkit 企业微信 SDK
English | 中文
wecomkit 是一个基于 asyncio + httpx 的企业微信异步 SDK。当前版本覆盖常用服务端 API 场景,包括通讯录、应用消息、素材、打卡、日程、审批、企业邮箱、人事助手、微盘和 WeDoc 的部分接口。
当前 SDK 不是企业微信全量 API 封装。接口状态以
src/wecomkit/services/中实际实现为准。
安装
从 PyPI 安装:
uv pip install wecomkit
# 或
pip install wecomkit
如果当前项目由 uv 管理,也可以直接添加依赖:
uv add wecomkit
需要 Redis token 缓存时安装可选依赖:
uv pip install "wecomkit[redis]"
# 或
uv add "wecomkit[redis]"
从源码开发安装:
git clone https://github.com/KevinFan1/wecomkit.git
cd wecomkit
uv pip install -e ".[dev]"
构建本地 wheel:
uv build
构建产物会输出到 dist/,可用于本地验证安装:
uv pip install ./dist/wecomkit-*.whl
# 或
pip install ./dist/wecomkit-*.whl
快速开始
import asyncio
from wecomkit import AsyncWeComClient, JSONFileTokenCache, WeComConfig
async def main() -> None:
config = WeComConfig(
corp_id="wwxxxxxx",
corp_secret="your-secret",
)
async with AsyncWeComClient(
config,
token_cache=JSONFileTokenCache(".cache/wecom-token.json"),
) as client:
token = await client.auth.get_access_token()
departments = await client.departments.list()
users = await client.users.simple_list_by_department(1, fetch_child=True)
print(token)
print(departments)
print(users)
asyncio.run(main())
Examples
示例代码位于 examples:
- examples/quickstart.py:初始化客户端、获取 token、读取部门和成员
- examples/contacts.py:成员、部门、标签常用操作
- examples/messages_and_media.py:上传素材并发送应用消息
- examples/mail.py:企业邮箱发送、未读数和邮件列表
- examples/hr.py:人事助手字段配置和员工花名册
- examples/robot.py:群机器人 webhook 消息
- examples/wedoc_smartsheet.py:WeDoc 文档和智能表格
- examples/approval.py:审批查询与假期余额
- examples/checkin.py:打卡记录、排班、规则和日报
- examples/wedrive.py:微盘空间、文件列表、下载链接
- examples/calendar_schedule.py:日历与日程查询
运行示例前设置环境变量:
export WECOM_CORP_ID="wwxxxxxx"
export WECOM_CORP_SECRET="your-secret"
export WECOM_AGENT_ID="1000001"
uv run python examples/quickstart.py
客户端结构
client.auth # 授权
client.users # 成员
client.departments # 部门
client.tags # 标签
client.agent # 应用
client.menu # 应用菜单
client.messages # 应用消息
client.media # 素材
client.mail # 企业邮箱
client.robot # 群机器人 webhook
client.hr # 人事助手
client.checkin # 打卡
client.approval # 审批
client.calendar # 日历
client.schedule # 日程
client.wedrive # 微盘
client.wedoc # WeDoc 聚合服务
WeDoc 子服务:
client.wedoc.documents # 文档
client.wedoc.smartsheets # 智能表格
client.wedoc.permissions # 权限
client.wedoc.forms # 收集表通用调用
client.wedoc.materials # 文档素材
已实现接口
授权
| SDK 方法 | 企业微信接口 |
|---|---|
client.auth.get_access_token() |
/cgi-bin/gettoken |
client.auth.get_user_info_by_code(code) |
/cgi-bin/auth/getuserinfo |
client.auth.get_user_detail(user_ticket) |
/cgi-bin/auth/getuserdetail |
通讯录
| 模块 | 已实现 |
|---|---|
| 成员 | create, update, delete, batch_delete, get, list_by_department, simple_list_by_department, list_id, get_join_qrcode, invite |
| 部门 | create, update, delete, list, simple_list, get |
| 标签 | create, update, delete, get, add_users, delete_users, list |
对应接口包括 /cgi-bin/user/*、/cgi-bin/department/*、/cgi-bin/tag/* 和 /cgi-bin/invite/send。
应用与消息
| 模块 | 已实现 |
|---|---|
| 应用 | agent.get, agent.list, agent.set |
| 菜单 | menu.create, menu.get, menu.delete |
| 应用消息 | messages.send, send_text, send_markdown, send_textcard, send_news, send_mpnews, send_miniprogram_notice, send_template_card, send_image, send_file, send_video, send_voice |
| 群机器人 | robot.send, send_text, send_markdown, send_image, send_file, send_news, send_textcard, send_template |
messages.send() 可直接发送任意已支持消息类型;常用消息类型也提供了便捷方法。
素材
| SDK 方法 | 企业微信接口 |
|---|---|
media.upload(file_path, media_type) |
/cgi-bin/media/upload |
media.upload_bytes(content, media_type, filename) |
/cgi-bin/media/upload |
media.get(media_id) |
/cgi-bin/media/get |
media.download(media_id, path) |
/cgi-bin/media/get |
media.upload_image(file_path) |
/cgi-bin/media/uploadimg |
media.upload_attachment(file_path) |
/cgi-bin/media/upload_attachment |
media.upload_jssdk_image(file_path) |
/cgi-bin/media/upload_attachment |
打卡
已实现:
get_checkin_dataget_checkin_schedulesget_checkin_optionadd_checkin_user_faceget_checkin_daydataget_checkin_monthdataget_checkin_all_optionsget_checkin_user_options
日历与日程
| 模块 | 已实现 |
|---|---|
| 日历 | calendar.create, calendar.get, calendar.update, calendar.delete |
| 日程 | schedule.create, schedule.get, schedule.list_by_calendar, schedule.update, schedule.cancel, schedule.add_attendees, schedule.del_attendees |
审批
已实现:
approval.get_templateapproval.applyapproval.get_approveapproval.get_approve_listapproval.get_vacation_configapproval.get_user_vacation
企业邮箱
| 模块 | 已实现 |
|---|---|
| 邮件收发 | mail.compose_send, send_mail, send_schedule_mail, send_meeting_mail, get_mail_list, get_mail_content, get_mail_unread_count |
| 应用邮箱 | mail.update_app_mailbox, get_app_mailbox |
| 邮件群组 | mail.create_mail_group, update_mail_group, delete_mail_group, get_mail_group, search_mail_group |
| 公共邮箱 | mail.create_public_mailbox, update_public_mailbox, delete_public_mailbox, get_public_mailbox, search_public_mailbox |
| 客户端密码 | mail.get_client_password_list, delete_client_password |
| 高级账号 | mail.allocate_mail_advanced_account, deallocate_mail_advanced_account, get_mail_advanced_account_list |
| 用户邮箱属性 | mail.toggle_mailbox_status, enable_mailbox, disable_mailbox, get_user_mail_attribute, update_user_mail_attribute |
人事助手
已实现:
hr.get_fieldshr.get_staff_infohr.update_staff_infohr.text_attrhr.uint32_attr
微盘
已实现:
wedrive.get_space_infowedrive.create_spacewedrive.rename_spacewedrive.dismiss_spacewedrive.set_spacewedrive.share_spacewedrive.add_space_aclwedrive.delete_space_aclwedrive.create_folderwedrive.upload_filewedrive.download_filewedrive.delete_filewedrive.list_filewedrive.get_file_metawedrive.rename_filewedrive.move_filewedrive.set_filewedrive.set_file_securitywedrive.add_file_aclwedrive.delete_file_aclwedrive.get_share_linkwedrive.cancel_share
WeDoc
| 模块 | 已实现 |
|---|---|
| 文档 | documents.create, rename, delete, get_base_info, share, get_content, edit_content |
| 智能表格记录 | smartsheets.add_records, delete_records, update_records, get_records |
| 智能表格子表 | smartsheets.add_sheet, delete_sheet, update_sheet, get_sheets |
| 智能表格视图 | smartsheets.add_view, delete_view, update_view, get_views |
| 智能表格字段 | smartsheets.add_fields, delete_fields, update_fields, get_fields |
| 智能表格编组 | smartsheets.add_group, delete_group, update_group, get_groups |
| 权限 | permissions.get_auth, set_auth, mod_member, mod_join_rule |
| 收集表 | forms.call(endpoint, payload), create_collect, modify_collect, get_info, get_answer, get_statistic |
| 素材 | materials.upload_image |
未实现或仅部分覆盖
以下模块当前没有封装,或只覆盖了少量常用接口:
- 客户联系、客户群、客户朋友圈
- 微信客服
- 会话内容存档
- 企业支付、红包
- 会议、会议室、直播
- 家校沟通
- 上下游、企业互联
- 汇报、公费电话、紧急通知
- 数据与智能专区
返回值与类型
服务方法默认返回企业微信原始 JSON 响应,即 dict[str, Any]。wecomkit.types 提供 Pydantic v2 模型,调用方可以按需自行校验:
from wecomkit.types import BaseResp, UserInfo
resp = await client.users.get("zhangsan")
user = UserInfo.model_validate(resp)
ok = BaseResp.model_validate({"errcode": 0, "errmsg": "ok"})
当前服务方法没有统一的 response_model 参数。
智能表格和 WeDoc 权限也提供了宽松模型,未知字段会保留,便于兼容接口扩展:
from wecomkit.types import SmartSheetRecordListResponse
resp = await client.wedoc.smartsheets.get_records(docid, sheet_id)
records = SmartSheetRecordListResponse.model_validate(resp)
重试策略
默认不重试。需要对网络错误或企业微信限流错误做简单重试时,在 WeComConfig 中配置:
config = WeComConfig(
corp_id="wwxx",
corp_secret="secret",
max_retries=2,
retry_backoff=0.5,
)
等待时间按 retry_backoff * 2 ** attempt 递增。
Token 缓存
SDK 默认使用内存缓存 token,并在过期前 token_refresh_buffer 秒提前刷新。生产环境建议使用文件缓存或自定义缓存:
from wecomkit import AsyncWeComClient, JSONFileTokenCache, WeComConfig
client = AsyncWeComClient(
WeComConfig(corp_id="wwxx", corp_secret="secret"),
token_cache=JSONFileTokenCache(".cache/wecom-token.json"),
)
多进程或多实例部署建议使用 Redis。Redis 是可选依赖:
uv pip install "wecomkit[redis]"
# 或
uv add "wecomkit[redis]"
from redis.asyncio import Redis
from wecomkit import AsyncWeComClient, RedisTokenCache, WeComConfig
redis = Redis.from_url("redis://localhost:6379/0")
client = AsyncWeComClient(
WeComConfig(corp_id="wwxx", corp_secret="secret"),
token_cache=RedisTokenCache(redis, key="wecomkit:access_token"),
)
也可以直接通过 URL 创建:
from wecomkit import RedisTokenCache
token_cache = RedisTokenCache.from_url(
"redis://localhost:6379/0",
key="wecomkit:access_token",
)
自定义缓存:
from wecomkit.token_cache import TokenCache
class RedisTokenCache(TokenCache):
async def get(self) -> str | None:
...
async def set(self, token: str, expires_in: int) -> None:
...
异常处理
from wecomkit.exceptions import WeComAPIError, WeComNetworkError
try:
user = await client.users.get("invalid_userid")
except WeComAPIError as e:
print(e.errcode, e.errmsg)
except WeComNetworkError as e:
print("network error", e)
开发
uv run pytest
uv run ruff check .
uv build
License
MIT
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file wecomkit-0.1.1.tar.gz.
File metadata
- Download URL: wecomkit-0.1.1.tar.gz
- Upload date:
- Size: 53.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.10
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
27430393f35cc2b40164190828048d4f4f6d908737c374a6d885a19cf17cdbf6
|
|
| MD5 |
d31d2e89880b4fc200cedbdae9ae25ff
|
|
| BLAKE2b-256 |
f9f9991f46b85bdc6a46ce19ec331fd468d13510d6947f897ffc7ee712c9e327
|
File details
Details for the file wecomkit-0.1.1-py3-none-any.whl.
File metadata
- Download URL: wecomkit-0.1.1-py3-none-any.whl
- Upload date:
- Size: 61.7 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.10
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
ed60be9f8e70a55a533e98f1b3d39613e1a84e11e09ee3b4e54523917cb778ee
|
|
| MD5 |
99fe48780d8910cdcc791b7a83f3e0f1
|
|
| BLAKE2b-256 |
1654b01d2b73084d46a7fb0c62548837001b9d0a451c27190fdb507939d95760
|