Scenario

A web service rarely lives in isolation. It serves HTTP requests, but it also needs to publish events, dispatch work to queues, and react to messages other services produce. With kubemq-faststream you do not stand up a separate worker process for that: the KubeMQBroker runs inside the same process as the web app, sharing its async event loop and startup/shutdown lifecycle.

The result is one deployable unit. An incoming HTTP request can publish to KubeMQ or issue a request-reply call; background @broker.subscriber handlers consume KubeMQ messages on the same loop and can push results back out to connected clients. This page walks through the integration pattern for FastAPI, Django, Flask, and Starlette, plus the lifecycle hooks that tie the broker to the application boundary.

The key idea is shared lifecycle: the broker is started when the web app starts and stopped when it shuts down, so connections open and close cleanly with the process.

Architecture

The following diagram shows the request path: an HTTP request hits a web route, the route calls the broker, KubeMQ delivers the message, and a background subscriber handles it on the same event loop.

FastAPI: Full Integration

The cleanest integration uses FastAPI's lifespan context manager. Start the FastStream app on entry and stop it on exit — the broker now lives and dies with the web app. Routes call broker.publish(...) to turn HTTP requests into KubeMQ messages, and a background @broker.subscriber consumes them on the same loop.

fastapi_full.py

from __future__ import annotations

import logging
import os
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

if TYPE_CHECKING:
    from collections.abc import AsyncIterator

import uvicorn
from fastapi import FastAPI, HTTPException

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")
CHANNEL = "example.integrations.fastapi"

broker = KubeMQBroker(KUBEMQ_ADDRESS)
faststream_app = FastStream(broker)


@broker.subscriber(events=CHANNEL)
async def handle_kubemq_message(msg: dict) -> None:
    """Background subscriber processes KubeMQ messages."""
    logger.info("[KubeMQ] Received: %s", msg)


@asynccontextmanager
async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
    """Start FastStream broker on startup, stop on shutdown."""
    await faststream_app.start()
    logger.info("FastStream broker started")
    try:
        yield
    finally:
        await faststream_app.stop()
        logger.info("FastStream broker stopped")


web_app = FastAPI(lifespan=lifespan, title="FastAPI + KubeMQ Full Integration")


@web_app.post("/publish")
async def publish_endpoint(payload: dict) -> dict:
    """Publish a message to KubeMQ via HTTP POST."""
    try:
        await broker.publish(payload, events=CHANNEL)
        return {"status": "published", "channel": CHANNEL}
    except Exception as exc:
        raise HTTPException(status_code=500, detail=f"Publish failed: {exc}") from exc


@web_app.get("/health")
async def health() -> dict:
    """Health check endpoint -- verifies broker connectivity."""
    is_healthy = await broker.ping(timeout=2.0)
    return {
        "broker": "connected" if is_healthy else "disconnected",
        "address": KUBEMQ_ADDRESS,
        "channel": CHANNEL,
    }


if __name__ == "__main__":
    uvicorn.run(web_app, host="0.0.0.0", port=8000)

Drive it with two HTTP calls — one to publish, one to check broker connectivity:

curl -X POST http://localhost:8000/publish \
     -H 'Content-Type: application/json' \
     -d '{"message": "hello from HTTP"}'

curl http://localhost:8000/health

The POST publishes the JSON body to KubeMQ; the background subscriber receives it on the same loop and logs [KubeMQ] Received: {'message': 'hello from HTTP'}. The /health route calls broker.ping(timeout=2.0) so a load balancer can probe broker connectivity, not just HTTP liveness.

FastAPI: Dependency Injection

Reaching for the module-level broker global works, but injecting it through FastAPI's Depends() decouples route logic from global state and makes routes easier to override in tests. Define a dependency that returns the broker, then declare it as a parameter on each route.

fastapi_dependency_injection.py

from fastapi import Depends, FastAPI, HTTPException

from kubemq_faststream import KubeMQBroker

broker = KubeMQBroker(KUBEMQ_ADDRESS)


async def get_broker() -> KubeMQBroker:
    """FastAPI dependency that provides the KubeMQ broker instance."""
    return broker


@web_app.post("/publish")
async def publish_endpoint(
    payload: dict,
    b: KubeMQBroker = Depends(get_broker),  # noqa: B008
) -> dict:
    """Publish a message using an injected broker instance."""
    try:
        await b.publish(payload, events=CHANNEL)
        return {"status": "published", "channel": CHANNEL}
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc)) from exc


@web_app.get("/health")
async def health(b: KubeMQBroker = Depends(get_broker)) -> dict:  # noqa: B008
    """Health check using an injected broker instance."""
    try:
        is_healthy = await b.ping(timeout=2.0)
    except Exception:
        is_healthy = False
    return {"broker": "connected" if is_healthy else "disconnected"}

The lifespan wiring is identical to the full example — only the routes change. In a test you can override get_broker with FastAPI's dependency_overrides to return a TestKubeMQBroker instead of opening a real connection.

WebSocket Bridge

