FastStream

https://github.com/ag2ai/faststream
Admin
FastStream is a Python framework for building event-driven microservices with support for multiple...
Tokens:357,440
Snippets:2,624
Trust Score:8.2
License:Apache-2.0
Update:1 week ago
Show doc for...
Tokens:
Raw
### Full Redis List Publishing Example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/redis/list/publishing.md

A complete example demonstrating the setup of a Redis list-based publish/subscribe pipeline.

```python
from pydantic import BaseModel
from faststream import FastStream
from faststream.redis import RedisBroker


class Data(BaseModel):
    data: str


broker = RedisBroker("redis://localhost:6379")
app = FastStream(broker)


@broker.subscriber(list="input-list")
@broker.publisher(list="output-list")
async def handle(data: Data) -> Data:
    return data
```

--------------------------------

### start()

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/lifespan/lifecycle.md

Runs the startup hooks and starts every broker. Returns as soon as they are consuming.

```APIDOC
## start()

### Description
Runs the startup hooks and starts every broker. Returns as soon as they are consuming.

### Method
start()

### Usage
Can be used with `async with app:` instead of calling start() and stop() manually.
```

--------------------------------

### serve_faststream.py with conditional Django setup for management command

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/integrations/django/index.md

Conditionally runs django.setup() only when started by FastStream CLI, allowing the same file to be used with Django's management command. The rest of the file defines the FastStream app and subscriber.

```Python
# serve_faststream.py
import sys

# From previous example, this run django.setup() only when started by FastStream CLI
# Remove this block entirely if you are not using FastStream CLI
if "bin/faststream" in sys.argv[0]:
    import os

    import django

    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "app.settings")
    django.setup()

from django.contrib.auth.models import User

from faststream import FastStream
from faststream.rabbit import RabbitBroker


broker = RabbitBroker("amqp://guest:guest@localhost:5672")

@broker.subscriber("demo")
async def faststream_django_orm_demo_handler(message: str):
    """
    This demonstrates how to access Django ORM from within a FastStream consumer.
    """
    qs = User.objects.all()
    async for user in qs:  # async django ORM is accessible
        print(user)
    print(message)

app = FastStream(broker)

```

--------------------------------

### RabbitBroker connect() and start()

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/rabbit/connection.md

Describes when to use `connect()` versus `start()`. `connect()` is for publishing or manual declarations, while `start()` is called by FastStream for full applications with subscribers.

```APIDOC
## RabbitBroker connect() and start()

### Description
Methods to establish a connection. `connect()` is used for publishing, RPC requests, or manual queue declarations. `start()` is called by FastStream on startup for full applications with subscribers.

### Method
async

### Endpoint
`await broker.connect()`
`await broker.start()`

### Parameters
None

### Request Example
```python
# For publishing only
await broker.connect()

# For full application (FastStream calls start() automatically)
# No need to call start() manually
```

### Response
#### Success Response
- **None** - The connection is established.

#### Response Example
```python
# Using async context manager
async with broker:
    # publish messages
    pass
```
```

--------------------------------

### Start local documentation server

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/contributing/docs.md

Starts the local development server for documentation. Use the --full flag for a complete build with all dependencies.

```bash
just docs-serve
```

```bash
just docs-serve --full
```

--------------------------------

### Install FastStream 0.7.0rc0 with MQTT support

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/release.md

Install the release candidate with MQTT support using pip or uv. The pip command installs the package, while the uv command adds it as a pre-release dependency.

```shell
pip install "faststream[mqtt]==0.7.0rc0"
```

```shell
uv add --pre "faststream[mqtt]==0.7.0rc0"
```

--------------------------------

### Topic Exchange Subscription Example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/rabbit/examples/topic.md

A complete example of setting up a Topic exchange and subscribing handlers.

```python
{! docs_src/rabbit/subscription/topic.py !}
```

--------------------------------

### Default Django asgi.py

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/integrations/django/index.md

The default Django ASGI application setup. This is the starting point before integrating FastStream.

```python
import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "app.settings")

application = get_asgi_application()
```

--------------------------------

### Start consumer with FastStream CLI

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/integrations/django/index.md

Command to start the FastStream consumer using the serve_faststream module.

```bash
faststream run serve_faststream:app

```

--------------------------------

### Full Example: Dynamic Config with Nats

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/howto/nats/dynaconf.md

Full example combining the KvWatch subscription and context retrieval, including startup and after-startup hooks to initialize the key-value store and publish a test message.

```python
from uuid import uuid4

from faststream import FastStream, Context
from faststream.nats import NatsMessage, NatsBroker, KvWatch

broker = NatsBroker()
app = FastStream(broker)

@broker.subscriber("create_sell", kv_watch=KvWatch("order_service"))
async def watch_kv_order_service(new_value: bool):
    app.context.set_global("create_sell", new_value)

@broker.subscriber("order_service.order.filled.buy")
async def handle_filled_buy(
    message: NatsMessage,
    create_sell: bool = Context("create_sell"),
):
    if create_sell:
        await broker.publish(b"", "order_service.order.create.sell")

@broker.subscriber("order_service.order.create.sell")
async def handle_create_sell(message: NatsMessage): ...

@app.on_startup
async def on_startup():
    await broker.connect()

    order_service_kv = await broker.key_value("order_service")

    initial_create_sell_value = b"1"
    await order_service_kv.put("create_sell", initial_create_sell_value)
    app.context.set_global("create_sell", bool(initial_create_sell_value))

@app.after_startup
async def after_startup():
    await broker.publish({"order_id": str(uuid4())}, "order_service.order.filled.buy")
```

