# Routers & Publishers (/integrations/faststream/how-to/composition)



## Overview [#overview]

As a FastStream app grows, registering every handler directly on a single `KubeMQBroker` becomes unwieldy. `kubemq-faststream` ships two composition primitives that keep large apps organized:

* **`KubeMQRouter`** groups related subscribers and publishers into modular units. A router can carry a `prefix` that propagates to every channel registered on it, so handler code stays short and channel namespaces stay consistent.
* **`@broker.publisher(...)`** stacks on top of a subscriber to automatically publish the handler's return value to another channel — no manual `broker.publish(...)` call inside the handler.

Both are standard FastStream constructs; the KubeMQ adapter implements them so they behave exactly like routers and publishers on any other FastStream broker.

## KubeMQRouter [#kubemqrouter]

`KubeMQRouter` is a container for handler registrations. You decorate subscribers (and publishers) on the router instead of the broker, then call `broker.include_router(router)` to fold those registrations into the broker at startup. This lets you split a large app into focused modules — one router per domain, per pattern, or per bounded context.

```python title="basic_router.py"
import asyncio
import logging

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, KubeMQRouter

logging.basicConfig(level=logging.INFO)

router = KubeMQRouter()


@router.subscriber(events="example.router.basic.orders")
async def handle_order(msg: dict) -> None:
    """Process incoming order events."""
    print(f"[Router] Order received: {msg}")


@router.subscriber(events="example.router.basic.payments")
async def handle_payment(msg: dict) -> None:
    """Process incoming payment events."""
    print(f"[Router] Payment received: {msg}")


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


@app.after_startup
async def run_demo() -> None:
    await broker.publish(
        {"order_id": "ORD-100", "amount": 49.99},
        events="example.router.basic.orders",
    )
    await broker.publish(
        {"payment_id": "PAY-200", "status": "completed"},
        events="example.router.basic.payments",
    )

    await asyncio.sleep(2)
    await app.stop()


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

You need a running KubeMQ broker for these examples. Start one locally with Docker — `kubemq-faststream` connects over the native gRPC port `50000`, and port `9090` is the shared HTTP server (REST and connector endpoints, not required here):

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

<Callout type="info">
  `kubemq-faststream` talks to KubeMQ over native gRPC on port `50000` — there is no HTTP connector to enable. The `9090` HTTP port is exposed above only for parity with other KubeMQ tooling; FastStream does not use it.
</Callout>

## Channel Namespacing with Prefixes [#channel-namespacing-with-prefixes]

A router constructed with `KubeMQRouter(prefix="orders.")` prepends that prefix to every channel registered on it. Inside the handler you decorate with a short, local name; the router resolves it to the full channel at registration time. A subscriber declared with `@orders.subscriber(events="created")` on a router with `prefix="orders."` listens on the effective channel `orders.created`.

This keeps each module's channel names short and guarantees a consistent namespace per domain — the prefix lives in exactly one place.

```python title="router_prefix.py"
import asyncio
import logging
import os

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, KubeMQRouter

logging.basicConfig(level=logging.INFO)

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")

broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)

# Router with prefix -- all channels get "orders." prepended
orders_router = KubeMQRouter(prefix="orders.")


@orders_router.subscriber(queues="new")  # effective channel: orders.new
async def handle_order(msg: dict) -> None:
    print(f"[orders] Received on orders.new: {msg}")


# Second router with a different prefix
payments_router = KubeMQRouter(prefix="payments.")


@payments_router.subscriber(events="received")  # effective channel: payments.received
async def handle_payment(msg: dict) -> None:
    print(f"[payments] Received on payments.received: {msg}")


broker.include_router(orders_router)
broker.include_router(payments_router)


@app.after_startup
async def run_demo() -> None:
    await broker.publish({"order_id": 1}, queues="orders.new")
    await broker.publish({"payment_id": "p-001"}, events="payments.received")
    await asyncio.sleep(2)
    await app.stop()


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

The prefix applies regardless of pattern: the `orders.` router above namespaces a `queues=` subscriber, and the `payments.` router namespaces an `events=` subscriber. When you publish, you use the full resolved channel name (`orders.new`, `payments.received`) — the prefix only rewrites the subscriber registration, not your publish calls.

