How should I properly to subscribe to real-time conversation events via the Notification API WebSocket endpoint? I am building a Grafana data source plugin to visualize queue metrics in real-time, specifically focusing on conversation state changes. The documentation suggests using /api/v2/analytics/events with a WebSocket connection, but I am struggling to scope the subscription correctly to avoid receiving global noise.
I have established a connection using the standard WebSocket URL with an OAuth bearer token in the query string. When I send the initial subscribe message, I receive a 200 OK response from the server, confirming the subscription is active. However, the event stream contains a massive volume of unrelated events, making it difficult to filter for specific conversation IDs or queue interactions on the client side. This impacts performance significantly for the dashboard rendering logic.
Here is the JSON payload I am using for the subscription request:
The issue persists even when I narrow the topic to conversation:call. I suspect the filter syntax might be incorrect or that I need to use a different mechanism for scoping. Is there a specific filter parameter or topic structure required to isolate events for a single conversation or a set of queue IDs? Any example of the correct subscription payload would be appreciated.
Check your subscription payload to ensure you are explicitly filtering by event type. The default WebSocket connection often returns excessive noise if you do not narrow the scope in the initial subscribe message. In Azure Functions, I handle this by sending a targeted JSON payload immediately after the upgrade, specifying only conversation:state events to reduce memory pressure on the worker.
var subscribePayload = new {
type = "subscribe",
events = new[] { "conversation:state" },
filters = new[] {
new { type = "queue", id = "your-queue-id" }
}
};
await websocket.SendJsonAsync(subscribePayload);
This approach ensures your .NET consumer only processes relevant state changes. You can further optimize by using a CancellationToken to manage long-lived connections in the Function App, preventing resource leaks during scale-out events in Europe/Paris data centers.
Yep, this is a known issue… You must explicitly filter by event.type in the initial subscribe payload to prevent global noise from overwhelming your Grafana plugin. The WebSocket does not support server-side filtering by queue ID, so you must handle that logic client-side.
The main issue here is that WebSocket subscriptions are inherently global by default. You cannot filter by queue ID or specific conversation attributes on the server side during the subscription handshake. If you subscribe to conversation:state without strict client-side filtering, your Grafana plugin will drown in noise from every agent in your organization.
I run heavy ETL pipelines against the Analytics API daily. Mixing parameters breaks the cursor logic, but for WebSockets, the issue is memory pressure. You must filter strictly in your message handler. Here is the correct workflow:
Subscribe to the specific event type only. Do not request generic all events.
Filter the incoming JSON payload immediately. Check for entity.id matching your target queues or state changes relevant to your metrics.
Discard irrelevant events before processing. This reduces memory overhead significantly.
The suggestion above about client-side matching is correct, but you need to implement it efficiently. Use a simple hash map for queue IDs to avoid string comparisons on every message.
Here is the Python logic I use in my data ingestion scripts to handle this noise:
import websockets
import asyncio
TARGET_QUEUE_IDS = ["queue-id-123", "queue-id-456"]
async def process_events(uri):
async with websockets.connect(uri) as websocket:
# 1. Subscribe to specific event type only
subscribe_payload = {
"type": "subscribe",
"data": {
"event": "conversation:state"
}
}
await websocket.send(subscribe_payload)
async for message in websocket:
data = json.loads(message)
# 2. Client-side filtering logic
entity_id = data.get("entity", {}).get("id")
# 3. Discard if not in target list
if entity_id not in TARGET_QUEUE_IDS:
continue
# Process only relevant events
update_grafana_metric(data)
See the Notification API Documentation for valid event types. Do not attempt to filter by queue in the subscribe payload; it will be ignored.
Hi there. The suggestion above regarding client-side filtering is correct, but it misses a critical performance bottleneck for high-volume environments. When you process raw WebSocket streams in a browser or light node worker, the sheer volume of conversation:state events can cause significant memory pressure and dropped frames.
The robust solution is to inject a distributed tracing context into your subscription handler. This allows you to correlate the subscription lifecycle with your backend processing logic, ensuring no events are lost during reconnections.