Skip to main content

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

运行示例前设置环境变量:

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_data
  • get_checkin_schedules
  • get_checkin_option
  • add_checkin_user_face
  • get_checkin_daydata
  • get_checkin_monthdata
  • get_checkin_all_options
  • get_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_template
  • approval.apply
  • approval.get_approve
  • approval.get_approve_list
  • approval.get_vacation_config
  • approval.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_fields
  • hr.get_staff_info
  • hr.update_staff_info
  • hr.text_attr
  • hr.uint32_attr

微盘

已实现:

  • wedrive.get_space_info
  • wedrive.create_space
  • wedrive.rename_space
  • wedrive.dismiss_space
  • wedrive.set_space
  • wedrive.share_space
  • wedrive.add_space_acl
  • wedrive.delete_space_acl
  • wedrive.create_folder
  • wedrive.upload_file
  • wedrive.download_file
  • wedrive.delete_file
  • wedrive.list_file
  • wedrive.get_file_meta
  • wedrive.rename_file
  • wedrive.move_file
  • wedrive.set_file
  • wedrive.set_file_security
  • wedrive.add_file_acl
  • wedrive.delete_file_acl
  • wedrive.get_share_link
  • wedrive.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


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

wecomkit-0.1.1.tar.gz (53.2 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

wecomkit-0.1.1-py3-none-any.whl (61.7 kB view details)

Uploaded Python 3

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

Hashes for wecomkit-0.1.1.tar.gz
Algorithm Hash digest
SHA256 27430393f35cc2b40164190828048d4f4f6d908737c374a6d885a19cf17cdbf6
MD5 d31d2e89880b4fc200cedbdae9ae25ff
BLAKE2b-256 f9f9991f46b85bdc6a46ce19ec331fd468d13510d6947f897ffc7ee712c9e327

See more details on using hashes here.

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

Hashes for wecomkit-0.1.1-py3-none-any.whl
Algorithm Hash digest
SHA256 ed60be9f8e70a55a533e98f1b3d39613e1a84e11e09ee3b4e54523917cb778ee
MD5 99fe48780d8910cdcc791b7a83f3e0f1
BLAKE2b-256 1654b01d2b73084d46a7fb0c62548837001b9d0a451c27190fdb507939d95760

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page