hobbs-iac/docs/plans/2026-03-04-frigate-video-alerts.md

Frigate Video Alerts Implementation Plan

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: Build a Python container that listens for Frigate detection events via MQTT, analyzes snapshots with Claude Vision, and posts rich notifications to a Matrix room.

Architecture: A single Python container (frigate-notify) on hf-pdocker-01 subscribes to Frigate's MQTT events, filters for person/car/animal detections, fetches snapshots from Frigate's API, sends them to Claude Haiku for natural language description, then posts the snapshot image and description to a dedicated #video-alerts Matrix room via a new @frigate:hobbs.farm bot.

Tech Stack: Python 3.12, paho-mqtt, anthropic SDK, aiohttp, Docker Compose

Design doc: docs/plans/2026-03-04-frigate-video-alerts-design.md

---

Task 1: Create Matrix bot account and room

This task runs on hf-pdocker-01 via SSH. No code files involved.

Step 1: Register @frigate:hobbs.farm bot account

Use the docker-swarm-ops agent to SSH to hf-pdocker-01 (192.168.86.192) and run:

# Register the bot via Synapse admin API
# First get an admin token (use @alertbot or register directly)
docker exec $(docker ps -q -f name=matrix_synapse) register_new_matrix_user \
  -u frigate \
  -p "FrigateBot2026!" \
  -a \
  --no-admin \
  -c /data/homeserver.yaml \
  http://localhost:8008

Step 2: Login to get access token

curl -s -X POST http://192.168.86.192:8008/_matrix/client/r0/login \
  -H "Content-Type: application/json" \
  -d '{"type":"m.login.password","user":"frigate","password":"FrigateBot2026!"}'

Save the access_token from the response.

Step 3: Create #video-alerts room

curl -s -X POST "http://192.168.86.192:8008/_matrix/client/r0/createRoom" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <TOKEN>" \
  -d '{
    "room_alias_name": "video-alerts",
    "name": "Video Alerts",
    "topic": "Frigate NVR detection alerts with AI descriptions",
    "visibility": "private",
    "invite": ["@jeremy:hobbs.farm"]
  }'

Save the room_id from the response.

Step 4: Send a test message to verify

curl -s -X PUT "http://192.168.86.192:8008/_matrix/client/r0/rooms/<ROOM_ID>/send/m.room.message/$(date +%s)" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <TOKEN>" \
  -d '{"msgtype":"m.text","body":"Frigate video alerts bot is online."}'

Step 5: Store credentials in vault

Use the vault-credential-manager agent to store:

• Name: "Frigate Matrix Bot"
• Username: @frigate:hobbs.farm
• Password: FrigateBot2026!
• Notes: access_token=, room_id=<ROOM_ID>

Step 6: Commit (nothing to commit yet - this is infrastructure setup)

---

Task 2: Create the Python application files

Files:

• Create: compose/hf-pdocker-01/frigate-notify/app/requirements.txt
• Create: compose/hf-pdocker-01/frigate-notify/app/vision.py
• Create: compose/hf-pdocker-01/frigate-notify/app/matrix_client.py
• Create: compose/hf-pdocker-01/frigate-notify/app/main.py

Step 1: Create requirements.txt

compose/hf-pdocker-01/frigate-notify/app/requirements.txt

paho-mqtt==2.1.0
anthropic>=0.40.0
aiohttp>=3.9.0

Step 2: Create vision.py - Claude Vision wrapper

compose/hf-pdocker-01/frigate-notify/app/vision.py

import base64
import logging
import anthropic

logger = logging.getLogger(__name__)

PROMPT = (
    "Briefly describe what's happening in this security camera image in 1-2 sentences. "
    "Focus on the people, vehicles, or animals and their activity. Be concise."
)


