kubemq-faststream is a first-class FastStream broker, not a thin transport shim. Every message — published or consumed, across all five patterns — flows through the full FastStream pipeline: parser, decoder, and the complete middleware chain. Because the broker runs that chain unchanged, standard FastStream middleware composes directly with KubeMQ messages. You attach middleware exactly as you would on a Kafka or Redis broker, and the same TestKubeMQBroker runs the pipeline end-to-end in tests.

This is the seam for observability and cross-cutting concerns: Prometheus metrics, OpenTelemetry tracing, request logging, timing, validation — all of it slots in without touching your handlers.

Prerequisites

A KubeMQBroker app already wired up (see Configuration & Security)
The optional backend package for whichever middleware you enable — prometheus-client for PrometheusMiddleware, opentelemetry-sdk / opentelemetry-api for TelemetryMiddleware

How Middleware Attaches

Middleware is registered on the broker constructor through the middlewares parameter. It accepts a sequence of FastStream middleware classes or instances:

broker.py constructor

broker = KubeMQBroker(
    "kubemq://localhost:50000",
    middlewares=(my_middleware,),
)

Each middleware wraps the handler call. The first middleware in the sequence is the outermost wrapper — it sees the message before any other middleware and returns after them. Both the live broker and TestKubeMQBroker execute the full chain, so the behavior you verify in tests matches production.

Install the optional dependency for each middleware before enabling it. The built-in Prometheus and OpenTelemetry middlewares live in FastStream but pull their backends from separate packages: prometheus-client for metrics and opentelemetry-sdk / opentelemetry-api for tracing. The examples below raise a clear SystemExit if the package is missing.

Prometheus Metrics

FastStream ships a PrometheusMiddleware that records message counts, processing latencies, and error counters automatically — no handler changes required. You expose the collected metrics over an HTTP endpoint that Prometheus scrapes.

Install the backend first:

pip install prometheus-client

Construct the middleware with a CollectorRegistry, attach it to the broker, and start an HTTP metrics server once the app is up:

prometheus_metrics.py

import asyncio

from faststream import FastStream
from faststream.prometheus import PrometheusMiddleware
from prometheus_client import CollectorRegistry, start_http_server

from kubemq_faststream import KubeMQBroker

registry = CollectorRegistry()
prometheus_middleware = PrometheusMiddleware(registry=registry)

broker = KubeMQBroker(
    "kubemq://localhost:50000",
    middlewares=(prometheus_middleware,),
)
app = FastStream(broker)

CHANNEL = "example.middleware.prometheus"


@broker.subscriber(events=CHANNEL)
async def handle_message(msg: dict) -> None:
    # Metrics are recorded automatically — no instrumentation here.
    print(f"[Prometheus] Processed: {msg}")


@app.after_startup
async def run_demo() -> None:
    start_http_server(port=9090, registry=registry)
    print("Prometheus metrics available at http://localhost:9090/metrics")

    for i in range(5):
        await broker.publish({"event_id": i, "type": "metric-demo"}, events=CHANNEL)


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

The middleware increments its counters as messages pass through the consume scope. Point a Prometheus scrape config at http://localhost:9090/metrics to collect them.

The metrics server in this example binds port 9090 on the local host. KubeMQ's own shared HTTP server also defaults to 9090. When you run both the broker and the metrics endpoint on the same machine, pick a different port for start_http_server (for example 8000) to avoid a collision — the broker itself connects only over gRPC on 50000.

OpenTelemetry Tracing

FastStream's TelemetryMiddleware creates a span for each KubeMQ message and propagates trace context through the message, so a trace can span multiple services connected via KubeMQ. You wire it to a standard OpenTelemetry TracerProvider.

Install the backend first:

pip install opentelemetry-sdk opentelemetry-api

Set up a provider (here exporting spans to the console for demonstration), build the middleware from it, and attach it to the broker:

opentelemetry_tracing.py

import asyncio

