> For the complete documentation index, see [llms.txt](https://docs.roboflow.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.roboflow.com/deployment/self-hosted/enterprise/stream-management-api.md).

# Stream Management API

{% hint style="warning" %}
**Discontinued as a standalone service.** Use the Inference SDK WebRTC client with an [Inference Server](/deployment/self-hosted/inference-server.md) for current video streaming deployments. The same client runs models and Workflows on self-hosted and Serverless runtimes. See [video processing](https://docs.roboflow.com/workflows/deploy/video-processing).
{% endhint %}

{% hint style="warning" %}
**Enterprise feature.** The standalone service on this page requires a Roboflow Enterprise license to use in production. See [Roboflow Licensing](https://roboflow.com/licensing) for details. This does not apply to the integrated video management API described above.
{% endhint %}

## About

The Stream Management API generated predictions from Roboflow object detection models on online video streams. It added an HTTP management layer for remotely controlling video workers.

This is useful for scenarios including, but not limited to:

* Performing inference across multiple online video streams simultaneously.
* Executing inference on multiple devices that need coordination.
* Establishing a monitoring layer to oversee video processing.

![Stream Management design](https://storage.googleapis.com/com-roboflow-marketing/inference/stream_management_api_design.jpg)

### Example use case

Joe wants to monitor objects in the footage captured by a fleet of IP cameras installed in his factory. After training an object detection model on the Roboflow platform, he is ready for deployment. With four cameras in his factory, Joe picks a model compact enough for over 30 inferences per second on his Jetson devices. Given that computational budget per device, he needs two Jetson devices to process footage from all cameras, at roughly 15 frames per second per video source.

To streamline deployment, Joe deploys Stream Management containers to all Jetson devices in his local network. This lets him communicate with each device over HTTP to orchestrate processing tasks. He builds a web app to send commands to the devices and retrieve metrics about the status of each video stream. Finally, he implements a UDP server to receive predictions, using the `supervision` package to track objects in the footage.

## How to run

### In Docker with `docker compose`

The most common use cases are packaged as Docker Compose configurations. Where custom configuration is required inside the containers, such as passing camera devices through, the separate-container options below may suit better.

{% tabs %}
{% tab title="CPU" %}

```bash
docker compose -f ./docker/dockerfiles/stream-management-api.compose-cpu.yaml up
```

{% endtab %}

{% tab title="GPU" %}

```bash
docker compose -f ./docker/dockerfiles/stream-management-api.compose-gpu.yaml up
```

{% endtab %}

{% tab title="Jetson (JetPack 5.1.1)" %}

```bash
docker compose -f ./docker/dockerfiles/stream-management-api.compose-jetson.5.1.1.yaml up
```

On Jetson devices, some operations (such as container bootstrap or model initialization) take longer than on other platforms. The Docker Compose definition does not currently wait for the Stream Manager TCP socket port to open, so initial requests to the HTTP API may be answered with HTTP 503.
{% endtab %}
{% endtabs %}

### In Docker, running the API and stream manager separately

{% tabs %}
{% tab title="CPU" %}

```bash
docker run -d --name stream_manager --network host roboflow/roboflow-inference-stream-manager-cpu:latest
docker run -d --name stream_management_api --network host roboflow/roboflow-inference-stream-management-api:latest
```

{% endtab %}

{% tab title="GPU" %}

```bash
docker run -d --name stream_manager --network host --runtime nvidia roboflow/roboflow-inference-stream-manager-gpu:latest
docker run -d --name stream_management_api --network host roboflow/roboflow-inference-stream-management-api:latest
```

{% endtab %}

{% tab title="Jetson (JetPack 5.1.1)" %}

```bash
docker run -d --name stream_manager --network host --runtime nvidia roboflow/roboflow-inference-stream-manager-jetson-5.1.1:latest
docker run -d --name stream_management_api --network host roboflow/roboflow-inference-stream-management-api:latest
```

{% endtab %}
{% endtabs %}

#### Configuration parameters

**Stream Management API**

* `STREAM_MANAGER_HOST` - hostname for the stream manager container. Change it to the container name if `--network host` is not used, or if you are targeting a remote machine.
* `STREAM_MANAGER_PORT` - port used to communicate with the stream manager. Must match the stream manager container.

**Stream Manager**

* `PORT` - the port the server runs on.
* Mount a volume at the container's `/tmp/cache` to enable permanent storage of models, for faster inference pipeline initialization.
* Camera connectivity must be enabled at the level of this container, so if devices need to be passed into Docker, do it here.

#### Building the images (optional)

```bash
# Stream Management API
docker build -t roboflow/roboflow-inference-stream-management-api:dev -f docker/dockerfiles/Dockerfile.stream_management_api .

# Stream Manager
docker build -t roboflow/roboflow-inference-stream-manager-{device}:dev -f docker/dockerfiles/Dockerfile.onnx.{device}.stream_manager .
```

### Bare-metal deployment

In some cases you need to deploy the application at host level. This is possible, but you must resolve the environment the way the Stream Manager and Stream Management API dockerfiles do for your platform. Once that is done, run:

```bash
python -m inference.enterprise.stream_management.manager.app  # runs the manager
python -m inference.enterprise.stream_management.api.app      # runs the management API
```

## How to integrate

After running the `roboflow-inference-stream-management-api` container, the HTTP API is available at `http://127.0.0.1:8080` with the default configuration.

Call `wget http://127.0.0.1:8080/openapi.json` to get the OpenAPI specification of the API, which you can render in [the Swagger editor](https://editor.swagger.io/).

An example Python client:

```python
import requests
from typing import Optional

URL = "http://127.0.0.1:8080"

def list_pipelines() -> dict:
    response = requests.get(f"{URL}/list_pipelines")
    return response.json()


def get_pipeline_status(pipeline_id: str) -> dict:
    response = requests.get(f"{URL}/status/{pipeline_id}")
    return response.json()


def pause_pipeline(pipeline_id: str) -> dict:
    response = requests.post(f"{URL}/pause/{pipeline_id}")
    return response.json()


def resume_pipeline(pipeline_id: str) -> dict:
    response = requests.post(f"{URL}/resume/{pipeline_id}")
    return response.json()


def terminate_pipeline(pipeline_id: str) -> dict:
    response = requests.post(f"{URL}/terminate/{pipeline_id}")
    return response.json()


def initialise_pipeline(
    video_reference: str,
    model_id: str,
    api_key: str,
    sink_host: str,
    sink_port: int,
    max_fps: Optional[int] = None,
) -> dict:
    response = requests.post(
        f"{URL}/initialise",
        json={
            "type": "init",
            "sink_configuration": {
                "type": "udp_sink",
                "host": sink_host,
                "port": sink_port,
            },
            "video_reference": video_reference,
            "model_id": model_id,
            "api_key": api_key,
            "max_fps": max_fps,
        },
    )
    return response.json()
```

{% hint style="info" %}
`initialise_pipeline()` must be given a `video_reference` and `sink_configuration` where every resource (video file or camera device) and URI (stream reference, sink reference) is **reachable from the Stream Manager environment**. For example, inside Docker containers `localhost` binds to the container's localhost, not the localhost of the host machine.
{% endhint %}

## Developer notes

The pivotal element of the implementation is the Stream Manager component, which operates as a single-threaded TCP server. It processes requests received from a TCP socket and supervises video worker processes. Multiprocessing queues carry commands and results between the workers and the Stream Manager.

Requests to the Stream Manager are handled sequentially in blocking mode, so each request must conclude before the next one starts.

### Communication protocol: requests

The Stream Manager accepts the following binary protocol. Each payload contains:

```
[HEADER: 4B, big-endian, unsigned - int value with message size][MESSAGE: utf-8 serialised json of size dictated by header]
```

The message must be valid JSON after decoding and must represent a valid command.

**`list_pipelines`**

```json
{
  "type": "list_pipelines"
}
```

**`init`**

```json
{
  "type": "init",
  "model_id": "some/1",
  "video_reference": "rtsp://192.168.0.1:554",
  "sink_configuration": {
    "type": "udp_sink",
    "host": "192.168.0.3",
    "port": 9999
  },
  "api_key": "YOUR_API_KEY",
  "max_fps": 16,
  "model_configuration": {
    "type": "object-detection",
    "class_agnostic_nms": true,
    "confidence": 0.5,
    "iou_threshold": 0.4,
    "max_candidates": 300,
    "max_detections": 3000
  },
  "video_source_properties": {
    "frame_width": 1920,
    "frame_height": 1080,
    "fps": 30
  }
}
```

{% hint style="info" %}
The model ID is composed of the string `<project_id>/<version_id>`. See [model IDs](https://docs.roboflow.com/models/model-ids) to find these values.
{% endhint %}

**`terminate`**

```json
{
  "type": "terminate",
  "pipeline_id": "my_pipeline"
}
```

**`pause`**

```json
{
  "type": "mute",
  "pipeline_id": "my_pipeline"
}
```

**`resume`**

```json
{
  "type": "resume",
  "pipeline_id": "my_pipeline"
}
```

**`status`**

```json
{
  "type": "status",
  "pipeline_id": "my_pipeline"
}
```

### Communication protocol: responses

For each request that can be processed (without timeout or source disconnection), the Stream Manager returns a result in this format:

```
[HEADER: 4B, big-endian, unsigned - int value with result size][RESULT: utf-8 serialised json of size dictated by header]
```

The result contains:

* `request_id` - a random string representing the request ID assigned by the Stream Manager, to ease debugging.
* `pipeline_id` - the pipeline the command is associated with, when applicable.
* `response` - the payload of the operation response.

Each `response` has a `status` key with one of two values: `success` or `failure`. Each failed response contains an `error_type` key to dispatch error handling, plus optional `error_class` and `error_message` fields with inner details of the error. The content of successful responses depends on the type of operation.

## Future work

* Securing the API connection layer, to enable safe remote control.
* Securing the TCP socket of the Stream Manager.