🚀 New issue to ag2ai/faststream by @ce1ebrimbor
📝 Feature: Support broker-level ack_policy default with per-subscriber override (#2826)


Problem

Currently ack_policy can only be set per-subscriber:

@broker.subscriber("orders.topic", ack_policy=AckPolicy.NACK_ON_ERROR)
async def handle_order(msg: OrderMessage) -> None:
    ...

There is no way to set a default ack_policy at the broker (or router) level. In practice, most services want the same policy across all subscribers — typically NACK_ON_ERROR for at-least-once delivery with DLQ. This forces every @broker.subscriber() call to repeat the same ack_policy= argument, which is error-prone: forgetting it on a single subscriber silently falls back to ACK_FIRST (at-most-once), which can cause silent message loss.

Proposed solution

Add an ack_policy parameter to KafkaBroker (and equivalently to RabbitBroker, NatsBroker, etc.) that sets the default for all subscribers registered on that broker. Individual subscribers can still override it.

broker = KafkaBroker("localhost:9092", ack_policy=AckPolicy.NACK_ON_ERROR)

# Inherits NACK_ON_ERROR from broker
@broker.subscriber("orders.topic")
async def handle_order(msg: OrderMessage) -> None:
    ...

# Overrides to ACK for this specific subscriber
@broker.subscriber("notifications.topic", ack_policy=AckPolicy.ACK)
async def handle_notification(msg: NotificationMessage) -> None:
    ...

The resolution order would be: subscriber-level > broker-level > built-in default (ACK_FIRST).

Why this matters
• Safety: ACK_FIRST as a silent default is dangerous for services that use DLQ or need at-least-once delivery. A broker-level default lets teams enforce their delivery guarantee in one place.
• DRY: Services with 10+ subscribers shouldn't need to repeat ack_policy=AckPolicy.NACK_ON_ERROR on every one.
• Consistency with other broker-level settings: decoder, middlewares, security, and logger are all broker-level defaults that subscribers inherit. ack_policy is the notable exception.

#enhancement #good_first_issue #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @BrikozO
📝 Bug: (#2842)



#bug #good_first_issue #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @zueve
📝 Bug: (#2868)


Describe the bug
Can't forward client_rack to aiokafka consumer through FastStream Kafka API.

How to reproduce
Include source code:

from faststream import FastStream

from faststream.kafka import KafkaBroker

# Attempt to configure at broker level

broker = KafkaBroker(
    "localhost:9092",
    client_rack="us-east-1a",  # not supported
)
app = FastStream(broker)

And/Or steps to reproduce the behaviour:

1. Create a KafkaBroker.
2. Try to pass client_rack to the broker constructor.
3. Observe that the parameter is not accepted.
4. There is no alternative way to configure it via subscriber(...) either.

Expected behavior
KafkaBroker should allow passing client_rack (or generic consumer kwargs) and forward it to the underlying aiokafka.AIOKafkaConsumer.

Observed behavior
client_rack is not supported in KafkaBroker constructor, nor in subscriber configuration, so it cannot be passed to aiokafka.

Screenshots
N/A.

Environment
Running FastStream 0.6.7 with CPython 3.14.5 on Linux

Additional context
aiokafka.AIOKafkaConsumer supports client_rack since 0.14.0 version


#bug #good_first_issue #aiokafka #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @dzeveloper
📝 Feature: Consumer-Only Mode for KafkaBroker (#2879)


When deploying a subscriber-only service, it's common to provision Kafka credentials scoped strictly to READ + DESCRIBE ACLs on specific topics. Today, KafkaBroker always starts a producer during broker.start(), regardless of whether any @broker.publisher is registered. This causes the producer to attempt a connection using consumer-only credentials, which fails or raises errors at the broker/ACL level.

Proposal:

# Option A: broker-level flag
broker = KafkaBroker("localhost:9092", consumer_only=True)

# Option B: skip producer if no publishers are registered (automatic)
# No API change needed — broker inspects registered publishers at start()

#enhancement #good_first_issue #confluent #aiokafka #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @DeoLeung
📝 Feature: relax typer version (#2969)


if no incompatibility, it will be great to relax the cap of typer, easy for projects using faststream cli to upgrade typer


#enhancement #good_first_issue #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @sobolevn
📝 Consider using `syrupy` for tests (#2978)


Link: https://pypi.org/project/syrupy/

Example usage:

• https://github.com/wemake-services/django-modern-rest/blob/master/tests/test_unit/test_plugins/test_msgspec/test_msgspec_snapshots.py
• https://github.com/wemake-services/django-modern-rest/blob/master/tests/test_unit/test_plugins/test_msgspec/__snapshots__/test_msgspec_snapshots.ambr

#good_first_issue #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @IvanKirpichnikov
📝 Feature: Add `RedisBatchStreamMessage` annotated alias (#3004)


Add to faststream/redis/annotations.py the alias for RedisBatchStreamMessage is similar to the aliases for RedisMessage, RedisChannelMessage, etc.


#enhancement #good_first_issue #redis #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @Lancetnik
📝 Feature: `StreamSub(declare=False)` to opt out of Redis Stream creation (#3013)


Is your feature request related to a problem? Please describe.

FastStream creates the Redis Stream for you whenever a subscriber uses a consumer group — mkstream=True is hardcoded and there is no way to turn it off:

faststream/faststream/redis/subscriber/usecases/stream_subscriber.py

Lines 133 to 138 in 2e36197

That is a fine default, but it means a typo in a stream name silently provisions a new empty stream instead of failing, and a service that is only supposed to consume a stream provisioned elsewhere has no way to say so. Every other broker in FastStream already has this opt-out — RabbitQueue(declare=False), RabbitExchange(declare=False), JStream(declare=False), KvWatch(declare=False), ObjWatch(declare=False). Redis is the only one missing it. The same knob is being added to Kafka in #1827.

Describe the solution you'd like

A declare: bool = True argument on StreamSub, passed through as mkstream=not declare. Default behaviour does not change.

Feature code example

from faststream import FastStream
from faststream.redis import RedisBroker, StreamSub

broker = RedisBroker()
app = FastStream(broker)


# unchanged default — stream is created if missing
@broker.subscriber(stream=StreamSub("orders", group="g", consumer="c"))
async def owned(msg: str) -> None: ...


# provisioned elsewhere — fail fast instead of creating an empty stream
@broker.subscriber(
    stream=StreamSub("external-events", group="g", consumer="c", declare=False)
)
async def consumed(msg: str) -> None: ...

Semantics

Redis only offers the "check, do not create" form, so declare=False fails when the stream is absent — the same as RabbitMQ's declare=False (AMQP passive=True), and unlike NATS' variant which just skips creation. Verified against redis:alpine:

mkstream=False -> ResponseError: The XGROUP subcommand requires the key to exist.
                  Note that for CREATE you may want to use the MKSTREAM option to
                  create an empty stream automatically.
exists after mkstream=False: 0
mkstream=True  -> exists after: 1

The existing except ResponseError only swallows "already exists", so this error propagates as-is today — which is the wanted behaviour. Wrapping it in a FastStream error with a message pointing at declare=False would be a nice touch, not a requirement.

Implementation notes

1. Add declare: bool = True to StreamSub.__init__ and to __slots__ —

 faststream/faststream/redis/schemas/stream_sub.py

 Lines 44 to 69 in 2e36197

 — plus an Args: entry in the class docstring.
2. Pass mkstream=not stream.declare at the xgroup_create call linked above.
3. declare only has an effect on the consumer-group path — that xgroup_create runs only when group and consumer are both set. Without a group the subscriber uses XREAD, which never creates anything. StreamSub already emits RuntimeWarning for argument combinations that have no effect (no_ack with a group, polling_interval with last_id != ">"); declare=False without a group deserves the same treatment.
4. Tests go in tests/brokers/redis/ — one real-broker case asserting the stream is not created and the error surfaces, one asserting the default still creates it.

Out of scope: publishing. XADD always creates the stream and Redis exposes no flag to prevent it, so there is no publisher-side equivalent.

Additional context

Prior art to copy the naming and docstring style from: RabbitQueue.declare and JStream.declare.


#enhancement #good_first_issue #redis #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @kumaranvpl
📝 Ability to configure topic params in confluent create_topics (#1827)


Current state: every topic FastStream creates is created with num_partitions=1, replication_factor=1, hardcoded:

faststream/faststream/confluent/helpers/admin.py

Lines 52 to 55 in 2e36197

The only knob users have is the broker-wide allow_auto_create_topics flag — all topics of a broker are created, or none of them are. There is no way to configure a single topic, nor to opt a single topic out of creation.

Request: introduce a Topic schema object accepted by subscriber() / publisher() alongside plain strings, carrying per-topic settings:

from faststream.confluent import KafkaBroker, Topic

broker = KafkaBroker()

@broker.subscriber(
    Topic("topic-name", num_partitions=3),
    Topic("topic-name2", num_partitions=1, replication_factor=2),
    Topic("externally-managed", declare=False),
    "topic-without-settings",
)
async def handler(msg: str) -> None: ...

Scope

1.

 num_partitions / replication_factor per topic — the original ask, from https://github.com/airtai/faststream/discussions/1821. Today both are pinned to 1, which makes FastStream-created topics unusable in any real deployment.

2.

 declare: bool = True — opt a single topic out of creation while auto-creation stays the default for everything else. This subsumes #2679: the global default is not changing, because a broker-wide switch is the wrong granularity — a service typically owns some of its topics and consumes others that are provisioned by a different team or by IaC.

Proposed semantics of declare=False: skip the create_topics call for that topic and nothing else — do not probe the cluster for existence, do not fail if the topic is missing. This matches what allow_auto_create_topics=False does today (a warning, then let the consumer proceed) and matches NATS' JStream(declare=False). Note this deliberately differs from RabbitMQ's RabbitQueue(declare=False), which maps to AMQP passive=True and does raise when the queue is absent — Kafka has no cheap equivalent of a passive declare.

Interaction with the broker-level flag: allow_auto_create_topics=False on the broker keeps winning over everything — it stays the "create nothing at all" switch. declare only narrows creation further when the broker-level flag is on.

Consistency across brokers: declare is the established name for this in FastStream — RabbitQueue(declare=...), JStream(declare=...), KvWatch(declare=...), ObjWatch(declare=...). Topic should use the same name rather than inventing a Kafka-specific one.

AioKafka is out of scope. FastStream never creates topics for faststream.kafka — there is no AdminClient.create_topics call on that path, and aiokafka (0.13.0) does not support an allow_auto_create_topics consumer option at all:

$ grep -rn "auto_create\|auto\.create" .venv/lib/python3.11/site-packages/aiokafka/
$ grep -rn "auto_create" faststream/kafka/

Both return nothing. Topic creation on that path is entirely the Kafka server's auto.create.topics.enable, which FastStream cannot influence. Topic may still be accepted there later for symmetry, but declare would be a no-op, so it should not block this issue.

Implementation notes

• create_subscriber() / create_publisher() signatures must accept str | Topic — *topics: str is what currently trips mypy on the branch in progress.
• Normalise str → Topic(name) at registration time, as RabbitMQ does with str → RabbitQueue.
• AsyncConfluentConsumer.topics_to_create should filter on declare —

 faststream/faststream/confluent/helpers/client.py

 Lines 292 to 294 in 2e36197

• Topic needs __hash__ / __eq__ consistent with each other, since topics end up as dict keys (see #2796 for the RabbitMQ precedent).

Related: #2679 (closed in favour of this), #1486, #1658, #2451.


#good_first_issue #confluent #kafka #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @IvanKirpichnikov
📝 Bug: Uncorrect typehint for `partition_assignment_strategy` parameter in `faststream.kafka` (#3019)


Now: partition_assignment_strategy: Sequence["AbstractPartitionAssignor"]
Need: partition_assignment_strategy: Sequence["type[AbstractPartitionAssignor]"]


#bug #good_first_issue #aiokafka #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @Lancetnik
📝 Feature: use aio_pika.Pool to connect RMQ (#975)


For now, FastStream uses a regular RobustConnection and RobustChannel to connect RabbitMQ, but we should support aio-pika connection Pool feature in this case

https://aio-pika.readthedocs.io/en/latest/quick-start.html#connection-pooling

Probably, regular behavior should be the Pool with size 1


#enhancement #good_first_issue #rabbitmq #100 #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @Lancetnik
📝 feature: document RPC responses in AsyncAPI (#1586)



#enhancement #good_first_issue #core #asyncapi #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @sobolevn
📝 Consider using `syrupy` for tests (#2978)


Link: https://pypi.org/project/syrupy/

Example usage:

• https://github.com/wemake-services/django-modern-rest/blob/master/tests/test_unit/test_plugins/test_msgspec/test_msgspec_snapshots.py
• https://github.com/wemake-services/django-modern-rest/blob/master/tests/test_unit/test_plugins/test_msgspec/__snapshots__/test_msgspec_snapshots.ambr

#enhancement #good_first_issue #dependencies #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @kumaranvpl
📝 Ability to configure topic params in confluent create_topics (#1827)


Current state: every topic FastStream creates is created with num_partitions=1, replication_factor=1, hardcoded:

faststream/faststream/confluent/helpers/admin.py

Lines 52 to 55 in 2e36197

The only knob users have is the broker-wide allow_auto_create_topics flag — all topics of a broker are created, or none of them are. There is no way to configure a single topic, nor to opt a single topic out of creation.

Request: introduce a Topic schema object accepted by subscriber() / publisher() alongside plain strings, carrying per-topic settings:

from faststream.confluent import KafkaBroker, Topic

broker = KafkaBroker()

@broker.subscriber(
    Topic("topic-name", num_partitions=3),
    Topic("topic-name2", num_partitions=1, replication_factor=2),
    Topic("externally-managed", declare=False),
    "topic-without-settings",
)
async def handler(msg: str) -> None: ...

Scope

1.

 num_partitions / replication_factor per topic — the original ask, from https://github.com/airtai/faststream/discussions/1821. Today both are pinned to 1, which makes FastStream-created topics unusable in any real deployment.

2.

 declare: bool = True — opt a single topic out of creation while auto-creation stays the default for everything else. This subsumes #2679: the global default is not changing, because a broker-wide switch is the wrong granularity — a service typically owns some of its topics and consumes others that are provisioned by a different team or by IaC.

Proposed semantics of declare=False: skip the create_topics call for that topic and nothing else — do not probe the cluster for existence, do not fail if the topic is missing. This matches what allow_auto_create_topics=False does today (a warning, then let the consumer proceed) and matches NATS' JStream(declare=False). Note this deliberately differs from RabbitMQ's RabbitQueue(declare=False), which maps to AMQP passive=True and does raise when the queue is absent — Kafka has no cheap equivalent of a passive declare.

Interaction with the broker-level flag: allow_auto_create_topics=False on the broker keeps winning over everything — it stays the "create nothing at all" switch. declare only narrows creation further when the broker-level flag is on.

Consistency across brokers: declare is the established name for this in FastStream — RabbitQueue(declare=...), JStream(declare=...), KvWatch(declare=...), ObjWatch(declare=...). Topic should use the same name rather than inventing a Kafka-specific one.

AioKafka is out of scope. FastStream never creates topics for faststream.kafka — there is no AdminClient.create_topics call on that path, and aiokafka (0.13.0) does not support an allow_auto_create_topics consumer option at all:

$ grep -rn "auto_create\|auto\.create" .venv/lib/python3.11/site-packages/aiokafka/
$ grep -rn "auto_create" faststream/kafka/

Both return nothing. Topic creation on that path is entirely the Kafka server's auto.create.topics.enable, which FastStream cannot influence. Topic may still be accepted there later for symmetry, but declare would be a no-op, so it should not block this issue.

Implementation notes

• create_subscriber() / create_publisher() signatures must accept str | Topic — *topics: str is what currently trips mypy on the branch in progress.
• Normalise str → Topic(name) at registration time, as RabbitMQ does with str → RabbitQueue.
• AsyncConfluentConsumer.topics_to_create should filter on declare —

 faststream/faststream/confluent/helpers/client.py

 Lines 292 to 294 in 2e36197

• Topic needs __hash__ / __eq__ consistent with each other, since topics end up as dict keys (see #2796 for the RabbitMQ precedent).

Related: #2679 (closed in favour of this), #1486, #1658, #2451.


#enhancement #good_first_issue #confluent #kafka #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @mahdialibi
📝 DOC - V 0.5 - Error in documentaion (#3047)


in documentation : https://faststream.ag2.ai/0.5/getting-started/observability/logging/#setting-logging-configuration-from-file
is stated :

If you use FastStream CLI, you have the option to use a file to configure your logging of the entire application directly from the command line.
faststream run serve:app --log-file config.json

But this is misleading , this feature is not available in v 0.5


#good_first_issue #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @Lancetnik
📝 Bug: `redis.asyncio.Redis` type hint is not injected and fails validation (#3065)


Describe the bug

Annotating a handler argument with redis.asyncio.Redis — the obvious type for the connection — does not inject it. Instead the argument is treated as a message field and fails validation. The working annotation is faststream.redis.annotations.Redis, which is a different symbol with the same name.

The two names are identical, the import paths differ, and nothing in the failure points at the fix.

How to reproduce

from redis.asyncio import Redis

@broker.subscriber(stream=StreamSub("stream", group="group", consumer="consumer"))
async def handler(msg: RedisStreamMessage, redis: Redis) -> None:
    ...

vs. the version that works:

from faststream.redis.annotations import Redis

Real usage

A worker running faststream[redis] in production hit this and left a warning to their future selves in the handler docstring — SW-Maestro-17th-HBB/Kkori-AI, worker/src/main.py#L186-L191:

redis 는 FastStream 이 Context 로 넣어주는 커넥션이다. redis.asyncio.Redis 를 그대로 힌트로 쓰면 주입되지 않고 검증 오류가 난다.

("redis is the connection FastStream injects via Context. If you use redis.asyncio.Redis directly as the hint, it is not injected and you get a validation error.")

They also import it aliased — from faststream.redis.annotations import Redis as InjectedRedis — which suggests the name collision cost them enough to want it visible at the call site.

Expected behavior

Two things would each be enough on their own:

1. Documentation — the Redis pages show Redis in examples without making the import path explicit enough to survive a copy-paste. A short note that the annotation comes from faststream.redis.annotations, and that the same-named driver class will not work, closes it.
2. A better failure — when an argument is annotated with a broker client class that has a matching FastStream annotation, say so in the error instead of failing validation on a missing message field.

Additional context

Found by reading a real user's source, not reported by them — they worked around it and moved on, which is the reason this kind of papercut stays invisible. Companion finding: #3064.

@IvanKirpichnikov — the docs half is cheap; the error-message half is your call.


#bug #documentation #good_first_issue #redis #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @davorrunje
📝 Docs: dealing with different schema registries (#1297)


It would be good to add documentation with examples how to deal with different schema registries. Again there are many registries and coupling router with a particular registry isn't a good idea, unless there will be a some Abstract class first, so later community can add implementation


#documentation #enhancement #good_first_issue #confluent #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @sobolevn
📝 Consider using `syrupy` for tests (#2978)


Link: https://pypi.org/project/syrupy/

Example usage:

• https://github.com/wemake-services/django-modern-rest/blob/master/tests/test_unit/test_plugins/test_msgspec/test_msgspec_snapshots.py
• https://github.com/wemake-services/django-modern-rest/blob/master/tests/test_unit/test_plugins/test_msgspec/__snapshots__/test_msgspec_snapshots.ambr

#enhancement #good_first_issue #dependencies #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @Lancetnik
📝 Bug: `StreamSub` warns that `no_ack` has no effect with a consumer group, but the flag is forwarded to XREADGROUP NOACK (#3126)


Describe the bug

StreamSub(..., group=..., consumer=..., no_ack=True) emits

RuntimeWarning: `no_ack` has no effect with consumer group

but the flag does have an effect: it reaches Redis as XREADGROUP ... NOACK, so entries never enter the PEL. The warning is a leftover from a time when the flag was not forwarded.

How to reproduce

from faststream.redis import RedisBroker, StreamSub

broker = RedisBroker()


@broker.subscriber(
    stream=StreamSub("my-stream", group="my-group", consumer="c1", no_ack=True),
)
async def handler(msg: str) -> None:
    ...

Constructing the StreamSub prints the warning. Running the app shows that the flag is honoured anyway: XPENDING my-stream my-group stays empty after messages are consumed.

Expected behavior

No warning. no_ack=True with a consumer group is a supported, documented mode ("equivalent to acknowledging the message when it is read", see docs/docs/en/redis/streams/groups.md).

Observed behavior

The warning is emitted in faststream/redis/schemas/stream_sub.py (the elif no_ack: branch under if group and consumer:), while the rest of the code base treats the flag as live:

• faststream/redis/subscriber/usecases/stream_subscriber.py passes noack=stream.no_ack to XREADGROUP;
• faststream/redis/subscriber/config.py switches the subscriber to AckPolicy.MANUAL when no_ack is set;
• faststream/redis/testing.py skips the PEL for no_ack subscribers;
• #3049 added a SetupError for claim_min_idle_time + no_ack, which only makes sense if no_ack works with groups.

History

• The warning was introduced in 0.3.5 (#1048), when no_ack was indeed not forwarded to XREADGROUP.
• #2309 kept the warning and reset no_ack = False right after it, so the flag really was ignored for a while.
• 0.6.0 (#1779) dropped that reset, and noack=stream.no_ack has been sent to Redis since, but the warning stayed.

The sibling warning, "no_ack is not supported by consumer group with last_id other than >", is a separate case and should be checked on its own: Redis documents NOACK for XREADGROUP regardless of the id, so it may be stale too.

Suggested fix

Remove the elif no_ack: warning branch in StreamSub.__init__ and add a test asserting that StreamSub("s", group="g", consumer="c", no_ack=True) raises no warning and that the subscriber calls xreadgroup with noack=True. Keep the claim_min_idle_time + no_ack SetupError as is.

Environment

Reproduced on main (commit 0307f09).


#bug #good_first_issue #redis #faststream #ag2ai
sent via relator

🚀 New issue to ag2ai/faststream by @IvanKirpichnikov
📝 Feature: Implementation of the `AsyncContextManager` interface for `Application` (#3228)


Is your feature request related to a problem? Please describe.
Instead of

await app.start()
try:
    ...
finally:
    await app.stop()

I want to do

async with app:
    ...

Describe the solution you'd like
Implementation of the AsyncContextManager interface for Application

This should work for Application subclassses, i.e., for FastStream and AsgiFastStream.

TestApp already has this capability. It just needs to be adjusted to use __aenter__ and __aexit__.

Feature code example

async with app:
    ...

#enhancement #good_first_issue #faststream #ag2ai
sent via relator