# FastStream (/integrations/faststream)



[`kubemq-faststream`](https://github.com/kubemq-io/kubemq-faststream) is a KubeMQ
broker adapter for the [FastStream](https://github.com/airtai/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](/integrations#what-an-integration-is)
for how SDK-level integrations differ from server-side [connectors](/connectors).

## Why FastStream + KubeMQ [#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 [#installation]

<Tabs groupId="py-installer" items="['uv', 'pip']">
  <Tab value="uv">
    ```bash
    uv add kubemq-faststream
    ```
  </Tab>

  <Tab value="pip">
    ```bash
    pip install kubemq-faststream
    ```
  </Tab>
</Tabs>

**Requirements**: Python 3.11+ and a running [KubeMQ](https://kubemq.io/) broker.

| Requirement | Supported versions        |
| ----------- | ------------------------- |
| Language    | Python 3.11, 3.12, 3.13   |
| FastStream  | >= 0.6.7, \< 0.7.0        |
| KubeMQ SDK  | >= 4.1.5, \< 5            |
| Package     | `kubemq-faststream` 0.1.0 |

Start a broker locally with Docker:

<RunKubeMQ ports="[50000, 9090]" />

<Callout type="info">
  `kubemq-faststream` is a native gRPC SDK client: it talks to KubeMQ over the gRPC port
  `50000`, the same transport the native SDKs use. It is **always on** — there is no
  server-side connector to enable and no HTTP flag to set. Port `9090` is the shared HTTP
  server (REST and the HTTP connectors) and is not used by FastStream.
</Callout>

## Architecture [#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.

<Mermaid
  chart="`
graph LR
APP[&#x22;FastStream app<br/>@broker.subscriber / publish / request&#x22;]
BROKER[&#x22;KubeMQBroker&#x22;]
PUBSUB[&#x22;AsyncPubSubClient<br/>Events · Events Store&#x22;]
QUEUES[&#x22;AsyncQueuesClient<br/>Queues&#x22;]
CQ[&#x22;AsyncCQClient<br/>Commands · Queries&#x22;]
KMQ[&#x22;KubeMQ broker<br/>gRPC :50000&#x22;]

APP --> BROKER
BROKER --> PUBSUB
BROKER --> QUEUES
BROKER --> CQ
PUBSUB --> KMQ
QUEUES --> KMQ
CQ --> KMQ

class APP client
class BROKER connector
class PUBSUB events
class QUEUES queue
class CQ command
class KMQ broker
`"
/>

*`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:

```python title="app.py"
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 [#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.

<Cards>
  <Card title="Events" href="/integrations/faststream/how-to/events" description="Fire-and-forget pub/sub with the events= subscriber and broker.publish, plus optional group load balancing." />

  <Card title="Events Store" href="/integrations/faststream/how-to/events-store" description="Persistent pub/sub with replay from first, a sequence, or a point in time via StartPosition." />

  <Card title="Queues" href="/integrations/faststream/how-to/queues" description="Point-to-point messaging with AckPolicy-controlled settlement, DLQ, TTL, delay, and batching." />

  <Card title="Commands & Queries" href="/integrations/faststream/how-to/commands-queries" description="Request-reply RPC with broker.request: void-response commands and cacheable data-response queries." />

  <Card title="Composition" href="/integrations/faststream/how-to/composition" description="KubeMQRouter prefix composition and @broker.publisher auto-publish decorators." />
</Cards>

## Capabilities [#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 [#next-steps]

<Cards>
  <Card title="Getting Started" href="/integrations/faststream/tutorials/getting-started" description="Install, start a broker, and run your first subscriber and publisher end to end." />

  <Card title="Concepts" href="/integrations/faststream/concepts/concepts" description="The broker lifecycle, the three-client model, dependency injection, and how patterns map to clients." />

  <Card title="Guides" href="/integrations/faststream/how-to/configuration" description="Configuration and security, testing with TestKubeMQBroker, and observability middleware." />

  <Card title="Reference" href="/integrations/faststream/reference/configuration" description="Constructor options, environment variables, URL formats, enums, and the public API surface." />
</Cards>

New to KubeMQ? Start with the [KubeMQ Getting Started guide](/deploy) for
the core concepts behind these patterns.