> ## 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()
coll = conn.get_collection()
rtstream = coll.get_rtstream("rts-xxx")
ws = conn.connect_websocket()
await ws.connect()

# Continuously create and store scene output
scene_understanding = rtstream.understand(
    segmentation={"type": "time", "window": "5s"},
    analyzers=[
        {
            "type": "vlm",
            "name": "scene",
            "sampling": {"frame_count": 2},
            "config": {"prompt": "Describe activity and detect unusual behavior"},
        }
    ],
    store=True,
)

# Index the stored scene output before attaching alerts
scene_index = rtstream.index(
    name="security_monitor",
    source=scene_understanding.outputs["scene"],
    use_for=["semantic"],
)

# Create detection rule (reusable)
event_id = conn.create_event(
    event_prompt="Detect intruder",
    label="security_alert"
)

# Wire to the V2 scene index; callback delivery remains required with WebSocket
scene_index.create_alert(
    event_id=event_id,
    callback_url="https://your-backend.com/alerts",  # Webhook
    ws_connection_id=ws.connection_id                 # Real-time
)

# Inspect WebSocket messages before accessing fields; their decoded payload shape is pipeline-specific.
async for event in ws.receive():
    print(event)
```

***

## 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}
# Reuse the same event with the V2 scene index
safety_event = conn.create_event(
    event_prompt="Detect safety violation",
    label="safety"
)

# Alerts wire events to specific indexes
scene_index.create_alert(
    event_id=safety_event,
    callback_url="https://your-backend.com/alerts",
)
```

***

## 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.receive():
    print(event)  # Inspect the decoded payload before accessing fields.
```

### Webhook

```python theme={null}
scene_index.create_alert(
    event_id=event_id,
    callback_url="https://your-backend.com/alerts"
)
```

Illustrative callback payload; inspect and validate the delivery your endpoint receives:

```json theme={null}
{
  "channel": "alert",
  "rtstream_id": "rts-xxx",
  "data": {
    "event_id": "event-xxx",
    "label": "intrusion",
    "triggered": true,
    "confidence": 0.92,
    "start": 1710000012340,
    "end": 1710000018900
  }
}
```

***

## 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

The current V2 alert channel is produced by the V2 scene index above:

| Channel | Source | Content |
| :- | :- | :- |
| `alert` | `scene_index.create_alert()` | Alert notifications |

The following channel names are available through RTStream helper APIs:

| Channel | Source | Content |
| :- | :- | :- |
| `transcript` | `start_transcript()` | Speech-to-text |
| `scene_index` | `index_visuals()` | Visual analysis |
| `audio_index` | `index_audio()` | Audio analysis |

***

## 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

<CardGroup cols={2}>
  <Card title="Baby Crib Monitoring" icon="baby" href="/examples-and-tutorials/live-intelligence/baby-crib-monitoring">
    Real-time infant monitoring with AI-powered alerts
  </Card>

  <Card title="Intrusion Detection" icon="shield" href="/examples-and-tutorials/live-intelligence/intrusion-detection">
    Sub-second alerts when unauthorized access is detected
  </Card>

  <Card title="Traffic Violations" icon="car" href="/examples-and-tutorials/live-intelligence/traffic-violations">
    Real-time detection of red light and stop sign violations
  </Card>
</CardGroup>

***

## Next Steps

<CardGroup cols={2}>
  <Card icon="video" title="RTStream Reference" href="/pages/ingest/live-streams/rtsp-ingest">
    Complete real-time API
  </Card>

  <Card icon="webhook" title="Webhooks Guide" href="/pages/act/live-action/alerts-and-callbacks">
    Setting up callbacks
  </Card>
</CardGroup>


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.