The first two posts were the engineering: part 1 rebuilt the connection
dispatcher so BlackBull's core stopped assuming HTTP, and part 2 put a lock-free,
single-owner MQTT 5 broker on top of that seam. This post is the payoff — what
all of it looks like from an application author's chair.
The short version: one file, one pip install, two protocols.
One decorator for HTTP, one for MQTT
from blackbull import BlackBull
from blackbull.mqtt import MQTTExtension, Message
app = BlackBull()
mqtt = app.add_extension(MQTTExtension(port=1883))
@app.route(path='/')
async def index():
return "HTTP here; MQTT broker on :1883."
@mqtt.on_message(topic='sensors/{room}/temperature')
async def on_temperature(msg: Message, room: str):
print(f"{room}: {msg.payload.decode()}") # {room} captured like a path param
app.run(port=8000) # HTTP on 8000, MQTT on 1883
mqtt.on_message deliberately mirrors app.route: both decorate an async
function, both match on an address pattern, and {room} captures a topic level
the same way {task_id} captures a URL path segment. The difference is
semantics — an HTTP route is the response; an MQTT tap observes the
broker's routing, which delivers to subscribers whether or not any handler is
registered.
MQTTExtension registers itself through the single extension seam
(app.add_extension) and binds :1883. The core BlackBull class carries zero
MQTT-specific code — the broker lives entirely in blackbull.mqtt.
Drive it with the tools you already have
It's a real MQTT 5 broker — CONNECT / SUBSCRIBE / PUBLISH at QoS 0–2, retained
messages, and Last-Will — so standard clients just work:
mosquitto_sub -t 'sensors/#' -p 1883 -V 5 # terminal 1
mosquitto_pub -t 'sensors/room1/temperature' \
-m '21.5' -p 1883 -V 5 # terminal 2
The message appears in the subscriber's terminal and fires the on_temperature
handler above. One process. No apt install mosquitto, no broker sidecar, no C
extension.
When HTTP and MQTT share a process
The real reason to put two protocols in one process isn't cleaner ops — it's
shared memory without a sidecar. Here's a tiny temperature dashboard that
would normally need a broker, a web server, and a Redis instance to bridge
them:
from blackbull import BlackBull
from blackbull.mqtt import MQTTExtension, Message
app = BlackBull()
mqtt = app.add_extension(MQTTExtension(port=1883))
# Plain dict — no locks, no Redis, no cross-process serialization.
latest: dict[str, float] = {}
@mqtt.on_message(topic='sensors/{room}/temperature')
async def ingest(msg: Message, room: str):
latest[room] = float(msg.payload)
@app.route(path='/sensors')
async def dashboard():
return latest # {"room1": 21.5, "room2": 30.1}
app.run(port=8000)
A sensor publishes 21.5 to sensors/room1/temperature via MQTT. The broker
routes it to subscribers and fires ingest, which updates the dictionary. A
browser hits /sensors and sees the current state as JSON. One variable, two
protocols, zero glue infrastructure.
This is what "multi-protocol" buys you in practice: the broker and the web
server live in the same memory space, so the bridge between them is just a
function call that writes to a dict. No redis-py, no pika, no
serialization — just Python.
Machine-readable docs for both protocols
BlackBull already auto-generates an OpenAPI 3.1 document from your route
signatures:
app.enable_openapi() # publishes /openapi.json and /docs
OpenAPI is an HTTP request/response vocabulary, though — it has nothing to say
about topics and publish/subscribe. The messaging world's counterpart is
AsyncAPI, and BlackBull ships the counterpart:
from blackbull.mqtt import AsyncAPIExtension
app.add_extension(AsyncAPIExtension(title='Sensor Gateway', version='1.0.0'))
After app.run() the document is served at /asyncapi.json, with a CDN-hosted
HTML viewer at /asyncapi (no new Python dependency). Each on_message topic
filter becomes a channel; each callback a receive operation. It's generated
lazily, so taps registered after the extension are still documented. Your HTTP
surface describes itself with OpenAPI; your MQTT surface describes itself with
AsyncAPI; same idea, right vocabulary for each.
The whole arc
Start to finish, that's the story:
- BlackBull decided to be a multi-protocol server.
- It prepared the ground — a protocol-agnostic dispatcher (peek-and-replay), built behind a regression oracle so the live HTTP server broke nothing.
- It built on that ground — a lock-free, single-owner MQTT 5 broker on the actor model, with HTTP throughput actually a little higher than before.
- And it stayed one
app.py: HTTP and MQTT side by side, sharing memory, each self-documenting — no sidecars, no C extensions.
What's next
MQTT is the first consumer of the multi-protocol seam, not the last.
@app.raw_handler already lets you bind raw TCP to a port and handle the
reader/writer yourself for custom protocols. A first-class gRPC handler — riding
the existing HTTP/2 stream, flow-control, and trailers machinery — is the
natural next step. Same pattern: register a ProtocolBinding, let the core
detect and dispatch. The core won't have to learn a thing about gRPC. That's
the whole point.
Try it
pip install 'blackbull[mqtt]'
python examples/mqtt_broker.py
Source: github.com/TOKUJI/BlackBull
Docs: tokuji.github.io/BlackBull
Top comments (0)