### Basic FastStream App Setup

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

This is a foundational example for setting up a basic FastStream application. It serves as a starting point before applying specific AsyncAPI customizations.

```python
from faststream import FastStream, Logger
from faststream.rabbit import RabbitBroker

broker = RabbitBroker()
app = FastStream(broker)

@app.subscriber("test_queue")
async def handle_message(body: dict, logger: Logger):
    logger.info(f"Received message: {body}")
    await broker.publish(
        {"message": "Hello from FastStream!"},
        "response_queue",
    )

```

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

### Basic AIOKafka Producer/Consumer Example

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

Demonstrates a simple producer and consumer setup using AIOKafka. Ensure AIOKafka is installed and a Kafka broker is running.

```python
from faststream.kafka import KafkaBroker

broker = KafkaBroker("localhost:9092")

@broker.subscriber("test-topic")
async def handle_message(msg: str):
    print(msg)

@broker.publisher("test-topic")
async def send_message(msg: str):
    pass

@broker.after_startup
async def startup():
    await send_message.publish("Hello Kafka!")

if __name__ == "__main__":
    broker.run_forever()

```

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

### Full Example: MQTT Subscription Test

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

Demonstrates a complete FastStream application setup and testing for MQTT.

```python
from faststream.mqtt import MQTTBroker
from faststream.test import TestMQTTBroker

broker = MQTTBroker()

@broker.subscriber("test-topic")
async def handle(user_id: int, name: str):
    pass


@broker.publisher("test-topic")
async def publish(user_id: int, name: str):
    pass


async def test_mqtt_full():
    async with TestMQTTBroker(broker, log_level="info") as test_broker:
        await test_broker.publish(
            {
                "user_id": 1,
                "name": "John",
            },
            "test-topic",
        )

        handle.mock.assert_called_once_with(user_id=1, name="John")

```

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

### Install FastStream with MQTT

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

Use this command to install FastStream with MQTT support. A test MQTT broker container setup is also provided.

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

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

### Full Example: Redis Subscription Test

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

Demonstrates a complete FastStream application setup and testing for Redis.

```python
from faststream.redis import RedisBroker
from faststream.test import TestRedisBroker

broker = RedisBroker()

@broker.subscriber("test-channel")
async def handle(user_id: int, name: str):
    pass


@broker.publisher("test-channel")
async def publish(user_id: int, name: str):
    pass


async def test_redis_full():
    async with TestRedisBroker(broker, log_level="info") as test_broker:
        await test_broker.publish(
            {
                "user_id": 1,
                "name": "John",
            },
            "test-channel",
        )

        handle.mock.assert_called_once_with(user_id=1, name="John")

```

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

### Full Example: RabbitMQ Subscription Test

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

Demonstrates a complete FastStream application setup and testing for RabbitMQ.

```python
from faststream.rabbit import RabbitBroker
from faststream.test import TestRabbitBroker

broker = RabbitBroker()

@broker.subscriber("test-queue")
async def handle(user_id: int, name: str):
    pass


@broker.publisher("test-queue")
async def publish(user_id: int, name: str):
    pass


async def test_rabbit_full():
    async with TestRabbitBroker(broker, log_level="info") as test_broker:
        await test_broker.publish(
            {
                "user_id": 1,
                "name": "John",
            },
            "test-queue",
        )

        handle.mock.assert_called_once_with(user_id=1, name="John")

```

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

### AIOKafka Basic Example

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

Demonstrates basic usage of AIOKafka with FastStream. Ensure AIOKafka is installed.

```python
from faststream.kafka import KafkaBroker

broker = KafkaBroker("localhost:9092")

@broker.subscriber("test-topic")
async def handle_message(msg: str):
    print(msg)

@broker.publisher("another-topic")
async def send_message(msg: str):
    return msg

@broker.after_startup
async def setup():
    await broker.publish("Hello, Kafka!", "test-topic")

@broker.after_server_stop
async def shutdown():
    print("Shutting down Kafka broker...")

```

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

### Initialize MQTT Broker

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

Import the MQTTBroker and initialize it. This example shows basic setup with optional parameters for version and security.

