MQTT Camera Runtime

Async MQTT runtime for INSTAR IP cameras with complete reconnect state machine, connection health separation, runtime hardening, and MCP integration.

Built on paho-mqtt>=2.1.0 with Paho sync client + isolated CommandExecutor worker. FastMCP-compatible async API.

Phase 11 Complete: Production-ready MCP server with Hermes integration support. Package installable, stdio transport verified, six camera tools exposed.

Hermes MCP Integration

The runtime includes a production-ready MCP server consumable by Hermes Agent and other MCP hosts.

Prerequisites

• Hermes Agent v0.21.0+ installed

• INSTAR camera reachable via MQTT broker

• camera-mcp installed locally

Install the package to make camera-mcp available:

pip install -e .

This installs the camera-mcp console script used by Hermes via stdio.

Installation hierarchy

Recommended — Catalog install (production)

Once the server is published to the Hermes catalog:

hermes mcp install instar-camera

The catalog manifest handles git checkout, pinned ref, venv bootstrap, and prompts for MQTT credentials. This is the supported production path.

Local development / testing

Use hermes mcp add when developing locally or testing without a published catalog entry:

hermes mcp add instar_camera \
  --command "$(which camera-mcp)" \
  --env INSTAR_MQTT_HOST=<broker_host> \
  --env INSTAR_MQTT_PORT=<broker_port> \
  --env INSTAR_MQTT_USERNAME=<mqtt_user> \
  --env INSTAR_MQTT_PASSWORD=<mqtt_pass> \
  --env INSTAR_MQTT_CLIENT_PREFIX=<prefix> \
  --env INSTAR_MQTT_CAMERA_ID=<camera_id>

Hermes will probe the server, list the six tools, and prompt to enable them. After saving, verify:

hermes mcp list
hermes mcp test instar_camera

Expected output shows 6 tools discovered and status ✓ enabled.

Environment variables required by the server:

• INSTAR_MQTT_HOST — MQTT broker hostname

• INSTAR_MQTT_PORT — MQTT broker port, default 1883

• INSTAR_MQTT_USERNAME — MQTT username

• INSTAR_MQTT_PASSWORD — MQTT password

• INSTAR_MQTT_CLIENT_PREFIX — MQTT topic prefix, default cameras

• INSTAR_MQTT_CAMERA_ID — Camera ID, default 224

You can also persist these in ~/.hermes/.env and reference them via --env INSTAR_MQTT_HOST=... on add, or edit ~/.hermes/config.yaml under mcp_servers.instar_camera.env.

Skill Integration

A Hermes skill instar-camera ships with the repo at skills/instar-camera/SKILL.md. The skill is independent of MCP installation and teaches the agent when and how to use the camera tools without constructing MQTT topics manually.

For local use, copy the skill to Hermes' skill path:

mkdir -p ~/.hermes/skills/home-automation/instar-camera
cp skills/instar-camera/SKILL.md ~/.hermes/skills/home-automation/instar-camera/

Hermes will discover it as a local skill and guide the agent to select the correct MCP tools. Once the catalog entry is upstream, the skill remains a separate user-facing asset.

The catalog manifest optional-mcps/instar-camera/manifest.yaml defines:

• git install with pinned ref

• bootstrap venv + pip install

• stdio transport command: "${INSTALL_DIR}/.venv/bin/camera-mcp"

• auth.env prompts for the six MQTT variables

• six tools default-enabled

See docs/hermes.md for complete configuration.

Available MCP Tools

The server exposes six tools:

• set_camera_value - Change camera setting with confirmation

• get_camera_state - Get observed state with confidence

• wait_camera_state - Wait for state change

• get_camera_health - Check broker/camera connectivity

• get_camera_image - Retrieve latest periodic snapshot

• list_camera_capabilities - Discover camera settings

Hermes registers these as mcp_instar_camera_*.

Skill Integration

A Hermes skill instar-camera ships with the repo at skills/instar-camera/SKILL.md. For local use, copy it to Hermes skill path:

mkdir -p ~/.hermes/skills/home-automation/instar-camera
cp skills/instar-camera/SKILL.md ~/.hermes/skills/home-automation/instar-camera/

Hermes will discover it as a local skill and guide the agent to use the MCP tools safely without manual MQTT topic construction.

See skills/instar-camera/SKILL.md for agent usage guidelines.

Related MCP server: MQTT MCP Server

Architecture

• Paho sync client with loop_start() background network thread

• Single CommandExecutor worker owns all Paho calls and state mutations

• Async callers enqueue operations and await futures via loop.call_soon_threadsafe

• Ownership invariant: Executor is the only component allowed to call Paho APIs or mutate application state. Paho callbacks only enqueue inbound events.

Project Structure

mqtt_runtime/
├── camera.py              # Public async API
├── executor.py            # CommandExecutor worker thread
├── paho_bridge.py         # Paho callbacks only enqueue
├── models.py              # StateEntry, Command, Confidence, Events
├── state_cache.py         # Confidence model, disconnect downgrade
├── subscription_registry.py # No wildcards, explicit topics only
├── command_manager.py     # FIFO, idempotency, superseding
├── topic_registry.py      # Topic metadata from topics.py + API docs
├── topics.py              # 740 topic definitions
├── event_queue.py         # Inbound/operation queues
└── experiments.py         # Paho behavior experiments