--------------------------------

### Install development requirements

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/template/index.md

Install all project dependencies required for development.

```bash
pip install --group dev -e .
```

--------------------------------

### Install FastStream with MQTT support

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/index.md

Install the FastStream package with MQTT support using pip. After installation, a test broker container is required.

```console
pip install "faststream[mqtt]"
```

```bash
docker run -d --rm -p 1883:1883 -e ANONYMOUS_LOGIN=true --name test-mq apache/activemq-artemis:latest-alpine
```

--------------------------------

### Install grpcio-tools for Protobuf

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/serialization/examples.md

Install the necessary dependency for Protobuf serialization.

```console
pip install grpcio-tools
```

--------------------------------

### Full example asgi.py with FastStream integration

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/integrations/django/index.md

Complete asgi.py example combining Django, Starlette, static files, and FastStream Kafka broker lifespan.

```python
import os
from contextlib import asynccontextmanager

from django.core.asgi import get_asgi_application
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.staticfiles import StaticFiles
from faststream.kafka import KafkaBroker


os.environ.setdefault("DJANGO_SETTINGS_MODULE", "app.settings")

broker = KafkaBroker()

@asynccontextmanager
async def broker_lifespan(app):
    await broker.start()
    try:
        yield
    finally:
        await broker.stop()

application = Starlette(
    routes=(
        Mount("/static", StaticFiles(directory="static"), name="static"),
        Mount("/", get_asgi_application()),
    ),
    lifespan=broker_lifespan,
)
```

--------------------------------

### MQTT Direct Publisher Example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/direct.md

This is a placeholder for the MQTT-specific direct publisher example. The actual code is included from an external file and is not displayed on this page.

```python
{!> docs_src/getting_started/publishing/mqtt/direct.py !}
```

--------------------------------

### Batch Consumption Example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/confluent/Subscriber/batch_subscriber.md

A complete example demonstrating how to consume messages in batches from the 'test_batch' topic.

```python
{! docs_src/confluent/batch_consuming_pydantic/app.py!}
```

--------------------------------

### Install msgpack

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/serialization/examples.md

Install the necessary dependency for Msgpack serialization.

```console
pip install msgpack
```

--------------------------------

### Full Direct Exchange example (direct.py)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/rabbit/examples/direct.md

Complete example from docs_src/rabbit/subscription/direct.py. It sets up a broker with a Direct exchange and multiple queues, demonstrating message routing and load balancing. The auto_delete=True argument is used to clear RabbitMQ state after the example runs.

```python
{! docs_src/rabbit/subscription/direct.py !}
```

--------------------------------

### Start and stop Docker dependencies

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/contributing/CONTRIBUTING.md

Start all dependencies as Docker containers with `just up`. Stop them with `just stop` or `just down`.

```bash
just up
```

```bash
just stop
# or
just down
```

--------------------------------

### Msgpack decoder example (msgpack_ex.py)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/serialization/examples.md

Shows how to write a Msgpack decoder without a schema.

```python
{! docs_src/getting_started/serialization/msgpack_ex.py !}
```

--------------------------------

### Install FastStream with RabbitMQ support

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/index.md

Install the FastStream package with RabbitMQ support using pip. After installation, a test broker container is required.

```console
pip install "faststream[rabbit]"
```

```bash
docker run -d --rm -p 5672:5672 --name test-mq rabbitmq:3.13-alpine
```

--------------------------------

### Install fastavro

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/serialization/examples.md

Install the necessary dependency for Avro serialization.

```console
pip install fastavro
```

--------------------------------

### Consumer Announcement and Queue Setup

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/rabbit/examples/topic.md

Defining the Topic exchange and queues with routing patterns.

```python
{! docs_src/rabbit/subscription/topic.py [ln:7-10]!}
```

--------------------------------

### Install cookiecutter

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/template/index.md

Install the cookiecutter package to enable project template generation.

```bash
pip install cookiecutter
```

--------------------------------

### Full Redis RPC example with FastStream

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/redis/rpc.md

Complete example of setting up Redis RPC with FastStream's RedisBroker. This is the full application combining all snippets from the page.

```python
{! docs_src/redis/rpc/app.py !}
```

--------------------------------

### Install FastStream CLI

Source: https://github.com/ag2ai/faststream/blob/main/README.md

Install the FastStream CLI using pip with the 'cli' extra.

```shell
pip install "faststream[cli]"
```

--------------------------------

### Install FastStream with prometheus-client

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/observability/prometheus.md

Install FastStream with the prometheus-client extra to enable Prometheus metrics collection.

```shell
pip install 'faststream[prometheus]'
```

--------------------------------

### Install FastStream with OpenTelemetry dependencies

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/observability/opentelemetry/index.md

Install FastStream with the OpenTelemetry extras and the OTLP exporter package.

```shell
pip install "faststream[otel]" opentelemetry-exporter-otlp
```

--------------------------------

### Publisher Object Example (MQTT)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/object.md

Example of using the Publisher Object with MQTT. The code is shown with highlighted lines 7 and 9.

```python
{!> docs_src/getting_started/publishing/mqtt/object.py !}
```

--------------------------------

### Basic Confluent KafkaBroker example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/confluent/index.md

Minimal example showing how to connect to Kafka using FastStream's KafkaBroker, with a subscriber for the in-topic and a publisher for the out-topic. The processing function is decorated with @broker.subscriber and @broker.publisher.

