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: an 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.
SOCIAL SHARE CARD GENERATOR