skills.rest
ricardolui
ricardoluiCOMMUNITY
Explore all skills by this author →
GitHub

developing-kafka-python-client

Generates Python Kafka producer and consumer projects with Schema Registry integration.

Updated Jul 7, 2026
Search Tags:#python#code generation#kafka#producer#consumer#confluent#schema registry
One-click install
npx skills add https://github.com/ricardolui/gcp-custom-agent-skills --skill developing-kafka-python-client-ricardolui
Or copy as Structured Prompt for Agent▼
Please help me install this Agent Skill.
Skill: developing-kafka-python-client
Source: https://github.com/ricardolui/gcp-custom-agent-skills/tree/main/developing-kafka-python-client
Command: npx skills add https://github.com/ricardolui/gcp-custom-agent-skills --skill developing-kafka-python-client-ricardolui

SYSTEM DOCUMENTATION & REQUIREMENTS

💡 This Skill requires confluent-kafka, requests, python-dotenv, and includes references (resource) components.

What problem does it solve? Building a correct Kafka client in Python involves many subtle decisions: choosing between async and sync producers, wiring up Schema Registry serialization, configuring authentication for different environments, and avoiding common pitfalls like per-message producer creation or lost messages on shutdown. This Skill scaffolds a complete, best-practice project so you avoid those mistakes. ## Core Features & Use Cases - Project Scaffolding: Generates producer.py, consumer.py, common.py, JSON Schema or Avro schemas, unit tests, .env.example, and requirements.txt tailored to your data fields. - Multi-Environment Support: Targets Confluent Cloud (SASL_SSL), local Docker (PLAINTEXT with docker-compose.yml), or WarpStream with optimized librdkafka settings. - Schema Registry Integration: Registers schemas explicitly, disables auto-registration, and uses JSON Schema by default (Avro/Protobuf for WarpStream's built-in registry). - Migration Path: Adds Schema Registry serialization to existing schemaless producer code instead of discarding it. - Use Case: You need a FastAPI service producing user signup events to Confluent Cloud. The Skill confirms your requirements, then generates an async AIOProducer project with a JSON Schema for your fields, tests, and graceful shutdown handling. ## Quick Start Ask the assistant to build a Python Kafka producer and consumer for your topic, mentioning your target environment and data fields.

Frequently Asked Questions about developing-kafka-python-client

High-intent search queries and answers about installing and using this skill.

FAQPage Schema
How do I build a Python Kafka producer with Schema Registry?▼

Use confluent-kafka-python with JSONSerializer or AsyncJSONSerializer connected to a Schema Registry client. Register your schema explicitly with register_schema(), set auto.register.schemas to False, and reuse a single producer instance across all messages.

Should I use AIOProducer or the synchronous Producer in Python?▼

Use AIOProducer when your code runs under an asyncio event loop, such as FastAPI, aiohttp, or Sanic applications. Use the synchronous Producer for scripts, batch jobs, and ETL pipelines where you control threads and can call poll() and flush() directly.

Does confluent-kafka-python support Avro and Protobuf serialization?▼

Yes, it provides AvroSerializer and ProtobufSerializer alongside JSONSerializer, in both sync and async variants. Note their constructor signatures differ: JSON takes schema_str first while Avro and Protobuf take schema_registry_client first, so always pass keyword arguments.

Can I use JSON Schema with WarpStream's built-in schema registry?▼

No, WarpStream's built-in schema registry only supports Avro and Protobuf. Use AvroSerializer with a .avsc schema file when targeting WarpStream SR, or point to a different registry like Confluent Cloud Schema Registry to keep using JSON Schema.

Why does AIOProducer raise NotImplementedError when I pass headers?▼

AIOProducer does not support custom headers in batch mode, so produce() raises NotImplementedError if headers are passed. Schema identification for async producers relies on the serializer's wire-format prefix instead of the confluent.value.schemaId header used by sync producers.

How do I run Kafka locally with Docker for Python development?▼

Use the confluentinc/confluent-local image with its built-in PLAINTEXT, PLAINTEXT_HOST, and CONTROLLER listeners, plus a cp-schema-registry container. Configure your Python client with PLAINTEXT protocol and no authentication, then run docker compose up -d.

Related Skills

Explore More Software Engineering Skills→
confluentincconfluentinc
42

developing-kafka-python-client

Create a Python Kafka client project with Schema Registry and JSON Schema validation.

Official
Intermediate
infantesromeroadrianinfantesromeroadrian

production

Deploy ML models to production with FastAPI, Docker, and Kubernetes templates.

Community
Advanced
machenjiemachenjie
4

contract-testing

Enforce executable contract evidence across API, schema, event, and integration boundaries.

Community
Advanced
vanstinatorvanstinator

golang-popular-libraries

Recommend production-ready Go libraries with a standard-library-first approach.

Community
Intermediate