Codecs

A codec turns a payload into bytes and bytes back into a payload, and declares the content type it writes. That content type is what makes this work across languages: a Java producer says application/json, a Python consumer picks the codec that answers for it, and neither end has been told about the other.

mq = await connect(url, codec=JsonCodec())              # the connection's default
mq.publisher("files", "uploaded", codec=BytesCodec())   # one publisher
await mq.consume("orders.dlq", inspect, codec=BytesCodec())  # one consumer

JSON is the default. Nothing has to be configured to get it.

The three that need nothing installed

The five behind an extra

EncryptedCodec is a sixth, and a different shape: it wraps one of the others rather than replacing it. See Encryption.

None of the five answers for a message with no content type. Only JsonCodec does that, and it is the only one that should: JSON is the default format, so an untyped message is far more likely to be JSON than anything else. A YAML codec that volunteered would be worse than useless — YAML parses JSON, so it would return the right value while recording that a YAML message had arrived.

They live one module each, and none of them is imported by acemq_amqp, so the core keeps its promise of depending on nothing:

pip install "acemq-amqp[yaml]"

from acemq_amqp import CompositeCodec, JsonCodec
from acemq_amqp.codecs.yaml import YamlCodec

mq = await connect(url, codec=CompositeCodec(JsonCodec(), YamlCodec()))

Why they exist

Not because Python could not parse YAML. It always could. The gap was that a Java or Go service publishing YAML produced a message this library refused — the bytes were readable and nothing here claimed the content type, so a perfectly good message was parked. Java and Go ship these five as separate optional modules; these are the same five, writing the same content types and accepting the same wider sets, so a message crosses either way.

What each codec reads is deliberately wider than what it writes, because a producer in another stack uses whichever spelling its own library picked. application/yaml was only registered by RFC 9512; application/x-yaml, text/yaml and text/x-yaml all predate it and are what a lot of tooling still emits. Getting that set wrong is the failure these codecs exist to prevent.

The tests are not round trips. tests/fixtures/codec-interop-fixtures.json holds message bodies produced by Java and by Go — by programs calling the same functions those codecs call, at the versions their build files pin — and the suite decodes those. Java and Go turned out to write byte-identical XML and byte-identical Avro; their YAML differs only in list indentation and their TOML only in quote style, which is exactly the difference a decoder has to absorb and a round trip would never have produced.

YAML

Block style, not flow style, which is the whole reason to pick YAML — flow style produces something very close to JSON and leaves nothing to justify the cost. No leading ---: a message body is a single document, so the marker separates nothing from nothing. Keys keep the order the payload was built in.

YamlCodec().encode({"orderId": "o-1", "items": ["widget", "gasket"]})
# b'orderId: o-1\nitems:\n- widget\n- gasket\n'

Loading is always safe loading, and that is not configurable. PyYAML's default loader constructs arbitrary Python objects out of tags like !!python/object/apply, and a message body is untrusted input.

YAML costs more to parse than JSON and is a poor choice for high volume. It earns its place where somebody will actually read the message — a configuration change broadcast to a fleet, a command replayed by hand from a dead-letter queue.

TOML

The top level has to be a mapping. TOML is a table format; a bare list or a bare number is not a TOML document, and encode says so rather than emitting something nothing can read:

TomlCodec().encode(["a", "b"])
# TypeError: ... the top level has to be a mapping ...

Refused at the publisher rather than discovered by the consumer, which is what Java and Go both do. Reading uses tomllib from the standard library on 3.11 and later and tomli — the same parser under its original name — on 3.10; writing has no standard-library answer at all, so the extra brings tomli-w.

XML

No extra, because there would be nothing in it: the codec is written against xml.etree.ElementTree and xml.parsers.expat.

Every document with a DTD is refused, and that is not configurable. A message body arrives from a queue, which is exactly the sort of place a message from somewhere unexpected turns up. Python's position here is better than it is often given credit for and worse than it needs to be:

• External entities are already inert. <!ENTITY x SYSTEM "file:///etc/passwd"> is not resolved by xml.etree; the reference fails as an undefined entity. So plain XXE and DTD retrieval are not live hazards.
• Internal entity expansion is not inert. The billion-laughs attack needs no network and no readable file — a few nested internal entities expand to gigabytes inside the parser — and ElementTree.fromstring expands them happily. That was checked against the interpreter this library is tested on rather than taken from a table.

