```
### Query Parameters
| Parameter | Type | Default | Description |
| ------------ | ---- | ------- | ------------------------------------------------- |
| `dataset_id` | UUID | `null` | Stream only notifications for a specific dataset. |
### Event Types
| Event | Description |
| -------------- | ---------------------------------------------------------------------------------- |
| `notification` | A new notification. The `data` field contains the full notification JSON object. |
| `ping` | Keepalive event sent every 3 seconds. No meaningful data. |
| `error` | An internal error occurred. The connection remains open and retries automatically. |
### JavaScript Example
```javascript theme={"theme":"monokai"}
const eventSource = new EventSource(
"https://app.visual-layer.com/api/v1/notifications/stream",
{ withCredentials: true }
);
eventSource.addEventListener("notification", (event) => {
const notification = JSON.parse(event.data);
console.log(`Alert: ${notification.metadata.view_name}`);
console.log(` ${notification.metadata.result_count} new matches`);
console.log(` Dataset: ${notification.metadata.dataset_name}`);
});
eventSource.addEventListener("ping", () => {
// Connection is alive — no action needed
});
eventSource.addEventListener("error", (event) => {
console.error("SSE error:", event);
});
```
### Python Example
```python theme={"theme":"monokai"}
import requests
import json
VL_BASE_URL = "https://app.visual-layer.com"
JWT_TOKEN = ""
headers = {"Authorization": f"Bearer {JWT_TOKEN}"}
# Open SSE stream
resp = requests.get(
f"{VL_BASE_URL}/api/v1/notifications/stream",
headers=headers,
stream=True,
)
for line in resp.iter_lines(decode_unicode=True):
if line.startswith("data: ") and line.strip() != "data:":
data = json.loads(line[6:])
if "metadata" in data and data["metadata"].get("is_alert"):
print(f"Alert: {data['metadata']['view_name']}")
print(f" {data['metadata']['result_count']} new results")
```
The SSE stream polls the database every 3 seconds. Notifications appear within one polling cycle of being generated. The connection stays open until the client disconnects.
***
## Python Example: Full Workflow
```python theme={"theme":"monokai"}
import requests
import time
VL_BASE_URL = "https://app.visual-layer.com"
JWT_TOKEN = ""
DATASET_ID = ""
headers = {"Authorization": f"Bearer {JWT_TOKEN}"}
# Check unread count
resp = requests.get(
f"{VL_BASE_URL}/api/v1/notifications/unread-count",
headers=headers,
params={"dataset_id": DATASET_ID},
)
resp.raise_for_status()
unread = resp.json()["count"]
print(f"Unread alerts: {unread}")
# List recent notifications for this dataset
resp = requests.get(
f"{VL_BASE_URL}/api/v1/notifications",
headers=headers,
params={"dataset_id": DATASET_ID, "limit": 10, "unread_only": True},
)
resp.raise_for_status()
data = resp.json()
for n in data["notifications"]:
meta = n["metadata"]
print(f" View: {meta['view_name']} — {meta['result_count']} new matches")
# Mark all as read for this dataset
if data["total"] > 0:
resp = requests.post(
f"{VL_BASE_URL}/api/v1/notifications/read-all",
headers=headers,
params={"dataset_id": DATASET_ID},
)
resp.raise_for_status()
print(f"Marked {resp.json()['count']} notifications as read")
```
***
## Response Codes
See [Error Handling](/api-reference/errors) for the error response format and Python handling patterns.
| HTTP Code | Meaning | Common Cause |
| --------- | ---------------------- | ------------------------------------------------------------------------------------------------ |
| **200** | Request successful. | |
| **401** | Unauthorized. | Invalid or expired JWT token. |
| **404** | Not found. | Notification does not exist, does not belong to the user, or is already read (for mark-as-read). |
| **422** | Validation error. | Invalid query parameter values (for example, `limit` exceeding 100). |
| **500** | Internal server error. | Contact support if this persists. |
***
## Best Practices
* **Use `unread-count` for lightweight polling.** The unread count endpoint returns a single integer and is suitable for frequent polling to update UI badge counts without fetching full notification payloads.
* **Filter by `dataset_id` when possible.** Scoping requests to a specific dataset reduces response size and improves relevance, especially for users with access to many datasets.
* **Use SSE for real-time integrations.** The streaming endpoint provides sub-second alert delivery without polling overhead. Use it for dashboards, Slack integrations, or automated response workflows.
* **Mark alerts as read after processing.** Unread counts continue to grow if alerts are not acknowledged. Use `read-all` after reviewing a batch of notifications, or mark individual notifications as you process them.
* **Use `has_more` for pagination.** Check the `has_more` field before requesting additional pages. Requesting pages beyond the total returns empty results.
***
## Related Resources
Create views and configure monitoring and alerting.
Understand how monitoring and alerting works across your datasets.
Add media to datasets, triggering monitoring evaluation on all views.
Track media addition tasks that trigger monitoring evaluation.