Serving a local model from a Raspberry Pi: one process for browsers (SSE) and devices (MQTT)

You have a small box - a Raspberry Pi, a Jetson, a cheap ARM VM - running a local model. Two kinds of traffic want in:

• People (browser tabs, curl, httpx scripts) call an HTTP endpoint and want the answer streamed token by token, not after the last token.
• Devices (sensors, cameras, controllers) speak MQTT. They report telemetry and drop batch jobs.

The conventional stack for that is three or four daemons: a web application server, a reverse proxy in front of it for HTTP/2, a separate MQTT broker, plus the glue that keeps them coordinated. On an edge box, every extra part is another thing to provision - and on ARM or RISC-V, potentially another native dependency to cross-compile.

I've been building BlackBull, a pure-Python ASGI framework whose protocol stack (HTTP/1.1, HTTP/2, WebSocket, gRPC, and an MQTT 5 broker) is all Python, all one process. Here is the whole edge-serving surface: one Python process

┌──────────────────────────────────────────────────────────┐
│ :8000 HTTP/1.1 · HTTP/2 (h2c, or ALPN with TLS)         │
│   GET /generate ──► SSE token stream                     │
│   GET /devices  ──► latest readings (JSON)               │
│                                                          │
│ :1883 MQTT 5 broker                                      │
│   sensors/{device}/{metric} ──► tap → in-memory          │
│   jobs/# ──► $share/… round-robin work queue             │
└──────────────────────────────────────────────────────────┘
   browsers · curl · httpx                     devices · workers

(Part of my ongoing BlackBull series. Earlier posts cover the HTTP/2, MQTT, and gRPC internals; this one is about what they add up to on a single small box.)

Token streaming is just a handler

Token-streaming inference is a plain streaming handler: iterate the model, yield one SSE event per token.

from blackbull import BlackBull, EventSourceResponse

app = BlackBull()

@app.route(path='/generate')
async def generate(prompt: str = 'Hello, edge'):
    async def events():
        async for token in fake_model(prompt):
            yield {'event': 'token', 'data': token}
        yield {'event': 'done', 'data': ''}
    return EventSourceResponse(events())

prompt resolves from the query string, which keeps the endpoint compatible with the browser's EventSource API (GET-only). Each yield is written to the wire as it happens - transport backpressure, not buffering, paces the stream - so the first token reaches the client while the rest are still being generated.

HTTP/2 without ceremony

Concurrent generations are where HTTP/2 earns its place: several /generate streams multiplex over one connection instead of queueing head-of-line behind each other. No extra setup is needed - cleartext h2c is detected from the connection preface:

curl --http2-prior-knowledge -N 'http://localhost:8000/generate?prompt=hi'

Zero TLS setup for the LAN case. Browsers require HTTP/2 over TLS; pass certfile= / keyfile= and ALPN negotiates h2 on the same port.

Swapping in a real model

The example's fake_model is an async generator that sleeps between tokens, so the demo runs dependency-free. A real model is CPU-bound (or NPU/GPU-bound behind a blocking call), and running it inline would stall the event loop - every other stream, and the MQTT broker, would freeze between tokens.

Keep the loop free by pushing inference off-thread and draining tokens through a queue:

import asyncio

async def real_model(prompt: str):
    q: asyncio.Queue[str | None] = asyncio.Queue()
    loop = asyncio.get_running_loop()

    def run():
        # blocking generation, worker thread
        for token in llm.generate(prompt):  # llama-cpp, ONNX, ...
            loop.call_soon_threadsafe(q.put_nowait, token)
        loop.call_soon_threadsafe(q.put_nowait, None)

    task = asyncio.create_task(asyncio.to_thread(run))
    try:
        while (token := await q.get()) is not None:
            yield token
    finally:
        task.cancel()

The serving code around it - the route, the SSE framing, the multiplexing - is unchanged.

Device ingest: the broker you already run

MQTTExtension puts a full MQTT 5 broker in the same process. Devices connect to :1883 with any standard client; on_message taps let the application observe the traffic - here, keeping the latest reading per device for the HTTP side to report:

from blackbull.mqtt import MQTTExtension, Message

mqtt = app.add_extension(MQTTExtension(port=1883))
_readings: dict[str, dict[str, str]] = {}

@mqtt.on_message(topic='sensors/{device}/{metric}')
async def on_reading(msg: Message, device: str, metric: str):
    _readings.setdefault(device, {})[metric] = msg.payload.decode()

@app.route(path='/devices')
async def devices():
    return _readings

{device} and {metric} are captures - they match one topic level like + and arrive as keyword arguments, the MQTT counterpart of HTTP path params. Taps observe routing; delivery to subscribed MQTT clients happens whether or not a tap matches.

Batch jobs: a work queue with no queue server

Interactive requests take the HTTP door; batch work takes the MQTT door. Shared subscriptions (MQTT 5 §4.8.2) turn a topic into a load-balanced work queue: every worker that subscribes to the same $share/{group}/{filter} joins a share group, and the broker gives each matching message to exactly one member, round-robin.

mosquitto_sub -t '$share/workers/jobs/#' -p 1883 -V 5   # worker 1
mosquitto_sub -t '$share/workers/jobs/#' -p 1883 -V 5   # worker 2

mosquitto_pub -t 'jobs/generate' -m '{"prompt": "job-1"}' -p 1883 -V 5
mosquitto_pub -t 'jobs/generate' -m '{"prompt": "job-2"}' -p 1883 -V 5

job-1 and job-2 land on different workers - a work queue with no queue server. Scaling the pool is starting another subscriber: no extra infrastructure, no client library beyond MQTT, and the broker is the process you already run.

What this is not

An honest-limits section, because the pitch is deliberately narrow:

• BlackBull moves the bytes; it does not batch, schedule, or accelerate inference. If GPU throughput is your bottleneck, a dedicated inference server in front of the GPU does more for you than any web framework choice - put it behind the HTTP surface, don't replace it with one.
• The broker is a device-gateway broker, not a clustered message bus. It runs on one worker process, sessions in memory; a restart clears in-flight session state.
• No offline buffering for an empty share group. If no worker is connected when a job is published, the job is dropped, not queued. Keep at least one worker connected whenever producers are publishing.

The docs keep an explicit known-limitations list.

Try it

The example is dependency-free - the "model" is canned tokens - so it runs anywhere, including the box you're about to deploy to:

pip install blackbull
python examples/edge_inference.py

Then, in other terminals:

curl -N 'http://localhost:8000/generate?prompt=hello+edge'   # SSE tokens
mosquitto_pub -t 'sensors/pi-cam-1/temperature' -m '41.2' -p 1883 -V 5
curl http://localhost:8000/devices

Full walkthrough with the TLS/ALPN details: Edge inference serving.

BlackBull is Early Alpha (ZeroVer) and a learning project first - issues and protocol nitpicks are very welcome.