```python
from faststream.mqtt.broker import MQTTBroker

broker = MQTTBroker(
    host="localhost",
    port=1883,
    version="5.0",
    client_id="faststream-client",
    security=None,
    keepalive=60,
    clean_session=True,
    reconnect=None,
    session_expiry_interval=None,
    mqtt_connect_timeout=30,
)


@broker.subscriber("test/topic")
async def handle_test_message():
    pass

```

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

### Basic MQTT Producer/Consumer Example

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

Shows a simple producer and consumer setup for MQTT. Requires an MQTT broker (like Mosquitto) running and the 'hbmqtt' library.

```python
from faststream.mqtt.faststream import MQTTFastStream
from faststream.mqtt.broker import TopicMessage

broker = MQTTFastStream("mqtt://localhost:1883")

@broker.subscriber("test-topic")
async def handle_message(msg: TopicMessage):
    print(msg.payload.decode())

@broker.publisher("test-topic")
async def send_message(msg: str):
    pass

@broker.after_startup
async def startup():
    await send_message.publish("Hello MQTT!")

if __name__ == "__main__":
    broker.run_forever()

```

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

### Subscribe to RabbitMQ

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

This example shows how to subscribe to a RabbitMQ queue. Install FastStream with the 'rabbit' extra for this functionality.

```python
from faststream.rabbit import RabbitBroker

broker = RabbitBroker()

@broker.subscriber("test")  # queue name
async def handle_msg(msg_body):
    ...
```

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

### Full Example: NATS Subscription Test

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

Demonstrates a complete FastStream application setup and testing for NATS.

```python
from faststream.nats import NatsBroker
from faststream.test import TestNatsBroker

broker = NatsBroker()

@broker.subscriber("test-subject")
async def handle(user_id: int, name: str):
    pass


@broker.publisher("test-subject")
async def publish(user_id: int, name: str):
    pass


async def test_nats_full():
    async with TestNatsBroker(broker, log_level="info") as test_broker:
        await test_broker.publish(
            {
                "user_id": 1,
                "name": "John",
            },
            "test-subject",
        )

        handle.mock.assert_called_once_with(user_id=1, name="John")

```

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

### Full Example: Confluent Subscription Test

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

Demonstrates a complete FastStream application setup and testing for Confluent.

```python
from faststream.confluent import KafkaBroker
from faststream.test import TestKafkaBroker

broker = KafkaBroker()

@broker.subscriber("test-topic")
async def handle(user_id: int, name: str):
    pass


@broker.publisher("test-topic")
async def publish(user_id: int, name: str):
    pass


async def test_confluent_full():
    async with TestKafkaBroker(broker, log_level="info") as test_broker:
        await test_broker.publish(
            {
                "user_id": 1,
                "name": "John",
            },
            "test-topic",
        )

        handle.mock.assert_called_once_with(user_id=1, name="John")

```

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

### FastStream Unified API Example

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

Demonstrates basic subscriber and publisher setup using FastStream's unified API. Use this for common messaging patterns.

```python
from faststream.[broker] import [Broker], [Broker]Message

broker = [Broker](*servers)

@broker.subscriber([source])  # Kafka topic / RMQ queue / NATS subject / MQTT topic / etc
@broker.publisher([destination])  # topic / routing key / subject / etc
async def handler(msg: [Broker]Message) -> None:
    await msg.ack()  # control brokers' acknowledgement policy

...

await broker.publish("Message", [destiination])
```

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

### Install FastStream with RabbitMQ

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

Use this command to install FastStream with RabbitMQ support. A test RabbitMQ broker container setup is also provided.

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

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

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

### Basic NATS Producer/Consumer Example

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

Provides a simple producer and consumer example for NATS. Ensure a NATS server is accessible and the 'nats-py' library is installed.

```python
from faststream.nats import NatsBroker

broker = NatsBroker("nats://localhost:4222")

@broker.subscriber("test-subject")
async def handle_message(msg: str):
    print(msg)

@broker.publisher("test-subject")
async def send_message(msg: str):
    pass

@broker.after_startup
async def startup():
    await send_message.publish("Hello NATS!")

if __name__ == "__main__":
    broker.run_forever()

```

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

### Full Example: AIOKafka Subscription Test

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

Demonstrates a complete FastStream application setup and testing for AIOKafka.