def analyze_snapshot(client: anthropic.Anthropic, image_bytes: bytes) -> str | None:
    """Send a snapshot to Claude Haiku for description. Returns None on failure."""
    try:
        b64 = base64.standard_b64encode(image_bytes).decode("utf-8")
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=150,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "image",
                            "source": {
                                "type": "base64",
                                "media_type": "image/jpeg",
                                "data": b64,
                            },
                        },
                        {"type": "text", "text": PROMPT},
                    ],
                }
            ],
        )
        description = response.content[0].text.strip()
        logger.info("Vision analysis complete: %d chars", len(description))
        return description
    except Exception:
        logger.exception("Claude Vision API failed")
        return None

Step 3: Create matrix_client.py - Matrix image upload and messaging

compose/hf-pdocker-01/frigate-notify/app/matrix_client.py

import logging
import time
import aiohttp

logger = logging.getLogger(__name__)

RETRY_DELAYS = [1, 2, 4]


class MatrixClient:
    def __init__(self, homeserver: str, token: str, room_id: str):
        self.homeserver = homeserver.rstrip("/")
        self.token = token
        self.room_id = room_id
        self.headers = {"Authorization": f"Bearer {token}"}

    async def _request(self, method: str, url: str, **kwargs) -> dict | None:
        for attempt, delay in enumerate(RETRY_DELAYS):
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.request(
                        method, url, headers=self.headers, **kwargs
                    ) as resp:
                        if resp.status < 300:
                            return await resp.json()
                        body = await resp.text()
                        logger.warning(
                            "Matrix API %s (attempt %d): %s", resp.status, attempt + 1, body
                        )
            except Exception:
                logger.exception("Matrix request failed (attempt %d)", attempt + 1)
            if attempt < len(RETRY_DELAYS) - 1:
                await _sleep(delay)
        logger.error("All Matrix retries exhausted for %s", url)
        return None

    async def upload_image(self, image_bytes: bytes, filename: str) -> str | None:
        url = f"{self.homeserver}/_matrix/media/v3/upload?filename={filename}"
        headers = {**self.headers, "Content-Type": "image/jpeg"}
        try:
            async with aiohttp.ClientSession() as session:
                async with session.post(url, headers=headers, data=image_bytes) as resp:
                    if resp.status < 300:
                        data = await resp.json()
                        mxc_uri = data.get("content_uri")
                        logger.info("Uploaded image: %s", mxc_uri)
                        return mxc_uri
                    body = await resp.text()
                    logger.error("Image upload failed %s: %s", resp.status, body)
        except Exception:
            logger.exception("Image upload failed")
        return None

    async def send_image(self, mxc_uri: str, filename: str, size: int) -> dict | None:
        txn_id = str(int(time.time() * 1000))
        url = f"{self.homeserver}/_matrix/client/r0/rooms/{self.room_id}/send/m.room.message/{txn_id}"
        body = {
            "msgtype": "m.image",
            "body": filename,
            "url": mxc_uri,
            "info": {"mimetype": "image/jpeg", "size": size},
        }
        return await self._request("PUT", url, json=body)

    async def send_text(self, plain: str, html: str) -> dict | None:
        txn_id = str(int(time.time() * 1000) + 1)
        url = f"{self.homeserver}/_matrix/client/r0/rooms/{self.room_id}/send/m.room.message/{txn_id}"
        body = {
            "msgtype": "m.text",
            "body": plain,
            "format": "org.matrix.custom.html",
            "formatted_body": html,
        }
        return await self._request("PUT", url, json=body)


async def _sleep(seconds: float):
    import asyncio
    await asyncio.sleep(seconds)

Step 4: Create main.py - MQTT listener and event processor

compose/hf-pdocker-01/frigate-notify/app/main.py

import asyncio
import json
import logging
import os
import signal
import sys
import time
from collections import defaultdict
from http.server import HTTPServer, BaseHTTPRequestHandler
from threading import Thread

import anthropic
import paho.mqtt.client as mqtt

from vision import analyze_snapshot
from matrix_client import MatrixClient

# --- Configuration ---
MQTT_HOST = os.environ.get("MQTT_HOST", "192.168.86.192")
MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
FRIGATE_URL = os.environ.get("FRIGATE_URL", "http://192.168.86.192:5000")
MATRIX_HOMESERVER = os.environ.get("MATRIX_HOMESERVER", "http://192.168.86.192:8008")
MATRIX_ACCESS_TOKEN = os.environ["MATRIX_ACCESS_TOKEN"]
MATRIX_ROOM_ID = os.environ["MATRIX_ROOM_ID"]
HEALTH_PORT = int(os.environ.get("HEALTH_PORT", "5199"))

