faststream Documentation

File: README.md Copy Chapter

FastStream

FastStream is an asynchronous Python framework for building event-driven applications. It brings together message broker integration, dependency injection, validation, testing utilities, and AsyncAPI documentation generation in a single toolkit, reducing boilerplate without hiding the capabilities of the underlying broker.

Features Copy Chapter

FastStream simplifies the process of writing producers and consumers for message queues, handling all the
parsing, networking and documentation generation automatically.

Making streaming microservices has never been easier. Designed with junior developers in mind, FastStream simplifies your work while keeping the door open for more advanced use cases. Here's a look at the core features that make FastStream a go-to framework for modern, data-centric microservices.

• Multiple Brokers: FastStream provides a suitable API to work across multiple message brokers (Kafka, RabbitMQ, NATS, Redis, MQTT support)

• Built-in Serialization: Leverage Pydantic or Msgspec validation capabilities to serialize and validate incoming messages

• Automatic Docs: Stay ahead with automatic AsyncAPI documentation

• Intuitive: Full-typed editor support makes your development experience smooth, catching errors before they reach runtime

• Powerful Dependency Injection System: Manage your service dependencies efficiently with FastStream's built-in DI system

• Testable: Supports in-memory tests, making your CI/CD pipeline faster and more reliable

• Extensible: Use extensions for lifespans, custom serialization and middleware

• Integrations: FastStream is fully compatible with any HTTP framework you want (FastAPI especially)

That's FastStream in a nutshell - easy, efficient, and powerful. Whether you're just starting with streaming microservices or looking to scale, FastStream has got you covered.

Documentation: https://faststream.ag2.ai/latest/

Table of Contents

• Features
• Versioning Policy
• Installation
• Quick Start
  • Pydantic serialization
  • Msgspec serialization
  • Unified API
• Testing
• CLI
• AsyncAPI Documentation
• Dependencies
• Integrations
  • Any Framework
  • FastAPI Plugin
• Stay in touch

Project History

FastStream is a package based on the ideas and experiences gained from FastKafka and Propan. By joining our forces, we picked up the best from both packages and created a unified way to write services capable of processing streamed data regardless of the underlying protocol.

Versioning Policy Copy Chapter

FastStream has a stable public API. Only major updates may introduce breaking changes.

Prior to FastStream's 1.0 release, each minor update is considered a major and can introduce breaking changes, but these changes were communicated through two-versions deprecation warnings prior to being fully removed. So features deprecated in the 0.4 version were only removed in version 0.6.

Our team is working toward the stable 1.0 version.

Installation Copy Chapter

FastStream works on Linux, macOS, Windows and most Unix-style operating systems.
You can install it with pip as usual:

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

Writing app code Copy Chapter

FastStream brokers provide convenient function decorators @broker.subscriber
and @broker.publisher to allow you to delegate the actual process of:

• consuming and producing data to Event queues, and

• decoding and encoding JSON-encoded messages

These decorators make it easy to specify the processing logic for your consumers and producers, allowing you to focus on the core business logic of your application without worrying about the underlying integration.

Also, FastStream uses Pydantic to parse input
JSON-encoded data into Python objects, making it easy to work with structured data in your applications, so you can serialize your input messages just using type annotations.

Here is an example Python app using FastStream that consumes data from an incoming data stream and outputs the data to another one:

from faststream import FastStream
from faststream.kafka import KafkaBroker
# from faststream.confluent import KafkaBroker
# from faststream.rabbit import RabbitBroker
# from faststream.nats import NatsBroker
# from faststream.redis import RedisBroker
# from faststream.mqtt import MQTTBroker

broker = KafkaBroker("localhost:9092")
# broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
# broker = NatsBroker("nats://localhost:4222/")
# broker = RedisBroker("redis://localhost:6379/")
# broker = MQTTBroker("localhost")

app = FastStream(broker)

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

Pydantic serialization Copy Section

Also, Pydantic’s BaseModel class allows you
to define messages using a declarative syntax, making it easy to specify the fields and types of your messages.