## Nested Routers and Multi-File Apps [#nested-routers-and-multi-file-apps]

Routers compose. A router can include another router, and prefixes stack: a child subscriber on `"placed"` under a parent prefix `"example.nested.orders."` resolves to `example.nested.orders.placed`. This lets you build deep, modular hierarchies where each level adds a namespace segment.

```python title="nested_routers.py"
import asyncio
import logging

from faststream import FastStream

from kubemq_faststream import KubeMQBroker, KubeMQRouter

logging.basicConfig(level=logging.INFO)

child_router = KubeMQRouter(prefix="example.nested.orders.")


@child_router.subscriber(events="placed")
async def on_order_placed(msg: dict) -> None:
    """Handle order-placed events (channel: example.nested.orders.placed)."""
    print(f"[Child] Order placed: {msg}")


@child_router.subscriber(events="shipped")
async def on_order_shipped(msg: dict) -> None:
    """Handle order-shipped events (channel: example.nested.orders.shipped)."""
    print(f"[Child] Order shipped: {msg}")


parent_router = KubeMQRouter()
parent_router.include_router(child_router)

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


@app.after_startup
async def run_demo() -> None:
    await broker.publish(
        {"order_id": "N-001", "items": 3},
        events="example.nested.orders.placed",
    )
    await broker.publish(
        {"order_id": "N-001", "carrier": "DHL"},
        events="example.nested.orders.shipped",
    )
    await asyncio.sleep(2)
    await app.stop()


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

The most common production layout puts each router in its own module and composes them from a single entry point. Define a router per concern in dedicated files:

```python title="events_router.py"
from kubemq_faststream import KubeMQRouter

events_router = KubeMQRouter(prefix="example.multifile.events.")


@events_router.subscriber(events="notifications")
async def on_notification(msg: dict) -> None:
    """Handle notification events."""
    print(f"[Events] Notification: {msg}")


@events_router.subscriber(events="alerts")
async def on_alert(msg: dict) -> None:
    """Handle alert events."""
    print(f"[Events] Alert: {msg}")
```

```python title="queues_router.py"
from kubemq_faststream import KubeMQRouter

queues_router = KubeMQRouter(prefix="example.multifile.queues.")


@queues_router.subscriber(queues="tasks")
async def process_task(msg: dict) -> None:
    """Process queued tasks with auto-ack."""
    print(f"[Queues] Task processed: {msg}")


@queues_router.subscriber(queues="emails")
async def send_email(msg: dict) -> None:
    """Process email send requests."""
    print(f"[Queues] Email sent to: {msg.get('to', 'unknown')}")
```

Then `main.py` imports each router module and includes them on the broker:

```python title="main.py"
import asyncio
import logging

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

from events_router import events_router
from queues_router import queues_router

logging.basicConfig(level=logging.INFO)

broker = KubeMQBroker("kubemq://localhost:50000")
broker.include_router(events_router)
broker.include_router(queues_router)
app = FastStream(broker)


@app.after_startup
async def run_demo() -> None:
    await broker.publish(
        {"type": "info", "text": "System started"},
        events="example.multifile.events.notifications",
    )
    await broker.publish(
        {"task": "generate-report", "params": {"format": "pdf"}},
        queues="example.multifile.queues.tasks",
    )
    await asyncio.sleep(3)
    await app.stop()


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

<Callout type="info">
  Run the multi-file app through its entry point — `python examples/router/multi_file_app/main.py` — so the router modules are imported and included before the broker connects.
</Callout>

## Publisher Decorator [#publisher-decorator]

Stacking `@broker.publisher(...)` on top of a `@broker.subscriber(...)` turns a handler into a transform stage: whatever the handler returns is automatically published to the publisher's channel. There is no `broker.publish(...)` call inside the handler — the return value is the message.

The decorator order matters. The subscriber must be the innermost decorator (closest to the function); the publisher wraps it. In the example below, `transform` consumes from the input channel and its return value is auto-published to `OUTPUT_CHANNEL`, where a second subscriber picks it up.