```python
from faststream.kafka import KafkaBroker
from faststream.test import TestKafkaBroker

broker = KafkaBroker()

@broker.subscriber("test-topic")
async def handle(user_id: int, name: str):
    pass


@broker.publisher("test-topic")
async def publish(user_id: int, name: str):
    pass


async def test_kafka_full():
    async with TestKafkaBroker(broker, log_level="info") as test_broker:
        await test_broker.publish(
            {
                "user_id": 1,
                "name": "John",
            },
            "test-topic",
        )

        handle.mock.assert_called_once_with(user_id=1, name="John")

```

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

### RabbitMQ Broker Example

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

Illustrates the setup of a FastStream RabbitMQ broker with subscriber and publisher functionalities. Use for standard RabbitMQ message handling.

```python
from faststream.rabbit import RabbitBroker, RabbitMessage

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

@broker.subscriber("in-queue")
@broker.publisher("out-queue")
async def handler(msg: RabbitMessage) -> None:
    await msg.ack()  # control brokers' acknowledgement policy

...

await broker.publish("Message", "in-queue")
```

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

### Install FastStream with Redis

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

Use this command to install FastStream with Redis support. A test Redis broker container setup is also provided.

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

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

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

### Install FastStream with Confluent

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

Use this command to install FastStream with Confluent support. A test Kafka broker container setup is also provided.

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

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

### Full Direct Subject Implementation

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

Complete example demonstrating the setup of a NATS broker and handlers for direct subjects.

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

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

### Full Example: Dynamic Configuration with NATS

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

A complete example demonstrating dynamic configuration using NATS Key-Value storage. It includes setting initial values, subscribing to changes, and reacting to events based on the configuration.

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

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

### FastStream MQTT Broker Example

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

Demonstrates setting up an MQTT broker with FastStream, including wildcard topic filters, path parameter capture, and message publishing.

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

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

### Install FastStream with AIOKafka

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

Use this command to install FastStream with AIOKafka support. A test Kafka broker container setup is also provided.

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

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

### MQTT Dependency Injection Setup

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

Sets up dependency injection for MQTT using the `Depends` class. This example highlights the declaration of a callable dependency and its subsequent use.

```python
from faststream.mqtt import MQTTBroker
from faststream.dependencies import Depends

def get_user_id() -> int:
    return 1


broker = MQTTBroker()


@broker.subscriber("test-topic")
async def handle(
    user_id: int = Depends(get_user_id),
):
    print(user_id)

```

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

### Install Development Requirements

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

Install all necessary dependencies for development.

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

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

### Confluent Kafka Broker Example

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

Shows how to configure a FastStream broker for Confluent Kafka, including subscriber and publisher setup. Suitable for basic Confluent Kafka usage.

```python
from faststream.confluent import KafkaBroker, KafkaMessage

broker = KafkaBroker("localhost:9092")

@broker.subscriber("in-topic")
@broker.publisher("out-topic")
async def handler(msg: KafkaMessage) -> None:
    await msg.ack()  # control brokers' acknowledgement policy

...

await broker.publish("Message", "in-topic")
```

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

### Subscriber Iteration Example

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

Demonstrates how to start a subscriber and iterate over incoming messages using an async for loop. This feature supports all middlewares and other FastStream functionalities.

```python
subscriber = broker.subscriber(..., persistent=False)

await subscriber.start()

async for msg in subscriber:
    ...
```

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

### Start Docker Dependencies

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

Starts all required dependencies as Docker containers for local development and testing.

```bash
just up
```

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

### Subscriber Iteration Example

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

Illustrates how to use the new subscriber iteration feature. After starting a subscriber, you can asynchronously iterate over incoming messages.

```python
subscriber = broker.subscriber(...)

await subscriber.start()

async for msg in subscriber:
    ...
```

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

### Install FastStream with MQTT Support

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

Use pip or uv to install the specific release version of FastStream with MQTT capabilities. This command installs the pre-release version.

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

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

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

### Install Protobuf dependencies

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

Install the necessary tools to work with Protobuf in Python.

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

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

### HTTP Framework Integration: Aiohttp

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

Example of using FastStream MQBrokers with the Aiohttp framework. The broker is started and stopped according to the application's lifespan.