```python
{! docs_src/index/confluent/basic.py!}
```

--------------------------------

### RabbitMQ Direct Publisher Example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/direct.md

This is a placeholder for the RabbitMQ-specific direct publisher example. The actual code is included from an external file and is not displayed on this page.

```python
{!> docs_src/getting_started/publishing/rabbit/direct.py !}
```

--------------------------------

### Install FastStream with Redis

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/index.md

Install FastStream with Redis support using pip.

```sh
pip install 'faststream[redis]'
```

--------------------------------

### Run Docker container

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/template/index.md

Start the local Docker container. Replace <username> and <repo-name> with your GitHub credentials.

```bash
docker run --rm --name faststream-app --net=host ghcr.io/<username>/<repo-name>:latest
```

--------------------------------

### MQTT basic example

Source: https://github.com/ag2ai/faststream/blob/main/docs/includes/en/simple-apps.md

Basic FastStream application for MQTT. The code is included from an external file and is not shown inline.

```python
{!> docs_src/index/mqtt/basic.py!}
```

--------------------------------

### Start consumer with Django management command

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/integrations/django/index.md

Command to start the FastStream consumer using Django's manage.py.

```bash
./manage.py faststream

```

--------------------------------

### Kafka Direct Publisher Example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/direct.md

This is a placeholder for the Kafka-specific direct publisher example. The actual code is included from an external file and is not displayed on this page.

```python
{!> docs_src/getting_started/publishing/kafka/direct.py !}
```

--------------------------------

### Install FastStream with Redis support

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/release.md

Use this command to install FastStream with the Redis extra, enabling Redis broker functionality.

```bash
pip install "faststream[redis]"
```

--------------------------------

### Install opentelemetry-exporter-otlp

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/observability/opentelemetry/sentry.md

Install the OpenTelemetry OTLP exporter package to export spans via gRPC.

```shell
pip install opentelemetry-exporter-otlp
```

--------------------------------

### Project structure for healthchecks example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/observability/healthchecks.md

Shows the file layout for the healthchecks example: a dummy directory containing docker-compose.yml, Dockerfile, and main.py.

```text
dummy
├── docker-compose.yml
├── Dockerfile
├── main.py
```

--------------------------------

### Redis Direct Publisher Example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/direct.md

This is a placeholder for the Redis-specific direct publisher example. The actual code is included from an external file and is not displayed on this page.

```python
{!> docs_src/getting_started/publishing/redis/direct.py !}
```

--------------------------------

### Install FastStream with OpenTelemetry SDK

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/observability/opentelemetry/tracing.md

Install FastStream with the OpenTelemetry SDK using pip. This is the first step to enable tracing for your broker.

```shell
pip install 'faststream[otel]'
```

--------------------------------

### Install taskiq-faststream

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/scheduling.md

Install the taskiq-faststream package using pip. This is a prerequisite for integrating Taskiq scheduling with FastStream.

```bash
pip install taskiq-faststream
```

--------------------------------

### Start Kafka container

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/template/index.md

Initialize the local Kafka Docker container required for application development.

```bash
just kafka-up
```

--------------------------------

### Custom ASGI routes with @get and @post decorators

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/asgi.md

Example of defining custom HTTP endpoints using the @get and @post decorators.

```python
{! docs_src/getting_started/asgi/custom_app.py !}
```

--------------------------------

### Install FastStream with Confluent Kafka support

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/confluent/index.md

Install the FastStream package with the 'confluent' extra to enable Confluent Kafka support.

```bash
pip install "faststream[confluent]"
```

--------------------------------

### Publisher Object Example (RabbitMQ)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/object.md

Example of using the Publisher Object with RabbitMQ. The code is shown with highlighted lines 7 and 9.

```python
{!> docs_src/getting_started/publishing/rabbit/object.py !}
```

--------------------------------

### NATS Direct Publisher Example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/direct.md

This is a placeholder for the NATS-specific direct publisher example. The actual code is included from an external file and is not displayed on this page.

```python
{!> docs_src/getting_started/publishing/nats/direct.py !}
```

--------------------------------

### Serve AsyncAPI documentation

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/template/index.md

Builds the AsyncAPI specification and serves the documentation at localhost:8000. Replace <directory-name> with the directory containing your app.py file.

```bash
faststream docs serve <directory-name>.application:app
```

--------------------------------

### Exclude custom ASGI route from AsyncAPI documentation

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/asgi.md

Example of excluding a custom route from AsyncAPI documentation using include_in_schema=False on the @get decorator.

```python
@get(include_in_schema=False)
async def liveness_ping(scope):
    return AsgiResponse(b"", status_code=200)

app = AsgiFastStream(
    broker,
    asgi_routes=[("/health", liveness_ping)]
)
```

--------------------------------

### Install FastStream with pip for various brokers

Source: https://github.com/ag2ai/faststream/blob/main/README.md

Install FastStream with the desired broker dependency using pip. Choose one of the options based on your broker.

```sh
pip install 'faststream[kafka]'
# or
pip install 'faststream[confluent]'
# or
pip install 'faststream[rabbit]'
# or
pip install 'faststream[nats]'
# or
pip install 'faststream[redis]'
# or
pip install 'faststream[mqtt]'
```

--------------------------------

### Register new subscribers at runtime

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/release.md

Shows how to register new subscribers at runtime with an already-started broker using `broker.setup_subscriber()` and the `async with` context manager.

