BlackBull goes multi-protocol (part 3) — one `app.py`, two protocols

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