ALLOWED_LABELS = {"person", "car", "dog", "cat", "bird"}
MIN_SCORE = 0.7
RATE_LIMIT_SECONDS = 60
DEDUP_WINDOW_SECONDS = 30

# --- Logging ---
logging.basicConfig(
    level=logging.INFO,
    format='{"time":"%(asctime)s","level":"%(levelname)s","module":"%(module)s","message":"%(message)s"}',
    stream=sys.stdout,
)
logger = logging.getLogger(__name__)

# --- State ---
recent_events: dict[str, float] = {}
last_notify_per_camera: defaultdict[str, float] = defaultdict(float)
events_processed = 0
last_event_time: str | None = None
mqtt_connected = False


# --- Health endpoint ---
class HealthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/health":
            body = json.dumps({
                "status": "ok",
                "mqtt_connected": mqtt_connected,
                "events_processed": events_processed,
                "last_event": last_event_time,
            })
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(body.encode())
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, format, *args):
        pass  # suppress access logs


def start_health_server():
    server = HTTPServer(("0.0.0.0", HEALTH_PORT), HealthHandler)
    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()
    logger.info("Health server started on port %d", HEALTH_PORT)


# --- Snapshot fetching ---
async def fetch_snapshot(event_id: str) -> bytes | None:
    import aiohttp
    url = f"{FRIGATE_URL}/api/events/{event_id}/snapshot.jpg"
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                if resp.status == 200:
                    data = await resp.read()
                    logger.info("Fetched snapshot for %s: %d bytes", event_id, len(data))
                    return data
                logger.warning("Snapshot fetch failed for %s: HTTP %d", event_id, resp.status)
    except Exception:
        logger.exception("Snapshot fetch error for %s", event_id)
    return None


# --- Event processing ---
async def process_event(event_data: dict, claude: anthropic.Anthropic, matrix: MatrixClient):
    global events_processed, last_event_time

    event_type = event_data.get("type")
    if event_type != "end":
        return

    after = event_data.get("after", {})
    event_id = after.get("id", "")
    camera = after.get("camera", "unknown")
    label = after.get("label", "")
    score = after.get("top_score", after.get("score", 0))
    has_snapshot = after.get("has_snapshot", False)

    # --- Filtering ---
    if label not in ALLOWED_LABELS:
        return
    if score < MIN_SCORE:
        return
    if not has_snapshot:
        return

    # Dedup
    now = time.time()
    if event_id in recent_events and now - recent_events[event_id] < DEDUP_WINDOW_SECONDS:
        return
    recent_events[event_id] = now

    # Rate limit per camera
    if now - last_notify_per_camera[camera] < RATE_LIMIT_SECONDS:
        logger.info("Rate limited: %s on %s", label, camera)
        return
    last_notify_per_camera[camera] = now

    # Cleanup old dedup entries
    cutoff = now - DEDUP_WINDOW_SECONDS * 2
    for eid in [k for k, v in recent_events.items() if v < cutoff]:
        del recent_events[eid]

    logger.info("Processing event: %s on %s (%s, score=%.2f)", event_id, camera, label, score)

    # --- Fetch snapshot ---
    snapshot = await fetch_snapshot(event_id)

    # --- AI analysis ---
    description = None
    if snapshot:
        description = analyze_snapshot(claude, snapshot)
    if not description:
        description = f"{label.capitalize()} detected on {camera}."

    # --- Format message ---
    score_pct = int(score * 100)
    ts = time.strftime("%I:%M:%S %p", time.localtime())
    date_str = time.strftime("%b %d, %Y", time.localtime())
    frigate_link = f"{FRIGATE_URL}/review?id={event_id}"

    plain = f"{camera} - {label.capitalize()} - {score_pct}%\n{description}\n\nView in Frigate: {frigate_link}\n{ts} - {date_str}"
    html = (
        f"<b>{camera}</b> &middot; {label.capitalize()} &middot; {score_pct}%<br/>"
        f"{description}<br/><br/>"
        f'<a href="{frigate_link}">View in Frigate</a><br/>'
        f"<small>{ts} &middot; {date_str}</small>"
    )

    # --- Send to Matrix ---
    if snapshot:
        filename = f"{camera}_{event_id}.jpg"
        mxc_uri = await matrix.upload_image(snapshot, filename)
        if mxc_uri:
            await matrix.send_image(mxc_uri, filename, len(snapshot))

    await matrix.send_text(plain, html)

    events_processed += 1
    last_event_time = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    logger.info("Notification sent for %s on %s", label, camera)


