Feature: unified codec configuration

[ce1ebrimbor]

Problem

Encoding and decoding use two different patterns in FastStream. Decoding is a function passed to the broker constructor (decoder=), called inside the handler DI chain with no middleware hook. Encoding is done through middleware via publish_scope. There is no decode_scope equivalent.

broker = KafkaBroker(
    "localhost:9092",
    decoder=my_avro_decoder,                   # function, wired into DI
    middlewares=[MyAvroEncoderMiddleware(...)],  # middleware class, wired into publish_scope
)

This means a developer working with a binary format like Confluent Avro or Protobuf has to implement the same serialization concern in two different places, with two different lifecycles, often sharing state awkwardly between them.

The workaround today is to decode in on_consume middleware and set a passthrough decoder= to prevent the default JSON decoder from re-processing the already-decoded body. It works, but requires understanding FastStream internals.

Proposal

Add a codec= parameter to the broker that handles both directions through a single object:

broker = KafkaBroker(
    "localhost:9092",
    codec=ConfluentAvroCodec(schema_registry_url="http://localhost:8081"),
)

Codec protocol

@runtime_checkable
class Codec(Protocol):
    async def decode(self, msg: StreamMessage[Any]) -> DecodedMessage: ...
    async def encode(self, cmd: PublishCommand) -> PublishCommand: ...

Both methods are async to support codecs that fetch schemas or perform IO during serialization.

Internal wiring

When codec= is provided, FastStream:

• Wires codec.decode as the decoder function, slotting into the existing _get_parser_and_decoder chain
• Auto-generates a middleware that calls codec.encode in publish_scope

No changes to the DI chain, middleware lifecycle, or parser composition. The codec is syntactic sugar over existing extension points.

Deprecation of decoder=

The codec= parameter replaces decoder=. Once codec is available, decoder= should be deprecated with a warning pointing users to codec= instead. The decoder= parameter would remain functional for a deprecation cycle but be removed in a future major version.

If both codec= and decoder= are provided, raise ValueError — the two are mutually exclusive to avoid ambiguous decode paths.

Example

codec = ConfluentAvroCodec(
    schema_registry_url="http://localhost:8081",
    topics_versions={"orders.events": "latest"},
)
broker = KafkaBroker("localhost:9092", codec=codec)

@broker.subscriber("orders.events")
async def handle_order(msg: OrderEvent) -> None:
    ...

await broker.publish(OrderEvent(...), topic="orders.events")

One object, one configuration site, both directions.

P.S. If the maintainers consider this a valuable addition, I'm happy to implement it and open a PR.

[Lancetnik]

@ce1ebrimbor thank you! It is a good suggestion and I was thinking about it many times - the main problem is in design and backward compatibility. If you have any ideas - I would like to discuss the design in PR
One last note: probably we should cover custom parsers by this class

[ce1ebrimbor]

Silly question @Lancetnik ^^
What do you mean by custom parser ?
Is there any difference between parser and decoder in this case ?

[Lancetnik]

Is there any difference between parser and decoder in this case ?

Yes. Parser is an entity, that maps incoming message to FastStream object. Decoder serializes body

• https://faststream.ag2.ai/latest/getting-started/serialization/parser/
• https://faststream.ag2.ai/latest/getting-started/serialization/decoder/

[ce1ebrimbor]

@Lancetnik
I've been going back and forth between two approaches and I'd like to share both before committing to one.

Option A: Add decode_scope / encode_scope to BaseMiddleware

This extends the existing middleware contract with two new hooks:

class BaseMiddleware:
    async def decode_scope(self, call_next, msg: StreamMessage) -> DecodedMessage:
        return await call_next(msg)
    async def encode_scope(self, call_next, msg: SendableMessage) -> tuple[bytes, str | None]:
        return await call_next(msg)

Pros: stays within the existing extension mechanism, no new concepts, composable (encryption middleware could wrap around codec middleware).

Cons: decode/encode is a strategy (you pick one codec), not a cross-cutting concern (you don't layer multiple decoders). Every middleware inherits hooks that only one layer would ever implement. It's the wrong abstraction shape — middleware is for things that compose horizontally (logging + auth + tracing), codecs don't.

Option B: Parser and Codec protocols as broker parameters

broker = KafkaBroker(
    "localhost:9092",
    parser=ConfluentWireParser(),
    codec=AvroCodec(schema_registry_url="http://localhost:8081"),
)

Pros: codec is a first-class concept, not a middleware side-effect. Broker-agnostic (Codec has no broker-specific types, reusable across Kafka/Rabbit/NATS). Clean separation — Parser handles wire framing (broker-specific), Codec handles content serialization (broker-agnostic). Backward compatible with existing parser=/decoder= functions.

Cons: introduces two new protocol interfaces. Requires wiring changes inside the producer (where encode_message is currently hardcoded) and the subscriber (where decoder runs inside the DI chain).

Where I'm leaning

I think the protocol approach is the right long-term design — it models what a codec actually is (a strategy you configure once) rather than forcing it into a middleware shape. But I wanted to surface the middleware option since it's a smaller delta to the existing codebase and might be a pragmatic first step.
Whichever pattern we choose for the codec, the parser would follow the same approach — either both are middleware scopes, or both are protocol interfaces. Keeping them consistent avoids a split where one extension point is a middleware hook and the other is a protocol object.
Would love to hear your thoughts on which direction makes more sense for the project.

[Lancetnik]

@ce1ebrimbor I like new codec and parser protocols - do you want to try to introduce it?

[ce1ebrimbor]

@ce1ebrimbor I like new codec and parser protocols - do you want to try to introduce it?

@Lancetnik yes !
Getting back with a PR this week 🙏