```python
subscriber = broker.subscriber("dynamic")
subscriber(handler_method)
...
broker.setup_subscriber(subscriber)
async with subscriber:
    ...
```

--------------------------------

### Full RPCWorker class example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/howto/kafka/rpc.md

Complete RPCWorker class with start and stop methods for managing the subscriber lifecycle. Includes the request method with timeout handling and the response handler.

```python
from uuid import uuid4
from asyncio import Future, wait_for

from faststream.types import SendableMessage
from faststream.kafka import KafkaBroker, KafkaMessage

class RPCWorker:
    responses: dict[str, Future[bytes]]

    def __init__(self, broker: KafkaBroker, reply_topic: str) -> None:
        self.responses = {}
        self.broker = broker
        self.reply_topic = reply_topic

        self.subscriber = broker.subscriber(reply_topic)
        self.subscriber(self._handle_responses)

    async def start(self) -> None:
        await self.subscriber.start()

    async def stop(self) -> None:
        await self.subscriber.stop()

    def _handle_responses(self, msg: KafkaMessage) -> None:
        if (future := self.responses.pop(msg.correlation_id, None)):
            future.set_result(msg.body)

    async def request(
        self,
        data: SendableMessage,
        topic: str,
        timeout: float = 10.0,
    ) -> bytes:
        correlation_id = str(uuid4())
        future = self.responses[correlation_id] = Future[bytes]()

        await self.broker.publish(
            data, topic,
            reply_to=self.reply_topic,
            correlation_id=correlation_id,
        )

        try:
            response: bytes = await wait_for(future, timeout=timeout)
        except TimeoutError:
            self.responses.pop(correlation_id, None)
            raise
        else:
            return response
```

--------------------------------

### Publisher Object Example (Redis)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/object.md

Example of using the Publisher Object with Redis. The code is shown with highlighted lines 7 and 9.

```python
{!> docs_src/getting_started/publishing/redis/object.py !}
```

--------------------------------

### Using FastStream MQBrokers with aiohttp

Source: https://github.com/ag2ai/faststream/blob/main/README.md

Shows how to use a KafkaBroker without a FastStream application by starting and stopping it according to the aiohttp application's lifespan. The broker is started on startup and stopped on cleanup.

```python
from aiohttp import web

from faststream.kafka import KafkaBroker

broker = KafkaBroker("localhost:9092")

@broker.subscriber("test")
async def base_handler(body):
    print(body)

async def start_broker(app):
    await broker.start()

async def stop_broker(app):
    await broker.stop()

async def hello(request):
    return web.Response(text="Hello, world")

app = web.Application()
app.add_routes([web.get("/", hello)])
app.on_startup.append(start_broker)
app.on_cleanup.append(stop_broker)

if __name__ == "__main__":
    web.run_app(app)
```

--------------------------------

### serve_faststream.py: Django setup before FastStream consumer

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/integrations/django/index.md

Sets up Django by setting DJANGO_SETTINGS_MODULE and calling django.setup() before importing Django models. Then defines a FastStream app with a RabbitMQ subscriber that queries the User model asynchronously.

```Python
# serve_faststream.py
# ruff: noqa: E402
###############################################################################
import os

import django

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "app.settings")
django.setup()
# These lines are necessary to set up Django, call them before any Django Related imports.
###############################################################################
from django.contrib.auth.models import User  # This line must be after django.setup()

from faststream import FastStream
from faststream.rabbit import RabbitBroker


broker = RabbitBroker("amqp://guest:guest@localhost:5672")

@broker.subscriber("demo")
async def faststream_django_orm_demo_handler(message: str):
    """
    This demonstrates how to access Django ORM from within a FastStream consumer.
    """
    qs = User.objects.all()
    async for user in qs:  # async django ORM is accessible
        print(user)
    print(message)

app = FastStream(broker)

```

--------------------------------

### Disable state setup in StreamRouter

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/integrations/fastapi/index.md

Disable the automatic state installation in the StreamRouter when the ASGI server does not support it. After disabling, the broker is not available from the application's state but remains accessible as router.broker.

```python
router = StreamRouter(..., setup_state=False)
```

--------------------------------

### Confluent Direct Publisher Example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/direct.md

This is a placeholder for the Confluent-specific direct publisher example. The actual code is included from an external file and is not displayed on this page.

```python
{!> docs_src/getting_started/publishing/confluent/direct.py !}
```

--------------------------------

### Integrate FastStream with Rocketry

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/scheduling.md

Example of using Rocketry with a FastStream NatsBroker. The broker is passed as a parameter to Rocketry, and a task publishes a message every second. The app is started by connecting the broker and then serving Rocketry.

```python
import asyncio

from rocketry import Rocketry
from rocketry.args import Arg

from faststream.nats import NatsBroker

app = Rocketry(execution="async")

broker = NatsBroker()      # regular broker
app.params(broker=broker)

async def start_app():
    async with broker:     # connect broker
        await app.serve()  # run rocketry

@app.task("every 1 second", execution="async")
async def publish(br: NatsBroker = Arg("broker")):
    await br.publish("Hi, Rocketry!", "test")

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

--------------------------------

### MQTT broker with wildcard topic and QoS

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/release.md

Shows how to set up an MQTT broker with a wildcard subscriber and a publisher that sends a message after startup. Requires `pip install faststream[mqtt]`. The subscriber uses `Path()` to capture the device_id from the topic and sets QoS to AT_LEAST_ONCE.

```python
from faststream import FastStream, Path
from faststream.mqtt import MQTTBroker, MQTTMessage, QoS