So rather than disabling the individual hazards, the codec refuses the construct they all need: expat's StartDoctypeDeclHandler, EntityDeclHandler, UnparsedEntityDeclHandler and ExternalEntityRefHandler each raise, and a body carrying <!DOCTYPE is a FatalError whatever the DTD would have said. There is no constructor argument to relax it. defusedxml was considered and not used: what it does is turn these handlers off, and this turns the same handlers off directly, in a way nobody can turn back on.

XML has no types, so everything decodes to a string, a nested dict, or a list where a tag repeats — the same thing Jackson gives when an XML message is read into a Map. A consumer that wants an int converts it, where it can decide what an unparseable field means.

A dataclass is written under its class name, the way Jackson and Go's encoding/xml write theirs; a plain mapping has no name to take, so XmlCodec(root="OrderPlaced") supplies one. The default is message.

Protobuf

A codec is built for one message type, the way it is in Java:

from acemq_amqp.codecs.protobuf import ProtobufCodec
from myapp.orders_pb2 import OrderPlaced

codec = ProtobufCodec(OrderPlaced)

Protobuf bytes carry no name — they are field numbers and wire types — so a reader must already know which message it is holding, or the bytes are not interpretable at all. A destination carrying several message types has to say which is which, and the protobuf answer to that is a wrapper message with a oneof: a decision about the schema, not about the transport.

decode hands back the generated message itself, with its fields typed, rather than a dict.

Avro

There is no schema-free Avro and there cannot be one. Avro's bytes describe nothing about themselves: a reader must already hold the schema the writer used. Two ways to say where it comes from, and the choice matters more than it looks.

from acemq_amqp.codecs.avro import AvroCodec

codec = AvroCodec(schema)                                        # avro/binary
codec = await AvroCodec.from_registry(registry, "order.placed", schema)
codec = await AvroCodec.from_registry(                           # and reads onto
    registry, "order.placed", schema, reader_schema=my_schema
)
codec = AvroCodec.reading(my_schema)                             # consume only

AvroCodec(schema) fixes one schema for the codec's whole life. Small, fast, nothing extra to run — and the writer's schema is whatever the reader happens to have. The moment a producer adds a field, every consumer still holding the old schema reads the new bytes wrongly and Avro will not always notice. Sound only where producer and consumer are released together.

AvroCodec.from_registry(...) writes the schema's identifier into the front of every message, so a reader can look up exactly what the writer used and let Avro resolve it against its own. That is what makes a field addition safe, and it is the mode to use unless there is a reason not to. The framing is one zero byte, four bytes of identifier big-endian, then the Avro body — Confluent's layout, and byte-for-byte the one Java and Go write.

The reader schema

The reader schema is the schema a consumer was written against, and it is what makes evolution mean anything on the read side. With the writer's schema alone a consumer receives whatever the producer sent: a field it has never heard of turns up, and a field it expects is simply missing until the producer starts sending it. Given both schemas, Avro resolves them — the unknown field is read past and discarded, the missing one is filled in from the reader schema's default — so the consumer always sees the shape it was written against, whichever version of the producer wrote the message.

reader_schema is what it is called here, and it is the same word in the rest of the family: Java's readerSchema, Ruby's reader_schema:, Go's ReaderSchema and .NET's ReaderSchema. One idea, one spelling, so a paragraph about it in any of the five reads in all of them.

A service that both publishes and consumes passes it alongside the schema it writes:

codec = await AvroCodec.from_registry(
    registry, "order.placed", schema, reader_schema=my_schema
)

The reader schema is a decoding instruction and nothing else. Only schema is registered and only schema is written, byte for byte, so the wire stays what the Java, Go, Ruby and .NET libraries read.

AvroCodec.reading(my_schema) is that same reader schema for a service that does not publish:

codec = AvroCodec.reading(my_schema)
await codec.learn_from(registry, 1)   # the version still in production
await codec.learn_from(registry, 2)   # and the one being rolled out

It registers nothing, because a service that only consumes writes no schema and has no identifier to frame; encode says so rather than inventing one. It is kept rather than folded into the constructor because it is the only way to say what it says: AvroCodec(my_schema) is a fixed-schema codec that writes avro/binary, and this is a registered codec with no identifier — a combination no constructor argument expresses, since schema_id is what puts a codec in registered mode and a read-only one has none. Java and .NET need no equivalent: their registries are synchronous, so one object holds readerSchema and still looks an identifier up from inside encode, which is why they keep one object for both directions where this library has two.

A change Avro will not resolve — a field whose type changed, a field added without a default — raises FatalError naming the identifier the message arrived with, the reader schema it was being read onto, and Avro's own account of what diverged. The alternative is a record of silent nonsense, which is the failure this whole mode exists to prevent.