```python
from faststream.rabbit import RabbitBroker
from aiohttp import web

broker = RabbitBroker()


@broker.subscriber("test-queue")
async def handle_message(body: str):
    print(body)


async def send_message(request):
    await broker.publish("Hello World!", queue="test-queue")
    return web.json_response({"message": "Message published"})


async def startup_handler():
    await broker.start()


async def shutdown_handler():
    await broker.stop()


app = web.Application()
app.on_startup.append(startup_handler)
app.on_shutdown.append(shutdown_handler)
app.add_routes([web.get("/send-message", send_message)])
```

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

### Complete Redis Stream Example

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

A full example demonstrating both subscription and publication within Redis streams.

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

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

### Install FastStream with MQTT

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

Install FastStream with support for the MQTT broker using pip.

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

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

### Serve AsyncAPI documentation

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

Starts a local server at localhost:8000 to preview AsyncAPI documentation. Replace <directory-name> with the project's directory containing app.py.

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

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

### Install Msgpack dependency

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

Install the msgpack library for binary serialization.

```console
pip install msgpack
```

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

### Subscribe to NATS Object Storage Watch

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

This example shows how to subscribe to changes in a NATS Object Storage bucket. It includes setting up a subscriber and interacting with the Object Storage after the application starts.

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

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

### Usage with Manual Start

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

Shows how to use the RPCWorker when it is initialized and started after the application lifecycle begins.

```python
from faststream import FastStream
from faststream.kafka import KafkaBroker

broker = KafkaBroker()
app = FastStream(broker)

@app.after_startup
async def send_request() -> None:
    worker = RPCWorker(broker, reply_topic="responses")
    await worker.start()

    data = await worker.request("echo", "echo-topic")
    assert data == "echo"
```

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

### Example .env File Content

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

A sample `.env` file demonstrating how to define environment variables for application configuration.

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

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

### Complete FastStream Scheduled Message Example

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

A full example demonstrating how to set up a FastStream application with JetStream message scheduling, including subscriber and publisher logic, and startup event for publishing.

```python
from datetime import UTC, datetime, timedelta
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=UTC)}")
  print(msg)


async def on_startup() -> None:
  current_time = datetime.now(tz=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())
```

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

### Serve Local Documentation with Just

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

This command serves the local documentation site, allowing you to preview changes in real-time. For a more comprehensive build including all dependencies and extended processing, use the --full flag.

```bash
just docs-serve
```

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

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

### FastStream Multi-Broker Support Example

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

Illustrates how to configure a single FastStream application to manage multiple brokers simultaneously, useful for bridging different messaging systems.

```python
from faststream import FastStream
from faststream.kafka import KafkaBroker
from faststream.nats import NatsBroker

kafka_broker = KafkaBroker("localhost:9092")
nats_broker = NatsBroker("nats://localhost:4222")

app = FastStream(kafka_broker, nats_broker)

@kafka_broker.subscriber("incoming")
@nats_broker.publisher("outgoing")
async def from_kafka(msg: str) -> str:
    # Bridge the message from Kafka to NATS
    return msg

@nats_broker.subscriber("outgoing")
async def from_nats(msg: str) -> None:
    print(f"Received from NATS: {msg}")
```

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

### Install FastStream with NATS

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

Use this command to install FastStream with NATS support. A test NATS broker container setup is also provided.

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

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

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

### Install FastStream CLI

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

Install the FastStream CLI to manage and run your applications. This is a prerequisite for the following steps.

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

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

### Full ASGI Example with FastStream and Django

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

A comprehensive example demonstrating the integration of FastStream with a Django project within a Starlette ASGI application. It includes static file handling and Django's default routing.

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

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

### FastStream Redis Broker Example

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

Example of setting up a FastStream application with a Redis broker. This snippet demonstrates how to define a subscriber that listens to a 'test' channel.

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

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

Install FastStream with the necessary OpenTelemetry packages and the OTLP exporter.

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

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

### Add Manual Start Method

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

Extends the RPCWorker to support manual subscriber startup.

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

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

### MQTT Basic Example

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

Shows basic usage of MQTT with FastStream. Requires an MQTT broker running.

