Getting Started with FastStream

Prerequisites

You need the following installed before you begin:

kubemq-faststream is the KubeMQ broker adapter for the FastStream async messaging framework. It talks to KubeMQ over the native gRPC port 50000 — the same transport the native SDKs use. There is no connector to enable and no HTTP flag to set.

Start a KubeMQ Broker

Run KubeMQ in 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

Port 50000 is the gRPC port that kubemq-faststream connects to. Port 9090 is the shared HTTP server (REST and connector endpoints) — not required for FastStream, but harmless to expose.

Install the Package

Add kubemq-faststream to your project:

uv add kubemq-faststream

pip install kubemq-faststream

This pulls in FastStream and the KubeMQ Python client as dependencies.

Write Your First App

Create app.py. This is the minimal end-to-end application: a broker connected over gRPC, a FastStream app, one subscriber on the orders queue, and a publisher that fires once the broker is connected.

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())

A few things to note:

• KubeMQBroker("kubemq://localhost:50000") opens the gRPC connection. The kubemq:// scheme is a plain (non-TLS) connection — use kubemq+tls:// for TLS.
• @broker.subscriber(queues="orders") registers handle_order as a consumer on the orders queue. The keyword (queues=) selects the messaging pattern.
• @app.after_startup runs once the broker is connected, making it the right place to publish a demo message.
• broker.publish(..., queues="orders") sends a message to the same queue, which the subscriber then receives and prints.

Run It

Start the app:

python app.py

The app connects to the broker, the subscriber registers on the orders queue, and the after_startup hook publishes one message. The subscriber receives it and prints:

Processing order: {'id': 'ORD-001', 'item': 'Widget'}

Press Ctrl+C to stop. You have now published and consumed a message end-to-end through KubeMQ.

Run via the FastStream CLI

For production-style running, FastStream ships a CLI that manages the process lifecycle, signal handling, and graceful shutdown for you. Point it at the module path and the app object (module:app):

faststream run app:app

The CLI imports app from app.py and runs it — no asyncio.run(app.run()) wiring of your own. The if __name__ == "__main__" block in the example above is the fallback for running the file directly with python; the CLI bypasses it and drives the app object instead.

Try All Five Patterns

KubeMQ exposes five messaging patterns, and kubemq-faststream maps each to a @broker.subscriber keyword:

The examples/quickstart/five_patterns.py script wires up all five side by side so you can see them in one app:

@broker.subscriber(events="example.quickstart.events")
async def events_handler(msg: dict) -> None:
    print(f"[Events] Received: {msg}")

@broker.subscriber(events_store="example.quickstart.events_store")
async def events_store_handler(msg: dict) -> None:
    print(f"[Events Store] Received: {msg}")

@broker.subscriber(queues="example.quickstart.queues")
async def queues_handler(msg: dict) -> None:
    print(f"[Queues] Received: {msg}")

@broker.subscriber(commands="example.quickstart.commands")
async def commands_handler(msg: dict) -> None:
    print(f"[Commands] Handling command: {msg}")

@broker.subscriber(queries="example.quickstart.queries")
async def queries_handler(msg: dict) -> dict:
    print(f"[Queries] Handling query: {msg}")
    return {"status": "ok", "uptime": 42}

Events, Events Store, and Queues are published with broker.publish(...). Commands and Queries are request-reply, so they use broker.request(...) and block until the handler responds:

await broker.publish({"pattern": "events"}, events="example.quickstart.events")
await broker.publish({"pattern": "events_store"}, events_store="example.quickstart.events_store")
await broker.publish({"pattern": "queues"}, queues="example.quickstart.queues")

await broker.request({"action": "restart"}, commands="example.quickstart.commands", timeout=10)
result = await broker.request({"question": "status?"}, queries="example.quickstart.queries", timeout=10)
print(f"[Queries] Response: {result}")

The companion examples/quickstart/full_app.py combines all five patterns into a single cohesive application. Run either one against your broker:

python examples/quickstart/five_patterns.py

Next Steps