I’m trying to track real-time adherence by subscribing to conversation events via the Notification API WebSocket. The connection establishes fine, but it drops after exactly 10 seconds without any error payload.
I’m using this snippet to get the token and connect:
import websockets
import asyncio
async def connect():
uri = f"wss://api.mypurecloud.com/api/v2/analytics/conversations?access_token={token}"
async with websockets.connect(uri) as websocket:
print("Connected")
await websocket.wait_closed()
asyncio.run(connect())
The docs mention a heartbeat but don’t specify the interval. Is there a keep-alive requirement I’m missing? The token is valid and has the correct scope. It just stops. Nothing in the logs.
The 10-second drop is almost certainly because you are not sending any subscription requests after the handshake. The Notification API WebSocket expects the client to define what it wants to listen to immediately. If the server doesn’t receive a subscribe message, it assumes the connection is idle or malformed and terminates it to free up resources.
You need to send a JSON payload defining your filters right after the connection opens. Here is how that looks in Python using websockets.
import websockets
import asyncio
import json
async def connect_and_subscribe():
# Ensure the access_token is valid and has the 'analytics:report:read' scope
uri = f"wss://api.mypurecloud.com/api/v2/analytics/conversations?access_token={token}"
async with websockets.connect(uri) as websocket:
# Define the subscription payload
# You must specify the resource type and filters
subscription = {
"resourceType": "conversations",
"filters": {
"conversationTypes": ["call", "chat"]
}
}
# Send the subscribe message
await websocket.send(json.dumps(subscription))
print("Subscription sent. Waiting for events...")
# Listen for incoming messages
async for message in websocket:
data = json.loads(message)
print(f"Received event: {data.get('eventType')}")
# Handle specific event types here
if data.get('eventType') == 'conversations.updated':
# Process adherence data
pass
# Run the async function
asyncio.run(connect_and_subscribe())
Make sure your OAuth token has the analytics:report:read scope. Without it, the subscription request will fail silently or cause a disconnect. Also, check if your environment has any idle timeout policies on the load balancer, though the 10-second mark is a strong indicator of the missing subscription payload issue.