broker = MQTTBroker("localhost:1883")
app = FastStream(broker)

@broker.subscriber(
    "sensors/{device_id}/temperature",
    qos=QoS.AT_LEAST_ONCE,
)
async def on_temperature(body: str, device_id: Annotated[str, Path()]) -> None:
    print(device_id, body)

@app.after_startup
async def publish_demo() -> None:
    await broker.publish(21.5, "sensors/room1/temperature", qos=QoS.AT_LEAST_ONCE)
```

--------------------------------

### Structlog configuration example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/observability/logging.md

Configures Structlog with console rendering for terminal sessions and JSON rendering for Docker containers, based on the Structlog guide. This logger prints messages in a user-friendly format during development and JSON in production.

```python
import sys
import structlog

shared_processors = (
    structlog.processors.add_log_level,
    structlog.processors.StackInfoRenderer(),
    structlog.dev.set_exc_info,
    structlog.processors.TimeStamper(fmt="iso"),
)

if sys.stderr.isatty():
    # terminal session
    processors = [
        *shared_processors,
        structlog.dev.ConsoleRenderer(),
    ]
else:
    # Docker container session
    processors = [
        *shared_processors,
        structlog.processors.dict_tracebacks,
        structlog.processors.JSONRenderer(),
    ]

structlog.configure(
    processors=processors,
    logger_factory=structlog.PrintLoggerFactory(),
    cache_logger_on_first_use=False,
)

logger = structlog.get_logger()
```

--------------------------------

### Initialize ML model with on_startup and on_shutdown (MQTT)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/lifespan/hooks.md

Shows how to initialize a machine learning model in the @app.on_startup hook and clean it up in the @app.on_shutdown hook for the MQTT broker. The highlighted lines (16 and 23) mark the hook definitions.

```python
{!> docs_src/getting_started/lifespan/mqtt/ml.py!}
```

--------------------------------

### Modify and clear baggage with CurrentBaggage in subscribers

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/observability/opentelemetry/baggage.md

Shows how to use the CurrentBaggage object, injected from context, to get, set, remove, and clear baggage entries across multiple subscribers. The example demonstrates that baggage is propagated through the chain of subscribers, and that clearing it removes all entries.

```python
from faststream.opentelemetry import CurrentBaggage

@broker.subscriber("first")
@broker.publisher("second")
async def response_handler_first(msg: str, baggage: CurrentBaggage):
    print(baggage.get_all())  # {'hello': 'world'}
    baggage.remove("hello")
    baggage.set("user-id", 1)
    baggage.set("request-id", "UUID")
    print(baggage.get("user-id"))  # 1
    return msg


@broker.subscriber("second")
@broker.publisher("third")
async def response_handler_second(msg: str, baggage: CurrentBaggage):
    print(baggage.get_all())  # {'user-id': '1', 'request-id': 'UUID'}
    baggage.clear()
    return msg


@broker.subscriber("third")
async def response_handler_third(msg: str, baggage: CurrentBaggage):
    print(baggage.get_all())  # {}
```

--------------------------------

### Serve FastStream app documentation

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/asyncapi/custom.md

Run this command to serve your FastStream app documentation. It is used after creating the basic app and after customizing app info.

```shell
{! docs_src/getting_started/asyncapi/serve.py [ln:17] !}
```

--------------------------------

### Adding manual start method to RPCWorker

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/howto/kafka/rpc.md

Adds a start method to RPCWorker that manually starts the subscriber. This is required when the worker is created after the application has already started.

```python
class RPCWorker:
    async def start(self) -> None:
        await self.subscriber.start()
```

--------------------------------

### Run FastStream application

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/template/index.md

Start the FastStream application locally. Replace <directory-name> with the directory containing your app.py file.

```bash
faststream run <directory-name>.application:app --workers 1
```

--------------------------------

### Complete example of scheduled message publishing

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/nats/jetstream/scheduling.md

Full working example combining subscriber and publisher. It schedules a message 3 seconds after startup and prints the delivery time. Note that NATS holds one pending schedule per publish subject, so use unique subjects for multiple schedules.

```python
from datetime import datetime, timedelta, timezone
from uuid import uuid4

from faststream import FastStream
from faststream.nats import JStream, NatsBroker, NatsMessage, Schedule

broker = NatsBroker()


@broker.subscriber(
  "test_stream.*",
  stream=JStream("test_stream", allow_msg_schedules=True)
)
async def handle_scheduled_message(msg: NatsMessage) -> None:
  print(f"Message received at {datetime.now(tz=timezone.utc)}")
  print(msg)


async def on_startup() -> None:
  current_time = datetime.now(tz=timezone.utc)
  schedule_time = current_time + timedelta(seconds=3)
  await broker.connect()

  schedule_target = f"test_stream.{uuid4()}"
  await broker.publish(
    message={"type": "do_something"},
    subject="test_stream.subject",
    schedule=Schedule(schedule_time, schedule_target),
    stream="test_stream",
    timeout=10,
  )
  print(f"Message scheduled for delivery at {schedule_time}")


app = FastStream(broker)
app.on_startup(on_startup)

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

--------------------------------

### Install FastStream with Redis support

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/index.md

Install the FastStream package with Redis support using pip. After installation, a test broker container is required.

```console
pip install "faststream[redis]"
```

```bash
docker run -d --rm -p 6379:6379 --name test-mq redis:8-alpine
```

