Payload Schemas¶

All cosalette MQTT payloads are either JSON objects or plain strings. This page documents the exact schema for each payload type. For topic routing and QoS details, see MQTT Topic Reference.

Device State¶

State payloads are user-defined JSON dicts. The framework imposes no schema — whatever dict you pass to ctx.publish_state() is serialised to JSON and published to {prefix}/{device}/state.

# In your device function:
await ctx.publish_state({"position": 75, "tilt": 45})

Produces:

{"position": 75, "tilt": 45}

The payload argument must be a dict[str, object]. The framework calls json.dumps() internally.

Error Payload¶

Published by the ErrorPublisher to {prefix}/error and (when a device name is known) to {prefix}/{device}/error. Defined in cosalette._errors.ErrorPayload.

Example¶

{
    "error_type": "invalid_command",
    "message": "InvalidCommandError",
    "device": "blind",
    "timestamp": "2026-02-14T12:34:56+00:00",
    "details": {}
}

Fields¶

Error Type Mapping¶

The error_type field is resolved from the error_type_map dict passed to the ErrorPublisher. The map keys are exact exception classes (no subclass matching). Unmapped exceptions produce "error" as the type. The map only assigns labels; it never discloses a message.

error_type_map = {
    ValueError: "invalid_command",
    TimeoutError: "timeout",
}

Common Error Types¶

The framework labels these exceptions itself. Framework labels are authoritative: an app error_type_map entry for one of these types is overridden, with a startup warning when its label differs.

ReturnValidationError (a handler returned a value that does not match its return contract) is an application fault and keeps the generic error label.

By default every one of these payloads publishes the exception class name as message (for example "message": "InvalidJsonError"). The examples below show the messages published when the app lists the type, for example with App(disclose_messages_for=cosalette.FRAMEWORK_DISCLOSED_ERRORS | {TimeoutError}):

# Invalid JSON
{"error_type": "invalid_json", "message": "unexpected character, expected a JSON value: line 1 column 13 (char 12)", "device": "light"}

# Missing routing field
{"error_type": "missing_sub_key", "message": "Missing field 'command' in command payload", "device": "light"}

# Unknown sub-command: only type, length and a SHA-256 fingerprint of the value
{"error_type": "unknown_sub_command", "message": "Unknown sub-command for 'light' (type=str, len=5, fp=e14a51fd)", "device": "light"}

# Rejected typed-command or typed telemetry trigger input: redacted locations and pydantic error codes only
{"error_type": "invalid_payload", "message": "Payload validation failed (parameter 'cmd'): [field]: type=literal_error", "device": "light"}

# Command watchdog expired (a TimeoutError raised by handler code keeps its own text)
{"error_type": "timeout", "message": "command handler exceeded 30.0s watchdog", "device": "light"}

The invalid_payload message never contains the rejected value or pydantic's msg text.

See Error Handling (concept) for the full error pipeline.

Heartbeat Payload¶

Published by HealthReporter.publish_heartbeat() to {prefix}/status. Defined in cosalette._health.HeartbeatPayload.

The framework publishes an initial heartbeat on connect, then repeats at the heartbeat_interval (default 60 s). Set App(heartbeat_interval=None) to disable periodic heartbeats.

Example¶

{
    "status": "online",
    "uptime_s": 3600,
    "version": "0.3.0",
    "devices": {
        "blind": {"status": "ok"},
        "temp": {"status": "ok"}
    }
}

Fields¶

DeviceStatus Fields¶

Each entry in the devices map is a DeviceStatus object:

A telemetry entity with no fresh cycle for its stale_after bound, or a stream with stale_after= and no item for that long, reports status: "stale", which outranks "error" (Staleness). Non-telemetry entries carry status only.

Devices are added to tracking when publish_device_available() is called and removed when publish_device_unavailable() is called.

Availability Messages¶

Published on {prefix}/{device}/availability and {prefix}/status (LWT). These are plain strings, not JSON.

Command Payloads¶

Inbound messages on {prefix}/{device}/set topics are plain strings. The framework passes the raw payload to the command handler. The recommended approach is @app.command() — handlers only declare the parameters they need:

@app.command("valve")
async def handle_valve(payload: str) -> dict[str, object]:
    return {"valve_state": payload}

If the handler also needs the full MQTT topic:

@app.command("blind")
async def handle_blind(topic: str, payload: str) -> dict[str, object]:
    position = int(payload)
    return {"position": position}

Alternatively, inside an @app.device() function you can use @ctx.on_command:

@ctx.on_command
async def handle(topic: str, payload: str) -> None:
    position = int(payload)  # User-defined decoding
    await set_blind_position(position)

The framework performs no parsing, validation, or transformation on command payloads — decoding is entirely the responsibility of the command handler.

See Also¶

• MQTT Topic Reference — topic patterns, QoS, and retain settings
• Error Handling (concept) — error pipeline and error_type_map
• Health & Availability (concept) — heartbeat scheduling and LWT integration
• ADR-011 — Error Handling
• ADR-012 — Health Reporting