Because the subscriber and the web routes run on the same loop, you can bridge a live transport like WebSockets to KubeMQ in both directions. Messages received from a WebSocket client are published to KubeMQ; KubeMQ events are forwarded out to every connected client by the background subscriber.

fastapi_websocket_bridge.py

from __future__ import annotations

import os
from contextlib import asynccontextmanager

from faststream import FastStream
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")
CHANNEL = "example.integrations.websocket"

broker = KubeMQBroker(KUBEMQ_ADDRESS)
faststream_app = FastStream(broker)

# Connected WebSocket clients
connected_clients: list[WebSocket] = []


@broker.subscriber(events=CHANNEL)
async def handle_kubemq_event(msg: dict) -> None:
    """Forward KubeMQ events to all connected WebSocket clients."""
    disconnected: list[WebSocket] = []
    for ws in connected_clients:
        try:
            await ws.send_json(msg)
        except Exception:
            disconnected.append(ws)
    for ws in disconnected:
        connected_clients.remove(ws)


@asynccontextmanager
async def lifespan(_app: FastAPI):
    """Start FastStream broker alongside FastAPI."""
    await faststream_app.start()
    yield
    await faststream_app.stop()


web_app = FastAPI(lifespan=lifespan, title="WebSocket-KubeMQ Bridge")


@web_app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket) -> None:
    """Handle WebSocket connections -- bridge messages to KubeMQ."""
    await ws.accept()
    connected_clients.append(ws)
    try:
        while True:
            data = await ws.receive_json()
            await broker.publish(data, events=CHANNEL)
    except WebSocketDisconnect:
        connected_clients.remove(ws)

Connect a client and the loop is complete:

websocat ws://localhost:8000/ws

Anything the client sends is published to the example.integrations.websocket event channel, and every event on that channel — including ones published by other services — fans out to all connected sockets. This makes KubeMQ a server-to-server fan-out hub behind a browser-facing WebSocket endpoint.

The WebSocket example needs the websockets package in addition to FastAPI and uvicorn: pip install fastapi uvicorn websockets.

Django, Flask, and Starlette

The integration shape changes with the framework's concurrency model. ASGI frameworks (Starlette, FastAPI) expose an async lifespan you hook directly. Synchronous frameworks (Flask, classic Django) need the async broker to run on its own loop.

Starlette has the same lifespan protocol as FastAPI, so the integration is nearly identical. Pass an asynccontextmanager as the lifespan= argument and publish from plain async route functions.

starlette_integration.py

import json
import os
from contextlib import asynccontextmanager

from faststream import FastStream
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")
CHANNEL = "example.integrations.starlette"

broker = KubeMQBroker(KUBEMQ_ADDRESS)
faststream_app = FastStream(broker)


@broker.subscriber(events=CHANNEL)
async def handle_kubemq_message(msg: dict) -> None:
    print(f"[KubeMQ] Received: {msg}")


@asynccontextmanager
async def lifespan(app: Starlette):
    await faststream_app.start()
    yield
    await faststream_app.stop()


async def publish_endpoint(request: Request) -> JSONResponse:
    payload = json.loads(await request.body())
    await broker.publish(payload, events=CHANNEL)
    return JSONResponse({"status": "published", "channel": CHANNEL})


routes = [Route("/publish", publish_endpoint, methods=["POST"])]
web_app = Starlette(routes=routes, lifespan=lifespan)

Flask is synchronous, so the async broker cannot share its request loop. Run the FastStream app in a daemon thread with its own event loop, started from the app factory. Routes hand publish coroutines to that loop with asyncio.run_coroutine_threadsafe.

flask_integration.py

import asyncio
import os
import threading

from faststream import FastStream
from flask import Flask, jsonify
from flask import request as flask_request

from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")
CHANNEL = "example.integrations.flask"

broker = KubeMQBroker(KUBEMQ_ADDRESS)
faststream_app = FastStream(broker)

# Background event loop for the async broker
_loop: asyncio.AbstractEventLoop | None = None


@broker.subscriber(events=CHANNEL)
async def handle_message(msg: dict) -> None:
    print(f"[KubeMQ] Received: {msg}")


def _run_broker() -> None:
    """Run the FastStream broker in a background thread."""
    global _loop
    _loop = asyncio.new_event_loop()
    asyncio.set_event_loop(_loop)
    _loop.run_until_complete(faststream_app.run())


def create_app() -> Flask:
    """Flask app factory."""
    app = Flask(__name__)

    broker_thread = threading.Thread(target=_run_broker, daemon=True)
    broker_thread.start()

    @app.route("/publish", methods=["POST"])
    def publish():
        payload = flask_request.get_json(force=True)
        if _loop is not None:
            future = asyncio.run_coroutine_threadsafe(
                broker.publish(payload, events=CHANNEL),
                _loop,
            )
            try:
                future.result(timeout=5.0)
            except Exception as exc:
                return jsonify({"error": str(exc)}), 500
        return jsonify({"status": "published", "channel": CHANNEL})

    return app

