aiokafka-foundation-kit for AI agents¶

One page holding everything a coding assistant needs to configure and drive aiokafka-foundation-kit correctly, plus a map of where the rest of the documentation keeps the details it leaves out. Give an agent this page rather than the whole site.

How to read this page¶

Every page of this site is also served as raw Markdown at its own URL with .md in place of the trailing slash — this page is /agents.md, the configuration guide is /guide/configuration.md — so anything the map below points at can be fetched as plain text rather than scraped out of HTML. The Copy page control at the top of a page does the same thing for a human with a chat window open. The one exception is the API reference: its Markdown is a list of instructions to a docstring renderer rather than the API, so it carries neither the control nor a .md twin — read it as HTML, or read the docstrings in the source.

Top to bottom before writing code. Rules that hold or break the code is the section correctness lives in — those are the things the library will not save you from. Every name used below is in the public API; if you need something not listed here, fetch the page the documentation map points at rather than guessing a method that sounds plausible.

Scope¶

It does turn a settings object into a configured AIOKafkaProducer or AIOKafkaConsumer, own the start/stop of that client as an async context manager, create topics idempotently before a producer starts, probe whether a cluster answers, serialise message values as JSON, and hand the same wiring to dishka, dependency-injector and OpenTelemetry.

It does not wrap aiokafka. Once a client is yielded you are holding an AIOKafkaProducer / AIOKafkaConsumer and every send, poll, commit, seek and rebalance callback is aiokafka's own API — read aiokafka's documentation for those. There is no publisher class, no consumer runner, no message router, no retry policy, no dead-letter handling, no outbox, no schema registry, no Prometheus metrics and no synchronous variant. It runs no loop of its own: consumer_lifecycle subscribes, and the async for over the consumer is yours to write.

Mental model¶

Four nouns:

• Settings — any object with the right attributes. Every entry point takes a typing.Protocol, not a base class, so a Pydantic model, a dataclass, a SimpleNamespace or a MagicMock all work. The models extra ships Pydantic implementations with the defaults below; nothing checks a hand-rolled object, so a missing attribute surfaces as AttributeError when the client is built.
• Factory — create_async_kafka_producer(settings) / create_async_kafka_consumer(settings, topics) read the settings and return a client that has not been started. Both go through build_kafka_common_config(settings), which is the one place connection, SASL and TLS keys are assembled.
• Lifecycle — producer_lifecycle / consumer_lifecycle are async context managers around managed_kafka_client: start, then in a try/finally the optional on_started hook, the yield, the stop and the optional on_stopped hook. producer_lifecycle can also run ensure_topics_async first.
• Topics — TopicConfig is a frozen dataclass describing one topic; ensure_topics_async(topics, settings) creates each one, reads the broker's per-topic reply, and treats an existing topic as success.

The contrib packages are three thin wrappers over that: contrib.di for dishka providers, contrib.dependency_injector for declarative containers, contrib.telemetry for one call to the OpenTelemetry aiokafka instrumentor. None of them is imported by aiokafka_foundation_kit/__init__.py; contrib/__init__.py exports nothing, so every contrib name is an explicit import from its own submodule and raises ImportError naming the extra when its dependency is missing.

Wiring¶

import asyncio

from aiokafka_foundation_kit import (
    TopicConfig,
    consumer_lifecycle,
    producer_lifecycle,
)
from aiokafka_foundation_kit.contrib.models import (
    BaseKafkaConsumerSettings,
    BaseKafkaProducerSettings,
)

producer_settings = BaseKafkaProducerSettings(bootstrap_servers="localhost:9092")
consumer_settings = BaseKafkaConsumerSettings(
    bootstrap_servers="localhost:9092",
    group_id="order-processor",
)

topics = [TopicConfig(name="orders", num_partitions=6, replication_factor=1)]


async def produce() -> None:
    async with producer_lifecycle(
        producer_settings,
        topics=topics,
        auto_create_topics=True,        # both arguments are needed; neither alone does anything
    ) as producer:
        await producer.send_and_wait("orders", {"id": 42})   # dict -> JSON bytes


async def consume() -> None:
    async with consumer_lifecycle(consumer_settings, topics=("orders",)) as consumer:
        async for message in consumer:                       # the loop is yours
            await handle(message.value)                      # already decoded from JSON
            await consumer.commit()                          # enable_auto_commit is False


asyncio.run(produce())

Everything happens inside a running loop on purpose: aiokafka's constructors take the running loop, so building a client at import time raises RuntimeError.

The API¶

The root package¶

from aiokafka_foundation_kit import …

Also at the root, for annotations only: KafkaSettingsProtocol, ProducerSettingsProtocol, ConsumerSettingsProtocol, TopicConfigProtocol.