```python
from faststream.mqtt import MQTTBroker

broker = MQTTBroker("mqtt://localhost:1883")

@broker.subscriber("test-topic")
async def handle_message(msg: str):
    print(msg)

@broker.publisher("another-topic")
async def send_message(msg: str):
    return msg

@broker.after_startup
async def setup():
    await broker.publish("Hello, MQTT!", "test-topic")

@broker.after_server_stop
async def shutdown():
    print("Shutting down MQTT broker...")

```

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

### Run FastStream Application

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

Start the FastStream application locally.

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

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

### Full Example of Redis List Publishing

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

A complete example demonstrating Redis list publishing with FastStream, including broker instantiation, app creation, Pydantic model, and publisher/subscriber functions.

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


class Ping(BaseModel):
    message: str


broker = RedisBroker()
app = FastStream(broker)


@broker.subscriber("input-list")
@broker.publisher("output-list")
async def handle_message(msg: Ping) -> Ping:
    return Ping(message=f"Processed: {msg.message}")
```

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

### Initialize settings using command-line arguments

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

This code demonstrates initializing application settings within a startup hook, using values passed from the command line via the 'env' argument.

```python
from faststream import FastStream, Context
from faststream.rabbit import RabbitBroker
from pydantic import BaseModel


class Settings(BaseModel):
    test: str


broker = RabbitBroker()
app = FastStream(broker, settings=Settings)


@app.on_startup
async def startup_handler(settings=Context()):
    print(settings)

```

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

### Install FastStream 0.5.0rc0

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

Manually install the release candidate version of FastStream using pip.

```bash
pip install faststream==0.5.0rc0
```

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

### Basic FastStream Application Setup

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

Create a standard FastStream application. This serves as the base for integrating Taskiq-FastStream.

```python
from faststream.kafka import KafkaBroker

broker = KafkaBroker("localhost:9092")


@broker.subscriber("test-topic")
async def handle_message():
    pass
```

```python
from faststream.confluent import ConfluentBroker

broker = ConfluentBroker("localhost:9092")


@broker.subscriber("test-topic")
async def handle_message():
    pass
```

```python
from faststream.rabbit import RabbitBroker

broker = RabbitBroker("localhost:5672")


@broker.subscriber("test-topic")
async def handle_message():
    pass
```

```python
from faststream.nats import NatsBroker

broker = NatsBroker("nats://localhost:4222")


@broker.subscriber("test-topic")
async def handle_message():
    pass
```

```python
from faststream.redis import RedisBroker

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


@broker.subscriber("test-topic")
async def handle_message():
    pass
```

```python
from faststream.mqtt import MQTTBroker

broker = MQTTBroker("localhost:1883")


@broker.subscriber("test-topic")
async def handle_message():
    pass
```

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

### Install FastStream with Kafka Support

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

Install FastStream with Kafka support using pip. Other broker options are available.

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

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

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

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

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

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

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

### Install FastStream with OpenTelemetry Support

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

Install the necessary dependencies to enable OpenTelemetry integration with FastStream. This command should be run in your project's environment.

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

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

### Start FastStream Consumer with Django Command

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

This Python code defines a Django management command to start the FastStream application. Ensure `serve_faststream` is correctly configured.

```python
import asyncio

from django.core.management.base import BaseCommand
from serve_faststream import app as faststream_app


class Command(BaseCommand):
    help = "Start FastStream consumer"

    def handle(self, *args, **options):
        asyncio.run(faststream_app.run())

```

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

### Include Router with Arguments in FastStream

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

Demonstrates how to use `broker.include_router` to pass arguments for router setup at the time of inclusion, rather than during creation. This allows for more flexible configuration of included routers.

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

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

### Full Redis List Batch Consumption Example

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

A complete example demonstrating the configuration of a subscriber to process messages in batches from a Redis list.

```python
{! docs_src/redis/list/sub_batch.py !}
```

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

### RabbitMQ Basic Example

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

Shows basic integration with RabbitMQ using FastStream. Ensure RabbitMQ is running.

```python
from faststream.rabbit import RabbitBroker

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

@broker.subscriber("test-queue")
async def handle_message(msg: str):
    print(msg)

@broker.publisher("another-queue")
async def send_message(msg: str):
    return msg

@broker.after_startup
async def setup():
    await broker.publish("Hello, RabbitMQ!", "test-queue")

