| name | azure-messaging-webpubsubservice-py |
| description | Azure Web PubSub Service SDK for Python. Use for real-time messaging, WebSocket connections, and pub/sub patterns.
Triggers: "azure-messaging-webpubsubservice", "WebPubSubServiceClient", "real-time", "WebSocket", "pub/sub".
|
| license | MIT |
| metadata | {"author":"Microsoft","version":"1.0.0","package":"azure-messaging-webpubsubservice"} |
Azure Web PubSub Service SDK for Python
Real-time messaging with WebSocket connections at scale.
Installation
pip install azure-messaging-webpubsubservice
pip install azure-messaging-webpubsubclient
Environment Variables
AZURE_WEBPUBSUB_HUB=my-hub
AZURE_TOKEN_CREDENTIALS=prod
Authentication & Lifecycle
🔑 Two rules apply to every code sample below:
- Prefer
DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.
- Local dev:
DefaultAzureCredential works as-is.
- Production: set
AZURE_TOKEN_CREDENTIALS=prod (or AZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.
- Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:
- Sync:
with <Client>(...) as client:
- Async:
async with <Client>(...) as client: and async with DefaultAzureCredential() as credential: (from azure.identity.aio)
Snippets may abbreviate this setup, but production code should always follow both rules.
Service Client (Server-Side)
Authentication
from azure.messaging.webpubsubservice import WebPubSubServiceClient
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
credential = DefaultAzureCredential(require_envvar=True)
with WebPubSubServiceClient(
endpoint="https://<name>.webpubsub.azure.com",
hub="my-hub",
credential=credential
) as client:
...
Generate Client Access Token
token = client.get_client_access_token()
print(f"URL: {token['url']}")
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"]
)
Send to All Clients
client.send_to_all(message="Hello everyone!", content_type="text/plain")
client.send_to_all(
message={"type": "notification", "data": "Hello"},
content_type="application/json"
)
Send to User
client.send_to_user(
user_id="user123",
message="Hello user!",
content_type="text/plain"
)
Send to Group
client.send_to_group(
group="my-group",
message="Hello group!",
content_type="text/plain"
)
Send to Connection
client.send_to_connection(
connection_id="abc123",
message="Hello connection!",
content_type="text/plain"
)
Group Management
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")
Connection Management
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")
Grant/Revoke Permissions
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"
)
Client SDK (Python WebSocket Client)
from azure.messaging.webpubsubclient import WebPubSubClient
with WebPubSubClient(credential=token["url"]) as client:
@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.send_to_group("my-group", "Hello from Python!")
Async Service Client
from azure.messaging.webpubsubservice.aio import WebPubSubServiceClient
from azure.identity.aio import DefaultAzureCredential
async def broadcast():
async with DefaultAzureCredential() as credential:
async with WebPubSubServiceClient(
endpoint="https://<name>.webpubsub.azure.com",
hub="my-hub",
credential=credential
) as client:
await client.send_to_all("Hello async!", content_type="text/plain")
Client Operations
| Operation | Description |
|---|
get_client_access_token | Generate WebSocket connection URL |
send_to_all | Broadcast to all connections |
send_to_user | Send to specific user |
send_to_group | Send to group members |
send_to_connection | Send to specific connection |
add_user_to_group | Add user to group |
remove_user_from_group | Remove user from group |
close_connection | Disconnect client |
connection_exists | Check connection status |
Best Practices
- Pick sync OR async and stay consistent. Do not mix
azure.xxx sync clients with azure.xxx.aio async clients in the same call path. Choose one mode per module.
- Always use context managers for clients and async credentials. Wrap every client in
with Client(...) as client: (sync) or async with Client(...) as client: (async). For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.
- Use
DefaultAzureCredential for portable auth across local dev and Azure (avoid connection strings / access keys when possible).
- Use roles to limit client permissions
- Use groups for targeted messaging
- Generate short-lived tokens for security
- Use user IDs to send to users across connections
- Handle reconnection in client applications
- Use JSON content type for structured data
- Close connections gracefully with reasons