--------------------------------

### Redis broker subscriber example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/release.md

This Python example shows how to create a Redis broker, attach it to a FastStream app, and define a subscriber that listens to a channel, list, or stream. The subscriber receives a string message and logs it.

```python
from faststream import FastStream, Logger
from faststream.redis import RedisBroker

broker = RedisBroker()
app = FastStream(broker)

@broker.subscriber(
    channel="test",  # or
    # list="test",     or
    # stream="test",
)
async def handle(msg: str, logger: Logger):
    logger.info(msg)
```

--------------------------------

### Install FastStream with NATS support

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/index.md

Install the FastStream package with NATS support using pip. After installation, a test broker container is required.

```console
pip install "faststream[nats]"
```

```bash
docker run -d --rm -p 4222:4222 --name test-mq nats -js
```

--------------------------------

### Complete Redis Stream Batch Consumption Example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/redis/streams/batch.md

A full example demonstrating how to configure a subscriber to process messages in batches from the test-stream Redis stream.

```python
{! docs_src/redis/stream/batch_sub.py !}
```

--------------------------------

### Async context manager middleware example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/release.md

Shows the breaking change that subscriber and publisher middlewares must be async context managers. The example defines `subscriber_middleware` and `publisher_middleware` using `@asynccontextmanager`.

```python
from contextlib import asynccontextmanager

@asynccontextmanager
async def subscriber_middleware(msg_body):
    yield msg_body

@asynccontextmanager
async def publisher_middleware(
    msg_to_publish,
    **publish_arguments,
):
    yield msg_to_publish

@broker.subscriber("in", middlewares=(subscriber_middleware,))
@broker.publisher("out", middlewares=(publisher_middleware,))
async def handler():
    ...
```

--------------------------------

### Run FastStream with --log-config JSON

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/observability/logging.md

Use this command to start the FastStream application with logging configuration from a JSON file.

```console
faststream run serve:app --log-config config.json
```

--------------------------------

### Install FastStream with AIOKafka support

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/index.md

Install the FastStream package with Kafka support using pip. After installation, a test broker container is required.

```console
pip install "faststream[kafka]"
```

```bash
docker run -d --rm -p 9092:9092 --name test-mq \
-e KAFKA_NODE_ID=1 \
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092 \
-e KAFKA_PROCESS_ROLES=broker,controller \
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
-e KAFKA_BROKER_ID=1 \
-e KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 \
-e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
-e KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT \
-e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
-e CLUSTER_ID=MkU3OEVBNTcwNTJENDM2Qk \
confluentinc/cp-kafka:8.0.0
```

--------------------------------

### Install FastStream with Confluent support

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/index.md

Install the FastStream package with Confluent Kafka support using pip. After installation, a test broker container is required.

```console
pip install "faststream[confluent]"
```

```bash
docker run -d --rm -p 9092:9092 --name test-mq \
-e KAFKA_NODE_ID=1 \
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092 \
-e KAFKA_PROCESS_ROLES=broker,controller \
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
-e KAFKA_BROKER_ID=1 \
-e KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 \
-e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
-e KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT \
-e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
-e CLUSTER_ID=MkU3OEVBNTcwNTJENDM2Qk \
confluentinc/cp-kafka:8.0.0
```

--------------------------------

### RabbitMQ basic example

Source: https://github.com/ag2ai/faststream/blob/main/docs/includes/en/simple-apps.md

Basic FastStream application for RabbitMQ. The code is included from an external file and is not shown inline.

```python
{!> docs_src/index/rabbit/basic.py!}
```

--------------------------------

### Run FastStream with --log-config YAML

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/observability/logging.md

Use this command to start the FastStream application with logging configuration from a YAML file.

```console
faststream run serve:app --log-config config.yaml
```

--------------------------------

### Initialize development environment with just init

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/contributing/CONTRIBUTING.md

Builds the FastStream image and installs all dependencies. By default, it builds Python 3.10; pass a version argument to use a different Python version.

```bash
just init
```

```bash
just init 3.11.5
```

--------------------------------

### Registering new subscribers at runtime

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/release.md

New subscribers can be registered at runtime with an already-started broker using `broker.setup_subscriber()` and `subscriber.start()`.

```python
subscriber = broker.subscriber("dynamic")
subscriber(handler_method)
...
broker.setup_subscriber(subscriber)
await subscriber.start()
...
await subscriber.close()

```

--------------------------------

### run()

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/lifespan/lifecycle.md

Starts the application and blocks until the process is told to stop. This is what the CLI calls. It handles Ctrl+C and SIGTERM for clean shutdown.

```APIDOC
## run()

### Description
Starts the application and blocks until the process is told to stop. This is what the CLI calls. It handles Ctrl+C and SIGTERM for clean shutdown.

### Method
run()

### Parameters
- **run_extra_options** (optional) - Extra options to pass to hooks that take CLI options.

### Usage Example
```python
import asyncio

asyncio.run(app.run())
```

### Notes
- Blocks until the process is told to stop.
- Handles Ctrl+C and SIGTERM.
- Enters the FastStream(lifespan=...) context manager if provided.
```

--------------------------------

### Project structure for FastStream CLI entry point

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/integrations/django/index.md

Shows the required project structure with manage.py and serve_faststream.py at the root.

```text
project-root/
├── manage.py
├── serve_faststream.py

```

--------------------------------

### Run FastStream with --log-config TOML

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/observability/logging.md