@broker.after_server_stop
async def shutdown():
    print("Shutting down RabbitMQ broker...")

```

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

### NATS Broker Example

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

Provides an example of configuring a FastStream NATS broker, including subscriber and publisher definitions. Ideal for basic NATS messaging.

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

broker = NatsBroker("nats://localhost:4222")

@broker.subscriber("in-subject")
@broker.publisher("out-subject")
async def handler(msg: NatsMessage) -> None:
    await msg.ack()  # control brokers' acknowledgement policy

...

await broker.publish("Message", "in-subject")
```

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

### Redis Broker Example

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

Demonstrates setting up a FastStream Redis broker with subscriber and publisher. Suitable for basic Redis Pub/Sub or Stream operations.

```python
from faststream.redis import RedisBroker, RedisMessage

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

@broker.subscriber("in-channel")
@broker.publisher("out-channel")
async def handler(msg: RedisMessage) -> None:
    await msg.ack()  # control brokers' acknowledgement policy

...

await broker.publish("Message", "in-channel")
```

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

### Install FastStream with Prometheus Support

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

Install the special distribution of FastStream that includes Prometheus support. This is required to use the Prometheus middleware.

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

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

### Include Direct Exchange Example File

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

Reference the full source file for the direct exchange implementation.

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

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

### Dynamic Subscribers with NATS

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

Provides an example of creating a dynamic subscriber for a NATS subject. `subscriber.get_one()` is not functional in this setup.

```python
broker = NatsBroker()

async with TestNatsBroker(broker) as br:
    subscriber = br.subscriber("test-subject", persistent=False)

    await subscriber.start()
    message = await subscriber.get_one()  # does not work
    await subscriber.stop()
```

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

### FastStream Redis Cluster Broker Example

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

Shows how to use the RedisClusterBroker for connecting to a Redis Cluster, featuring automatic node discovery and basic event handling.

```python
from faststream import FastStream
from faststream.redis import RedisClusterBroker

# A single URL is enough — the cluster auto-discovers all remaining nodes
broker = RedisClusterBroker("redis://node1:7000")
app = FastStream(broker)

@broker.subscriber("events")
async def handle_event(msg: str) -> None:
    print(f"Received: {msg}")

@app.after_startup
async def publish_event() -> None:
    await broker.publish("hello from cluster", "events")
```

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

### Install FastStream with Redis Support

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

Install FastStream with Redis support using pip. This command is used to add Redis capabilities to your FastStream project.

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

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

### Establish a connection with KafkaBroker

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

A basic example demonstrating how to initialize a KafkaBroker and use subscriber and publisher decorators.

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

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

### Serve AsyncAPI Documentation with CLI

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

Use this command to serve the AsyncAPI documentation. Specify the path in the format `python_module:FastStream` or directly to an `asyncapi.json` or `asyncapi.yaml` file.

```shell
faststream asyncapi serve docs_src.asyncapi:FastStream
```

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

### Consume Single Message with MQTT

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

This example illustrates consuming a single message from an MQTT broker. Ensure you manually start and stop the subscriber.

```python
from faststream.mqtt import MQTTBroker

broker = MQTTBroker("mqtt://localhost:1883")

@broker.subscriber("test-topic")
async def handle_message():
    pass

async def main():
    await broker.start()
    # Process a single message
    await broker.consume_message()
    await broker.stop()

```

```python
await broker.start()
    # Process a single message
    await broker.consume_message()
    await broker.stop()

```

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

### Initializing Broker with MsgSpecSerializer

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

Example of how to configure a FastStream Broker to use MsgSpecSerializer for serialization.

```python
from fast_depends.msgspec import MsgSpecSerializer

broker = Broker(serializer=MsgSpecSerializer())
```

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

### MQTT Broker Example

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

Shows the configuration of a FastStream MQTT broker with subscriber and publisher. Use for standard MQTT message handling.

```python
from faststream.mqtt import MQTTBroker, MQTTMessage

broker = MQTTBroker("localhost", port=1883)

@broker.subscriber("in-topic")
@broker.publisher("out-topic")
async def handler(msg: MQTTMessage) -> None:
    await msg.ack()  # control brokers' acknowledgement policy

...

await broker.publish("Message", "in-topic")
```

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

