Azure Messaging Web PubSub 服务 Python SDK
Azure Web PubSub Service Python SDK
实现大规模 WebSocket 连接的实时消息传递。
安装
# 服务端 SDK (server-side)
pip install azure-messaging-webpubsubservice
客户端 SDK (用于 Python WebSocket 客户端)
pip install azure-messaging-webpubsubclient环境变量
AZURE_WEBPUBSUB_CONNECTION_STRING=Endpoint=https://<name>.webpubsub.azure.com;AccessKey=...
AZURE_WEBPUBSUB_HUB=my-hub服务客户端 (服务端)
身份验证
from azure.messaging.webpubsubservice import WebPubSubServiceClient
使用连接字符串
client = WebPubSubServiceClient.from_connection_string(
connection_string=os.environ["AZURE_WEBPUBSUB_CONNECTION_STRING"],
hub="my-hub"
)
使用 Entra ID
from azure.identity import DefaultAzureCredential
client = WebPubSubServiceClient(
endpoint="https://<name>.webpubsub.azure.com",
hub="my-hub",
credential=DefaultAzureCredential()
)
生成客户端访问令牌
# 为匿名用户生成令牌
token = client.get_client_access_token()
print(f"URL: {token['url']}")
为指定用户 ID 生成令牌
token = client.get_client_access_token(
user_id="user123",
roles=["webpubsub.sendToGroup", "webpubsub.joinLeaveGroup"]
)
为指定组生成令牌
token = client.get_client_access_token(
user_id="user123",
groups=["group1", "group2"]
)发送给所有客户端
# 发送文本
client.send_to_all(message="Hello everyone!", content_type="text/plain")
发送 JSON
client.send_to_all(
message={"type": "notification", "data": "Hello"},
content_type="application/json"
)发送给指定用户
client.send_to_user(
user_id="user123",
message="Hello user!",
content_type="text/plain"
)发送给指定组
client.send_to_group(
group="my-group",
message="Hello group!",
content_type="text/plain"
)发送给指定连接
client.send_to_connection(
connection_id="abc123",
message="Hello connection!",
content_type="text/plain"
)组管理
# 将用户添加到组
client.add_user_to_group(group="my-group", user_id="user123")
将用户从组中移除
client.remove_user_from_group(group="my-group", user_id="user123")
将连接添加到组
client.add_connection_to_group(group="my-group", connection_id="abc123")
将连接从组中移除
client.remove_connection_from_group(group="my-group", connection_id="abc123")连接管理
# 检查连接是否存在
exists = client.connection_exists(connection_id="abc123")
检查用户是否有连接
exists = client.user_exists(user_id="user123")
检查组是否有连接
exists = client.group_exists(group="my-group")
关闭连接
client.close_connection(connection_id="abc123", reason="Session ended")
关闭用户的所有连接
client.close_all_connections(user_id="user123")授予/撤销权限
from azure.messaging.webpubsubservice import WebPubSubServiceClient
授予权限
client.grant_permission(
permission="joinLeaveGroup",
connection_id="abc123",
target_name="my-group"
)
撤销权限
client.revoke_permission(
permission="joinLeaveG检查权限
has_permission = client.check_permission( permission="joinLeaveGroup", connection_id="abc123", target_name="my-group" )## 客户端 SDK (Python WebSocket 客户端)client = WebPubSubClient(credential=token["url"])
事件处理器
@client.on("connected") def on_connected(e): print(f"已连接: {e.connection_id}")@client.on("server-message")
def on_message(e):
print(f"消息: {e.data}")
@client.on("group-message")
def on_group_message(e):
print(f"群组 {e.group}: {e.data}")
连接并发送
client.open() client.send_to_group("my-group", "Hello from Python!")## 异步服务客户端async def broadcast():
credential = DefaultAzureCredential()
client = WebPubSubServiceClient(
endpoint="https://<name>.webpubsub.azure.com",
hub="my-hub",
credential=credential
)
await client.send_to_all("Hello async!", content_type="text/plain")
await client.close()
await credential.close()
``
客户端操作
| 操作 | 描述 |
|-----------|-------------|
| get_client_access_token | 生成 WebSocket 连接 URL |send_to_all
| | 向所有连接广播 |send_to_user
| | 发送给特定用户 |send_to_group
| | 发送给群组成员 |send_to_connection
| | 发送给特定连接 |add_user_to_group
| | 将用户添加到群组 |remove_user_from_group
| | 将用户从群组中移除 |close_connection
| | 断开客户端连接 |connection_exists` | 检查连接状态 |
|
最佳实践
1. 使用角色 来限制客户端权限
2. 使用群组 进行定向消息推送
3. 生成短效令牌 以增强安全性
4. 使用用户 ID 向跨连接的用户发送消息
5. 在客户端应用程序中 处理重连机制
6. 对于结构化数据，使用 JSON 内容类型
7. 优雅地 关闭连接 并提供原因
适用场景
本技能适用于执行概览中所描述的工作流或操作。局限性
- 仅在任务与上述范围明确匹配时使用此技能。
- 不要将输出视为针对特定环境的验证、测试或专家评审的替代方案。
- 如果缺少必要的输入、权限、安全边界或成功标准，请停止操作并请求澄清。
