用途
Azure Web PubSub Service SDK for Python 用于在规模化的 WebSocket 连接上实现实时消息传递，适用于实时消息、WebSocket 连接和发布/订阅（pub/sub）模式。
安装
服务端 SDK：
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")
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="joinLeaveGroup",
connection_id="abc123",
target_name="my-group"
)
has_permission = client.check_permission(
permission="joinLeaveGroup",
connection_id="abc123",
target_name="my-group"
)
客户端 SDK（Python WebSocket 客户端）
from azure.messaging.webpubsubclient import WebPubSubClient
client = WebPubSubClient(credential=token["url"])
@client.on("connected")
def on_connected(e):
print(f"Connected: {e.connection_id}")
@client.on("server-message")
def on_message(e):
print(f"Message: {e.data}")
@client.on("group-message")
def on_group_message(e):
print(f"Group {e.group}: {e.data}")
client.open()
client.send_to_group("my-group", "Hello from Python!")
异步服务客户端
from azure.messaging.webpubsubservice.aio import WebPubSubServiceClient
from azure.identity.aio import DefaultAzureCredential
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 | 检查连接状态 |
最佳实践
- 使用 roles 限制客户端权限
- 使用 groups 进行定向消息传递
- 生成短期令牌以保障安全
- 使用用户 ID 跨连接向用户发送消息
- 在客户端应用中处理重连
- 对结构化数据使用 JSON 内容类型
- 使用原因优雅地关闭连接
适用场景
该技能适用于执行概述中描述的工作流或操作。
限制
- 仅在任务明确匹配上述范围时使用此技能。
- 不要将输出视为环境特定验证、测试或专家审查的替代品。
- 如果缺少必需的输入、权限、安全边界或成功标准，请停止并请求澄清。