FastStream

kubemq-faststream is a KubeMQ broker adapter for the FastStream async messaging framework. It registers KubeMQBroker as a first-class FastStream broker, so all five KubeMQ patterns — Events, Events Store, Queues, Commands, and Queries — become ordinary @broker.subscriber(...) endpoints, fully wired into FastStream's lifecycle, dependency injection, middleware, and testing infrastructure.

New to the idea? See what is an integration for how SDK-level integrations differ from server-side connectors.

Why FastStream + KubeMQ

• All five patterns as FastStream endpoints — one keyword on the subscriber (events=, events_store=, queues=, commands=, queries=) selects the pattern; no per-pattern client wiring.
• Idiomatic decorator API — register handlers with @broker.subscriber(...) and auto-publish results with @broker.publisher(...), exactly like every other FastStream broker.
• Full FastStream pipeline — parser, decoder, broker and subscriber middleware, and FastDepends dependency injection run for every message.
• In-memory testing — TestKubeMQBroker routes published messages to matching subscribers without a live broker, exercising the real parse/decode/middleware path.
• Runs inside your web app — drop the broker into a FastAPI, Starlette, Django, or Flask process and consume KubeMQ messages on the same event loop.

Installation

uv add kubemq-faststream

pip install kubemq-faststream

Requirements: Python 3.11+ and a running KubeMQ broker.

Start a broker locally with Docker:

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

Architecture

When it connects, KubeMQBroker creates three KubeMQ SDK async clients — one per transport family — each on its own gRPC channel and all sharing a single set of connection settings (URL, client ID, auth token, TLS, message-size limits, keepalive). Your subscriber keyword routes each publish, request, and handler registration to the matching client.

KubeMQBroker fans your handlers across three SDK clients, all reaching the KubeMQ broker over gRPC on :50000.

A minimal app wires the broker into FastStream, registers a subscriber, and publishes after startup:

import asyncio
from faststream import FastStream
from kubemq_faststream import KubeMQBroker

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


@broker.subscriber(queues="orders")
async def handle_order(order: dict) -> None:
    print(f"Processing order: {order}")


@app.after_startup
async def publish() -> None:
    await broker.publish({"id": "ORD-001", "item": "Widget"}, queues="orders")


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

Messaging patterns

Each KubeMQ pattern maps to a subscriber keyword. The pages below document the FastStream API for each — the decorators, options, and broker.publish/broker.request calls — and link to the underlying KubeMQ concept.

Capabilities

• KubeMQBroker over native gRPC — connect with kubemq://host:50000 (or kubemq+tls:// for TLS) straight to the KubeMQ gRPC port.
• KubeMQRouter composition — group handlers into routers whose prefix propagates to every channel.
• @broker.publisher auto-publish — stack on a subscriber to publish its return value to another channel.
• AckPolicy settlement — choose how queue messages are acked, nacked, or rejected.
• StartPosition replay — replay Events Store streams from first, a sequence, or a point in time.
• Server-side query caching — pass cache_key and cache_ttl on a query request.
• Health check — await broker.ping() verifies broker connectivity.

Next steps

New to KubeMQ? Start with the KubeMQ Getting Started guide for the core concepts behind these patterns.