topics on producer_lifecycle is a Sequence[TopicConfigProtocol]; topics on consumer_lifecycle is a tuple[str, ...] of names. on_started and on_stopped are async callables taking the client. name appears only in log messages and matters when one process runs several clients.

Public, but not at the root¶

managed_kafka_client(client, *, name, on_started=None, on_stopped=None) wraps anything with async start() / stop(); both lifecycles are built on it. dumps_bytes(value) returns bytes unchanged and JSON-encodes everything else; loads_bytes(value) returns None for value is None (a tombstone) and JSON-decodes otherwise. Both use orjson when the orjson extra is installed and the standard library json when it is not.

Settings protocols¶

Structural, so any object with these attributes qualifies. The hierarchy:

KafkaConnectionSettingsProtocol   bootstrap_servers, client_id, security_protocol,
│                                 metadata_max_age_ms, get_sasl_password()
├── KafkaSaslSettingsProtocol     + sasl_mechanism, sasl_username
└── KafkaSslSettingsProtocol      + ssl_cafile, ssl_certfile, ssl_keyfile, ssl_check_hostname

KafkaSettingsProtocol             = SASL + SSL, the base every entry point accepts
├── ProducerSettingsProtocol      + acks, compression_type, enable_idempotence,
│   │                               max_batch_size, linger_ms, request_timeout_ms
│   └── ProducerLifecycleSettingsProtocol   + auto_create_topics
└── ConsumerSettingsProtocol      + group_id, auto_offset_reset, enable_auto_commit,
                                    session_timeout_ms, heartbeat_interval_ms,
                                    max_poll_records, max_poll_interval_ms,
                                    fetch_max_wait_ms, fetch_min_bytes, fetch_max_bytes

Note that the password is a method, get_sasl_password() -> str | None, not an attribute. Protocol attributes are invariant, so a hand-rolled settings class has to use the Literal aliases themselves: security_protocol: str does not satisfy security_protocol: KafkaSecurityProtocol, and the same goes for acks, compression_type, sasl_mechanism and auto_offset_reset.

TopicConfigProtocol is name, num_partitions, replication_factor, replica_assignment, topic_configs, declared read-only so that a frozen dataclass — the kit's own TopicConfig among them — type-checks where the protocol is expected. A config carrying a replica_assignment is created from it: the assignment fixes both the partition count and the replication factor, so the two count fields may only repeat what it says or be -1 — see rule 15.

Pydantic settings — contrib.models¶

pip install "aiokafka-foundation-kit[models]". These are plain BaseModels, so extra keyword arguments are ignored rather than rejected, and bootstrap_servers (and group_id) are the only required fields.

BaseKafkaSettings — connection, SASL and TLS:

Two validators run after construction: a SASL_* protocol requires sasl_mechanism, sasl_username and a non-empty sasl_password; SSL and SASL_SSL require ssl_cafile. Either failure is a pydantic.ValidationError.

BaseKafkaProducerSettings = BaseKafkaSettings + KafkaAutoCreateMixin:

BaseKafkaConsumerSettings = BaseKafkaSettings +:

BaseKafkaInfraSettings describes a topic namespace rather than a connection: topic_prefix: str | None = None, topic_catalog: dict[str, KafkaTopicSettings] | None = None, consumer_subscriptions: list[str] | None = None. KafkaTopicSettings is num_partitions=3, replication_factor=3, topic_configs=None. normalize_kafka_topic_prefix_value(value) is the prefix coercion on its own: None, "", whitespace, "None" and "null" in any case all become None, and so does any non-string.

The mixins are exported separately for composing your own model: KafkaConnectionMixin, KafkaSaslMixin, KafkaSslMixin, KafkaAutoCreateMixin. The same composition is the place for fields the library does not read. Neither model carries an observability flag such as metrics_enabled: the library collects no metrics, so there is nothing for it to switch. Put it in a mixin of your own and inherit that in both your producer and your consumer subclass; one Protocol then describes both. Set model_config = ConfigDict(extra="forbid") on those subclasses if an unknown key should be a ValidationError rather than silently dropped.

dishka — contrib.di¶

pip install "aiokafka-foundation-kit[dishka]". Three Provider subclasses, all Scope.APP, all raising ImportError from __init__ when dishka is absent.

AsyncKafkaProducerProvider is the only place settings.auto_create_topics is read. KafkaInfraProvider turns topic_catalog into TopicConfig objects and consumer_subscriptions into names, applying topic_prefix as f"{prefix}.{name}". The infrastructure protocols — KafkaInfraBaseSettingsProtocol, KafkaProducerInfraSettingsProtocol, KafkaConsumerInfraSettingsProtocol, KafkaTopicSettingsProtocol — are exported for the settings side.

Adding KafkaInfraProvider to a container is therefore enough to satisfy the topics argument of both client providers; see rule 10 for a container without it.