Use this command to start the FastStream application with logging configuration from a TOML file.

```console
faststream run serve:app --log-config config.toml
```

--------------------------------

### broker.include_router with prefix, dependencies, middlewares, include_in_schema

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/release.md

Shows how to pass setup arguments to `include_router` at inclusion time instead of router creation. Supports `prefix`, `dependencies`, `middlewares`, and `include_in_schema`.

```python
broker.include_router(
   router,
   prefix="test_",
   dependencies=[Depends(...)],
   middlewares=[BrokerMiddleware],
   include_in_schema=False,
)
```

--------------------------------

### Run FastStream application

Source: https://github.com/ag2ai/faststream/blob/main/README.md

Start the FastStream application using the CLI command, passing the module and app symbol.

```shell
faststream run basic:app
```

--------------------------------

### Change directory

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/template/index.md

Navigate into the newly created project directory.

```bash
cd <directory-name>
```

--------------------------------

### Install FastStream with Confluent

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/index.md

Install FastStream with Confluent support using pip.

```sh
pip install 'faststream[confluent]'
```

--------------------------------

### Redis basic example

Source: https://github.com/ag2ai/faststream/blob/main/docs/includes/en/simple-apps.md

Basic FastStream application for Redis. The code is included from an external file and is not shown inline.

```python
{!> docs_src/index/redis/basic.py!}
```

--------------------------------

### Install FastStream with AIOKafka

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/index.md

Install FastStream with AIOKafka support using pip.

```sh
pip install 'faststream[kafka]'
```

--------------------------------

### Initialize ML model with on_startup and on_shutdown (Redis)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/lifespan/hooks.md

Shows how to initialize a machine learning model in the @app.on_startup hook and clean it up in the @app.on_shutdown hook for the Redis broker. The highlighted lines (16 and 23) mark the hook definitions.

```python
{!> docs_src/getting_started/lifespan/redis/ml.py!}
```

--------------------------------

### Create .env file

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/config/index.md

Create a .env file with settings. This file is read as if it were environment variables.

```bash
URL="amqp://guest:guest@localhost:5672/"
QUEUE="test-queue"
```

--------------------------------

### NATS basic example

Source: https://github.com/ag2ai/faststream/blob/main/docs/includes/en/simple-apps.md

Basic FastStream application for NATS. The code is included from an external file and is not shown inline.

```python
{!> docs_src/index/nats/basic.py!}
```

--------------------------------

### Initialize ML model with on_startup and on_shutdown (Confluent)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/lifespan/hooks.md

Shows how to initialize a machine learning model in the @app.on_startup hook and clean it up in the @app.on_shutdown hook for the Confluent broker. The highlighted lines (16 and 23) mark the hook definitions.

```python
{!> docs_src/getting_started/lifespan/confluent/ml.py!}
```

--------------------------------

### NATS KeyValue watch subscription example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/release.md

Shows how to subscribe to a NATS KeyValue bucket using kv_watch and put a value after startup. Requires a NATS broker and the faststream.nats module.

```python
from faststream import FastStream, Logger
from faststream.nats import NatsBroker

broker = NatsBroker()
app = FastStream(broker)

@broker.subscriber("some-key", kv_watch="bucket")
async def handler(msg: int, logger: Logger):
    logger.info(msg)

@app.after_startup
async def test():
    kv = await broker.key_value("bucket")
    await kv.put("some-key", b"1")
```

--------------------------------

### Initialize ML model with on_startup and on_shutdown (RabbitMQ)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/lifespan/hooks.md

Shows how to initialize a machine learning model in the @app.on_startup hook and clean it up in the @app.on_shutdown hook for the RabbitMQ broker. The highlighted lines (16 and 23) mark the hook definitions.

```python
{!> docs_src/getting_started/lifespan/rabbit/ml.py!}
```

--------------------------------

### Install FastStream CLI dependencies

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/cli.md

Install the required dependencies for the FastStream CLI using pip.

```shell
pip install 'faststream[cli]'
```

--------------------------------

### Initialize and push Git repository

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/template/index.md

Initialize a local Git repository, commit the project files, and push them to a remote GitHub repository.

```bash
git init
git add .
git commit -m "first commit"
git branch -M main
git remote add origin git@github.com:<username>/<repo-name>.git
git push -u origin main
```

--------------------------------

### Install FastStream 0.5.0rc0 via pip

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/release.md

Command to manually install the release candidate version of FastStream.

```console
pip install faststream==0.5.0rc0
```

--------------------------------

### Create a basic FastStream app (basic.py)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/asyncapi/custom.md

Copy this code into your basic.py file to create a simple FastStream application instance. This is the prerequisite for customizing AsyncAPI documentation.

```python
{! docs_src/getting_started/asyncapi/asyncapi_customization/basic.py !}
```

--------------------------------

### Serve AsyncAPI docs with FastStream CLI

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/asyncapi/hosting.md

Run the FastStream CLI to serve AsyncAPI documentation. Requires an internet connection to fetch the AsyncAPI HTML via CDN. The path can be specified as python_module:FastStream, asyncapi.json, or asyncapi.yaml.

```shell
faststream asyncapi serve python_module:FastStream
```

```shell
faststream asyncapi serve asyncapi.json
```

```shell
faststream asyncapi serve asyncapi.yaml
```

--------------------------------

### Install PyYAML for YAML support

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/asyncapi/export.md

Install the PyYAML dependency required to generate documentation in YAML format.

```shell
pip install PyYAML
```

--------------------------------