Where this library lands against the other four, and the bytes both behaviours are pinned to, are in schema resolution.

Each mode claims only its own framing type. The two are not interchangeable and the difference is invisible in the bytes: a framed message begins with five bytes a fixed-schema codec would read as the first field. That does not throw — Avro decodes the shifted bytes into whatever they happen to mean — so a codec that accepted the other framing would hand back a record full of silent nonsense.

The content type decides the framing; the leading zero byte is a last resort. The rule, which all five libraries follow:

• a content type naming Avro — avro/binary, application/avro, anything ending +avro — is believed, and the body is read as a fixed-schema one whatever its first byte is. A legitimate Avro body starts with a zero byte whenever its first field encodes to one: an empty string does, and so do 0, false and the first branch of a union. Refusing those would be refusing real messages over a guess the sender already answered.
• application/vnd.acemq.avro is refused by a fixed-schema codec, because reading five bytes of schema identifier as the first field produces a record whose every value is wrong, silently.
• only when the content type is absent, or names something that is not Avro at all — application/octet-stream is as uninformative as silence — does the zero byte get a vote, and then it refuses rather than guesses.

The registry here is async and a codec is not. Java's registry is a synchronous interface and Go's takes a context, so both can look a schema up from inside encode. SchemaRegistry in this library is a set of coroutines, and encode cannot await one — it runs on the publisher's hot path and, in the sync facade, on a worker thread with no loop. So resolution is lifted out of the message path and done once, explicitly:

codec = await AvroCodec.from_registry(registry, "order.placed", schema)
await codec.learn_from(registry, 7)   # a writer version this consumer will meet

A message carrying an identifier the codec has not been taught raises FatalError naming the identifier rather than guessing.

JSON

Dataclasses are encoded through dataclasses.asdict, so a payload written as a dataclass goes on the wire as an object with its field names:

@dataclass
class OrderPlaced:
    order_id: str
    total_cents: int


await orders.send(OrderPlaced("o-1", 4250))
# {"order_id": "o-1", "total_cents": 4250}

Everything else is handed to json unchanged, which covers dicts, lists and the scalars. A payload json cannot carry raises TypeError naming the type, and nothing is published.

Decoding produces dicts and lists, not your class. The wire says nothing about which class was meant, and a library that guessed would be guessing on every message. An application that wants its own type back constructs it from what it got, where it can decide what a missing field means:

async def ship(message: Message) -> Ack:
    order = OrderPlaced(**message.payload)

It answers for a message with no content type. JSON is the default format, so an untyped message is far more likely to be JSON than anything else, and something has to read it. That is the rule the .NET library uses too.

A malformed body is fatal, not retryable. A body that is not JSON is not JSON the next time either, so JsonCodec.decode raises FatalError — and a decode failure never reaches the retry path anyway: the consumer parks it in {queue}.parked before the handler runs.

Text

For messages that really are text — a line of a log, a command somebody typed — rather than a structure that happens to be readable. Prefer JSON for anything with fields.

It deliberately does not answer for a message with no content type. There is no reason to think an untyped message is text, and a codec that decodes anything into a string never fails, so it would quietly hand a handler the printable rendering of a body it should have refused.

Bytes

Passes bodies through untouched. For a payload that is already encoded — an image, something another system's serializer produced — and for reading a message this process has no type for.

It is what to read a dead-letter queue with. The message that went there may be exactly the one nothing could decode, and a codec that fails on it would park it a second time. Replaying uses it for the same reason: the bytes that were committed are the bytes that should go back, and re-encoding through a class that has since gained a field would produce something else.

It answers for every content type, which is why it has to be asked for rather than found. In a CompositeCodec it must go last, or nothing after it is ever reached.

Several formats on one queue

from acemq_amqp import CompositeCodec, JsonCodec, TextCodec

mq = await connect(url, codec=CompositeCodec(JsonCodec(), TextCodec()))

The first codec is what it writes. All of them are offered a message to read, in the order given, and the first that claims the content type and succeeds wins.

With no content type nothing can be ruled out, so every codec is a candidate rather than the first being assumed right. A message that claims one format and is written in another is still read, because a candidate that refuses says nothing about the next one. Only when every candidate has refused is anything raised — and it carries all the reasons rather than whichever came last:

acemq: no codec could read this message. Tried JsonCodec: acemq: this message is
not JSON: ...; TextCodec: acemq: this message is not UTF-8 text: ...

That is the shape a queue needs during a migration, and the shape it needs where several producers were written at different times.

Schema resolution

