# Getting Started with FastStream (/integrations/faststream/tutorials/getting-started)



## Build your first app [#build-your-first-app]

<Steps>
  <Step>
    ### Prerequisites [#prerequisites]

    You need the following installed before you begin:

    | Requirement | Version            |
    | ----------- | ------------------ |
    | Python      | 3.11+              |
    | Docker      | Any recent version |

    `kubemq-faststream` is the KubeMQ broker adapter for the [FastStream](https://github.com/airtai/faststream) async messaging framework. It talks to KubeMQ over the &#x2A;*native gRPC port `50000`** — the same transport the native SDKs use. There is no connector to enable and no HTTP flag to set.

    <Callout type="info">
      This is different from the HTTP-based connectors (CloudEvents, REST, MCP, A2A), which share the HTTP server on port `9090` and require an enable flag. FastStream connects directly over gRPC on `50000`, so the broker works out of the box with no extra configuration.
    </Callout>
  </Step>

  <Step>
    ### Start a KubeMQ Broker [#start-a-kubemq-broker]

    Run KubeMQ in Docker:

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

    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.
  </Step>

  <Step>
    ### Install the Package [#install-the-package]

    Add `kubemq-faststream` to your project:

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

    This pulls in FastStream and the KubeMQ Python client as dependencies.
  </Step>

  <Step>
    ### Write Your First App [#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.

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

    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.
  </Step>

  <Step>
    ### Run It [#run-it]

    Start the app:

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

    ```text
    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.
  </Step>

  <Step>
    ### Run via the FastStream CLI [#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`):

    ```bash
    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.

    <Callout type="info">
      The module path uses Python import syntax, not a file path. For a script nested in a package — for example `examples/lifecycle/standalone_cli.py` — you would run `faststream run examples.lifecycle.standalone_cli:app`.
    </Callout>
  </Step>

  <Step>
    ### Try All Five Patterns [#try-all-five-patterns]

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

    | Pattern      | Subscriber keyword | Style                          |
    | ------------ | ------------------ | ------------------------------ |
    | Events       | `events=`          | Fire-and-forget pub/sub        |
    | Events Store | `events_store=`    | Persistent pub/sub with replay |
    | Queues       | `queues=`          | Point-to-point with ack/nack   |
    | Commands     | `commands=`        | Request-reply, void response   |
    | Queries      | `queries=`         | Request-reply, data response   |

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

    ```python title="five_patterns.py"
    @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:

    ```python title="five_patterns.py"
    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:

    ```bash
    python examples/quickstart/five_patterns.py
    ```
  </Step>

  <Step>
    ### Next Steps [#next-steps]

    <Cards>
      <Card title="Concepts" href="/integrations/faststream/concepts/concepts" description="The broker model, the three-client mapping, and the FastStream message pipeline." />

      <Card title="Events" href="/integrations/faststream/how-to/events" description="Fire-and-forget pub/sub with the events= subscriber and broker.publish." />

      <Card title="Events Store" href="/integrations/faststream/how-to/events-store" description="Persistent events with replay and configurable start positions." />

      <Card title="Queues" href="/integrations/faststream/how-to/queues" description="Point-to-point messaging with ack policies and batch publish." />

      <Card title="Commands & Queries" href="/integrations/faststream/how-to/commands-queries" description="Request-reply RPC with void and data responses, plus query caching." />

      <Card title="Reference" href="/integrations/faststream/reference/configuration" description="Broker constructor options, environment variables, and URL formats." />
    </Cards>
  </Step>
</Steps>

<Callout type="info">
  **Override the broker address without touching code.** Every constructor option can be set with an environment variable, and env vars take precedence when present. Point any example at a different broker with `KUBEMQ_ADDRESS`:

  ```bash
  KUBEMQ_ADDRESS=kubemq://my-broker:50000 python examples/quickstart/hello_world.py
  ```
</Callout>