from pydantic import BaseModel, Field, PositiveInt
from faststream import FastStream
from faststream.kafka import KafkaBroker

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

class User(BaseModel):
    user: str = Field(..., examples=["John"])
    user_id: PositiveInt = Field(..., examples=["1"])

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

By default we use PydanticV2 written in Rust as serialization library, but you can downgrade it manually, if your platform has no Rust support - FastStream will work correctly with PydanticV1 as well.

To choose the Pydantic version, you can install the required one using the regular

pip install pydantic==1.X.Y

FastStream (and FastDepends inside) should work correctly with almost any version.

Msgspec serialization Copy Section

Moreover, FastStream is not tied to any specific serialization library, so you can use any preferred one. Fortunately, we provide a built‑in alternative for the most popular Pydantic replacement - Msgspec.

from fast_depends.msgspec import MsgSpecSerializer
from faststream.kafka import KafkaBroker

broker = KafkaBroker(serializer=MsgSpecSerializer())

You can read more about the feature in the documentation.

Unified API Copy Section

At first glance, FastStream unifies various broker backends under a single API. However, a completely unified API inevitably results in missing features. We do not want to limit users' choices. If you prefer Kafka over Redis, there is a reason. Therefore, we support all native broker features you need.

Consequently, our unified API has a relatively limited scope:

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

Beyond this scope you can use any broker-native features you need:

• Kafka - specific partition reads, partitioner control, consumer groups, batch processing, etc.
• RabbitMQ - all exchange types, Redis Streams, RPC, manual channel configuration, DLQ, etc.
• NATS - core and Push/Pull JetStream subscribers, KeyValue, ObjectStorage, RPC, etc.
• Redis - Pub/Sub, List, Stream subscribers, consumer groups, acknowledgements, etc.
• MQTT - topic subscriptions (including wildcards), QoS and retain, MQTT 3.1.1 and 5.0, request/reply (RPC), TLS, etc.

You can find detailed information about all supported features in FastStream’s broker‑specific documentation.

If a particular feature is missing or not yet supported, you can always fall back to the native broker client/connection for those operations.

Testing the service Copy Chapter

The service can be tested using the TestBroker context managers, which, by default, puts the Broker into "testing mode".

The Tester will redirect your subscriber and publisher decorated functions to the InMemory brokers, allowing you to quickly test your app without the need for a running broker and all its dependencies.

Using pytest, the test for our service would look like this:

import pytest
import pydantic
from faststream.kafka import TestKafkaBroker


@pytest.mark.asyncio
async def test_correct():
    async with TestKafkaBroker(broker) as br:
        await br.publish({
            "user": "John",
            "user_id": 1,
        }, "in")

@pytest.mark.asyncio
async def test_invalid():
    async with TestKafkaBroker(broker) as br:
        with pytest.raises(pydantic.ValidationError):
            await br.publish("wrong message", "in")

Running the application Copy Chapter

The application can be started using built-in FastStream CLI command.

Before running the service, install FastStream CLI using the following command:

pip install "faststream[cli]"

To run the service, use the FastStream CLI command and pass the module (in this case, the file where the app implementation is located) and the app symbol to the command.

faststream run basic:app

After running the command, you should see the following output:

INFO     - FastStream app starting...
INFO     - input_data |            - `HandleMsg` waiting for messages
INFO     - FastStream app started successfully! To exit press CTRL+C

Also, FastStream provides you with a great hot reload feature to improve your Development Experience

faststream run basic:app --reload

And multiprocessing horizontal scaling feature as well:

faststream run basic:app --workers 3

You can learn more about CLI features here

Project Documentation Copy Chapter

FastStream automatically generates documentation for your project according to the AsyncAPI specification. You can work with both generated artifacts and place a web view of your documentation on resources available to related teams.

The availability of such documentation significantly simplifies the integration of services: you can immediately see what channels and message formats the application works with. And most importantly, it won't cost anything - FastStream has already created the docs for you!

Dependencies Copy Chapter

FastStream (thanks to FastDepends) has a dependency management system similar to pytest fixtures and FastAPI Depends at the same time. Function arguments declare which dependencies you want are needed, and a special decorator delivers them from the global Context object.