from faststream import FastStream
from faststream.opentelemetry import TelemetryMiddleware
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor

from kubemq_faststream import KubeMQBroker

resource = Resource.create({"service.name": "kubemq-faststream-demo"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)

telemetry_middleware = TelemetryMiddleware(tracer_provider=provider)

broker = KubeMQBroker(
    "kubemq://localhost:50000",
    middlewares=(telemetry_middleware,),
)
app = FastStream(broker)

CHANNEL = "example.middleware.otel"


@broker.subscriber(events=CHANNEL)
async def handle_message(msg: dict) -> None:
    # Spans are created and linked automatically.
    print(f"[OTel] Traced message: {msg}")


@app.after_startup
async def run_demo() -> None:
    for i in range(3):
        await broker.publish({"trace_id": i, "action": "otel-demo"}, events=CHANNEL)


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

In production, swap ConsoleSpanExporter for an OTLP exporter pointed at your collector. Trace context rides along with each published message, so a consumer in another service continues the same trace.

Custom Middleware

For logging, timing, validation, or any application-specific concern, subclass BaseMiddleware and override consume_scope. The method receives call_next (the next stage in the chain) and the message; you wrap the call however you need. No external packages are required.

custom_middleware.py

import asyncio
import logging
import time
from collections.abc import Awaitable, Callable
from typing import Any

from faststream import BaseMiddleware, FastStream

from kubemq_faststream import KubeMQBroker

logging.basicConfig(level=logging.INFO)
mw_logger = logging.getLogger("custom_middleware")


class TimingMiddleware(BaseMiddleware):
    """Logs wall-clock time spent in each handler."""

    async def consume_scope(
        self,
        call_next: Callable[[Any], Awaitable[Any]],
        msg: Any,
    ) -> Any:
        start = time.monotonic()
        try:
            result = await call_next(msg)
        finally:
            elapsed_ms = (time.monotonic() - start) * 1000
            mw_logger.info("Handler took %.1f ms", elapsed_ms)
        return result


broker = KubeMQBroker(
    "kubemq://localhost:50000",
    middlewares=(TimingMiddleware,),
)
app = FastStream(broker)

CHANNEL = "example.middleware.custom"


@broker.subscriber(events=CHANNEL)
async def handle_message(msg: dict) -> None:
    await asyncio.sleep(0.05)
    print(f"[Custom] Processed: {msg}")


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

The try/finally ensures the timing log fires even when the handler raises. A few common variations:

Logging — log the message before and after call_next, tagging each with a request ID.
Validation — inspect the message and raise before calling call_next to reject malformed input.
Error capture — wrap call_next in try/except to record or swallow handler exceptions.

Composing Multiple Middlewares

Pass several middlewares as one sequence and they stack in order — the first is the outermost wrapper. This lets you run a custom logging layer, an error-catch layer, Prometheus, and OpenTelemetry together in a single chain. A robust pattern is to add the optional middlewares only when their backend package is importable, so the app degrades gracefully when a dependency is absent.

middleware_chain.py

import asyncio
import logging
from collections.abc import Awaitable, Callable
from typing import Any

from faststream import BaseMiddleware, FastStream

from kubemq_faststream import KubeMQBroker

logging.basicConfig(level=logging.INFO)
chain_logger = logging.getLogger("middleware_chain")


class RequestIdMiddleware(BaseMiddleware):
    """Assigns a sequential request ID to each processed message."""

    _counter: int = 0

    async def consume_scope(
        self,
        call_next: Callable[[Any], Awaitable[Any]],
        msg: Any,
    ) -> Any:
        RequestIdMiddleware._counter += 1
        req_id = RequestIdMiddleware._counter
        chain_logger.info("[ReqID=%d] Before handler", req_id)
        result = await call_next(msg)
        chain_logger.info("[ReqID=%d] After handler", req_id)
        return result


class ErrorCatchMiddleware(BaseMiddleware):
    """Catches handler exceptions and logs them without crashing."""

    async def consume_scope(
        self,
        call_next: Callable[[Any], Awaitable[Any]],
        msg: Any,
    ) -> Any:
        try:
            return await call_next(msg)
        except Exception:
            chain_logger.exception("Handler error caught by middleware")
            return None


middlewares: list[Any] = [RequestIdMiddleware, ErrorCatchMiddleware]

try:
    from faststream.prometheus import PrometheusMiddleware
    from prometheus_client import CollectorRegistry

    middlewares.append(PrometheusMiddleware(registry=CollectorRegistry()))
    chain_logger.info("Prometheus middleware added to chain")
except ImportError:
    chain_logger.info("prometheus-client not installed — skipping Prometheus middleware")

try:
    from faststream.opentelemetry import TelemetryMiddleware
    from opentelemetry.sdk.trace import TracerProvider

    middlewares.append(TelemetryMiddleware(tracer_provider=TracerProvider()))
    chain_logger.info("OpenTelemetry middleware added to chain")
except ImportError:
    chain_logger.info("opentelemetry-sdk not installed — skipping OTel middleware")


broker = KubeMQBroker(
    "kubemq://localhost:50000",
    middlewares=middlewares,
)
app = FastStream(broker)


@broker.subscriber(events="example.middleware.chain")
async def handle_message(msg: dict) -> None:
    print(f"[Chain] Handled: {msg}")


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

With both optional backends installed, a message passes through RequestIdMiddleware → ErrorCatchMiddleware → Prometheus → OpenTelemetry → your handler, then unwinds in reverse. Note that ErrorCatchMiddleware sits outside the metrics and tracing layers, so the inner middlewares still observe and record the failing message before the error is caught.

Error Handling & Graceful Shutdown

Observability is most valuable around failures, so it helps to understand how the broker behaves when things go wrong.

Handler Exceptions

When a subscriber raises, FastStream catches the exception, logs it, and the application keeps processing subsequent messages — one bad message does not crash the app. For events (fire-and-forget), the failing message is not re-delivered. For queues, settlement follows the subscriber's AckPolicy (the default AckPolicy.ACK nacks and requeues on error). An ErrorCatchMiddleware like the one above can also intercept exceptions before they reach FastStream's default logging.

handler_exception.py

@broker.subscriber(events="example.error.exception")
async def handle_message(msg: dict) -> None:
    global counter  # noqa: PLW0603
    counter += 1
    if counter == 2:
        raise ValueError(f"Simulated failure on message #{counter}")
    print(f"Message #{counter} processed OK: {msg}")

The app continues running after the ValueError and processes message #3 normally.

Automatic Reconnection

The underlying KubeMQ SDK handles gRPC reconnection transparently. If the broker restarts mid-run, the client re-establishes its channels and delivery resumes — your code does not need retry wiring for the connection itself. gRPC keepalive (configured via keepalive_time_ms / keepalive_timeout_ms) keeps the channel healthy between messages.

reconnection.py

@app.after_startup
async def run_demo() -> None:
    for i in range(1, 11):
        try:
            await broker.publish({"seq": i}, events="example.error.reconnect")
            print(f"Published #{i}")
        except Exception as exc:
            print(f"Publish #{i} failed (will retry): {exc}")
        await asyncio.sleep(1)

Graceful Shutdown & Draining

On SIGTERM or SIGINT, the broker stops accepting new work and waits for in-flight handlers to finish before closing its connections. The graceful_timeout constructor option (default 15.0 seconds) caps how long it waits — handlers still running when the timeout elapses are cancelled. Set it to match your slowest handler so in-flight messages drain cleanly:

graceful_shutdown.py

import signal

broker = KubeMQBroker("kubemq://localhost:50000", graceful_timeout=10.0)
app = FastStream(broker)


@broker.subscriber(events="example.error.shutdown")
async def handle_message(msg: dict) -> None:
    print(f"Processing: {msg}")
    await asyncio.sleep(0.5)
    print(f"Done: {msg}")


@app.after_startup
async def run_demo() -> None:
    def request_stop(signum: int, _frame: object) -> None:
        sig_name = signal.Signals(signum).name
        print(f"\n{sig_name} received — requesting graceful stop...")
        asyncio.get_event_loop().call_soon_threadsafe(asyncio.ensure_future, app.stop())

    signal.signal(signal.SIGINT, request_stop)
    signal.signal(signal.SIGTERM, request_stop)

Running under the FastStream CLI (faststream run app:app) wires signal handling and graceful shutdown for you, honoring the same graceful_timeout. The manual signal handlers above are for scripts run directly with python.

Serialization Through the Pipeline

The parser and decoder are part of the same pipeline that runs the middleware chain, so your choice of serialization travels with every message — published and consumed alike. By default, messages are encoded and decoded as JSON, which transparently handles dicts, dataclasses, and Pydantic models.

To use a different format, supply a custom decoder (and, for outbound encoding, a matching parser) on the broker constructor. The decoder receives the raw FastStream message and returns the object handed to your handler.

No configuration needed — JSON is the default codec.

json_default.py

broker = KubeMQBroker("kubemq://localhost:50000")


@broker.subscriber(events="example.serial.json")
async def handle_message(msg: dict) -> None:
    print(f"[JSON] {msg}")

Compact binary encoding. Requires pip install msgpack.

msgpack_serializer.py

from typing import Any

import msgpack


async def msgpack_decoder(msg: Any) -> Any:
    raw = msg.body if hasattr(msg, "body") else msg
    if isinstance(raw, bytes):
        return msgpack.unpackb(raw, raw=False)
    return raw


broker = KubeMQBroker(
    "kubemq://localhost:50000",
    decoder=msgpack_decoder,
)

Strongly-typed contracts. Requires pip install protobuf.

protobuf_messages.py

from typing import Any

from google.protobuf import json_format, struct_pb2


def protobuf_decode(raw: bytes) -> dict:
    struct = struct_pb2.Struct()
    struct.ParseFromString(raw)
    return json_format.MessageToDict(struct)


async def protobuf_decoder(msg: Any) -> Any:
    raw = msg.body if hasattr(msg, "body") else msg
    if isinstance(raw, bytes):
        return protobuf_decode(raw)
    return raw


broker = KubeMQBroker(
    "kubemq://localhost:50000",
    decoder=protobuf_decoder,
)

Any application-specific format — here a simple key=value|key=value codec.

custom_serializer.py

from typing import Any

SEPARATOR = "|"


def custom_decode(raw: bytes) -> dict:
    text = raw.decode() if isinstance(raw, bytes) else str(raw)
    result = {}
    for pair in text.split(SEPARATOR):
        if "=" in pair:
            k, v = pair.split("=", 1)
            result[k] = v
    return result


async def custom_decoder(msg: Any) -> Any:
    raw = msg.body if hasattr(msg, "body") else msg
    if isinstance(raw, bytes):
        return custom_decode(raw)
    return raw


broker = KubeMQBroker(
    "kubemq://localhost:50000",
    decoder=custom_decoder,
)

Because the decoder runs inside the pipeline, it composes with middleware: a metrics or tracing middleware sees every message regardless of which codec decoded it, and TestKubeMQBroker exercises the same decoder you ship.

Running the Examples

All of the snippets above run against a local KubeMQ broker. Start one in Docker — kubemq-faststream needs only the gRPC port:

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

There is no connector to enable and no HTTP flag to set: FastStream connects directly over gRPC on 50000. Install the relevant optional backend (prometheus-client, opentelemetry-sdk, msgpack, or protobuf) before running an example that requires it.

Testing

Run the full parser, decoder, and middleware pipeline in-memory with TestKubeMQBroker — no live broker required.

Configuration & Security

Broker constructor options and environment variables: graceful_timeout, keepalive, TLS, message size limits.

Observability & Middleware