# --- MQTT ---
def create_mqtt_client(loop: asyncio.AbstractEventLoop, claude: anthropic.Anthropic, matrix: MatrixClient):
    global mqtt_connected

    def on_connect(client, userdata, flags, reason_code, properties):
        global mqtt_connected
        mqtt_connected = True
        client.subscribe("frigate/events")
        logger.info("MQTT connected, subscribed to frigate/events")

    def on_disconnect(client, userdata, flags, reason_code, properties):
        global mqtt_connected
        mqtt_connected = False
        logger.warning("MQTT disconnected: %s", reason_code)

    def on_message(client, userdata, msg):
        try:
            payload = json.loads(msg.payload)
            asyncio.run_coroutine_threadsafe(process_event(payload, claude, matrix), loop)
        except json.JSONDecodeError:
            logger.warning("Invalid JSON on %s", msg.topic)
        except Exception:
            logger.exception("Error processing MQTT message")

    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.on_message = on_message
    return client


# --- Main ---
def main():
    logger.info("Starting frigate-notify")

    claude = anthropic.Anthropic()
    matrix = MatrixClient(MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN, MATRIX_ROOM_ID)

    loop = asyncio.new_event_loop()
    loop_thread = Thread(target=loop.run_forever, daemon=True)
    loop_thread.start()

    start_health_server()

    mqtt_client = create_mqtt_client(loop, claude, matrix)
    mqtt_client.connect(MQTT_HOST, MQTT_PORT)

    def shutdown(signum, frame):
        logger.info("Shutting down...")
        mqtt_client.disconnect()
        loop.call_soon_threadsafe(loop.stop)
        sys.exit(0)

    signal.signal(signal.SIGTERM, shutdown)
    signal.signal(signal.SIGINT, shutdown)

    mqtt_client.loop_forever()


if __name__ == "__main__":
    main()

Step 5: Commit

git add compose/hf-pdocker-01/frigate-notify/app/
git commit -m "feat: add frigate-notify Python application

MQTT listener for Frigate events with Claude Vision analysis
and Matrix notification delivery."

---

Task 3: Create Dockerfile and docker-compose.yml

Files:

• Create: compose/hf-pdocker-01/frigate-notify/Dockerfile
• Create: compose/hf-pdocker-01/frigate-notify/docker-compose.yml
• Create: compose/hf-pdocker-01/frigate-notify/.env

Step 1: Create Dockerfile

compose/hf-pdocker-01/frigate-notify/Dockerfile

FROM python:3.12-slim

WORKDIR /app

COPY app/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app/ .

EXPOSE 5199

CMD ["python", "-u", "main.py"]

Step 2: Create docker-compose.yml

compose/hf-pdocker-01/frigate-notify/docker-compose.yml