from typing import Annotated
from faststream import Depends, Logger

async def base_dep(user_id: int) -> bool:
    return True

@broker.subscriber("in-test")
async def base_handler(user: str,
                       logger: Logger,
                       dep: Annotated[bool, Depends(base_dep)]):
    assert dep is True
    logger.info(user)

HTTP Frameworks integrations Copy Chapter

Any Framework Copy Section

You can use FastStream MQBrokers without a FastStream application.
Just start and stop them according to your application's lifespan.

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)

**FastAPI** Plugin (deprecated) Copy Section

The integration has been moved to the
faststream_fastapi
package and will be removed in 1.0.0 version.

Also, FastStream can be used as part of FastAPI.

Just import a StreamRouter you need and declare the message handler with the same @router.subscriber(...) and @router.publisher(...) decorators.

from fastapi import FastAPI
from pydantic import BaseModel

from faststream.kafka.fastapi import KafkaRouter

router = KafkaRouter("localhost:9092")

class Incoming(BaseModel):
    m: dict

@router.subscriber("test")
@router.publisher("response")
async def hello(m: Incoming):
    return {"response": "Hello, world!"}

app = FastAPI()
app.include_router(router)

More integration features can be found here

Benchmarks Copy Chapter

We use codspeed to run benchmarks for both FastStream itself and raw clients.

Stay in touch Copy Chapter

Please show your support and stay in touch by:

• giving our GitHub repository a star, and

• joining our EN Discord server

• joining our RU Telegram group

Your support helps us to stay in touch with you and encourages us to
continue developing and improving the framework. Thank you for your
support!

Contributors Copy Chapter

Thanks to all of these amazing people who made the project better!

File: docs/docs/en/confluent/Publisher/batch_publisher.md Copy Chapter

0.5 - API

2 - Release

3 - Contributing

5 - Template Page

10 - Default

search: boost: 10 Copy Chapter

Publishing in Batches

General Overview Copy Chapter

If you need to send your data in batches, the #!python @broker.publisher(...) decorator offers a convenient way to achieve this. To enable batch production, you need to perform two crucial steps:

1. When creating your publisher, set the batch argument to True. This configuration tells the publisher that you intend to send messages in batches.

2. In your producer function, return a tuple containing the messages you want to send as a batch. This action triggers the producer to gather the messages and transmit them as a batch to a Kafka broker.

Let's delve into a detailed example illustrating how to produce messages in batches to the #!python "output_data" topic while consuming from the #!python "input_data_1" topic.

Code Example Copy Chapter

First, let's take a look at the whole app creation and then dive deep into the steps for producing in batches. Here is the application code:

{! docs_src/confluent/publish_batch/app.py!}

Below, we have highlighted key lines of code that demonstrate the steps involved in creating and using a batch publisher:

Step 1: Creation of the Publisher

{! docs_src/confluent/publish_batch/app.py [ln:19] !}

Step 2: Publishing an Actual Batch of Messages

You can publish a batch by directly calling the publisher with a batch of messages you want to publish, as shown here:

{! docs_src/confluent/publish_batch/app.py [ln:32.5,33.5,34.5] !}

Or you can decorate your processing function and return a batch of messages, as shown here:

{! docs_src/confluent/publish_batch/app.py [ln:22-26] !}

The application in the example implements both of these ways, so feel free to use whichever option fits your needs better.

!!! note
Also, you can publishes messages in batches right from a broker object: just call
#!python broker.publish_batch("msg2", "msg2", topic="output_data")

Per-Message Attributes with `KafkaPublishMessage` Copy Chapter

When publishing in batches, you often need to assign different keys, headers, or timestamps to each message within the same batch. To support this scenario, FastStream provides the #!python KafkaPublishMessage helper — a semantic alias for #!python KafkaResponse that is intended to be used directly inside #!python publish_batch(...) calls.

By wrapping a payload into #!python KafkaPublishMessage, you can attach per-message attributes such as:

