# Process a Video Stream

This recipe runs a [Roboflow Workflow](https://docs.roboflow.com/workflows/create-a-workflow) against a video — either a file or a live RTSP/HLS stream. The runtime side is documented in depth in the product docs; this page focuses on wiring the pieces together from a developer perspective.

There are two practical paths:

* **Hosted (Serverless v2)** — submit a video file or URL and get back per-frame predictions. Best for batch / one-shot processing.
* **Self-hosted Inference** — stream frames into a local Inference server and consume predictions in real time. Best for low-latency or edge use cases.

## Path A — Hosted: process a video file

Use the [Serverless Video API](https://docs.roboflow.com/deploy/serverless-video-streaming-api) for hosted batch video processing.

```python
import os
import time
import requests

API_KEY = os.environ["ROBOFLOW_API_KEY"]
WORKSPACE = "my-workspace"
WORKFLOW = "my-detector-workflow"

# 1. Get a signed upload URL.
signed = requests.post(
    "https://api.roboflow.com/video_upload_signed_url/",
    params={"api_key": API_KEY, "file_name": "input.mp4"},
).json()

# 2. Upload the file.
with open("input.mp4", "rb") as f:
    requests.put(signed["signedUrl"], data=f, headers={"Content-Type": "video/mp4"})

# 3. Submit the inference job pointing at the uploaded file.
job = requests.post(
    "https://api.roboflow.com/videoinfer/",
    json={
        "api_key": API_KEY,
        "workspace": WORKSPACE,
        "workflow": WORKFLOW,
        "video_url": signed["fileUrl"],
        "fps": 5,
    },
).json()
job_id = job["jobId"]

# 4. Poll until the job is done.
while True:
    status = requests.get(
        f"https://api.roboflow.com/videoinfer?api_key={API_KEY}&jobId={job_id}"
    ).json()
    if status["status"] in ("complete", "failed"):
        break
    time.sleep(5)

print(status)
```

The output structure (per-frame predictions, timing) is documented in the [Serverless Video API reference](https://docs.roboflow.com/deploy/serverless-video-streaming-api).

## Path B — Self-hosted: stream frames to local Inference

For real-time / low-latency, run [Roboflow Inference](https://inference.roboflow.com) locally and stream frames into it.

```bash
docker run -it --rm -p 9001:9001 roboflow/roboflow-inference-server-cpu
```

Then from Python:

```python
import cv2
from inference import InferencePipeline
from inference.core.interfaces.stream.sinks import render_boxes

pipeline = InferencePipeline.init_with_workflow(
    api_key=os.environ["ROBOFLOW_API_KEY"],
    workspace_name="my-workspace",
    workflow_id="my-detector-workflow",
    video_reference="rtsp://camera.local:554/stream",  # or a file path, or 0 for webcam
    on_prediction=render_boxes,
)

pipeline.start()
pipeline.join()
```

`InferencePipeline` handles frame decoding, batching, and async predictions. Replace `render_boxes` with your own callback to do something useful with the predictions (push to a webhook, write to a DB, raise an alert).

## When to use which

|          | Hosted                           | Self-hosted                |
| -------- | -------------------------------- | -------------------------- |
| Latency  | Seconds (per-frame after upload) | Tens of milliseconds       |
| Setup    | Just an API key                  | Docker host or edge device |
| Cost     | Per-frame credits                | Hardware + electricity     |
| Best for | Batch analysis, one-off footage  | Live streams, alerts, edge |

## Logging predictions to Vision Events

Either path can log structured predictions to [Vision Events](/developer/python-sdk/vision-events.md) for dashboards and alerting:

```python
ws.write_vision_event({
    "use_case": "store-traffic",
    "timestamp": "2026-05-01T17:05:33Z",
    "metadata": {"camera": "front-door", "person_count": 4},
})
```

That's the same code path the [Vision Events SDK](/developer/python-sdk/vision-events.md) and [REST](/developer/rest-api/vision-events.md) docs go into in detail.


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.roboflow.com/developer/recipes/process-a-video-stream.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.