# API Reference (/integrations/faststream/reference/api)



This page is the API reference for `kubemq-faststream`: every exported symbol, the
messaging-pattern and acknowledgement enums, the Events Store start positions, the
subscriber configuration fields, and the `broker.request()` request-reply signature. For
connection options, URL formats, and environment variables, see the
[configuration reference](/integrations/faststream/reference/configuration).

## Public API Surface [#public-api-surface]

Everything in the package's `__all__` is importable directly from `kubemq_faststream`. The
classes are loaded lazily — the package only checks that the `kubemq` SDK is installed when
one of these symbols is first accessed.

```python
from kubemq_faststream import (
    KubeMQBroker,
    KubeMQRouter,
    KubeMQPublisher,
    KubeMQMessage,
    KubeMQPublishCommand,
    TestKubeMQBroker,
    AckPolicy,
    StartPosition,
    KubeMQPattern,
    FeatureNotSupportedException,
    __version__,
)
```

| Symbol                         | Kind      | Purpose                                                                                            |
| ------------------------------ | --------- | -------------------------------------------------------------------------------------------------- |
| `KubeMQBroker`                 | class     | Main broker adapter — lifecycle, `publish`, `request`, and `subscriber`/`publisher` decorators     |
| `KubeMQRouter`                 | class     | Modular handler group with `prefix` propagation, included on a broker via `include_router`         |
| `KubeMQPublisher`              | class     | Publisher object created by `@broker.publisher(...)` for auto-publishing handler return values     |
| `KubeMQMessage`                | class     | Decoded message wrapper passed to handlers (headers, body, ack/nack)                               |
| `KubeMQPublishCommand`         | class     | Internal publish command describing destination, pattern, and per-message options                  |
| `TestKubeMQBroker`             | class     | In-memory test broker that routes published messages to matching subscribers without a live broker |
| `AckPolicy`                    | enum      | Message acknowledgement strategy for queue subscribers (re-exported from FastStream)               |
| `StartPosition`                | enum      | Events Store replay start position                                                                 |
| `KubeMQPattern`                | enum      | The five KubeMQ messaging patterns                                                                 |
| `FeatureNotSupportedException` | exception | Raised when a feature is unsupported for the selected pattern                                      |
| `__version__`                  | str       | Installed package version                                                                          |

## KubeMQPattern [#kubemqpattern]

The `KubeMQPattern` enum names the five KubeMQ messaging patterns. You rarely reference it
directly — the `@broker.subscriber(...)` keyword and the
`broker.publish(...)`/`broker.request(...)` keyword select the pattern for you — but it
appears in subscriber configuration and `KubeMQPublishCommand`.

| Value                        | String           | Subscriber keyword |
| ---------------------------- | ---------------- | ------------------ |
| `KubeMQPattern.EVENTS`       | `"events"`       | `events=`          |
| `KubeMQPattern.EVENTS_STORE` | `"events_store"` | `events_store=`    |
| `KubeMQPattern.QUEUES`       | `"queues"`       | `queues=`          |
| `KubeMQPattern.COMMANDS`     | `"commands"`     | `commands=`        |
| `KubeMQPattern.QUERIES`      | `"queries"`      | `queries=`         |

`KubeMQPattern` is a `StrEnum`, so each member compares equal to its string value.

## AckPolicy [#ackpolicy]

`AckPolicy` is re-exported from `faststream.middlewares` and controls how queue messages
are settled. It applies to the **Queues** pattern; for Events and Events Store
(fire-and-forget) the ack/nack operations are no-ops.

| AckPolicy                   | Behavior                                                             |
| --------------------------- | -------------------------------------------------------------------- |
| `AckPolicy.ACK`             | Ack on success, nack (requeue) on handler error. **Default.**   |
| `AckPolicy.NACK_ON_ERROR`   | Nack (requeue) on handler error.                                     |
| `AckPolicy.REJECT_ON_ERROR` | Reject on handler error (server-configured disposition).             |
| `AckPolicy.ACK_FIRST`       | Ack immediately before the handler runs (at-most-once).              |
| `AckPolicy.MANUAL`          | No automatic ack — the handler must call `msg.ack()` / `msg.nack()`. |

```python
from kubemq_faststream import KubeMQBroker, AckPolicy

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


@broker.subscriber(queues="tasks", ack_policy=AckPolicy.ACK)
async def process_task(task: dict) -> None:
    ...  # auto-acked on success, nacked on exception
```

See [Queues](/integrations/faststream/how-to/queues) for the full settlement walkthrough.

## StartPosition [#startposition]

`StartPosition` selects where an Events Store subscriber begins reading. Three of the six
positions require a companion `start_value` (passed to
`@broker.subscriber(events_store=..., start_position=..., start_value=...)`). `start_value`
is typed `int | float | None` and must be a positive number for the positions that require
it.

| StartPosition         | String                  | Requires `start_value` | Description                                                |
| --------------------- | ----------------------- | ---------------------- | ---------------------------------------------------------- |
| `START_FROM_NEW`      | `"start_from_new"`      | No                     | Only messages published after subscribe. **Default.** |
| `START_FROM_FIRST`    | `"start_from_first"`    | No                     | Replay from the first stored message.                      |
| `START_FROM_LAST`     | `"start_from_last"`     | No                     | Start from the last stored message.                        |
| `START_AT_SEQUENCE`   | `"start_at_sequence"`   | Yes                    | Start at a specific sequence number.                       |
| `START_AT_TIME`       | `"start_at_time"`       | Yes                    | Start at a Unix timestamp (seconds).                       |
| `START_AT_TIME_DELTA` | `"start_at_time_delta"` | Yes                    | Start from N seconds ago.                                  |

