用途：该技能生成一个可直接运行的 Python Kafka 项目，实现向某个 Kafka 主题生产消息并从该主题消费，使用 confluent-kafka-python，并预置 Schema Registry（JSON Schema）、优雅关闭、幂等生产者、基于 header 的模式标识以及测试。目标环境与关键词通过 $ARGUMENTS 传入。
主要能力与工作流：先从已接入的 Kafka MCP（如 Lenses、Confluent、Aiven 或自定义服务）发现目标主题、分区数、副本因子、已注册模式和可用的消费者组；若没有 Kafka MCP 或发现无结果，则退回向用户询问缺失信息。发现后必须先向用户复述确认（硬性关卡），确认后才生成文件。生成的项目结构包括 producer.py、consumer.py、common.py、schemas/value.schema.json、tests/test_project.py、.env.example、pyproject.toml 与 README.md。随后用 pytest 运行测试并修复代码；再执行验证关卡，调用 kafka-topic-audit 审计主题、kafka-perf-review 审查生成的生产者与消费者代码。
使用方式：依赖用 uv 管理，通过 uv sync 创建环境，uv run 执行命令，例如 uv sync 与 uv run pytest tests/。默认生产者使用同步 Producer，若检测到异步运行时（如 asyncio、FastAPI、Quart 等）则改用 AIOProducer；消费者始终使用异步 AIOConsumer。模式固定为 JSON Schema，且必须接入 Schema Registry。配置写入 .env.example，如 KAFKA_ENV、BOOTSTRAP_SERVER、TOPIC、SCHEMA_REGISTRY_URL、CLIENT_ID、GROUP_ID。
限制：资料未提供具体收费、价格或平台支持范围；若未接入 Kafka MCP，则无法验证模式与主题注册是否一致，需要用户补充信息。生成代码要求每个进程一个生产者、幂等生产者（enable.idempotence=true、acks=all）、优雅关闭、使用 kwargs 构造序列化器等，具体细节见资料说明。