用途：当用户想构建 Python Kafka 生产者或消费者、为现有 Python 代码添加 Schema Registry、或从裸 JSON 迁移到 schema 时使用。开始时会声明正在使用该技能。
主要能力：生成生产级 Python 项目，支持三种环境（Confluent Cloud 托管、本地 Docker 开源 Kafka、WarpStream 对象存储后端）和两种生产者风格（AsyncIO 非阻塞、同步阻塞）。默认始终集成 Schema Registry：Confluent Cloud 与本地 Docker 用 JSON Schema；WarpStream 内置 SR 仅支持 Avro 和 Protobuf，默认 Avro。消费者始终使用异步 AIOConsumer。生成结构含 producer.py、consumer.py、common.py、schemas/、tests/、.env.example、requirements.txt，本地 Docker 另含 docker-compose.yml。
使用方式：先收集需求（是否已有应用、目标环境、生产者/消费者、异步或同步、是否已有 schema、数据字段、topic、消费组），必须发送确认摘要并等待用户回复后才生成文件。已有代码时只做迁移与增量修改，不重建项目。schema 注册为独立显式步骤，auto.register.schemas=False、use.latest.version=True。异步生产者退出前需 await flush() 与 close()，同步生产者需 flush()，消费者需 unsubscribe() 后 close()。
限制：不得读取或显示 .env 文件，只生成占位模板。AIOProducer.produce() 不支持 headers。JSONSerializer 与 AvroSerializer 构造参数顺序不同，需用关键字参数。librdkafka 中 message.max.bytes 为客户端全局配置，消费者构造时要求 fetch.max.bytes >= message.max.bytes。WarpStream 需应用 references/warpstream-optimization.md 中的 librdkafka 覆盖项。具体价格、平台细节与效果资料暂未提供。