```python title="basic_publisher.py"
import asyncio
import logging

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

logging.basicConfig(level=logging.INFO)

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

INPUT_CHANNEL = "example.publisher.basic.input"
OUTPUT_CHANNEL = "example.publisher.basic.output"


@broker.subscriber(events=OUTPUT_CHANNEL)
async def receive_processed(msg: dict) -> None:
    """Consume the auto-published output."""
    print(f"[Output] Processed message received: {msg}")


@broker.publisher(events=OUTPUT_CHANNEL)
@broker.subscriber(events=INPUT_CHANNEL)
async def transform(msg: dict) -> dict:
    """Transform input and auto-publish the return value to the output channel."""
    print(f"[Transform] Input: {msg}")
    return {"original": msg, "processed": True, "uppercase_name": msg.get("name", "").upper()}


@app.after_startup
async def run_demo() -> None:
    await broker.publish(
        {"name": "widget", "quantity": 5},
        events=INPUT_CHANNEL,
    )
    await asyncio.sleep(2)
    await app.stop()


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

## Publisher Headers and Chaining [#publisher-headers-and-chaining]

Create a publisher object up front with `broker.publisher(...)` and assign its `headers` attribute to attach a default headers dict to every message it auto-publishes. This is the place to stamp metadata — source, version, content-type — onto all outgoing messages from that stage. Apply the publisher to a handler by using it as a decorator (`@output_publisher`).

```python title="publisher_headers.py"
import asyncio
import logging
import os

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

logging.basicConfig(level=logging.INFO)

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")

broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)

INPUT_CHANNEL = "example.publisher.headers.input"
OUTPUT_CHANNEL = "example.publisher.headers.output"

# Create the publisher and set default headers
output_publisher = broker.publisher(events=OUTPUT_CHANNEL)
output_publisher.headers = {
    "x-source": "header-demo",
    "x-version": "1.0",
    "x-content-type": "application/json",
}


@broker.subscriber(events=OUTPUT_CHANNEL)
async def receive_output(msg: dict) -> None:
    """Consume messages that were auto-published with headers."""
    print(f"[output] Received message: {msg}")


@output_publisher
@broker.subscriber(events=INPUT_CHANNEL)
async def transform(msg: dict) -> dict:
    """Transform input and auto-publish to the output channel with headers."""
    print(f"[transform] Processing: {msg}")
    return {**msg, "processed": True}
```

To fan a single handler's output out to several destinations, stack multiple `@broker.publisher(...)` decorators. The return value is published once to each publisher's channel, so two downstream consumers — analytics and archive, say — each receive a copy.

```python title="publisher_chain.py"
ANALYTICS_CHANNEL = "example.publisher.chain.analytics"
ARCHIVE_CHANNEL = "example.publisher.chain.archive"


@broker.publisher(events=ANALYTICS_CHANNEL)
@broker.publisher(events=ARCHIVE_CHANNEL)
@broker.subscriber(events=INPUT_CHANNEL)
async def transform(msg: dict) -> dict:
    """Process input and auto-publish to both analytics and archive channels."""
    print(f"[transform] Processing: {msg}")
    return {**msg, "processed": True}
```

## Cross-Pattern and Dynamic Publishing [#cross-pattern-and-dynamic-publishing]

A single app can publish to any KubeMQ pattern — the keyword on `broker.publish(...)` selects the target. Use `events=` for fire-and-forget broadcast and `queues=` for reliable point-to-point delivery, side by side, with the same payload.

```python title="publish_to_pattern.py"
EVENT_CHANNEL = "example.publisher.pattern.notifications"
QUEUE_CHANNEL = "example.publisher.pattern.tasks"


@app.after_startup
async def run_demo() -> None:
    payload = {"action": "user_signup", "user_id": "U-42"}

    await broker.publish(payload, events=EVENT_CHANNEL)
    print("Published to events pattern (fire-and-forget)")

    await broker.publish(payload, queues=QUEUE_CHANNEL)
    print("Published to queues pattern (reliable delivery)")
```

Channel names do not have to be static. Compute the target channel at runtime from message content — for example, route an order to a region-specific channel based on a field in the payload.

```python title="dynamic_routing.py"
@app.after_startup
async def run_demo() -> None:
    orders = [
        {"order_id": "ORD-1", "region": "us", "total": 29.99},
        {"order_id": "ORD-2", "region": "eu", "total": 45.00},
        {"order_id": "ORD-3", "region": "us", "total": 12.50},
    ]

    for order in orders:
        channel = f"example.publisher.dynamic.{order['region']}"
        await broker.publish(order, events=channel)
        print(f"Routed order {order['order_id']} to {channel}")