Handed the writer's schema and the reader's, Avro reconciles the two: a field the writer added that the reader does not declare is skipped, and a field the writer omitted is filled in from the reader's default. Handed only the writer's, there is nothing to reconcile, and the record arrives in the shape it was written.

So there is one rule, and it is the same rule in all five AceMQ libraries:

Resolution happens when the library has a reader schema to resolve onto.

What differs between the languages is where a reader schema comes from, and therefore how often there is one. Nothing about the bytes differs.

This is not an inconsistency waiting to be flattened. A library that resolves and a library that does not are both right about the same bytes — they are answering different questions, because only one of them was told what the reader expects.

The case that bites is a field the writer removed that the reader declares with a default. With resolution, the field arrives carrying that default. Without it, the field is simply absent: a missing key, whatever the language calls one. A consumer written against the reader schema then reads a value that was never on the wire, or fails to read a field it is sure it declared, and which of those happens is decided entirely by whether a reader schema was in play.

The other direction is the one people expect to be dangerous and is not. A field the writer added that the reader does not declare is skipped under resolution and present without it, and either way the fields the reader does declare come back correct — the unknown field does not shift the ones after it.

Both cases are pinned, with the bytes, in tests/fixtures/avro-resolution-fixtures.json, which every AceMQ library carries a copy of. It records the decoded value under each behaviour, as resolved and writerShape, and which library lands on which.

Asking for resolution in Python

There is nothing to ask for. A codec here cannot be built without a schema, so it always has a reader schema and every message is resolved onto it:

# Resolves onto the schema it was built with, which is the only kind there is.
codec = await AvroCodec.from_registry(registry, "order.placed", my_schema)

# Resolves onto a different one: publish the version you write, read onto yours.
codec = await AvroCodec.from_registry(
    registry, "order.placed", schema, reader_schema=my_schema
)

# The same reader schema for a service that only consumes.
codec = AvroCodec.reading(my_schema)

reader_schema= does not switch resolution on — it says which schema to resolve onto. Left out, that is the schema being written, so a consumer reading its own producer's messages gets the shape it registered; given, the two are separated, which is what a service that publishes one version and consumes another needs.

The writerShape column is reachable from here, and only by saying so: build the codec with the writer's schema as its reader schema and resolution has nothing left to do, so the record arrives as it was written. That is worth knowing mainly because it is what a consumer accidentally does when it pins itself to the producer's current schema — it is how an estate quietly loses the defaults it thought it had. tests/test_avro_resolution.py asserts both columns against the shared fixture for exactly that reason.

One edge worth knowing before you meet it: a codec resolves only against writer schemas it has been taught. A message framing an identifier that has not reached learn_from(registry, id) raises FatalError naming the identifier rather than guessing at a schema, and a change Avro will not resolve at all — a field whose type changed, a field added without a default — raises FatalError naming both schemas. Neither is a resolution that silently did not happen, which is the failure mode this section exists to rule out.

Writing one

Codec is a runtime-checkable Protocol, so there is nothing to inherit from. Four members:

from typing import Any

import msgpack   # for the sake of the example


class MsgPackCodec:
    @property
    def content_type(self) -> str:
        return "application/msgpack"

    def encode(self, payload: Any) -> bytes:
        return msgpack.packb(payload)

    def decode(self, body: bytes, content_type: str | None = None) -> Any:
        try:
            return msgpack.unpackb(body)
        except Exception as failure:
            raise FatalError(f"not msgpack: {failure}") from failure

    def can_decode(self, content_type: str | None) -> bool:
        return content_type is not None and "msgpack" in content_type.lower()

Two rules worth following, because the rest of the library relies on them:

• Raise FatalError for a body this codec will never read. Anything else is treated as a transient failure by the code that catches it, and a body that is not msgpack is not msgpack the next time either.
• Be honest in can_decode. A codec that answers True for everything makes a CompositeCodec stop at it.

decode takes the content type as a defaulted argument rather than needing a second interface for it. That is where this departs from Go: Go's Decode takes a pointer, because that is how every Go decoder works, and it then needs something else to get the content type to a composite codec that has to choose.

The registry

from acemq_amqp import codec_by_name, codec_names, register_codec

register_codec("msgpack", MsgPackCodec)

codec_names()                    # ['bytes', 'json', 'msgpack', 'text']
codec = codec_by_name(config.codec)

So that configuration can name a format without the code that reads the configuration importing every format it might name. json, bytes and text are registered by the library, and text also answers to string — the name .NET and Ruby use — so configuration shared across languages works in all of them; importing acemq_amqp.codecs.yaml, .toml or .xml adds yaml, toml and xml, which is how Go's init() does it.