services:
  frigate-notify:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: frigate-notify
    restart: unless-stopped
    network_mode: host
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - MATRIX_ACCESS_TOKEN=${MATRIX_ACCESS_TOKEN}
      - MATRIX_ROOM_ID=${MATRIX_ROOM_ID}
      - FRIGATE_URL=http://192.168.86.192:5000
      - MQTT_HOST=192.168.86.192
      - MQTT_PORT=1883
      - MATRIX_HOMESERVER=http://192.168.86.192:8008
    mem_limit: 256M
    memswap_limit: 256M
    healthcheck:
      test: ["CMD-SHELL", "curl -sf http://localhost:5199/health || exit 1"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 30s
    logging:
      driver: "json-file"
      options:
        max-size: "50m"
        max-file: "3"

Step 3: Create .env file

compose/hf-pdocker-01/frigate-notify/.env

ANTHROPIC_API_KEY=<get from existing netclaw .env or vault>
MATRIX_ACCESS_TOKEN=<from Task 1 step 2>
MATRIX_ROOM_ID=<from Task 1 step 3>

Note: Get the ANTHROPIC_API_KEY from the existing netclaw deployment:

ssh glyptic@192.168.86.100 "cat /mnt/docker-data/compose/hf-pdocker-02/netclaw/.env | grep ANTHROPIC"

Step 4: Commit

git add compose/hf-pdocker-01/frigate-notify/Dockerfile
git add compose/hf-pdocker-01/frigate-notify/docker-compose.yml
# Do NOT commit .env (contains secrets)
git commit -m "feat: add frigate-notify Docker configuration

Standalone Compose service on hf-pdocker-01 with host networking,
256M memory limit, health check on port 5199."

---

Task 4: Build and deploy on hf-pdocker-01

This task uses the docker-swarm-ops agent to SSH to hf-pdocker-01.

Step 1: Copy files to hf-pdocker-01

SCP the entire compose/hf-pdocker-01/frigate-notify/ directory to /mnt/docker-data/compose/hf-pdocker-01/frigate-notify/ on hf-pdocker-01.

Step 2: Build the container

cd /mnt/docker-data/compose/hf-pdocker-01/frigate-notify
docker compose build

Step 3: Start the container

docker compose up -d

Step 4: Verify container is running

docker ps --filter name=frigate-notify --format "table {{.Names}}\t{{.Status}}"

Expected: frigate-notify Up X seconds (health: starting)

Step 5: Check logs for successful startup

docker logs frigate-notify --tail 20

Expected: Health server started, MQTT connected, subscribed to frigate/events.

Step 6: Verify health endpoint

curl -s http://localhost:5199/health | python3 -m json.tool

Expected: {"status": "ok", "mqtt_connected": true, "events_processed": 0, "last_event": null}

---

Task 5: Add monitoring and verify end-to-end

Files:

• Modify: stacks/monitoring/prometheus.yml - add health probe target

Step 1: Add blackbox probe target

In stacks/monitoring/prometheus.yml, add to the http-pdocker01 targets list:

- http://192.168.86.192:5199/health   # Frigate Notify

Step 2: Deploy updated prometheus config

SCP to hf-pdocker-01, copy to NAS, hot-reload Prometheus:

sudo cp /data/monitoring/prometheus.yml /mnt/docker-data/configs/prometheus/prometheus.yml
# Reload inside container
docker exec $(docker ps -q -f name=monitoring_prometheus) wget -qO- --post-data='' http://localhost:9090/-/reload

Step 3: Trigger a test detection

Walk in front of a camera, or wait for an organic detection event. Check logs:

docker logs frigate-notify --tail 30 -f

Expected: Event processing log, snapshot fetch, vision analysis, Matrix notification sent.

Step 4: Verify Matrix message received

Check #video-alerts:hobbs.farm in Matrix for the snapshot image and AI-generated description.

Step 5: Store credentials in vault

Use vault-credential-manager agent to store the Frigate bot credentials.

Step 6: Commit monitoring update

git add stacks/monitoring/prometheus.yml
git commit -m "feat: add frigate-notify to Prometheus blackbox probes"

---

Task 6: Update documentation and memory

Step 1: Update MEMORY.md

Add entry for frigate-notify service:

• Container location, port, purpose
• Matrix bot identity and room
• Key config notes

Step 2: Update CLAUDE.md service distribution

Add frigate-notify to hf-pdocker-01 Compose services list.

Step 3: Update alert-manager agent

Add frigate-notify to the service list and probe targets documentation.

Step 4: Commit documentation updates

git add -A
git commit -m "docs: add frigate-notify to infrastructure documentation"