dependency-injector — contrib.dependency_injector¶

pip install "aiokafka-foundation-kit[dependency-injector]".

Override the dependencies, await container.init_resources(), then producer = await container.producer(); await container.shutdown_resources() stops the client. See rule 11 — kafka_settings is the one override every container needs.

OpenTelemetry — contrib.telemetry¶

pip install "aiokafka-foundation-kit[telemetry]". instrument_aiokafka(*, tracer_provider=None, async_produce_hook=None, async_consume_hook=None, **kwargs) calls AIOKafkaInstrumentor().instrument(...), which patches aiokafka globally — call it once at startup, before any client is built. Only the arguments you pass are forwarded, so omitting tracer_provider leaves the global provider in place. async_produce_hook is async (span, args, kwargs), async_consume_hook is async (span, record, args, kwargs). Extra **kwargs go straight to the instrumentor.

Rules that hold or break the code¶

1. Build clients inside a running event loop. create_async_kafka_producer, create_async_kafka_consumer and both lifecycles construct aiokafka objects, and aiokafka's constructor takes the running loop. Called at import time or from synchronous module scope it raises RuntimeError: The object should be created within an async function or provide loop directly.
2. TLS reaches aiokafka as one ssl_context key. aiokafka takes a prepared ssl.SSLContext and nothing else, so for SSL and SASL_SSL build_kafka_common_config loads ssl_cafile, ssl_certfile and ssl_keyfile through aiokafka.helpers.create_ssl_context and returns ssl_context; the four ssl_* keys never appear in the dict. The files are read at that moment, so a missing or unreadable one raises FileNotFoundError (a malformed one ssl.SSLError) from the factory rather than at connect time, and with no ssl_cafile the system trust store is used. ssl_check_hostname=False sets check_hostname = False on the context — the certificate is still verified against the CA, only the hostname is not. SASL_PLAINTEXT and PLAINTEXT build no context at all.
3. producer_lifecycle does not read settings.auto_create_topics. It has its own keyword, defaulting to False, and creates topics only when auto_create_topics=True and topics is non-empty. Passing one without the other silently creates nothing. The settings field is honoured only by AsyncKafkaProducerProvider and by whatever you pass to the dependency-injector container.
4. enable_auto_commit is False by default on BaseKafkaConsumerSettings. Nothing commits for you: call await consumer.commit() after processing, or the group re-reads from the last committed offset on every restart and rebalance. The factory forwards the flag as-is, so a settings object that omits it fails with AttributeError, not a default.
5. enable_idempotence=True (the default) allows only acks="all". aiokafka raises ValueError: Invalid ACKS parameter for "0" or "1". The KafkaAcks alias offers three values; two of them require enable_idempotence=False.
6. Only gzip compression works out of the box. snappy, lz4 and zstd need cramjam, which this package does not depend on — without it aiokafka raises RuntimeError: Compression library for lz4 not found at construction. Install aiokafka[lz4] (or [snappy], [zstd]) alongside.
7. Values are JSON, keys are bytes. The producer factory installs value_serializer=dumps_bytes, so a dict is encoded for you and bytes pass through untouched. It installs no key_serializer, so key= must already be bytes or aiokafka raises. On the consumer side value_deserializer=loads_bytes means message.value is a decoded object; a topic carrying Avro, protobuf or plain text needs value_deserializer=lambda raw: raw or the decode raises inside aiokafka's fetcher.
8. An on_started hook that raises aborts the context. The hook runs inside the try/finally, so the client is stopped and the hook's exception comes out of the async with statement itself — the body never runs, and nothing is left connected.
9. Only KafkaError is swallowed on stop. managed_kafka_client catches KafkaError from stop(), logs it, and skips on_stopped; anything else — OSError, asyncio.CancelledError, a TimeoutError — propagates out of the context manager and can mask the body's own exception.
10. dishka resolves by exact type, and the kit's providers ask for optionals. AsyncKafkaProducerProvider wants Sequence[TopicConfigProtocol] | None and AsyncKafkaConsumerProvider wants tuple[str, ...] | None. Python's = None default is not a dishka default, so something in the container has to provide those exact types or the build fails with GraphMissingFactoryError. KafkaInfraProvider provides both; without it, write the one-line factory yourself (see Common mistakes).
11. kafka_settings is the only override a container requires. It is a providers.Dependency with no default, so leaving it out raises Error: Dependency "…kafka_settings" is not defined the moment the resource is resolved. topics and auto_create_topics are providers.Object, so they already hold None and False; override them only when you have something to pass. (providers.Dependency(default=None) would not be a default — dependency-injector reads None as "unset" — which is why these are Object.)
12. default_partitions and default_replication_factor are validated and never used. Nothing in the library reads them; the shape of a created topic comes from TopicConfig (where num_partitions and replication_factor are both required) or from KafkaTopicSettings in a topic_catalog. The validator that demands default_replication_factor when auto_create_topics=True is the only thing they do.
13. The consumer lifecycle subscribes; it does not consume. You write the async for, and your per-batch work has to finish inside max_poll_interval_ms (300 s by default) or the broker evicts the member and the partitions rebalance under you.
14. ensure_topics_async tolerates exactly one failure. Each topic is created on its own, and the broker's answer for it is read: an existing topic is logged at DEBUG and ignored, every other refusal is raised as the aiokafka error for that error code — InvalidReplicationFactorError for a replication_factor above the broker count, InvalidTopicError for an unusable name — and aborts the rest of the sequence. It also never changes an existing topic: partition counts and topic_configs of a topic that already exists are left alone.
15. A replica_assignment decides the shape of the topic. Give one and it fixes both the partition count and the replication factor; num_partitions and replication_factor may then only repeat what it says (num_partitions == len(replica_assignment), replication_factor == the length of each replica list) or be -1, which is how aiokafka spells "the assignment decides". Anything else is a ValueError naming the topic, raised before the request is sent. An empty dict counts as no assignment.
16. check_kafka_health_async opens a real producer connection, and its timeout_seconds becomes request_timeout_ms, not an asyncio.timeout around the call — a DNS or TCP stall can outlast it. It returns False (and logs a traceback at ERROR) for TimeoutError, KafkaError and OSError, and lets anything else out. It is a probe, not a cheap one; do not call it per request.
17. A topic_prefix joins with a dot — prefix + "." + name — and only KafkaInfraProvider applies it. TopicConfig.name and the topic names you hand consumer_lifecycle are physical names, already prefixed.
18. The protocols are structural and unenforced. Nothing validates a hand-rolled settings object; a typo or a missing attribute is an AttributeError at client construction, and get_sasl_password must be a callable, not a string.

