Frigate NVR Telegram Notifier

Type: Custom Python Docker container
Image: frigate-notifier:latest (locally built)
Source: /opt/frigate-notifier/ on the Frigate host (10.0.20.15)

This container listens to Frigate's MQTT events and pushes detection clips/snapshots to a Telegram chat.

---

How It Works

Frigate detects object
       │
       ▼
MQTT event published to frigate/events
       │
       ▼
notifier.py receives event (paho-mqtt)
       │
       ▼
Fetches /api/events/{id}/clip.mp4 from Frigate API
       │
   ┌───┴───┐
   ▼       ▼
 CLIP    SNAPSHOT (fallback)
   │       │
   ▼       ▼
Telegram Bot API → sendVideo / sendPhoto / sendMessage

Priority: Clip → Snapshot → Text-only message

---

Project Structure

/opt/frigate-notifier/
├── Dockerfile          # Build definition
├── notifier.py         # Python script

---

Files

Dockerfile

FROM python:3.12-alpine

RUN pip install --no-cache-dir paho-mqtt requests

COPY notifier.py /notifier.py

CMD ["python", "-u", "/notifier.py"]

notifier.py

import json
import os
import requests
import logging
import io

import paho.mqtt.client as mqtt

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger(__name__)

FRIGATE_URL = os.getenv("FRIGATE_URL", "http://frigate:5000")
MQTT_HOST = os.getenv("MQTT_HOST", "mosquitto")
MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
MQTT_USER = os.getenv("MQTT_USERNAME", "")
MQTT_PASS = os.getenv("MQTT_PASSWORD", "")
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "")
CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", "")
MQTT_TOPIC = os.getenv("MQTT_TOPIC", "frigate/events")
CAMERA_FILTER = os.getenv("CAMERA_FILTER", "")

TELEGRAM_API = f"https://api.telegram.org/bot{BOT_TOKEN}"

def telegram_send_video(video_bytes, caption):
    url = f"{TELEGRAM_API}/sendVideo"
    files = {"video": ("clip.mp4", video_bytes, "video/mp4")}
    data = {"chat_id": CHAT_ID, "caption": caption}
    r = requests.post(url, files=files, data=data, timeout=30)
    log.info(f"sendVideo: {r.status_code} {r.text[:200]}")
    return r

def telegram_send_photo(photo_bytes, caption):
    url = f"{TELEGRAM_API}/sendPhoto"
    files = {"photo": ("snap.jpg", photo_bytes, "image/jpeg")}
    data = {"chat_id": CHAT_ID, "caption": caption}
    r = requests.post(url, files=files, data=data, timeout=30)
    log.info(f"sendPhoto: {r.status_code} {r.text[:200]}")
    return r

def telegram_send_text(text):
    url = f"{TELEGRAM_API}/sendMessage"
    data = {"chat_id": CHAT_ID, "text": text}
    r = requests.post(url, json=data, timeout=15)
    log.info(f"sendMessage: {r.status_code} {r.text[:200]}")
    return r

def send_clip(camera, label, clip_path):
    log.info(f"Fetching clip: clip_path={clip_path}")

    clip_url = f"{FRIGATE_URL}/api/events/{clip_path}/clip.mp4"
    log.info(f"Clip URL: {clip_url}")
    try:
        r = requests.get(clip_url, stream=True, timeout=30)
        log.info(f"Clip response: {r.status_code}")
        if r.status_code == 200:
            telegram_send_video(r.content, f"\U0001f4f9 {label} detected on {camera}")
            return
    except Exception as e:
        log.warning(f"clip fetch failed: {e}")

    snap_url = f"{FRIGATE_URL}/api/events/{clip_path}/snapshot.jpg"
    log.info(f"Snapshot URL: {snap_url}")
    try:
        rs = requests.get(snap_url, stream=True, timeout=15)
        log.info(f"Snapshot response: {rs.status_code}")
        if rs.status_code == 200:
            telegram_send_photo(rs.content, f"\U0001f4f8 {label} on {camera}")
            return
    except Exception as e:
        log.warning(f"snapshot fetch failed: {e}")

    telegram_send_text(f"\U0001f514 {label} detected on {camera} (no media)")

