sickn33/antigravity-awesome-skills/azure-messaging-webpubsubservice-py

Claude Code and agent skill by sickn33/antigravity-awesome-skills

Trust tier 3
  • community
Source check pending

Checking that the SKILL.md file is still in the repository.

Azure Web PubSub Service SDK for Python. Use for real-time messaging, WebSocket connections, and pub/sub patterns.

Install

npx vskill@latest install sickn33/antigravity-awesome-skills/azure-messaging-webpubsubservice-py

SKILL.md

Azure Web PubSub Service SDK for Python

Real-time messaging with WebSocket connections at scale.

Installation

bash
# Service SDK (server-side)
pip install azure-messaging-webpubsubservice
 
# Client SDK (for Python WebSocket clients)
pip install azure-messaging-webpubsubclient

Environment Variables

bash
AZURE_WEBPUBSUB_CONNECTION_STRING=Endpoint=https://<name>.webpubsub.azure.com;AccessKey=...
AZURE_WEBPUBSUB_HUB=my-hub

Service Client (Server-Side)

Authentication
python
import os
from azure.messaging.webpubsubservice import WebPubSubServiceClient
# Connection string
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()
)
Generate Client Access Token
python
# Token for anonymous user
token = client.get_client_access_token()
# Treat token["url"] as a bearer secret; return it only to the authenticated intended client.
# Token with user ID
token = client.get_client_access_token(
    user_id="user123",
    roles=["webpubsub.sendToGroup.my-group", "webpubsub.joinLeaveGroup.my-group"]
)
# Token with groups
token = client.get_client_access_token(
    user_id="user123",
    groups=["group1", "group2"]
)
Send to All Clients
python
# Send text
client.send_to_all(message="Hello everyone!", content_type="text/plain")
# Send JSON
client.send_to_all(
    message={"type": "notification", "data": "Hello"},
    content_type="application/json"
)
Send to User
python
client.send_to_user(
    user_id="user123",
    message="Hello user!",
    content_type="text/plain"
)
Send to Group
python
client.send_to_group(
    group="my-group",
    message="Hello group!",
    content_type="text/plain"
)
Send to Connection
python
client.send_to_connection(
    connection_id="abc123",
    message="Hello connection!",
    content_type="text/plain"
)
Group Management
python
# Add user to group
client.add_user_to_group(group="my-group", user_id="user123")
# Remove user from group
client.remove_user_from_group(group="my-group", user_id="user123")
# Add connection to group
client.add_connection_to_group(group="my-group", connection_id="abc123")
# Remove connection from group
client.remove_connection_from_group(group="my-group", connection_id="abc123")
Connection Management
python
# Check if connection exists
exists = client.connection_exists(connection_id="abc123")
# Check if user has connections
exists = client.user_exists(user_id="user123")
# Check if group has connections
exists = client.group_exists(group="my-group")
# Close connection
client.close_connection(connection_id="abc123", reason="Session ended")
# Close all connections for user
client.close_all_connections(user_id="user123")
Grant/Revoke Permissions
python
from azure.messaging.webpubsubservice import WebPubSubServiceClient
# Grant permission
client.grant_permission(
    permission="joinLeaveGroup",
    connection_id="abc123",
    target_name="my-group"
)
# Revoke permission
client.revoke_permission(
    permission="joinLeaveGroup",
    connection_id="abc123",
    target_name="my-group"
)
# Check permission
has_permission = client.check_permission(
    permission="joinLeaveGroup",
    connection_id="abc123",
    target_name="my-group"
)

Client SDK (Python WebSocket Client)

python
from azure.messaging.webpubsubclient import WebPubSubClient
from azure.messaging.webpubsubclient.models import CallbackType, WebPubSubDataType
with WebPubSubClient(token["url"]) as ws_client:
    ws_client.subscribe(CallbackType.CONNECTED, lambda e: print("Connected"))
    ws_client.subscribe(CallbackType.GROUP_MESSAGE, lambda e: print("Group message received"))
    ws_client.join_group("my-group")
    ws_client.send_to_group("my-group", "Hello from Python!", WebPubSubDataType.TEXT)

Authorize token issuance on the server; derive user IDs and allowed groups from the authenticated session, never untrusted request fields. Use a short explicit expiration supported by the pinned SDK, avoid logging access URLs or message bodies, and close service clients and credentials when done. These sends/group changes affect live clients; use a test hub for examples.

Async Service Client

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()

Client Operations

OperationDescription
get_client_access_tokenGenerate WebSocket connection URL
send_to_allBroadcast to all connections
send_to_userSend to specific user
send_to_groupSend to group members
send_to_connectionSend to specific connection
add_user_to_groupAdd user to group
remove_user_from_groupRemove user from group
close_connectionDisconnect client
connection_existsCheck connection status

Best Practices

  1. Use roles to limit client permissions
  2. Use groups for targeted messaging
  3. Generate short-lived tokens for security
  4. Use user IDs to send to users across connections
  5. Handle reconnection in client applications
  6. Use JSON content type for structured data
  7. Close connections gracefully with reasons

When to Use

This skill is applicable to execute the workflow or actions described in the overview.

Limitations

  • Use this skill only when the task clearly matches the scope described above.
  • Do not treat the output as a substitute for environment-specific validation, testing, or expert review.
  • Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.

Security

  • Security scanPassed, 95 of 100Pattern checks on the skill's files.
  • Model reviewPassed

See the full security report

Quality evaluation

  • Task evaluationNot runNo evaluation yet. Run one in Skill Studio.

See evaluation results

Source

Popularity

Repository stars
23,134
Repository forks
3,985
Repository npm downloads, 7 days
11,772
7-day trend score
72

Stars and forks are for the whole GitHub repository, not this skill alone. Downloads are for the repository's npm package. Refreshed 17 Mar 2026, 7 months ago.

Works with

Any agent vskill installs to: 56 targets in all. These 39 have a page on this site.

  • Claude Code
  • Cursor
  • GitHub Copilot
  • Windsurf
  • Codex
  • Gemini CLI
  • Cline
  • Amp
  • Roo Code
  • Goose
  • Continue
  • Kilo Code
  • Junie
  • Trae
  • OpenHands
  • Qwen Code
  • AI Agent
  • Tabnine Chat
  • Sourcegraph Cody
  • Amazon Q
  • Firebase Genkit
  • Sweep
  • Devin
  • Phind
  • Mentat
  • Claude Desktop
  • Aider
  • GPT Pilot
  • Plandex
  • bolt.new
  • v0
  • Replit Agent
  • Anse
  • TypingMind
  • BetterChatGPT
  • LibreChat
  • Warp AI
  • Pieces
  • Fig AI
Embed badge
vskill VERIFIED badge
[![vskill VERIFIED](https://verified-skill.com/api/v1/skills/sickn33/antigravity-awesome-skills/azure-messaging-webpubsubservice-py/badge)](https://verified-skill.com/skills/sickn33/antigravity-awesome-skills/azure-messaging-webpubsubservice-py)

Something wrong with this skill or its page? Tell us.