### Initialize ML model with on_startup and on_shutdown (NATS)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/lifespan/hooks.md

Shows how to initialize a machine learning model in the @app.on_startup hook and clean it up in the @app.on_shutdown hook for the NATS broker. The highlighted lines (16 and 23) mark the hook definitions.

```python
{!> docs_src/getting_started/lifespan/nats/ml.py!}
```

--------------------------------

### Install pydantic-settings

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/config/index.md

Install the pydantic-settings package. In Pydantic v1, this functionality was included with the main package; now it is distributed separately.

```console
pip install pydantic-settings
```

--------------------------------

### Subscribe with KvWatch for Detailed Settings

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/nats/jetstream/key-value.md

Shows how to use the KvWatch object to configure key-value subscription with more detailed settings, such as declaring the bucket.

```python
from faststream.nats import NatsBroker, KvWatch

@broker.subscriber(
    "key",
    kv_watch=KvWatch("bucket", declare=False),
)
async def handler(msg: str):
    ...
```

--------------------------------

### Install specific Pydantic version

Source: https://github.com/ag2ai/faststream/blob/main/README.md

Command to install a specific Pydantic version, e.g., to downgrade to PydanticV1 if needed.

```shell
pip install pydantic==1.X.Y
```

--------------------------------

### Pass extra startup arguments with --port and --foo

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/cli.md

Passes custom startup arguments to the application. The output shows the port and foo values being printed.

```shell
faststream run main:app --port 5000 --foo bar
```

```console
2025-08-20 17:53:44,224 INFO     - FastStream app starting...
Port: 5000
Foo: bar
```

--------------------------------

### Run with .env file (production)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/config/index.md

Run the application with a production .env file.

```console
ENV=.production.env faststream run serve:app
```

--------------------------------

### Full Direct Subject example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/nats/examples/direct.md

Complete example of using the Direct subject in FastStream, including consumer declarations and message publishing.

```python
{! docs_src/nats/direct.py [ln:1-12.42,13-] !}
```

--------------------------------

### Basic ASGI app with AsyncAPI docs

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/asyncapi/hosting.md

Creates a FastStream ASGI app with a Kafka broker and serves AsyncAPI documentation at /docs/asyncapi. The app is run with uvicorn on port 8000.

```python
from faststream.asgi import AsgiFastStream
from faststream.kafka import KafkaBroker
from faststream.specification import AsyncAPI

broker = KafkaBroker()

@broker.subscriber('topic')
async def my_handler(msg: str) -> None:
    print(msg)

app = AsgiFastStream(
    broker,
    specification=AsyncAPI(),
    asyncapi_path="/docs/asyncapi",
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

--------------------------------

### Startup function with env parameter

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/lifespan/hooks.md

Python code for the @app.on_startup decorator that declares a function to run at startup, with an 'env' parameter passed from command-line arguments.

```python
@app.on_startup
async def setup(env: str): ...
```

--------------------------------

### NATS Object Storage watch subscription example

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/release.md

Shows how to subscribe to a NATS Object Storage bucket using obj_watch and put an object after startup. Requires a NATS broker and the faststream.nats module.

```python
from faststream import FastStream, Logger
from faststream.nats import NatsBroker

broker = NatsBroker()
app = FastStream(broker)

@broker.subscriber("file-bucket", obj_watch=True)
async def handler(filename: str, logger: Logger):
    logger.info(filename)

@app.after_startup
async def test():
    object_store = await broker.object_storage("file-bucket")
    await object_store.put("some-file.txt", b"1")
```

--------------------------------

### Build Docker image

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/template/index.md

Build the Docker image locally using the provided initialization script.

```bash
just init
```

--------------------------------

### Publisher Object Example (NATS)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/object.md

Example of using the Publisher Object with NATS. The code is shown with highlighted lines 7 and 9.

```python
{!> docs_src/getting_started/publishing/nats/object.py !}
```

--------------------------------

### Publisher Object Example (Confluent)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/object.md

Example of using the Publisher Object with Confluent. The code is shown with highlighted lines 7 and 9.

```python
{!> docs_src/getting_started/publishing/confluent/object.py !}
```

--------------------------------

### Initialize ML model with on_startup and on_shutdown (AIOKafka)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/lifespan/hooks.md

Shows how to initialize a machine learning model in the @app.on_startup hook and clean it up in the @app.on_shutdown hook for the AIOKafka broker. The highlighted lines (16 and 23) mark the hook definitions.

```python
{!> docs_src/getting_started/lifespan/kafka/ml.py!}
```

--------------------------------

### Publisher Object Example (AIOKafka)

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/publishing/object.md

Example of using the Publisher Object with AIOKafka. The code is shown with highlighted lines 7 and 9.

```python
{!> docs_src/getting_started/publishing/kafka/object.py !}
```

--------------------------------

### Filtered subscriber example for MQTT

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/getting-started/subscription/filtering.md

Same filtering example as AIOKafka but for MQTT. Uses a filter to handle JSON messages and a default handler for others.

```python
{!> docs_src/getting_started/subscription/mqtt/filter.py [ln:1-18] !}
```

--------------------------------

### Create KafkaBroker instance for basic publishing

Source: https://github.com/ag2ai/faststream/blob/main/docs/docs/en/kafka/Publisher/index.md

Creates a KafkaBroker instance. This is the first step for basic publishing.

```python
{!> docs_src/kafka/raw_publish/example.py [ln:7] !}
```
Tokens:
Raw