tests/
├── run_tests.py
└── unit/                  # 14 test suites
    ├── test_ownership.py
    ├── test_state_cache.py
    ├── test_command_lifecycle.py
    ├── test_async_boundary.py
    ├── test_connection_state_machine.py
    ├── test_subscription_registry.py
    ├── test_shutdown.py
    ├── test_concurrency.py
    ├── test_event_stream.py
    ├── test_payload_parsing.py
    ├── test_phase2_reconnect.py
    ├── test_phase2_health.py
    ├── test_phase2_events.py
    └── test_phase2_images.py

main.py                    # Live demo: toggle red alarm area

Quick Start

Install

pip install -r requirements.txt

Broker Defaults

IP: 192.168.2.224
Port: 1883 (plain) / 8883 (TLS)
User: admin
Password: instar
Prefix: cameras
ID: 224

Live Demo

python3 main.py

Toggles alarm/areas/red/enable via plain MQTT and TLS (insecure), verifies state with confidence tracking.

Programmatic Usage

from mqtt_runtime.camera import Camera

camera = Camera(client_prefix="cameras", camera_id="224")

await camera.connect(
    host="192.168.2.224",
    port=1883,
    username="admin",
    password="instar",
    use_tls=False,
)

# Relative path resolves via TopicRegistry
state = await camera.get_state("alarm/areas/red/enable")
print(state)  # {'value': '1', 'confidence': 'observed', ...}

await camera.set_value("alarm/areas/red/enable", "0")
await camera.subscribe("alarm/areas/red/enable")

await camera.close()

Public API

• await camera.connect(host, port, username, password, client_prefix, camera_id, use_tls, tls_insecure)

• await camera.close()

• await camera.set_value(topic, value, timeout=...) — topic resolves via registry

• await camera.get_state(topic)

• await camera.wait_for_state(topic, expected, timeout=...)

• await camera.subscribe(topic)

• await camera.unsubscribe(topic)

• await camera.check_health() — returns HealthResult with broker_connected and camera_online

• await camera.get_image(topic) — decode image outside callback path

• camera.events() — async iterator for CameraEvent (STATE_CHANGED, CONNECTION_CHANGED, COMMAND_COMPLETED, ERROR)

Topic Metadata

Catalogue of 663 stateful topics from:

• mqtt_runtime/topics.py — authoritative topic list + type/flags

• topics/api/_extracted.json — descriptions + value hints

Topic naming:

• Read (status): {prefix}/{id}/status/<path> — retained

• Write (command): {prefix}/{id}/<path> — publishes {"val": "..."}

Phase 2 Features

Phase 0 (Complete)

• ✅ Executor ownership model

• ✅ Async bridge with loop.call_soon_threadsafe

• ✅ Worker supervision

• ✅ Connection state machine

• ✅ Subscription registry (no wildcards)

• ✅ State confidence model (UNKNOWN / KNOWN_BUT_STALE / OBSERVED)

• ✅ Disconnect downgrade: OBSERVED → KNOWN_BUT_STALE

• ✅ Retained messages never confirm commands

• ✅ Command lifecycle with FIFO per topic

• ✅ Idempotency and superseding

• ✅ Event stream

• ✅ Shutdown resolves all futures

Phase 1 (Complete)

• ✅ Command confirmation requires OBSERVED confidence tier and non-retained message

• ✅ Same-topic commands produce last-wins final camera state

• ✅ UNCHANGED decided before publish from fresh OBSERVED state

• ✅ CLAMPED as valid observed different value

• ✅ Worker failures resolve pending futures

• ✅ Queued commands may resume after reconnect

Phase 2 (Complete)

• ✅ Complete reconnect state machine with exponential backoff

• ✅ Subscription restoration hardening with pending additions/removals

• ✅ Connection health separation: broker_connected vs camera_online

• ✅ StateCache production hardening with immutable snapshots and bounded LRU

• ✅ Public event stream API with bounded queue and diagnostics

• ✅ Public async API stabilization

• ✅ Image/content support with LRU cache and base64 validation

• ✅ MCP adapter boundary with Result model

Testing

python3 tests/run_tests.py

14 test suites, all passing. Tests verify ownership boundaries, async isolation, state transitions, command lifecycle, reconnect behavior, health separation, event ordering, image handling.

Phase 2 Verification Report

• Test Suites: 14 / 14 passing

• Phase 0 Guarantees: Preserved

• Phase 1 Guarantees: Preserved

• Phase 2 Guarantees: Reconnect deterministic, state confidence enforced, MCP adapter ready

• Remaining: Empirical Paho runtime verification against live broker

Migration from mcp_server.py

mcp_server.py deleted. Functionality migrated to:

• mqtt_runtime/topic_registry.py — TopicRegistry + CameraTopic

• mqtt_runtime/camera.py — Async API

• mqtt_runtime/topics.py — Moved from topics/topics.py

Old synchronous API replaced with async ownership model.

Security

• Credentials via parameters, never source

• TLS opt-in, cert verification default

• No wildcards allowed in subscriptions

• Worker failure isolates per-operation errors