Skip to content

blackbull.mqtt.asyncapi

blackbull.mqtt.asyncapi

AsyncAPI 3.0 documentation for the MQTT broker — :class:AsyncAPIExtension.

OpenAPI documents BlackBull's HTTP surface; it has no vocabulary for topics, QoS, retain, or the publish/subscribe direction, so the MQTT broker is — by design — invisible to it. AsyncAPI (https://www.asyncapi.com/) is the messaging-world counterpart, and this extension emits an AsyncAPI 3.0 document describing the topic taps the application registered via MQTTExtension.on_message, served over HTTP exactly as /openapi.json is.

It is a normal :class:~blackbull.extension.Extension, parallel to OpenAPIExtension and coexisting with it::

app = BlackBull()
mqtt = app.add_extension(MQTTExtension(port=1883))
app.add_extension(AsyncAPIExtension(title='Sensor Gateway', version='1.0.0'))

@mqtt.on_message(topic='sensors/{room}/temperature')
async def on_temp(msg, room): ...

After app.run() the document is at /asyncapi.json and an HTML viewer at /asyncapi.

Honest caveats (also stated in the document's info.description):

  1. It documents the application's taps, not "the broker's API". A broker accepts any topic from any client; on_message filters describe only what this app observes.
  2. QoS / retain are not captured — on_message taps regardless of QoS, so MQTT channel bindings are omitted until the tap API carries that metadata.
  3. Payloads are opaque bytes (application/octet-stream) until an optional schema= on on_message lands (a deliberate fast-follow).

AsyncAPIExtension

Bases: Extension

Mount an AsyncAPI 3.0 spec endpoint (+ HTML viewer) describing the MQTT taps.

Requires an MQTTExtension on the same app; it is read lazily (per request) via app.extensions['mqtt'], so there is no ordering constraint at registration — and taps added after this extension are still documented. The MQTT extension must, however, be present when the spec route is hit, or the request raises RuntimeError.

Two GET routes are registered (mirroring OpenAPIExtension):

  • spec_path (default /asyncapi.json) → the document as JSON;
  • docs_path (default /asyncapi, None to skip) → an HTML page hosting the CDN AsyncAPI renderer.

init_app(app)

Wire the spec + docs routes onto app through the public API.

asyncapi_html(spec_url, title='BlackBull — AsyncAPI')

Return a self-contained HTML page hosting the AsyncAPI renderer.

generate_asyncapi(mqtt, *, title, version, description, server_host)

Build an AsyncAPI 3.0 document for mqtt's registered on_message taps.

mqtt is an MQTTExtension (anything exposing iter_subscriptions() and port). The returned dict is JSON-serialisable.