The minimum server

POST /api/devices/heartbeat
  → 200 { "status": "ok", "last_seen_at": "<iso8601>", "pending_commands": 0 }
  → 403 if you want the device to stop (it treats this as terminal)
  → 404 if the MAC is unknown

POST  /api/devices/{mac}/incident             → 200 { "incident_id": 1, "status": "recorded" }
GET   /api/devices/{mac}/commands/pending     → 200 { "commands": [] }
PATCH /api/devices/{mac}/commands/{id}/ack    → 200
PATCH /api/devices/{mac}/commands/{id}/result → 200
GET   /api/devices/{mac}/ota/check            → 200 { "update_available": false }
PATCH /api/devices/{mac}/ota/{id}/progress    → 200

import asyncio, re

MAC_RE = re.compile(r"^[0-9A-F]{2}(:[0-9A-F]{2}){5}$")

def normalise_mac(raw):
    s = raw.strip().upper().replace("-", ":")
    if MAC_RE.match(s):
        return s
    hexonly = s.replace(":", "")
    if len(hexonly) == 12 and all(c in "0123456789ABCDEF" for c in hexonly):
        return ":".join(hexonly[i:i + 2] for i in range(0, 12, 2))
    return None

def parse(datagram):
    """Returns (meta, pcm) or None. Invalid packets are skipped, never fatal."""
    nl = datagram.find(b"\x0a")
    if nl < 1:
        return None
    header = datagram[:nl].rstrip(b"\r").decode("ascii", "ignore")
    pcm    = datagram[nl + 1:]
    if not pcm:
        return None                                  # empty payload: drop

    parts = header.split(",")
    if len(parts) < 9 or parts[0] != "v1":
        return None                                  # wrong version or too few fields

    _, seq, session, mac, org, prop, team, user = parts[:8]
    mac = normalise_mac(mac)
    if not mac or not seq.isdigit() or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", session):
        return None

    def as_int(v):
        return int(v) if v.strip().isdigit() else None

    return ({"seq": int(seq), "session": session, "mac": mac,
             "org_id": as_int(org), "property_id": as_int(prop),
             "team_id": as_int(team), "user_id": as_int(user)}, pcm)

class Ingest(asyncio.DatagramProtocol):
    def __init__(self):
        self.sessions = {}                           # keyed on mac|device_session

    def datagram_received(self, data, addr):
        parsed = parse(data)
        if not parsed:
            return                                   # log it; never raise
        meta, pcm = parsed
        key = f"{meta['mac']}|{meta['session']}"     # a reboot is a NEW session
        stream = self.sessions.get(key)
        if stream is None:
            stream = self.sessions[key] = open_recogniser(meta)
        stream.feed(pcm)                             # your STT provider

async def main():
    loop = asyncio.get_running_loop()
    await loop.create_datagram_endpoint(Ingest, local_addr=("0.0.0.0", 12345))
    await asyncio.Event().wait()

asyncio.run(main())

from fastapi import FastAPI, Request, Response, Header

app   = FastAPI()
START = b"\xFF\xFF\xFF\xFF"
END   = b"\xEE\xEE\xEE\xEE"
downlink = {}          # mac -> [(text, pcm), ...]  your business logic pushes here

@app.post("/voice")
async def voice(request: Request,
                x_device_id: str = Header(...),
                x_property_id: int | None = Header(None)):
    buf = bytearray()
    async for chunk in request.stream():
        buf += chunk

    if buf.startswith(START):
        buf = buf[len(START):]
    if buf.endswith(END):
        buf = buf[:-len(END)]                 # a missing end marker is tolerated

    await handle_utterance(x_device_id, x_property_id, bytes(buf))
    return Response("ok", media_type="text/plain")   # the entire success contract

@app.get("/sse/notifications")
async def notifications(x_device_id: str = Header(...)):
    queue = downlink.get(x_device_id.upper(), [])
    if not queue:
        return Response(status_code=204)      # normal, cheap, expected
    text, pcm = queue.pop(0)                  # delivered once — popping is the ack
    return Response(START + pcm + END,
                    media_type="application/octet-stream",
                    headers={"x-notification-type": "reply",
                             "x-notification-text": text})

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import json

app = FastAPI()

@app.websocket("/ws")
async def ws(sock: WebSocket):
    await sock.accept()
    hello = json.loads(await sock.receive_text())

    device = lookup(hello.get("mac"))
    if device is None:
        # Permanent. Say so precisely — the device must not retry this.
        await sock.send_json({"type": "error", "code": "unknown_device",
                              "message": "MAC not provisioned"})
        return await sock.close()

    await sock.send_json({
        "type": "ready", "session_id": new_session_id(),
        "property_id": device.property_id, "table": device.table_id,
        "currency": device.currency,
        "tts": {"encoding": "pcm_s16le", "sample_rate": 16000,
                "channels": 1, "frame_ms": 40},
        "tail_ms": 300, "heartbeat_interval_s": 20,
    })

    turn = None
    try:
        while True:
            msg = await sock.receive()
            if "bytes" in msg and msg["bytes"] is not None:
                await recogniser.feed(msg["bytes"])
                continue

            evt = json.loads(msg["text"])
            if evt["type"] == "interrupt" and turn:
                turn.cancel()          # abandon the rest of this reply at once
                turn = None
            elif evt["type"] == "bye":
                break
    except WebSocketDisconnect:
        pass

async def speak(sock, pcm_frames):
    """Always bracket agent audio with these two — the device mutes its mic on
       speaking_start and unmutes tail_ms after speaking_end."""
    await sock.send_json({"type": "speaking_start"})
    for frame in pcm_frames:                 # 1280-byte s16le frames
        await sock.send_bytes(frame)
    await sock.send_json({"type": "speaking_end"})