### HTTP Framework Integration: Sanic

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

Example of using FastStream MQBrokers with the Sanic framework. The broker is started and stopped according to the application's lifespan.

```python
from faststream.rabbit import RabbitBroker
from sanic import Sanic

broker = RabbitBroker()


@broker.subscriber("test-queue")
async def handle_message(body: str):
    print(body)


app = Sanic(__name__)


@app.get("/send-message")
async def send_message(request):
    await broker.publish("Hello World!", queue="test-queue")
    return app.json({"message": "Message published"})


app.register_listener(broker.start, "before_server_start")
app.register_listener(broker.stop, "after_server_stop")
```

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

### FastStream KafkaBroker with Confluent Client

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

Example demonstrating how to initialize KafkaBroker with Confluent's client and set up a message handler. This function consumes from 'in-topic' and publishes to 'out-topic'.

```python
from faststream import FastStream
from faststream.confluent import KafkaBroker

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

@broker.subscriber("in-topic")
@broker.publisher("out-topic")
async def handle_msg(user: str, user_id: int) -> str:
    return f"User: {user_id} - {user} registered"
```

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

### Complete Kafka Partition Key Application

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

A full example demonstrating consuming from an input topic and publishing to an output topic using a partition key.

```python
{! docs_src/confluent/publish_with_partition_key/app.py [ln:1-25] !}
```

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

### HTTP Framework Integration: Quart

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

Example of using FastStream MQBrokers with the Quart framework. The broker is started and stopped according to the application's lifespan.

```python
from faststream.rabbit import RabbitBroker
from quart import Quart

broker = RabbitBroker()


@broker.subscriber("test-queue")
async def handle_message(body: str):
    print(body)


app = Quart(__name__)


@app.route("/send-message")
async def send_message():
    await broker.publish("Hello World!", queue="test-queue")
    return {"message": "Message published"}


@app.before_serving
async def startup_handler():
    await broker.start()


@app.after_serving
async def shutdown_handler():
    await broker.stop()
```

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

### Implementing Async Context Manager Middlewares

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

Provides examples of async subscriber and publisher middlewares that must be implemented as async context managers. These middlewares can intercept and modify messages during subscription and publication.

```python
async def subscriber_middleware(call_next, msg):
    return await call_next(msg)

async def publisher_middleware(call_next, msg, **kwargs):
    return await call_next(msg, **kwargs)

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

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

### HTTP Framework Integration: Falcon

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

Example of using FastStream MQBrokers with the Falcon framework. The broker is started and stopped according to the application's lifespan.

```python
import falcon
from faststream.rabbit import RabbitBroker

broker = RabbitBroker()


@broker.subscriber("test-queue")
async def handle_message(body: str):
    print(body)


class MessageResource:
    async def on_get(self, req, resp):
        await broker.publish("Hello World!", queue="test-queue")
        resp.text = "Message published"


app = falcon.asgi.App()
app.add_route("/send-message", MessageResource())


@app.lifespan
async def lifespan(scope, receive, send):
    await broker.start()
    yield
    await broker.stop()
```

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

### Async Context Manager Middlewares

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

Demonstrates the breaking change where both subscriber and publisher middlewares must now be async context managers. Includes examples for both subscriber and publisher middleware implementations.

```python
from contextlib import asynccontextmanager

@asynccontextmanager
def subscriber_middleware(msg_body):
    yield msg_body

@asynccontextmanager
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():
    ...
```

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

### HTTP Framework Integration: Blacksheep

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

Example of using FastStream MQBrokers with the Blacksheep framework. The broker is started and stopped according to the application's lifespan.

```python
from faststream.rabbit import RabbitBroker
from blacksheep import Application, Response

broker = RabbitBroker()


@broker.subscriber("test-queue")
async def handle_message(body: str):
    print(body)


async def send_message(request):
    await broker.publish("Hello World!", queue="test-queue")
    return Response(b"Message published")


app = Application()
app.add_event_handler("startup", broker.start)
app.add_event_handler("shutdown", broker.stop)
app.router.add_route("GET", "/send-message", send_message)
```

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

### Basic NATS Subscriber with Pull Subscription

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

Demonstrates a basic NATS subscriber using `pull_sub=True` for simplified pull subscription setup. This is an alternative to using `pull_sub=PullSub()`.

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

broker = NatsBroker()
app = FastStream(broker)

@broker.subscriber("test", stream="stream", pull_sub=True)
async def handler(msg, logger: Logger):
    logger.info(msg)
```

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

