> ## Documentation Index
> Fetch the complete documentation index at: https://docs.videodb.io/llms.txt
> Use this file to discover all available pages before exploring further.
# Events & Real-time
> Events and alerts turn understanding into action. VideoDB processes live media with sub-second latency, delivering events via WebSocket or webhook.
## Quick Example
```python theme={null}
import videodb
conn = videodb.connect()
ws = conn.connect_websocket()
await ws.connect()
# Create detection rule (reusable)
event_id = conn.create_event(
event_prompt="Detect intruder",
label="security_alert"
)
# Wire to index with delivery method
index.create_alert(
event_id=event_id,
callback_url="https://your-backend.com/alerts", # Webhook
ws_connection_id=ws.connection_id # Real-time
)
# Alerts fire in <1 second
async for event in ws.stream():
if event.get("channel") == "alert":
print(f"ALERT: {event['data']['text']}")
```
***
## Events vs Alerts
| Concept | Purpose | Example |
| :-------- | :----------------------------- | :----------------------------- |
| **Event** | What to detect (reusable rule) | "Detect person without helmet" |
| **Alert** | Where to deliver (wiring) | Webhook URL, WebSocket ID |
```python theme={null}
# Events are reusable across streams
safety_event = conn.create_event(
event_prompt="Detect safety violation",
label="safety"
)
# Alerts wire events to specific indexes
for rtstream in streams:
index = rtstream.get_scene_index(index_id)
index.create_alert(event_id=safety_event)
```
***
## Delivery Methods
| Method | Latency | Use Case |
| :------------ | :------------- | :--------------------------- |
| **WebSocket** | Real-time | Dashboards, interactive apps |
| **Webhook** | Near real-time | Automation, integrations |
### WebSocket
```python theme={null}
ws = conn.connect_websocket()
await ws.connect()
async for event in ws.stream():
channel = event.get("channel")
# transcript, scene_index, audio_index, alert
```
### Webhook
```python theme={null}
index.create_alert(
event_id=event_id,
callback_url="https://your-backend.com/alerts"
)
```
Payload:
```json theme={null}
{
"event_label": "intrusion",
"rtstream_id": "rts-xxx",
"timestamp": 1710000012340,
"data": {"text": "Person in restricted area"}
}
```
***
## Latency Profile
| Operation | Typical Latency |
| :----------------- | :-------------- |
| Alert trigger | \< 1 second |
| Transcript event | 1-2 seconds |
| Visual index event | 2-5 seconds |
| Search query | \< 500ms |
***
## Delivery Guarantees
| Method | Guarantee | Handle |
| :-------- | :------------ | :----------------- |
| WebSocket | At-most-once | Client reconnects |
| Webhook | At-least-once | Idempotency checks |
```python theme={null}
# Webhook idempotency
@app.post("/webhooks")
async def handle(request):
event = await request.json()
if already_processed(event["event_id"]):
return {"status": "ok"}
process(event)
```
***
## Event Channels
| Channel | Source | Content |
| :------------ | :------------------- | :------------------ |
| `transcript` | `start_transcript()` | Speech-to-text |
| `scene_index` | `index_visuals()` | Visual analysis |
| `audio_index` | `index_audio()` | Audio analysis |
| `alert` | `create_alert()` | Alert notifications |
***
## Best Practices
1. **Make events reusable** - Define once, use across streams
2. **Be specific in prompts** - "Detect person falling" beats "detect problems"
3. **Use both methods** - WebSocket for UI, webhooks for automation
4. **Handle idempotency** - Webhooks may deliver duplicates
## What You Can Build
Real-time infant monitoring with AI-powered alerts
Sub-second alerts when unauthorized access is detected
Real-time detection of red light and stop sign violations
***
## Next Steps
Complete real-time API
Setting up callbacks