aiokafka provides asyncio-native Kafka producer and consumer. pip install aiokafka. Producer: from aiokafka import AIOKafkaProducer; producer = AIOKafkaProducer(bootstrap_servers="localhost:9092"); await producer.start(); await producer.send_and_wait("topic", b"hello"). Consumer: from aiokafka import AIOKafkaConsumer; consumer = AIOKafkaConsumer("topic", bootstrap_servers="localhost:9092"); await consumer.start(); async for msg in consumer: print(msg.value). Group: AIOKafkaConsumer("topic", group_id="workers", bootstrap_servers="..."). Key: await producer.send("topic", key=b"k1", value=b"v1"). Serializer: value_serializer=lambda v: json.dumps(v).encode(). Deserializer: value_deserializer=lambda v: json.loads(v). key_serializer, key_deserializer. auto_offset_reset="earliest". enable_auto_commit=False. Manual commit: await consumer.commit(). Seek: consumer.seek(TopicPartition("t",0), 0). Partitions: consumer.assignment(). Batch: producer.create_batch(). Multiple topics: AIOKafkaConsumer("t1","t2",...). Stop: await producer.stop(); await consumer.stop(). Context manager: async with producer: .... SSL: pass ssl_context param. SASL: sasl_mechanism="PLAIN". compression_type="gzip". acks="all". max_batch_size. Claude Code generates aiokafka stream processors, event bus producers, and consumer group workers.
CLAUDE.md for aiokafka
## aiokafka Stack
- Version: aiokafka >= 0.10 | pip install aiokafka
- Producer: async with AIOKafkaProducer(bootstrap_servers="host:9092") as p: await p.send_and_wait("topic", val)
- Consumer: async with AIOKafkaConsumer("topic", group_id="g", bootstrap_servers="...") as c: async for msg in c: ...
- Serializer: value_serializer=lambda v: json.dumps(v).encode()
- Commit: enable_auto_commit=False; await consumer.commit() after processingaiokafka Kafka Streaming Pipeline
# app/kafka_stream.py — aiokafka producer, consumer, batch, and FastAPI event bus
from __future__ import annotations
import asyncio
import json
import logging
from contextlib import asynccontextmanager
from dataclasses import asdict, dataclass
from typing import Any, AsyncGenerator, Callable
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer, TopicPartition
from aiokafka.errors import KafkaConnectionError, TopicAuthorizationFailedError
log = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# 1. Producer helpers
# ─────────────────────────────────────────────────────────────────────────────
def make_producer(
bootstrap_servers: str = "localhost:9092",
acks: str = "all",
compression_type: str | None = "gzip",
max_batch_size: int = 16384,
linger_ms: int = 0,
request_timeout_ms: int = 30_000,
) -> AIOKafkaProducer:
"""
Create an AIOKafkaProducer with JSON serialization.
acks="all": wait for all in-sync replicas (safest).
linger_ms: batch messages arriving within X ms together.
compression_type: "gzip" | "snappy" | "lz4" | "zstd" | None.
"""
return AIOKafkaProducer(
bootstrap_servers=bootstrap_servers,
value_serializer=lambda v: json.dumps(v).encode(),
key_serializer=lambda k: k.encode() if isinstance(k, str) else k,
acks=acks,
compression_type=compression_type,
max_batch_size=max_batch_size,
linger_ms=linger_ms,
request_timeout_ms=request_timeout_ms,
)
@asynccontextmanager
async def producer_ctx(
bootstrap_servers: str = "localhost:9092",
**kwargs: Any,
) -> AsyncGenerator[AIOKafkaProducer, None]:
"""
Async context manager for a producer — start/stop lifecycle.
Usage:
async with producer_ctx("localhost:9092") as p:
await send(p, "events", {"type": "user.created", "id": 42})
"""
p = make_producer(bootstrap_servers, **kwargs)
await p.start()
try:
yield p
finally:
await p.stop()
async def send(
producer: AIOKafkaProducer,
topic: str,
value: Any,
key: str | bytes | None = None,
partition: int | None = None,
headers: list[tuple[str, bytes]] | None = None,
wait: bool = True,
) -> Any:
"""
Send a message. If wait=True, awaits broker acknowledgement.
Returns RecordMetadata.
Example:
meta = await send(p, "orders", {"order_id": 123, "status": "paid"}, key="order-123")
print(meta.offset)
"""
kwargs: dict = {}
if partition is not None:
kwargs["partition"] = partition
if headers:
kwargs["headers"] = headers
if wait:
return await producer.send_and_wait(topic, value=value, key=key, **kwargs)
else:
return await producer.send(topic, value=value, key=key, **kwargs)
async def send_batch(
producer: AIOKafkaProducer,
topic: str,
messages: list[tuple[str | None, Any]],
) -> None:
"""
Send a batch of (key, value) pairs efficiently.
Falls back to individual sends if batch API not available.
Example:
await send_batch(p, "logs", [("svc-a", {"msg": "ok"}), ("svc-b", {"msg": "err"})])
"""
for key, value in messages:
await producer.send(topic, value=value, key=key)
await producer.flush()
# ─────────────────────────────────────────────────────────────────────────────
# 2. Consumer helpers
# ─────────────────────────────────────────────────────────────────────────────
def make_consumer(
*topics: str,
bootstrap_servers: str = "localhost:9092",
group_id: str | None = None,
auto_offset_reset: str = "latest",
enable_auto_commit: bool = True,
auto_commit_interval_ms: int = 5000,
session_timeout_ms: int = 30_000,
heartbeat_interval_ms: int = 3000,
max_poll_records: int = 500,
) -> AIOKafkaConsumer:
"""
Create an AIOKafkaConsumer with JSON deserialization.
group_id: consumer group for parallel processing across instances.
auto_offset_reset: "latest" (new messages only) | "earliest" (all messages).
enable_auto_commit=False: control offset commits manually.
"""
return AIOKafkaConsumer(
*topics,
bootstrap_servers=bootstrap_servers,
group_id=group_id,
value_deserializer=lambda v: json.loads(v),
key_deserializer=lambda k: k.decode() if k else None,
auto_offset_reset=auto_offset_reset,
enable_auto_commit=enable_auto_commit,
auto_commit_interval_ms=auto_commit_interval_ms,
session_timeout_ms=session_timeout_ms,
heartbeat_interval_ms=heartbeat_interval_ms,
max_poll_records=max_poll_records,
)
@asynccontextmanager
async def consumer_ctx(
*topics: str,
bootstrap_servers: str = "localhost:9092",
group_id: str | None = None,
auto_offset_reset: str = "latest",
**kwargs: Any,
) -> AsyncGenerator[AIOKafkaConsumer, None]:
"""
Async context manager for a consumer — start/stop lifecycle.
Usage:
async with consumer_ctx("events", group_id="processors") as c:
async for msg in c:
process(msg.value)
"""
c = make_consumer(
*topics,
bootstrap_servers=bootstrap_servers,
group_id=group_id,
auto_offset_reset=auto_offset_reset,
**kwargs,
)
await c.start()
try:
yield c
finally:
await c.stop()
async def consume_loop(
consumer: AIOKafkaConsumer,
handler: Callable[[Any, str | None], None],
batch_commit: bool = False,
poll_timeout_ms: int = 1000,
) -> None:
"""
Process messages in a loop.
handler(value, key) — called for each message.
batch_commit=True: commit offsets after each batch poll.
Example:
async def my_handler(value, key):
await db.insert(value)
await consume_loop(consumer, my_handler)
"""
async for msg in consumer:
try:
if asyncio.iscoroutinefunction(handler):
await handler(msg.value, msg.key)
else:
handler(msg.value, msg.key)
except Exception as e:
log.error("Handler error for msg at offset %s: %s", msg.offset, e)
continue
if batch_commit:
await consumer.commit()
# ─────────────────────────────────────────────────────────────────────────────
# 3. Event bus pattern
# ─────────────────────────────────────────────────────────────────────────────
class KafkaEventBus:
"""
Simple async event bus backed by Kafka.
Publish events as JSON; subscribe with handler callbacks.
Usage:
bus = KafkaEventBus("localhost:9092")
await bus.start()
await bus.publish("user.created", {"id": 42, "name": "Alice"})
await bus.consume_forever("user.created", handler, group_id="auth-svc")
await bus.stop()
"""
def __init__(self, bootstrap_servers: str = "localhost:9092") -> None:
self._servers = bootstrap_servers
self._producer: AIOKafkaProducer | None = None
async def start(self) -> None:
self._producer = make_producer(self._servers)
await self._producer.start()
async def stop(self) -> None:
if self._producer:
await self._producer.stop()
async def publish(
self,
topic: str,
payload: dict,
key: str | None = None,
) -> None:
if self._producer is None:
raise RuntimeError("Call start() first")
await send(self._producer, topic, payload, key=key)
async def consume_forever(
self,
topic: str,
handler: Callable[[dict, str | None], None],
group_id: str = "default",
auto_offset_reset: str = "earliest",
) -> None:
"""
Consume messages from topic indefinitely.
Stops when cancelled (e.g. asyncio.CancelledError).
"""
async with consumer_ctx(topic, bootstrap_servers=self._servers,
group_id=group_id,
auto_offset_reset=auto_offset_reset) as c:
await consume_loop(c, handler)
# ─────────────────────────────────────────────────────────────────────────────
# 4. FastAPI lifespan integration
# ─────────────────────────────────────────────────────────────────────────────
FASTAPI_EXAMPLE = '''
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.kafka_stream import KafkaEventBus
bus = KafkaEventBus("localhost:9092")
@asynccontextmanager
async def lifespan(app: FastAPI):
await bus.start()
yield
await bus.stop()
app = FastAPI(lifespan=lifespan)
@app.post("/orders/")
async def create_order(order: dict):
order_id = 42 # from DB
await bus.publish("orders.created", {**order, "id": order_id})
return {"id": order_id}
# Background consumer task (start separately):
# asyncio.create_task(bus.consume_forever("orders.created", process_order, group_id="fulfillment"))
'''
# ─────────────────────────────────────────────────────────────────────────────
# Demo (smoke test without Kafka — shows API only)
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
async def demo():
print("aiokafka API demo (no Kafka broker needed for this output)")
print("=== Producer setup ===")
p = make_producer("localhost:9092")
print(f" Producer created: {type(p).__name__}")
print("\n=== Consumer setup ===")
c = make_consumer("events", bootstrap_servers="localhost:9092",
group_id="demo-group")
print(f" Consumer created: {type(c).__name__}")
print("\n=== KafkaEventBus ===")
bus = KafkaEventBus("localhost:9092")
print(f" Bus created: {type(bus).__name__}")
print("\nTo test with real Kafka:")
print(" docker run -d -p 9092:9092 apache/kafka:3.7.0")
asyncio.run(demo())For the kafka-python alternative — kafka-python is the original pure-Python Kafka client using a synchronous threading model; aiokafka is built on asyncio giving async for msg in consumer iteration and await producer.send_and_wait() — it integrates naturally with FastAPI, aiohttp, and asyncio-based microservices without needing additional threading. For the confluent-kafka-python alternative — confluent-kafka uses the high-performance librdkafka C library for maximum throughput (>1M msgs/sec) with a callback-based API; aiokafka is pure Python and integrates seamlessly with asyncio coroutines — use aiokafka for asyncio microservices, confluent-kafka when you need production-grade throughput and official Confluent support. The Claude Skills 360 bundle includes aiokafka skill sets covering make_producer() with acks/compression/linger, producer_ctx()/consumer_ctx() async context managers, send()/send_batch() helpers, make_consumer() with group_id/auto_offset_reset, consume_loop() with error handling, KafkaEventBus publish/consume pattern, FastAPI lifespan integration, manual offset commit, and SSL/SASL configuration notes. Start with the free tier to try async Kafka streaming code generation.