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/ 中实际实现为准。

安装

从源码安装:

uv pip install -e .
# 或
pip install -e .

构建 wheel:

uv build

构建产物会输出到 dist/

uv add ./dist/wecomkit-0.1.0-py3-none-any.whl
# 或
pip install ./dist/wecomkit-0.1.0-py3-none-any.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 是可选依赖:

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.0.tar.gz (52.0 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.0-py3-none-any.whl (60.7 kB view details)

Uploaded Python 3

File details

Details for the file wecomkit-0.1.0.tar.gz.

File metadata

  • Download URL: wecomkit-0.1.0.tar.gz
  • Upload date:
  • Size: 52.0 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.0.tar.gz
Algorithm Hash digest
SHA256 2f7cfb766c18f50e0917ad5ef84be0c1f13234914e5b1e8bf0721cf93c29c763
MD5 3591c71968d89054f98ec1be821604d7
BLAKE2b-256 3f154769a7e65d7b92029a5ea7f8f9ebe7cff9cd3c8715c230c838362676f30a

See more details on using hashes here.

File details

Details for the file wecomkit-0.1.0-py3-none-any.whl.

File metadata

  • Download URL: wecomkit-0.1.0-py3-none-any.whl
  • Upload date:
  • Size: 60.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.0-py3-none-any.whl
Algorithm Hash digest
SHA256 a413e5e4ef639b1a822793ab19586272b80ec431e156176f0f3ddf4079c468d3
MD5 546797873df064227c5f120c716302bd
BLAKE2b-256 ff7e70fca8645fbc5128321fab78312ecbc3a07602436b73d0a6bb644c63cb84

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