protobuf and avro are deliberately not registrable. Neither format's bytes describe themselves, so a codec needs a message type or a schema before it can read anything, and a no-argument factory has nothing to hand back.

register_codec takes a factory, not an instance, so codec_by_name returns a new one each time. Registering the same name twice replaces the first, which is what lets a test override a default. codec_by_name raises KeyError listing the names it does know.

Encryption

EncryptedCodec wraps another codec and encrypts what it produced, so the broker, its disk, its backups and everybody who can read its management interface hold ciphertext.

pip install "acemq-amqp[crypto]"

from acemq_amqp import JsonCodec, connect
from acemq_amqp.codecs.encrypted import EncryptedCodec, Keyring, generate_key

keyring = Keyring.of("2026-01", generate_key())
mq = await connect(url, codec=EncryptedCodec(JsonCodec(), keyring))

It wraps a delegate rather than serialising anything itself, so choosing a format and choosing to encrypt stay independent: JSON in, AES-GCM out, and Avro just as well. The content type on the wire is application/vnd.acemq.encrypted — deliberately not ...+json, whatever the plaintext underneath is, because a +json suffix would make every JSON-aware consumer volunteer to parse ciphertext.

What is on the wire

0xAE  0x01  len  key identifier   12-byte nonce   ciphertext + 16-byte tag

The key identifier travels in the clear. That is the point: a consumer reads which key a message needs rather than assuming the current one, so a key can be rotated while messages written with the old one are still queued. key_id_of() reads it back from the bytes alone, which is what an operator staring at a dead-letter queue they can no longer read actually wants — and it needs no key to do it.

The header is authenticated but not encrypted: GCM binds all ten-or-so bytes of it as associated data, so a key identifier altered in flight makes the message fail to open rather than quietly opening as something else.

Rotation

keyring.add(EncryptionKey("2026-02", generate_key()))   # every consumer first
keyring.use("2026-02")                                  # then one publisher

That order, always. A keyring holding one key cannot rotate without an outage.

Interoperability

All five AceMQ libraries write application/vnd.acemq.encrypted, and since the 0.5.0 round they all write the same bytes underneath it:

A body written by any of the five opens in any of the other four, given the key — a .NET producer and a Python consumer share a queue without either of them knowing about the other. tests/test_encrypted.py holds the vector that pins it, a known key and a known nonce and a known plaintext with the exact bytes written out, and the same vector is in all five test suites, so a library that drifts off the framing fails its own tests rather than somebody's dead-letter queue.

.NET used to be the exception and no longer is. Up to its own 0.3.0 it wrote AES-256-CBC with a separate HMAC-SHA-256 — sound cryptography under a content type that promised the family framing — and it moved to AES-GCM in this framing in the same round Go dropped the magic-less variant of its own. .NET still reads its old bodies, so a queue filled before that change can be drained by the library that filled it; nothing writes that framing any more, and no other library has ever read it.

This library reads one framing and writes the same one. A body that does not begin 0xAE is refused as what it is — an error naming the framing, not a decryption failure — so a consumer pointed at a plaintext queue, or at bodies one of those retired framings left behind, is told what has actually happened rather than being left to suspect its keys.

What it does not do

The broker can no longer read the message, and neither can the people who operate it. Decide what they do instead before turning this on: a dead-letter queue full of ciphertext is a queue nobody can triage.

Encryption is not authorisation — every service holding the keyring reads every message, so separate audiences mean separate keys — and it is not a signature: anybody holding a key can write a message this codec will decrypt without complaint. It does not hide the routing either. Exchange, routing key, headers and message size stay in the clear, and for many systems the routing key is the sensitive part.

Nothing in the module ever puts a plaintext, a key, or any part of either into an exception, a log line or a repr, and a failure to decrypt says the same thing whether the key was wrong or the bytes were altered. Both are pinned by tests.

What is not here

Compression. A codec that compresses is a codec; it wraps another one and needs nothing from this library that is not on this page.

A payload too large for a broker is also a codec's problem, and it does ship: ClaimCheckCodec wraps another codec, sends anything over a threshold to a store, and puts the key on the wire instead. See the claim check.

Schema evolution is mostly a different problem from serialization and has its own answer: see the schema registry. The half of it that is a codec's problem does ship, and ships switched on: an AvroCodec cannot be built without a schema, so it always has a reader schema and always resolves a writer's schema onto it. reader_schema= and AvroCodec.reading(...) say which schema to resolve onto rather than whether to. See the reader schema and schema resolution, which pins the behaviour against the fixture the other four libraries assert against.