```

For full control over routing metadata, build a `KubeMQPublishCommand` directly. It is the wire-level alternative to `broker.publish(...)`, exposing pattern, headers, a metadata string, correlation ID, and a custom message ID. Publish it through the broker's producer.

```python title="publish_command.py"
from kubemq_faststream import KubeMQBroker, KubeMQPattern, KubeMQPublishCommand

CHANNEL = "example.publisher.command"


@app.after_startup
async def run_demo() -> None:
    cmd = KubeMQPublishCommand(
        {"action": "deploy", "version": "2.1.0"},
        destination=CHANNEL,
        pattern=KubeMQPattern.EVENTS,
        metadata="deployment-event",
        headers={"env": "staging", "region": "us-east-1"},
        correlation_id="deploy-corr-001",
        message_id="custom-msg-001",
    )

    if broker.config.producer is None:
        raise RuntimeError("Broker not connected")
    await broker.config.producer.publish(cmd)
    print("Published via KubeMQPublishCommand with custom metadata and headers")
```

<Callout type="info">
  `KubeMQPublishCommand` is a low-level escape hatch. For everyday publishing — including auto-publish via `@broker.publisher` — `broker.publish(...)` and the publisher decorators are the idiomatic API.
</Callout>

## AsyncAPI Metadata [#asyncapi-metadata]

FastStream generates an AsyncAPI schema from your subscribers and publishers. Both `@broker.subscriber(...)` and `@broker.publisher(...)` accept three documentation parameters:

| Parameter           | Purpose                                                        |
| ------------------- | -------------------------------------------------------------- |
| `title`             | Human-readable name for the channel in the generated schema    |
| `description`       | Longer explanation shown in the AsyncAPI docs                  |
| `include_in_schema` | Set `False` to hide an internal channel from the public schema |

The example below documents the public output channel and the input intake, while hiding an internal audit channel from the schema with `include_in_schema=False`.

```python title="asyncapi_schema.py"
INPUT_CHANNEL = "example.config.asyncapi.orders"
OUTPUT_CHANNEL = "example.config.asyncapi.processed"
AUDIT_CHANNEL = "example.config.asyncapi.audit"


# Subscriber with full AsyncAPI metadata -- visible in docs
@broker.subscriber(
    events=OUTPUT_CHANNEL,
    title="Processed Orders",
    description="Receives enriched/processed order data",
    include_in_schema=True,
)
async def processed_handler(msg: dict) -> None:
    """Consume processed orders (documented in AsyncAPI schema)."""
    print(f"[processed] Enriched order: {msg}")


# Publisher with AsyncAPI metadata, publishing to documented and internal channels
@broker.publisher(
    events=OUTPUT_CHANNEL,
    title="Order Processing Output",
    description="Publishes enriched orders after processing",
    include_in_schema=True,
)
@broker.publisher(
    events=AUDIT_CHANNEL,
    title="Audit Publisher",
    description="Internal audit log publisher",
    include_in_schema=False,
)
@broker.subscriber(
    events=INPUT_CHANNEL,
    title="Order Intake",
    description="Receives raw orders for processing",
    include_in_schema=True,
)
async def process_order(msg: dict) -> dict:
    """Process the order, auto-publish to output and audit channels."""
    print(f"[public] Received order: {msg}")
    return {**msg, "processed": True}
```

Setting `include_in_schema=False` on the audit publisher keeps the runtime behavior identical — the message is still published — but omits the channel from the generated documentation, which is the right default for internal-only channels.

## Related [#related]

<Cards>
  <Card title="Events (Pub/Sub)" href="/integrations/faststream/how-to/events" description="Fire-and-forget broadcast to all subscribers, with optional group load balancing." />

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

  <Card title="Queues" href="/integrations/faststream/how-to/queues" description="Point-to-point messaging with AckPolicy-controlled settlement and batch publishing." />

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

  <Card title="Reference" href="/integrations/faststream/reference/api" description="The KubeMQRouter, KubeMQPublisher, and KubeMQPublishCommand API, plus broker constructor options." />
</Cards>