RTSP + AI: turn any camera stream into agent-readable events.

RTSP stream AI analysis means connecting a live camera feed to infrastructure that continuously understands it. You get rolling-window analysis of what’s happening, a live index you can search, and events you define in plain language that push alerts the moment they occur. No frame-grabbing loops, no training a classifier per event, no pipeline babysitting. Here’s the whole thing, working, in one page.

The DIY wall

Search this topic and you’ll find forum threads and half-finished repos. The do-it-yourself version is a grind: OpenCV to pull frames off the stream, YOLO or a VLM call per frame, then your own logic to decide when detections add up to an event. Three problems never go away:

1. Frames aren’t events. A person in frame 4,000 is not information. A person who wasn’t there before, or who’s been near the loading dock for ten minutes, is information. Events live in time windows, and per-frame pipelines throw time away.
2. Streams misbehave. RTSP drops, reconnects, drifts. Keeping a pipeline alive 24/7 is an SRE job you didn’t sign up for.
3. No history. If you don’t also build storage + indexing, you can’t ask “what happened last night?” The footage evaporated through your pipeline.

Connect a stream, define an event, get alerts

There’s a public sample stream below, so you can run this without owning a camera.

pip install videodb   # free API key at console.videodb.io

import asyncio
import videodb

RTSP_URL = "rtsp://samples.rts.videodb.io:8554/intruder"  # public test stream

async def main():
    conn = videodb.connect()
    coll = conn.get_collection()

    # 1. Ingest the live stream (store=True keeps searchable history)
    rtstream = coll.connect_rtstream(url=RTSP_URL, name="Dock Cam", store=True)

    # 2. Continuous understanding in rolling windows
    understanding = rtstream.understand(
        segmentation={"type": "time", "window": "5s"},
        analyzers=[{
            "type": "vlm",
            "name": "scene",
            "sampling": {"frame_count": 2},
            "config": {"prompt": "Describe the people, vehicles, movement, and activity."}
        }],
        store=True
    )

    # 3. A live index - queryable while the stream runs
    index = rtstream.index(name="dock-scenes",
                           source=understanding.outputs["scene"],
                           use_for=["semantic"])

    # 4. Define the event in plain language
    event_id = conn.create_event(
        event_prompt="Detect when a person appears in the monitored area.",
        label="person_appears"
    )

    # 5. Get pushed alerts (webhook and/or websocket)
    websocket = conn.connect_websocket(coll.id)
    await websocket.connect()
    alert_id = index.create_alert(event_id=event_id,
                                  callback_url="https://your-app.example.com/hooks/camera",
                                  ws_connection_id=websocket.connection_id)

    async for message in websocket.receive():
        if message.get("channel") == "alert":
            alert = message["data"]
            print(alert["label"], alert["confidence"])
            print(alert["explanation"])   # why it fired
            print(alert["stream_url"])    # playable clip of the moment
            break

asyncio.run(main())

The event definition is the part that changes what’s possible: it’s a sentence, not a trained model.

“Detect water pooling on the floor.” “Detect a vehicle blocking the fire lane.” “Detect when the machine’s status light turns red.” Changing what you monitor for is editing a prompt, not retraining a model.

Every alert arrives with an explanation and a clip of the exact moment. That is evidence your team (or your agent) can act on directly.

Because store=True keeps history flowing into the index, the questions work backwards too: search “delivery truck at the dock” against last week and get timestamped matches.

The lifecycle details that separate demos from production

Streams and their processing are resources with on/off switches. Run them deliberately:

index.disable_alert(alert_id)
index.stop()          # stop indexing
understanding.stop()  # stop analysis (stop paying for it)
rtstream.stop()       # disconnect the stream
await websocket.close()

Rolling windows and frame sampling are tunable per stream, which is your cost dial. A lobby camera might analyze 2 frames every 5 seconds. A compliance-critical line might run denser.

For camera fleets, the same pattern scales across thousands of concurrent feeds. And for footage that can’t leave your walls, VideoDB deploys in your own cloud (AWS/GCP/Azure), VPC, or edge, with Zero Data Retention options and SOC 2 Type II. See Live Camera Intelligence for the fleet-scale picture.

Things people build with this

Intrusion detection, flood and leak detection, baby and elder-care monitors, ICU observation, retail queue analytics, drone patrol review, manufacturing QC. Several have runnable versions in the cookbook. Clone one, point it at the sample stream, and adapt the prompt.

FAQ

What is RTSP, and will my camera work?

RTSP (Real Time Streaming Protocol) is the standard streaming protocol spoken by virtually every IP camera, NVR, and drone gateway. If your device exposes an RTSP URL, it can connect. No vendor SDK required.

Do I need to train a model for my use case?

No. Analyzers use vision-language models driven by your prompts, and events are defined in natural language. Custom CV models are supported when you need them, but the starting point is a sentence, not a dataset.

How fast are alerts?

Understanding runs continuously in rolling windows you configure. The example uses 5-second windows. Alerts push over WebSocket/webhook as events are detected, so you get seconds-scale awareness rather than batch processing. Tune window size and sampling to your latency and cost needs.

Can I search past footage?

Yes. With store=True, the stream’s history is indexed and searchable with the same semantic queries as recorded video, and results return playable clips.