• #!python key — the Kafka message key used for partitioning;
• #!python headers — custom headers for the individual message;
• #!python timestamp_ms — an explicit message timestamp;
• #!python correlation_id — a custom correlation identifier.

You can freely mix #!python KafkaPublishMessage instances with raw payloads in the same call. Plain values (strings, bytes, dicts, models, etc.) are published as-is and use the default key (#!python None) inherited from the #!python publish_batch(...) call:

from faststream.confluent import KafkaBroker, KafkaPublishMessage

broker = KafkaBroker()


@broker.subscriber("input")
async def handler() -> None:
    await broker.publish_batch(
        KafkaPublishMessage("user:1", key=b"user1"),
        KafkaPublishMessage("user:2", key=b"user2"),
        "user:3",  # Uses default key (None)
        topic="output",
    )

In the example above, the first two messages are sent with their own dedicated keys (#!python b"user1" and #!python b"user2"), while the third message — passed as a plain string — falls back to the default key.

!!! note
#!python KafkaPublishMessage is the recommended, more semantic name when constructing outgoing messages for #!python publish_batch(...). Under the hood it is the same object as #!python KafkaResponse, so both names are fully interchangeable.

!!! tip
Use #!python KafkaPublishMessage whenever messages within a single batch need to be routed to different partitions via distinct keys, or when you need to attach per-message metadata — this is the cleanest way to control individual message attributes without splitting the batch into separate #!python publish(...) calls.

Why Publish in Batches? Copy Chapter

In the above example, we've explored how to leverage the #!python @broker.publisher(...) decorator to efficiently publish messages in batches using FastStream and Kafka. By following the two key steps outlined in the previous sections, you can significantly enhance the performance and reliability of your Kafka-based applications.

Publishing messages in batches offers several advantages when working with Kafka:

1. Improved Throughput: Batch publishing allows you to send multiple messages in a single transmission, reducing the overhead associated with individual message delivery. This leads to improved throughput and lower latency in your Kafka applications.

2. Reduced Network and Broker Load: Sending messages in batches reduces the number of network calls and broker interactions. This optimization minimizes the load on the Kafka brokers and network resources, making your Kafka cluster more efficient.

3. Atomicity: Batches ensure that a group of related messages is processed together or not at all. This atomicity can be crucial in scenarios where message processing needs to maintain data consistency and integrity.

4. Enhanced Scalability: With batch publishing, you can efficiently scale your Kafka applications to handle high message volumes. By sending messages in larger chunks, you can make the most of Kafka's parallelism and partitioning capabilities.

File: docs/docs/en/confluent/Publisher/index.md Copy Chapter

0.5 - API

2 - Release

3 - Contributing

5 - Template Page

10 - Default

search: boost: 10 Copy Chapter

Publishing

The FastStream KafkaBroker supports all regular publishing use cases{.internal-link}, and you can use them without any changes.

However, if you wish to further customize the publishing logic, you should take a closer look at specific KafkaBroker parameters.

Basic Kafka Publishing Copy Chapter

The KafkaBroker uses the unified publish method (from a producer object) to send messages.

In this case, you can use Python primitives and pydantic.BaseModel to define the content of the message you want to publish to the Kafka broker.

You can specify the topic to send by its name.

1. Create your KafkaBroker instance

   python linenums="1" Copy

   {!> docs_src/confluent/raw_publish/example.py [ln:7] !}

2. Publish a message using the publish method

   python linenums="1" Copy

   {!> docs_src/confluent/raw_publish/example.py [ln:25-30] !}

This is the most basic way of using the KafkaBroker to publish a message.

Creating a publisher object Copy Chapter

The simplest way to use a KafkaBroker for publishing has a significant limitation: your publishers won't be documented in the AsyncAPI documentation. This might be acceptable for sending occasional one-off messages. However, if you're building a comprehensive service, it's recommended to create publisher objects. These objects can then be parsed and documented in your service's AsyncAPI documentation. Let's go ahead and create those publisher objects!

1. Create your KafkaBroker instance

   python linenums="1" Copy

   {!> docs_src/confluent/publisher_object/example.py [ln:8] !}

2. Create a publisher instance

   python linenums="1" Copy

   {!> docs_src/confluent/publisher_object/example.py [ln:17] !}

3. Publish a message using the publish method of the prepared publisher

   python linenums="1" Copy

   {!> docs_src/confluent/publisher_object/example.py [ln:26.9,27.7,28.9,29.9,30.9,31.9] !}

Now, when you wrap your broker into a FastStream object, the publisher will be exported to the AsyncAPI documentation.

Decorating your publishing functions Copy Chapter

To publish messages effectively in the Kafka context, consider utilizing the Publisher Decorator. This approach offers an AsyncAPI representation and is ideal for rapidly developing applications.

The Publisher Decorator creates a structured DataPipeline unit with both input and output components. The sequence in which you apply Subscriber and Publisher decorators does not affect their functionality. However, note that these decorators can only be applied to functions decorated by a Subscriber as well.

This method relies on the return type annotation of the handler function to properly interpret the function's return value before sending it. Hence, it's important to ensure accuracy in defining the return type.

Let's start by examining the entire application that utilizes the Publisher Decorator and then proceed to walk through it step by step.

{! docs_src/confluent/publish_example/app.py [ln:1-26] !}

1. Initialize the KafkaBroker instance: Start by initializing a KafkaBroker instance with the necessary configuration, including Kafka broker address.

   python linenums="1" Copy

   {!> docs_src/confluent/publish_example/app.py [ln:13] !}

2. Prepare your publisher object to use later as a decorator:

   python linenums="1" Copy

   {!> docs_src/confluent/publish_example/app.py [ln:17] !}

3. Create your processing logic: Write a function that will consume the incoming messages in the defined format and produce a response to the defined topic

   python linenums="1" Copy

   {!> docs_src/confluent/publish_example/app.py [ln:22-23] !}

4. Decorate your processing function: To connect your processing function to the desired Kafka topics you need to decorate it with #!python @broker.subscriber(...) and #!python @broker.publisher(...) decorators. Now, after you start your application, your processing function will be called whenever a new message in the subscribed topic is available and produce the function return value to the topic defined in the publisher decorator.

   python linenums="1" Copy

   {!> docs_src/confluent/publish_example/app.py [ln:20-23] !}

File: docs/docs/en/confluent/Publisher/using_a_key.md Copy Chapter

0.5 - API

2 - Release

3 - Contributing

5 - Template Page

10 - Default

search: boost: 10 Copy Chapter

Using a Partition Key

Partition keys are a crucial concept in Apache Kafka, enabling you to determine the appropriate partition for a message. This ensures that related messages are kept together in the same partition, which can be invaluable for maintaining order or grouping related messages for efficient processing. Additionally, Kafka utilizes partitioning to distribute load across multiple brokers and scale horizontally, while replicating data across brokers provides fault tolerance.

You can specify your partition keys when utilizing the #!python @KafkaBroker.publisher(...) decorator in FastStream. This guide will walk you through the process of using partition keys effectively.

Publishing with a Partition Key Copy Chapter

To publish a message to a Kafka topic using a partition key, follow these steps:

Step 1: Define the Publisher Copy Section

In your FastStream application, define the publisher using the #!python @KafkaBroker.publisher(...) decorator. This decorator allows you to configure various aspects of message publishing, including the partition key.

{! docs_src/confluent/publish_with_partition_key/app.py [ln:17] !}

Step 2: Pass the Key Copy Section

When you're ready to publish a message with a specific key, simply include the key parameter in the publish function call. This key parameter is used to determine the appropriate partition for the message.

{! docs_src/confluent/publish_with_partition_key/app.py [ln:25.5] !}

Example Application Copy Chapter

Let's examine a complete application example that consumes messages from the #!python "input_data" topic and publishes them with a specified key to the #!python "output_data" topic. This example will illustrate how to incorporate partition keys into your Kafka-based applications:

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

As you can see, the primary difference from standard publishing is the inclusion of the key parameter in the publish call. This key parameter is essential for controlling how Kafka partitions and processes your messages.

In summary, using partition keys in Apache Kafka is a fundamental practice for optimizing message distribution, maintaining order, and achieving efficient processing. It is a key technique for ensuring that your Kafka-based applications scale gracefully and handle large volumes of data effectively.

File: docs/docs/en/confluent/Subscriber/batch_subscriber.md Copy Chapter

0.5 - API

2 - Release

3 - Contributing

5 - Template Page

10 - Default

search: boost: 10 Copy Chapter

Batch Subscriber

If you want to consume data in batches, the #!python @broker.subscriber(...) decorator makes it possible. By defining your consumed msg object as a list of messages and setting the batch parameter to True, the subscriber will call your consuming function with a batch of messages consumed from a single partition. Let's walk through how to achieve this.

Using the Subscriber with Batching Copy Chapter

To consume messages in batches, follow these steps:

Step 1: Define Your Subscriber Copy Section

In your FastStream application, define the subscriber using the #!python @broker.subscriber(...) decorator. Ensure that you configure the msg object as a list and set the batch parameter to True. This configuration tells the subscriber to handle message consumption in batches.

{! docs_src/confluent/batch_consuming_pydantic/app.py [ln:20] !}

Step 2: Implement Your Consuming Function Copy Section

Create a consuming function that accepts the list of messages. The #!python @broker.subscriber(...) decorator will take care of collecting and grouping messages into batches based on the partition.

{! docs_src/confluent/batch_consuming_pydantic/app.py [ln:20-22] !}

Example of Consuming in Batches Copy Chapter

Let's illustrate how to consume messages in batches from the #!python "test_batch" topic with a practical example:

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

In this example, the subscriber is configured to process messages in batches, and the consuming function is designed to handle these batches efficiently.

Consuming messages in batches is a valuable technique when you need to optimize the processing of high volumes of data in your Kafka-based applications. It allows for more efficient resource utilization and can enhance the overall performance of your data pipelines.

File: docs/docs/en/confluent/Subscriber/index.md Copy Chapter

0.5 - API

2 - Release

3 - Contributing

5 - Template Page

10 - Default

search: boost: 10 Copy Chapter

Basic Subscriber

To start consuming from a Kafka topic, simply decorate your consuming function with a #!python @broker.subscriber(...) decorator, passing a string as a topic key.

In the following example, we will create a simple FastStream app that will consume HelloWorld messages from a #!python "hello_world" topic.

The full app code looks like this:

{! docs_src/confluent/consumes_basics/app.py!}

Import FastStream and KafkaBroker Copy Chapter

To use the #!python @broker.subscriber(...) decorator, first, we need to import the base FastStream app KafkaBroker to create our broker.

{! docs_src/confluent/consumes_basics/app.py [ln:3-4] !}

Define the HelloWorld Message Structure Copy Chapter

Next, you need to define the structure of the messages you want to consume from the topic using Pydantic. For the guide, we’ll stick to something basic, but you are free to define any complex message structure you wish in your project.

{! docs_src/confluent/consumes_basics/app.py [ln:7-12] !}

Create a KafkaBroker Copy Chapter

Next, we will create a KafkaBroker object and wrap it into the FastStream object so that we can start our app using CLI later.

{! docs_src/confluent/consumes_basics/app.py [ln:15-16] !}

Create a Function that will Consume Messages from a Kafka hello-world Topic Copy Chapter

Let’s create a consumer function that will consume HelloWorld messages from #!python "hello_world" topic and log them.

{! docs_src/confluent/consumes_basics/app.py [ln:19-21] !}

The function decorated with the #!python @broker.subscriber(...) decorator will be called when a message is produced to Kafka.

The message will then be injected into the typed msg argument of the function, and its type will be used to parse the message.

In this example case, when the message is sent to a #!python "hello_world" topic, it will be parsed into a HelloWorld class, and the on_hello_world function will be called with the parsed class as the msg argument value.

Multiple Topics Copy Chapter

You can subscribe to multiple topics with a single #!python @broker.subscriber(...) call by passing multiple topic names as positional arguments:

{! docs_src/confluent/multiple_topics_subscription/app.py !}

A single handler will receive messages from all listed topics, under one consumer group.

This differs from stacking multiple #!python @broker.subscriber(...) decorators, which creates separate independent handlers.

!!! warning
max_workers > 1 is only compatible with AckPolicy.ACK_FIRST. Using any other ack policy with max_workers > 1 will raise a SetupError at startup, regardless of how many topics are subscribed to.

File: docs/docs/en/confluent/ack.md Copy Chapter

0.5 - API

2 - Release

3 - Contributing

5 - Template Page

10 - Default

search: boost: 10 Copy Chapter

Consuming Acknowledgements

As you may know, Kafka consumer should commit a topic offset when consuming a message.

The default behaviour, also implemented as such in the FastStream, uses the #!python AckPolicy.ACK_FIRST policy which automatically commits (acks) topic offset using enable.auto.commit{.external-link target="_blank"} setting. This is the at most once consuming strategy.

However, if you wish to use at least once strategy, you should commit offset AFTER the message is processed correctly. To accomplish that, set a consumer group and use #!python AckPolicy.ACK strategy:

@broker.subscriber(
    "test", group_id="group", ack_policy=AckPolicy.ACK
)
async def base_handler(body: str):
    ...

This way, the message processed will be acknowledged after handler execution. In the case of an exception being raised, the message will also be acknowledged.

If you want to retry on error, you can use #!python AckPolicy.NACK_ON_ERROR strategy. In this way offset will not be committed and consumer seeks to read this message again:

@broker.subscriber(
    "test", group_id="group", ack_policy=AckPolicy.NACK_ON_ERROR
)
async def base_handler(body: str):
    ...

However, there are situations where you might want to use a different acknowledgement logic.

Manual Acknowledgement Copy Chapter

If you want to acknowledge a message manually, you can get direct access to the message object via the Context{.internal-link} and acknowledge the message by calling the ack method:

from faststream.confluent.annotations import KafkaMessage


@broker.subscriber(
    "test", group_id="group", ack_policy=AckPolicy.MANUAL
)
async def base_handler(body: str, msg: KafkaMessage):
    await msg.ack()
    # or
    await msg.nack()
    # or any custom logic

!!! tip
You can use the nack method to prevent offset commit and the message can be consumed by another consumer within the same group.

FastStream will see that the message was already acknowledged and will do nothing at the end of the process.

Summary **FastStream** [AckPolicy](../getting-started/acknowledgement.md){.internal-link} behavior Copy Chapter

Interrupt Process Copy Chapter

If you wish to interrupt the processing of a message at any call stack level and acknowledge the message, you can achieve that by raising the faststream.exceptions.AckMessage.

{! docs_src/confluent/ack/errors.py !}

This way, FastStream interrupts the current message processing and acknowledges it immediately. Similarly, you can raise NackMessage as well to prevent the message from being committed.

{! includes/en/no_ack.md !}

File: docs/docs/en/confluent/additional-configuration.md Copy Chapter

0.5 - API

2 - Release

3 - Contributing

5 - Template Page

10 - Default

search: boost: 10 Copy Chapter

Passing Additional Configuration to confluent-kafka-python

The confluent-kafka-python package is a Python wrapper around librdkakfa, which is a C/C++ client library for Apache Kafka.

confluent-kafka-python accepts a config dictionary that is then passed on to librdkafka. librdkafka provides plenty of configuration properties to configure the Kafka client.

FastStream also provides users with the ability to pass the config dictionary to librdkafka to provide greater customizability.

Example Copy Chapter

In the following example, we are setting the parameter topic.metadata.refresh.fast.interval.ms's value to 300 instead of the default value 100 via the config parameter.

{! docs_src/confluent/additional_config/app.py !}

Similarly, you could use the config parameter to pass any configuration properties to librdkafka.

File: docs/docs/en/confluent/index.md Copy Chapter

0.5 - API

2 - Release

3 - Contributing

5 - Template Page

10 - Default

search: boost: 10 Copy Chapter

Confluent Kafka Routing

Confluent's Python Client for Apache Kafka Copy Chapter

The Confluent Kafka Python library is developed by Confluent, the company founded by the creators of Apache Kafka. It offers a high-level Kafka producer and consumer API that integrates well with the Kafka ecosystem. The Confluent library provides a comprehensive set of features, including support for Avro serialization, schema registry integration, and various configurations to fine-tune performance.

Developed by Confluent, this library enjoys strong support from the core team behind Kafka. This often translates to better compatibility with the latest Kafka releases and a more robust feature set.

!!! note ""
If you prefer the aiokafka library instead, then please refer to aiokafka's KafkaBroker

Installation Copy Chapter

!!! warning "Available starting from v0.4.0rc0"
Support for Confluent is present starting from the version v0.4.0rc0.

Please run the following command to install it:

```bash
pip install "faststream[confluent]>=0.4.0"
```

FastStream Confluent KafkaBroker Copy Section

The FastStream Confluent KafkaBroker is a key component of the FastStream framework that enables seamless integration with Apache Kafka using confluent kafka{.external-link target="_blank"} python library. With the KafkaBroker, developers can easily connect to Kafka brokers, produce messages to Kafka topics, and consume messages from Kafka topics within their FastStream applications.

Establishing a Connection Copy Section

To connect to Kafka using the FastStream KafkaBroker module, follow these steps:

1. Initialize the KafkaBroker instance: Start by initializing a KafkaBroker instance with the necessary configuration, including Kafka broker address.

2. Create your processing logic: Write a function that will consume the incoming messages in the defined format and produce a response to the defined topic

3. Decorate your processing function: To connect your processing function to the desired Kafka topics you need to decorate it with #!python @broker.subscriber(...) and #!python @broker.publisher(...) decorators. Now, after you start your application, your processing function will be called whenever a new message in the subscribed topic is available and produce the function return value to the topic defined in the publisher decorator.

Here's a simplified code example demonstrating how to establish a connection to Kafka using FastStream's KafkaBroker module:

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

This minimal example illustrates how FastStream simplifies the process of connecting to Kafka and performing basic message processing from the in_topic to the out-topic. Depending on your specific use case and requirements, you can further customize your Kafka integration with FastStream to build robust and efficient streaming applications.

For more advanced configuration options and detailed usage instructions, please refer to the FastStream Kafka documentation and the official Kafka documentation{.external-link target="_blank"}.

File: docs/docs/en/confluent/message.md Copy Chapter

0.5 - API

2 - Release

3 - Contributing

5 - Template Page

10 - Default

search: boost: 10 Copy Chapter

Access to Message Information

As you may know, FastStream serializes a message body and provides you access to it through function arguments. However, there are times when you need to access additional message attributes such as offsets, headers, or other metadata.

Message Access Copy Chapter

You can easily access this information by referring to the message object in the Context

This object serves as a unified FastStream wrapper around the native broker library message (for example, confluent_kafka.Message in this case which uses Confluent python library). It contains most of the required information, including:

• #!python headers(): Sequence[Tuple[str, bytes]]
• #!python key(): Optional[Union[str, bytes]]
• #!python offset(): int
• #!python partition(): int
• #!python timestamp(): Tuple[int, int]
• #!python topic(): str
• #!python value(): Optional[Union[str, bytes]]

For example, if you would like to access the headers of an incoming message, you would do so like this:

from faststream.confluent import KafkaMessage

@broker.subscriber("test")
async def base_handler(
    body: str,
    msg: KafkaMessage,
):
    print(msg.headers)

Raw Message Access Copy Chapter

In some cases, you may want to access the raw confluent_kafka.Message created by confluent_kafka library. In such cases, you can do so by:

from faststream.confluent import KafkaMessage

@broker.subscriber("test")
async def base_handler(
    body: str,
    msg: KafkaMessage,
):
    print(msg.raw_message.headers())

Message Fields Access Copy Chapter

In most cases, you don't need all message fields; you need to know just a part of them.
You can use Context Fields access feature for this.

For example, you can get access to the headers like this:

from faststream import Context

@broker.subscriber("test")
async def base_handler(
    body: str,
    headers: str = Context("message.headers"),
):
    print(headers)

{! includes/message/headers.md !}