### Configure confluent-kafka-python with Custom Settings

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

Pass a configuration dictionary to KafkaBroker to customize confluent-kafka-python settings. This example sets the topic metadata refresh interval to 300ms.

```python
from faststream import FastStream
from faststream.confluent import KafkaBroker

config = {"topic.metadata.refresh.fast.interval.ms": 300}
broker = KafkaBroker("localhost:9092", config=config)
app = FastStream(broker)
```

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

### Configure Prometheus Middleware and ASGI App

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

Integrate the NatsPrometheusMiddleware into your FastStream application to collect metrics. This example also shows how to serve a metrics endpoint using ASGI and Prometheus client.

```python
from prometheus_client import CollectorRegistry, make_asgi_app
from faststream.asgi import AsgiFastStream
from faststream.nats import NatsBroker
from faststream.nats.prometheus import NatsPrometheusMiddleware

registry = CollectorRegistry()

broker = NatsBroker(
    middlewares=(
        NatsPrometheusMiddleware(registry=registry),
    )
)

app = AsgiFastStream(
    broker,
    asgi_routes=[
        ("/metrics", make_asgi_app(registry)),
    ]
)
```

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

### HTTP Framework Integration: Litestar

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

Example of using FastStream MQBrokers with the Litestar framework. The broker is started and stopped according to the application's lifespan.

```python
from faststream.rabbit import RabbitBroker
from litestar import Litestar

broker = RabbitBroker()


@broker.subscriber("test-queue")
async def handle_message(body: str):
    print(body)


app = Litestar(
    route_list=[],
    lifespan_plugins=[
        broker
    ],
)


@app.get("/send-message")
async def send_message():
    await broker.publish("Hello World!", queue="test-queue")
    return {"message": "Message published"}
```

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

### Serve AsyncAPI Documentation from YAML

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

Serve AsyncAPI documentation directly from a YAML file using the CLI.

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

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

### NATS Basic Example

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

Demonstrates basic usage of NATS with FastStream. Requires a running NATS server.

```python
from faststream.nats import NatsBroker

broker = NatsBroker("nats://localhost:4222")

@broker.subscriber("test-subject")
async def handle_message(msg: str):
    print(msg)

@broker.publisher("another-subject")
async def send_message(msg: str):
    return msg

@broker.after_startup
async def setup():
    await broker.publish("Hello, NATS!", "test-subject")

@broker.after_server_stop
async def shutdown():
    print("Shutting down NATS broker...")

```

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

### FastStream Redis Router with FastAPI Context

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

Integrates FastStream's Redis router with a FastAPI application, utilizing the lifespan context for setup and including a subscriber handler.

```python
from fastapi import FastAPI
from faststream.redis.fastapi import RedisRouter, Logger

router = RedisRouter()

@router.subscriber("test")
async def handler(msg, logger: Logger):
    logger.info(msg)

app = FastAPI(lifespan=router.lifespan_context)
app.include_router(router)
```

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

### Redis Dependency Injection Setup

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

Sets up dependency injection for Redis using the `Depends` class. This example highlights the declaration of a callable dependency and its subsequent use.

```python
from faststream.redis import RedisBroker
from faststream.dependencies import Depends

def get_user_id() -> int:
    return 1


broker = RedisBroker()


@broker.subscriber("test-channel")
async def handle(
    user_id: int = Depends(get_user_id),
):
    print(user_id)

```

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

### Configure External Backend for 'Try It Out'

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

Point the 'Try It Out' UI to an external backend by passing a custom URL to `try_it_out_path` via AsyncAPIRoute. This is useful for separate services or production broker URLs.

```python
from faststream.nats import NatsBroker
from faststream.asgi import AsgiFastStream, AsyncAPIRoute

broker = NatsBroker()

app = AsgiFastStream(
    broker,
    asyncapi_path=AsyncAPIRoute(
        "/docs/asyncapi",
        try_it_out_path="https://api.example.com/asyncapi/try",
    ),
)
```