def on_message(client, userdata, msg):
    try:
        payload = json.loads(msg.payload)
        log.info(f"MQTT payload type={payload.get('type')}")
        if payload.get("type") == "new":
            after = payload.get("after", {})
            camera = after.get("camera", "")
            label = after.get("label", "unknown")
            if CAMERA_FILTER and CAMERA_FILTER not in camera:
                return
            clip_id = after.get("id", "")
            if clip_id:
                log.info(f"Detection: {label} on {camera} (id={clip_id})")
                send_clip(camera, label, clip_id)
    except Exception as e:
        log.warning(f"parse error: {e}")

def main():
    if not BOT_TOKEN or not CHAT_ID:
        log.error("TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID are required")
        return

    client = mqtt.Client()
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    client.on_message = on_message
    client.connect(MQTT_HOST, MQTT_PORT, 60)
    client.subscribe(MQTT_TOPIC)
    log.info(f"Listening on {MQTT_HOST}:{MQTT_PORT} -> {MQTT_TOPIC}")
    log.info(f"Telegram chat_id: {CHAT_ID}")
    client.loop_forever()

if __name__ == "__main__":
    main()

---

Replication Guide

Prerequisites

• Docker on the target host
• Access to the Frigate MQTT broker
• A Telegram bot token (from @BotFather)
• The Telegram chat ID where notifications should be sent

Step 1: Create project directory

mkdir -p /opt/frigate-notifier

Step 2: Add files

Copy Dockerfile and notifier.py into /opt/frigate-notifier/.

Step 3: Build the image

cd /opt/frigate-notifier
docker build -t frigate-notifier:latest .

Step 4: Run the container

docker run -d \
  --name frigate-notifier \
  --restart unless-stopped \
  --network frigate_default \
  -e MQTT_HOST=mosquitto \
  -e MQTT_USERNAME=frigate \
  -e MQTT_PASSWORD=MQTT_PASSWORD \
  -e TELEGRAM_BOT_TOKEN=TG_BOT_TOKEN \
  -e TELEGRAM_CHAT_ID=-1003972495209 \
  -e FRIGATE_URL=http://frigate:5000 \
  frigate-notifier:latest

Note: Replace MQTT_PASSWORD and TG_BOT_TOKEN with actual values. Use a .env file or Docker secrets for production.

Step 5: Verify

docker logs frigate-notifier

Expected output:

Listening on mosquitto:1883 -> frigate/events
Telegram chat_id: -1003972495209

Step 6 (optional): Add to docker-compose

If you want the notifier managed alongside Frigate, add it as a service in docker-compose.yml:

services:
  # ... existing frigate and mosquitto services ...

  frigate-notifier:
    build: /opt/frigate-notifier
    container_name: frigate-notifier
    restart: unless-stopped
    networks:
      - frigate_default
    environment:
      MQTT_HOST: mosquitto
      MQTT_USERNAME: frigate
      MQTT_PASSWORD: ${MQTT_PASSWORD}
      TELEGRAM_BOT_TOKEN: ${TG_BOT_TOKEN}
      TELEGRAM_CHAT_ID: "-1003972495209"
      FRIGATE_URL: http://frigate:5000

Then .env:

MQTT_PASSWORD=actual_mqtt_password
TG_BOT_TOKEN=actual_bot_token

---

Configuration Reference

Env Variable	Default	Required	Description
MQTT_HOST	mosquitto	No	MQTT broker hostname
MQTT_PORT	1883	No	MQTT broker port
MQTT_USERNAME	""	No	MQTT username
MQTT_PASSWORD	""	No	MQTT password
TELEGRAM_BOT_TOKEN	""	Yes	Telegram bot token
TELEGRAM_CHAT_ID	""	Yes	Target chat ID
FRIGATE_URL	http://frigate:5000	No	Frigate API base URL
MQTT_TOPIC	frigate/events	No	MQTT topic to subscribe to
CAMERA_FILTER	""	No	Only send alerts for cameras containing this string

---

Current Deployment (this host)

• Source: /opt/frigate-notifier/
• Image tag: frigate-notifier:latest
• Network: frigate_default (same as Frigate + Mosquitto)
• Chat ID: -1003972495209
• Restart: unless-stopped
• Not in docker-compose — started manually via docker run