Concepts

kubemq-faststream is a real FastStream broker, not a thin transport shim. Understanding three things — the broker model, the pattern-to-client mapping, and the message pipeline — explains everything the API does. This page covers the model; the capability pages document the per-pattern API and the reference lists every option.

The broker model

KubeMQBroker extends FastStream's BrokerUsecase, so it behaves like the KafkaBroker or RedisBroker you may already know: you construct it with a URL, register handlers with @broker.subscriber(...), publish with broker.publish(...), and call request-reply RPC with broker.request(...). What makes it KubeMQ-specific is the pattern keyword on the subscriber and publish calls — it selects which of KubeMQ's five messaging patterns the message uses.

from kubemq_faststream import KubeMQBroker

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

A single set of connection settings (URL, client ID, auth token, TLS, message-size limits, keepalive, timeouts) is configured once on the constructor and shared across everything the broker does. See Configuration & Security for the full surface.

Three clients, one broker

On connect, KubeMQBroker creates three KubeMQ SDK async clients, each on its own gRPC channel, all sharing the one connection configuration. Each client serves a family of patterns:

Each publish, request, and subscriber registration is routed to the matching client by its pattern keyword.

You never address these clients directly. The keyword you use selects the client for you:

The KubeMQPattern enum names these five patterns; you rarely reference it directly because the keyword sets it for you. See the API reference for the enum and the full subscriber configuration.

Connection lifecycle

The clients are created when the broker connects and closed when it stops. There are two ways to drive that lifecycle:

• A FastStream app — app = FastStream(broker) ties the broker to the app's lifecycle. The clients connect on app.start() (or the faststream run CLI) and close on app.stop(). Attach work to @app.on_startup, @app.after_startup, and @app.on_shutdown.
• An async context manager — async with KubeMQBroker(...) as broker: connects on entry and closes on exit, which is convenient for scripts and one-off publishers.

import asyncio
from kubemq_faststream import KubeMQBroker


async def main() -> None:
    async with KubeMQBroker("kubemq://localhost:50000") as broker:
        await broker.publish({"hello": "world"}, events="demo")


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

On shutdown, the broker stops accepting new work and waits up to graceful_timeout (default 15.0 seconds) for in-flight handlers to finish before closing the channels.

The message pipeline

Every message — published or consumed, across all five patterns — flows through the full FastStream pipeline: the parser, the decoder, and the complete middleware chain, with FastDepends dependency injection into each handler. The KubeMQ adapter runs this chain unchanged, which has two practical consequences:

• Standard FastStream middleware (Prometheus, OpenTelemetry, your own BaseMiddleware) composes directly with KubeMQ messages — see Observability & Middleware.
• TestKubeMQBroker runs the same pipeline in-memory, so unit tests exercise the real parse/decode/middleware path with no live broker.

By default messages are encoded and decoded as JSON, which transparently handles dicts, dataclasses, and Pydantic models. Supply a custom decoder/parser on the constructor to use another format.

Next steps