Common mistakes¶

# WRONG — a class the library has never had, imported from the root
from aiokafka_foundation_kit import AsyncKafkaPublisher, BaseKafkaProducerSettings

# RIGHT — settings live in contrib.models, and the producer is aiokafka's own
from aiokafka_foundation_kit import producer_lifecycle
from aiokafka_foundation_kit.contrib.models import BaseKafkaProducerSettings

async with producer_lifecycle(settings) as producer:
    await producer.send_and_wait("orders", {"id": 42})

# WRONG — built at import time, outside any loop
producer = create_async_kafka_producer(settings)   # RuntimeError

# RIGHT — inside a coroutine, and let the context manager own start/stop
async def main() -> None:
    async with producer_lifecycle(settings) as producer:
        ...

# WRONG — topics without the flag; nothing is created, and send fails on a
# cluster with auto.create.topics.enable=false
async with producer_lifecycle(settings, topics=topics) as producer:
    ...

# RIGHT — both, every time
async with producer_lifecycle(settings, topics=topics, auto_create_topics=True) as producer:
    ...

# WRONG — at-least-once assumed, never delivered: enable_auto_commit is False
async with consumer_lifecycle(settings, topics=("orders",)) as consumer:
    async for message in consumer:
        await handle(message.value)

# RIGHT — commit after the work, so a crash replays rather than skips
async with consumer_lifecycle(settings, topics=("orders",)) as consumer:
    async for message in consumer:
        await handle(message.value)
        await consumer.commit()

# WRONG — a str key, and acks the idempotent producer will not accept
settings = BaseKafkaProducerSettings(bootstrap_servers="localhost:9092", acks="1")
await producer.send_and_wait("orders", {"id": 42}, key="order-42")

# RIGHT — bytes for the key; turn idempotence off if you really want acks="1"
settings = BaseKafkaProducerSettings(bootstrap_servers="localhost:9092")
await producer.send_and_wait("orders", {"id": 42}, key=b"order-42")

# WRONG — nothing provides Sequence[TopicConfigProtocol] | None, and the `= None`
# on the provider's parameter is not a dishka default; the container refuses to build
container = make_async_container(SettingsProvider(), AsyncKafkaProducerProvider())

# RIGHT — KafkaInfraProvider provides that exact type from the topic catalog
container = make_async_container(SettingsProvider(), KafkaInfraProvider(),
                                 AsyncKafkaProducerProvider())

# ALSO RIGHT — no catalog, so provide the type yourself
class TopicsProvider(Provider):
    scope = Scope.APP

    @provide
    def producer_topics(self) -> Sequence[TopicConfigProtocol] | None:
        return None

container = make_async_container(SettingsProvider(), TopicsProvider(),
                                 AsyncKafkaProducerProvider())

Errors¶

The library defines no exception class of its own. What you will see:

Documentation map¶

Fetch a page when the task is the one named beside it.