Install & Compatibility
Where this runs
tested against v0.7.5 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
40MB installed
● package 40MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
FastStream
✓ from faststream import FastStream
Logger
✓ from faststream import Logger
KafkaBroker
✓ from faststream.kafka import KafkaBroker
✗ from faststream.broker.kafka import KafkaBroker
Broker imports are now directly from `faststream.<broker_name>` as of v0.7.0rc0 and likely stable 0.7.0. Older versions used `faststream.broker.<broker_name>`.
MQTTBroker
✓ from faststream.mqtt import MQTTBroker
✗ from faststream.broker.mqtt import MQTTBroker
Broker imports are now directly from `faststream.<broker_name>` as of v0.7.0rc0 and likely stable 0.7.0. Older versions used `faststream.broker.<broker_name>`.
ContextRef
✓ from faststream.types import ContextRef
This quickstart demonstrates a basic FastStream application with Kafka. It defines a Kafka broker, an application instance, and a subscriber that listens to 'test-topic'. A message is published on application startup for immediate testing. Ensure Kafka is running and FastStream is installed with Kafka extras to run.
import os
from faststream import FastStream, Logger
from faststream.kafka import KafkaBroker
# Configure Kafka broker connection
# Use 'localhost:9092' for local Kafka, or environment variable for production
KAFKA_BROKER_URL = os.environ.get('KAFKA_BROKER_URL', 'localhost:9092')
broker = KafkaBroker(KAFKA_BROKER_URL)
app = FastStream(broker)
@broker.subscriber('test-topic')
async def handle_message(msg: str, logger: Logger):
logger.info(f"Received message: {msg}")
@app.on_startup
async def app_startup():
logger = Logger().bind(service="startup")
logger.info("FastStream application started")
# Example: publish a message on startup
await broker.publish("Hello, FastStream!", topic='test-topic')
@app.on_shutdown
async def app_shutdown():
logger = Logger().bind(service="shutdown")
logger.info("FastStream application stopped")
# To run this application:
# 1. Ensure Kafka is running.
# 2. Save as `app.py`.
# 3. `pip install "faststream[kafka]"`
# 4. `faststream run app:app`
faststream --version
Debug
Known issues
breakingVersion 0.7.0 (release candidate available) removes several deprecated features. Specifically: publisher/subscriber-level middlewares, `ack_policy` now replaces several deprecated options, `RedisJSONMessageParser` is removed (Redis services must now use binary message format), and `broker.close` is replaced by `broker.stop`.fixReview your application for usage of deprecated features. For Redis, ensure your message formats are binary. Replace `broker.close()` with `broker.stop()`.
affects: <0.7.0 to 0.7.0+
breakingAs of 0.7.0rc0 (and likely stable 0.7.0), import paths for specific brokers have changed. They are now directly under `faststream.<broker_name>` (e.g., `from faststream.kafka import KafkaBroker`) instead of `faststream.broker.<broker_name>`.fixUpdate all broker import statements to the new direct path, e.g., `from faststream.kafka import KafkaBroker`.
affects: <0.7.0 to 0.7.0+
gotchaFastStream's core installation (`pip install faststream`) does not include broker-specific client libraries. You must install FastStream with the appropriate extras, e.g., `pip install "faststream[kafka]"`, `"faststream[redis]"`, `"faststream[mqtt]"`, etc. Failing to do so will result in `ModuleNotFoundError` when trying to import or use a broker.fixAlways install FastStream with the necessary broker extras for your chosen messaging queue (e.g., `pip install "faststream[kafka,redis]"` for both).
affects: all
breakingVersion 0.6.0 introduced significant changes to the Middleware API and Router API. Applications upgrading from versions prior to 0.6.0 may experience breaking changes related to how middlewares are defined and applied, and how routers are configured.fixConsult the official FastStream 0.6.0 release notes and documentation regarding the new Middleware and Router APIs. Adjust your code to conform to the finalized API designs.
affects: <0.6.0 to 0.6.0+
Errors
Common errors & fixes
ModuleNotFoundError: No module named 'your_module_name'
This error often occurs when `faststream run` or other CLI commands cannot locate your application module or a module imported within it. This can be due to an incorrect `--app-dir` or issues with Python's path, especially when running from a different working directory.
fixEnsure your application's directory is correctly added to `PYTHONPATH` or use the `--app-dir` argument in the FastStream CLI to specify the directory containing your application. For example, if your app is in `my_app/main.py`, run `faststream run --app-dir my_app main:app` or set `PYTHONPATH=./my_app` before running `faststream run main:app`.
AttributeError: 'NatsMessage' object has no attribute 'metadata'
Users encounter this when trying to access specific metadata attributes (like `num_delivered` or `timestamp`) directly on broker-specific message objects (e.g., `NatsMessage`, `KafkaMessage`) that might not expose these properties directly or have them under a different access pattern in FastStream.
fixAccess message-specific metadata or properties through the appropriate FastStream-provided methods or attributes, which might abstract away the native client's message structure. For NATS, you might need to access `msg.raw_message.metadata` or use context variables, depending on the specific property needed. Refer to the FastStream documentation for the specific broker's message object for correct attribute access.
ValidationError: 1 validation error for YourModelName Input should be a valid string
This Pydantic `ValidationError` indicates that an incoming message, which FastStream attempts to deserialize into a Python object based on your type hints, does not conform to the expected Pydantic model or field type. For instance, expecting a `str` but receiving `bytes` or malformed JSON.
fixReview the Pydantic model or type hint in your subscriber function signature to match the actual incoming message format. Ensure the producer sends data in a format (e.g., JSON) that can be correctly parsed into your Pydantic model. If raw bytes are expected, update your type hint accordingly or implement custom deserialization.
FastStream(broker=broker) is no longer valid. You should always pass the broker as a separate positional argument, like FastStream(brokers)
This 'error' is a result of a breaking change introduced in FastStream 0.6.x (or prior versions leading up to it), where the `broker` argument to the `FastStream` application constructor became positional-only. Passing it as a keyword argument (`broker=broker`) is no longer supported.
fixUpdate your `FastStream` application initialization to pass the broker as a positional argument. Change `app = FastStream(broker=my_broker)` to `app = FastStream(my_broker)`.
Error: The specified application 'app:app' could not be loaded. Is it a valid Python path?
The `faststream run` command cannot find the specified application object (`app`) within the specified module (`app.py`), or the file itself is not found.
fixEnsure your FastStream application is defined as 'app' in an 'app.py' file, or adjust the command to match your file and object names (e.g., `faststream run main:my_app`). Also, check for syntax errors in your application file.
Upgrade
Version history
0.7.5latest on PyPI · released Aug 27, 2026
Audit
Dependencies
aiokafkaoptionalRequired for Kafka integration via `faststream[kafka]`
pydanticrequiredUsed for data validation and serialization in messages