```python
from kubemq_faststream import KubeMQBroker, StartPosition

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


@broker.subscriber(
    events_store="audit-log",
    start_position=StartPosition.START_FROM_FIRST,
)
async def on_audit(msg: dict) -> None:
    print(f"Audit event: {msg}")
```

See [Events Store](/integrations/faststream/how-to/events-store) for replay walkthroughs.

## Subscriber Configuration [#subscriber-configuration]

The `@broker.subscriber(...)` decorator accepts a pattern keyword (`events=`,
`events_store=`, `queues=`, `commands=`, or `queries=`) plus the fields below. The pattern
keyword's value is the channel name. The configuration is backed by
`KubeMQSubscriberConfig`.

| Field          | Type            | Default  | Applies to     | Description                                                              |
| -------------- | --------------- | -------- | -------------- | ------------------------------------------------------------------------ |
| `channel`      | `str`           | —        | All            | Channel name (the value of the pattern keyword).                         |
| `pattern`      | `KubeMQPattern` | `EVENTS` | All            | Resolved messaging pattern (set by the keyword used).                    |
| `group`        | `str \| None`   | `None`   | Events, Queues | Load-balancing group — only one member of a group receives each message. |
| `max_messages` | `int`           | `1`      | Queues         | Maximum messages pulled per receive batch.                               |
| `wait_timeout` | `int`           | `60`     | Queues         | Seconds to wait for queue messages per poll.                             |
| `ack_policy`   | `AckPolicy`     | `ACK`    | Queues         | Acknowledgement strategy (see [AckPolicy](#ackpolicy)).                  |

For Events Store subscribers, also pass `start_position` and (where required) `start_value`
as described in [StartPosition](#startposition).

## broker.request() — RPC Reference [#brokerrequest--rpc-reference]

`broker.request(...)` performs request-reply RPC. It is valid only for the **Commands** and
**Queries** patterns — calling it without `commands=` or `queries=` raises
`FeatureNotSupportedException`. The call blocks until the handler responds or the timeout
elapses, and returns the handler's response.

```python
async def request(
    message,
    /,
    *,
    commands: str | None = None,
    queries: str | None = None,
    timeout: int | None = None,
    cache_key: str | None = None,
    cache_ttl: int | None = None,
    headers: dict[str, str] | None = None,
    metadata: str = "",
    correlation_id: str | None = None,
) -> Any: ...
```

| Argument               | Type                     | Applies to        | Description                                                                                                         |
| ---------------------- | ------------------------ | ----------------- | ------------------------------------------------------------------------------------------------------------------- |
| `commands` / `queries` | `str \| None`            | —                 | Target channel. Exactly one selects the pattern.                                                                    |
| `timeout`              | `int \| None`            | Commands, Queries | Seconds to wait for the response. Falls back to `default_cq_timeout` (30) when unset.                               |
| `cache_key`            | `str \| None`            | Queries           | Server-side cache key. A second request with the same key returns the cached response without invoking the handler. |
| `cache_ttl`            | `int \| None`            | Queries           | Cache lifetime in seconds.                                                                                          |
| `headers`              | `dict[str, str] \| None` | Commands, Queries | Message tags.                                                                                                       |
| `metadata`             | `str`                    | Commands, Queries | Free-form metadata string.                                                                                          |
| `correlation_id`       | `str \| None`            | Commands, Queries | Correlation identifier.                                                                                             |

```python title="basic_command.py"
@broker.subscriber(commands="example.commands.basic")
async def handle_command(msg: dict) -> dict:
    return {"status": "executed", "action": msg["action"], "success": True}

# caller
response = await broker.request(
    {"action": "create_user", "name": "Alice"},
    commands="example.commands.basic",
)
```

<Callout type="info">
  For fire-and-forget commands or queries where no reply is expected, subscribe with
  `no_reply=True` and send with `broker.publish(commands=...)` /
  `broker.publish(queries=...)` instead of `broker.request(...)`. The handler runs, but no
  response is returned to the caller.
</Callout>

See [Commands & Queries](/integrations/faststream/how-to/commands-queries) for the full
walkthrough, including caching and timeout handling.

## Internal Architecture [#internal-architecture]

On connect, `KubeMQBroker` creates **three** KubeMQ SDK async clients, each on its own
gRPC channel, sharing one set of connection settings (URL, client ID, auth token, TLS, size
limits, keepalive):

<Mermaid
  chart="`
graph LR
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;]

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

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

*Each `publish`, `request`, and `@broker.subscriber` call is routed to the matching client by its pattern keyword.*

The clients are created when the broker connects (via `start()` or `async with broker:`)
and closed on `stop()`. See [Concepts](/integrations/faststream/concepts/concepts) for the
lifecycle and the message pipeline.

## Project Links [#project-links]

<Cards>
  <Card title="PyPI — kubemq-faststream" href="https://pypi.org/project/kubemq-faststream/" description="Install the published package." />

  <Card title="GitHub — kubemq-io/kubemq-faststream" href="https://github.com/kubemq-io/kubemq-faststream" description="Source, examples, and issue tracker." />

  <Card title="FastStream Documentation" href="https://faststream.airt.ai/" description="The async messaging framework this adapter plugs into." />

  <Card title="Configuration Reference" href="/integrations/faststream/reference/configuration" description="Constructor options, URL formats, validation rules, and environment variables." />
</Cards>