open-event-sdk-python

WPS Open Platform Event SDK for Python — 通过 WebSocket 长连接实时接收平台事件。

安装

pip install open-event-sdk

快速开始

基础用法 — 单一处理器

from open_event_sdk import Client, Event


def handle_event(event: Event) -> None:
    print(f"收到事件: code={event.event_code}, data={event.data}")


client = Client(
    app_id="your_app_id",
    app_secret="your_app_secret",
    handler=handle_event,
)
client.start()

按事件类型分发 — Dispatcher

from open_event_sdk import Client, Dispatcher, Event

dispatcher = Dispatcher()


def on_message(event: Event) -> None:
    print(f"收到消息: {event.data}")


dispatcher.register_func("kso.app_chat.message.create", on_message)

client = Client(
    app_id="your_app_id",
    app_secret="your_app_secret",
    dispatcher=dispatcher,
)
client.start()

类型化事件处理 — Typed Events

from open_event_sdk import Client, Dispatcher
from open_event_sdk.event.typed_event import V7AppChatMessageCreateEvent


async def on_message(event: V7AppChatMessageCreateEvent) -> None:
    data = event.parsed_data
    print(f"消息来自: {data.sender.id}")
    print(f"消息内容: {data.message.content}")


dispatcher = Dispatcher()
dispatcher.on_v7_app_chat_message_create(on_message)

client = Client(
    app_id="your_app_id",
    app_secret="your_app_secret",
    dispatcher=dispatcher,
)
client.start()

异步启动（适用于已有 event loop 的场景）

import asyncio
from open_event_sdk import Client, Event


async def handle_event(event: Event) -> None:
    print(f"收到事件: {event.event_code}")


async def main() -> None:
    client = Client(
        app_id="your_app_id",
        app_secret="your_app_secret",
        handler=handle_event,
    )
    await client.astart()


asyncio.run(main())

配置选项

参数	类型	默认值	说明
app_id	str	必填	应用 ID
app_secret	str	必填	应用密钥
handler	Handler | HandlerFunc	None	事件处理器（与 dispatcher 二选一）
dispatcher	Dispatcher	None	事件分发器（与 handler 二选一）
endpoint	str	wss://openapi.wps.cn/v7/event/ws	WebSocket 端点
base_url	str	None	基础 URL（会自动拼接事件路径）
auto_reconnect	bool	True	是否自动重连
reconnect_base_interval	float	1.0	重连基础间隔（秒）
reconnect_max_interval	float	60.0	重连最大间隔（秒）
reconnect_multiplier	float	2.0	重连间隔倍数
reconnect_max_retry	int	-1	最大重试次数（-1 无限）
reconnect_jitter	float	0.2	重连抖动系数
write_timeout	float	10.0	写超时（秒）
pong_timeout	float	90.0	Pong 超时（秒）
ack_mode	bool	True	是否启用 ACK 模式
tls_verify	bool	False	是否校验 TLS 证书（也可通过环境变量 OPEN_EVENT_TLS_VERIFY 开启）
logger	Logger	None	自定义日志实例
log_level	LogLevel	LogLevel.INFO	日志级别

TLS 证书校验

SDK 默认跳过 TLS 证书校验，方便私有化部署或连接使用自签证书的服务。连接公网正式环境时，建议开启校验。

方式	说明
默认	跳过证书校验
tls_verify=True	构造参数开启证书校验
OPEN_EVENT_TLS_VERIFY=1 或 true	环境变量开启证书校验

优先级：构造参数 > 环境变量 > 默认值。

# 生产环境建议开启证书校验
client = Client(
    app_id="your_app_id",
    app_secret="your_app_secret",
    handler=handle_event,
    tls_verify=True,
)

# 或通过环境变量开启
export OPEN_EVENT_TLS_VERIFY=1

ACK 模式

启用 ACK 模式后（默认开启），SDK 会向服务端报告事件处理结果：

• 处理器正常返回 → 发送 code=200 的 ACK
• 处理器抛出异常 → 发送 code=500 的 ACK，服务端会触发重试

安全机制

• KSO-1 签名认证：WebSocket 握手使用 HMAC-SHA256 签名
• 消息签名验证：每条事件消息都经过 HMAC-SHA256 签名验证
• AES-256-CBC 加密：事件数据使用 AES-CBC 加密传输

重连策略

SDK 使用指数退避 + 抖动策略自动重连：

delay = min(base_interval × multiplier^(n-1), max_interval) × (1 ± jitter)

依赖

• Python >= 3.9
• websockets >= 13.0
• pycryptodome >= 3.20

License

MIT