Django's management commands are the natural place to run long-lived async work. This self-contained script configures Django inline, then starts the broker, publishes a few demo messages, and stops cleanly inside a single asyncio.run(...) entry point — the same shape you would put inside a BaseCommand.handle().

django_integration.py

import asyncio
import os

import django
from django.conf import settings

if not settings.configured:
    settings.configure(
        DEBUG=True,
        SECRET_KEY="example-secret-key-not-for-production",
        INSTALLED_APPS=["django.contrib.contenttypes"],
        DATABASES={
            "default": {
                "ENGINE": "django.db.backends.sqlite3",
                "NAME": ":memory:",
            }
        },
        DEFAULT_AUTO_FIELD="django.db.models.BigAutoField",
    )
    django.setup()

from faststream import FastStream  # noqa: E402

from kubemq_faststream import KubeMQBroker  # noqa: E402

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")
CHANNEL = "example.integrations.django"

broker = KubeMQBroker(KUBEMQ_ADDRESS)
faststream_app = FastStream(broker)


@broker.subscriber(events=CHANNEL)
async def handle_message(msg: dict) -> None:
    print(f"[KubeMQ] Received: {msg}")


async def run_broker() -> None:
    """Start the broker, publish demo messages, then stop."""
    await faststream_app.start()
    try:
        for i in range(3):
            await broker.publish(
                {"source": "django", "action": "demo", "index": i},
                events=CHANNEL,
            )
        await asyncio.sleep(2)
    finally:
        await faststream_app.stop()


if __name__ == "__main__":
    asyncio.run(run_broker())

The dividing line is the event loop. Starlette and FastAPI already run an async loop, so await faststream_app.start() joins it directly through the lifespan. Flask serves requests synchronously, so the broker gets a dedicated thread and loop, and routes marshal coroutines onto it. Django's command runner owns the entry point, so asyncio.run(...) drives the broker for the life of the command.

Lifecycle Hooks

Whichever framework you use, the broker is started and stopped by some lifecycle boundary. When you run a FastStream app directly (the Django and CLI cases) you can attach work to three hooks instead of writing your own context manager:

Hook	When it runs	Use for
`@app.on_startup`	Before the broker connects	Initialize shared resources
`@app.after_startup`	After the broker is connected	Publish, prime caches, kick off work
`@app.on_shutdown`	During shutdown	Clean up resources

lifespan_management.py

import asyncio
import os

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")
CHANNEL = "example.web.lifespan"

broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)

shared_state: dict[str, str] = {}


@broker.subscriber(events=CHANNEL)
async def handle_message(msg: dict) -> None:
    env = shared_state.get("environment", "unknown")
    print(f"[{env}] Received: {msg}")


@app.on_startup
async def setup_resources() -> None:
    """Called BEFORE broker connects -- initialize shared resources."""
    shared_state["environment"] = "production"
    shared_state["version"] = "1.0.0"


@app.after_startup
async def run_demo() -> None:
    """Called AFTER broker connects -- safe to publish."""
    await broker.publish(
        {"action": "test", "env": shared_state["environment"]},
        events=CHANNEL,
    )
    await asyncio.sleep(2)
    await app.stop()


@app.on_shutdown
async def cleanup_resources() -> None:
    """Called during shutdown -- clean up resources."""
    shared_state.clear()


if __name__ == "__main__":
    asyncio.run(app.run())

The ordering matters: on_startup runs before the connection is live, so it is for preparing state you will need (configuration, shared dicts, clients). after_startup runs once the broker is connected, so it is the only safe place to call broker.publish(...).

For scripts and one-off publishers that do not need a full FastStream app, use the broker directly as an async context manager. Entering the block starts the connection; leaving it stops cleanly, even on error.

context_manager.py

import asyncio
import os

from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")


async def main() -> None:
    async with KubeMQBroker(KUBEMQ_ADDRESS) as broker:
        await broker.publish(
            {"source": "context_manager"},
            events="example.lifecycle.ctx",
        )


if __name__ == "__main__":
    asyncio.run(main())

Running the Examples

Start a KubeMQ broker

kubemq-faststream connects over native gRPC on port 50000 — there is no HTTP connector to enable. Port 9090 is the shared HTTP server (REST and connector endpoints); it is not required for FastStream but is harmless to expose.

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Install the web framework dependencies

kubemq-faststream brings in FastStream and the KubeMQ client, but each web framework is a separate dependency you install per example.

Example	Install
FastAPI (full / DI)	`pip install kubemq-faststream fastapi uvicorn`
FastAPI (WebSocket bridge)	`pip install kubemq-faststream fastapi uvicorn websockets`
Starlette	`pip install kubemq-faststream starlette uvicorn`
Flask	`pip install kubemq-faststream flask`
Django	`pip install kubemq-faststream django`

Run an example

ASGI apps run under uvicorn on port 8000; Flask runs on port 5000. Point any example at a different broker with the KUBEMQ_ADDRESS environment variable.

python fastapi_full.py

Install only the web framework you need. The examples guard their imports and exit with an install hint if the framework package is missing, so a single environment can hold several of them without conflict.

Web Framework Integration