kafka-python-client

Kafka Python 客户端脚手架

使用 confluent-kafka-python 生成一个可用于生产环境的 Python 项目，用于向 Kafka 主题生产消息和从中消费消息，并支持 Schema Registry（JSON Schema）、优雅关闭、幂等生产者、基于标头的架构识别和测试。Agent 应通过所连接的任意 Kafka MCP 服务（Lenses MCP、Confluent、Aiven 或其他服务），在询问用户之前，从实时集群中发现目标主题的所有信息（名称、分区数、已注册的架构）。只有在未连接 Kafka MCP 服务或未发现任何信息时，才向用户提问。

目标环境和关键词：$ARGUMENTS

请以以下内容作为第一条回复的开头：“正在运行 kafka-python-client 技能以搭建此项目。”

工作流

复制此清单并跟踪进度：

Scaffold Progress:
- [ ] Step 1: Discover topic + schema via the attached Kafka MCP
- [ ] Step 2: Hard gate - confirm with user before generating
- [ ] Step 3: Generate the project files
- [ ] Step 4: Run pytest against the generated tests
- [ ] Step 5: Run validation gate (kafka-topic-audit + kafka-perf-review)
- [ ] Step 6: Hand back with run instructions

1. 发现主题和架构：通过所连接的任意 Kafka MCP 服务进行发现（参见 references/mcp-discovery.md）
2. 硬性门槛：概述发现结果，并在生成任何内容之前向用户确认
3. 生成项目：使用 references/ 中的模板
4. 运行 pytest：对生成的 tests/ 运行测试，并修复所有失败项（修复代码，不要修改测试）
5. 运行验证门槛：针对目标主题调用 kafka-topic-audit，并针对生成的 producer.py 和 consumer.py 调用 kafka-perf-review
6. 交付：提供运行说明，并附上验证门槛的检查结果

步骤 1：通过所连接的任意 Kafka MCP 服务发现主题和架构

阅读 references/mcp-discovery.md，了解完整的探测流程、特定厂商的工具名称提示和回退方案。大致流程如下：

1. 识别所连接的 Kafka MCP 服务：查看当前会话中可用的服务。常见服务包括：mcp__Lenses__*（Lenses MCP——参考实现）、mcp__Confluent__*、mcp__Aiven__*，以及标记为 Kafka 的自定义服务。
2. 发现环境／集群：使用 MCP 提供的工具（Lenses：list_environments；Confluent：list_clusters；其他服务各不相同）。
3. 按用户提示中的关键词搜索候选主题（Lenses：list_datasets(search=...)；原生 Kafka 管理 MCP 服务：调用 list_topics 后进行筛选）。
4. 获取所选主题值的 subject 中注册的架构（Lenses：get_dataset；Confluent：get_schema(subject=<topic>-value)；独立的 Schema Registry MCP 服务：对 /subjects/<topic>-value/versions/latest 发起 HTTP GET 请求）。
5. 读取分区数和副本因子（Lenses：get_topic_metadata；大多数其他服务会提供 describe_topic 或 get_topic 等效工具）。
6. 检查现有消费者组，以建议不冲突的 GROUP_ID（Lenses：list_consumer_groups_by_topic；大多数其他服务：list_consumer_groups）。

如果有多个候选主题与关键词匹配，询问用户要选哪一个，不要猜测。列出这些主题及其分区数和字段名称，以便用户明确选择。

如果未连接 Kafka MCP 服务，或未发现任何信息，则回退到步骤 2 中的硬性门槛，并直接向用户询问缺失的信息。明确说明无法获取哪些信息：“由于未连接 Kafka MCP 服务，我无法验证架构是否与主题中注册的架构相符。”

步骤 1 的预期输出：类似下面的结果概述

Discovered via <MCP server name> (Lenses MCP / Confluent MCP / ...):
- Cluster:        staging
- Topic:          orders.payment.completed
- Partitions:     12
- RF:             1 (single-broker cluster — kafka-topic-audit will flag this)
- Value schema:   JSON Schema, 6 required fields + 1 optional (pulled from registry)
- Suggested group: payments-service-<8-char-random>

步骤 2：硬性门槛——生成前确认

强制要求。 在写入任何文件前，先发送一条消息，其中：

1. 以项目符号列表的形式回顾步骤 1 中发现的值。
2. 列出发现过程无法解答的任何未决问题（例如项目目录，或用户是否特别偏好同步还是异步的生产者风格）。
3. 明确请用户确认或更正。

然后停止并等待用户回复。回顾是为了让用户在写入任何文件前发现误读；即使提示看起来已提供全部信息，这一步也不可省略。

只有在用户已于本次对话中确认过该回顾时，才可以跳过此门槛。

确认后应用以下默认值：

• 生产者风格： 默认使用 confluent_kafka 中的同步 Producer；如果用户表示采用异步运行时，则提供 AIOProducer。异步信号包括：明确提到 asyncio；任何异步优先的 Web 框架（FastAPI、Starlette、Quart、Litestar、BlackSheep、Robyn、aiohttp、Sanic、Tornado）；通常意义上的 ASGI 应用；或异步任务系统（ARQ、Dramatiq、使用异步任务的 Celery）。如果用户提到 Django，默认使用同步 Producer，除非用户明确说明是异步 ASGI。如果不确定，请询问。
• 消费者风格： 始终使用异步 AIOConsumer。轮询循环通常会运行数小时或数天，因此绝不能阻塞事件循环。
• 架构格式： JSON Schema。始终接入 Schema Registry；此技能不会生成无架构的项目。
• 项目目录： ./payments-service/（或根据用户的表述推断）。

步骤 3：生成项目文件

在选定的项目目录中创建以下布局：

<project-dir>/
├── producer.py
├── consumer.py
├── common.py
├── schemas/
│   └── value.schema.json
├── tests/
│   └── test_project.py
├── .env.example
├── pyproject.toml
└── README.md

项目使用 uv 管理 Python 环境和依赖项——而非 pip、venv、pipenv、poetry 或 conda。依赖项在 pyproject.toml 中声明；使用 uv sync 配置环境；通过 uv run 运行命令。生成的文档必须始终遵循这一点。

使用 references/ 中的模板：

• producer-template.py 用于 producer.py
• consumer-template.py 用于 consumer.py
• common-template.py 用于 common.py
• tests-template.py 用于 tests/test_project.py

直接将步骤 1 中获取的架构写入 schemas/value.schema.json——不要重新生成或改述。MCP 发现的核心目的，就是获取真实的架构。

.env.example

尽可能填入 MCP 发现的值。对于本地 Lenses CE 设置：

KAFKA_ENV=local
BOOTSTRAP_SERVER=localhost:9092
TOPIC=<topic from discovery>
SCHEMA_REGISTRY_URL=http://localhost:8081
CLIENT_ID=payments-service
GROUP_ID=<suggested group from discovery>

对于远程托管集群，请将 BOOTSTRAP_SERVER 和 SCHEMA_REGISTRY_URL 替换为 MCP 的数据集／集群元数据调用返回的值，并添加 SASL 凭据占位符。

pyproject.toml

在 PEP 621 的 [project] 表中声明依赖项，以便 uv 解析并锁定依赖项。不要同时生成 requirements.txt——pyproject.toml 是唯一可信来源，uv sync 会在其旁边生成 uv.lock。

[project]
name = "<project-name>"
version = "0.1.0"
description = "Kafka producer and consumer scaffolded by kafka-python-client"
requires-python = ">=3.10"
dependencies = [
    "confluent-kafka[json,schema_registry]>=2.13.2",
    "jsonschema",
    "python-dotenv",
    "requests>=2.25.0",
    "httpx",
    "authlib",
    "cachetools",
    "attrs",
    "typing_extensions",
]

[dependency-groups]
dev = [
    "pytest",
    "pytest-asyncio",
]

[tool.uv]
package = false

由于脚手架是一组扁平的 .py 文件（应用程序，而非可导入的库），请设置 [tool.uv] package = false，让 uv sync 将项目视为应用程序，而不尝试构建和安装它。在这种布局中，单独使用 [build-system] hatchling 配置块会因“无法确定要打包哪些文件”而失败。

producer.py、consumer.py 或 common.py 中导入的每个第三方包都必须出现在 [project].dependencies 中。仅供测试使用的包（pytest、pytest-asyncio）应放在 dev 依赖组中，由 uv sync 自动安装，并通过 uv sync --no-dev 从生产安装中排除。由于消费者始终是异步的，因此请包含 pytest-asyncio。异步 Schema Registry 客户端在运行时会导入 httpx 和 authlib，尽管它们并未声明为 confluent-kafka 依赖项——请将它们明确列出。

可以手动编辑 pyproject.toml，也可以先运行 uv init --bare <project-dir>，再逐步调用 uv add 来生成，例如：

uv init --bare <project-dir>
cd <project-dir>
uv add "confluent-kafka[json,schema_registry]>=2.13.2" jsonschema python-dotenv "requests>=2.25.0" httpx authlib cachetools attrs typing_extensions
uv add --dev pytest pytest-asyncio

README.md

一份简短的 README，包含：前置条件（已安装 uv、Python 3.10+——如果未检测到 Python 解释器，uv 会自动获取受管理的解释器——如果要在本地运行 Kafka，还需安装 Docker）、设置命令（运行 uv sync 创建 .venv 并安装所有依赖）、如果尚未注册 schema，如何注册（链接到 schemas/value.schema.json）、如何使用 uv run 运行生产者和消费者，以及如何使用 uv run pytest 运行测试。README 不得提及 pip、python -m venv、source .venv/bin/activate 或 requirements.txt——这些都由 uv 处理。

生成的代码必须遵循的规则

完整的查找表请阅读 references/common-mistakes.md。简要说明：

1. 每个进程只创建一个生产者。 在 main() 中构造，并将其作为参数传给 produce()。切勿每次调用都创建新的生产者。
2. 将 schema 注册作为独立步骤。 将序列化器配置为 auto.register.schemas=False 和 use.latest.version=True，并通过一个小型 register_schema() 辅助函数发布 schema，该函数直接调用 sr_client.register_schema()，并让任何异常向上抛出。
3. 构造序列化器时使用关键字参数。 在 confluent-kafka-python 中，JSONSerializer、AvroSerializer 和 ProtobufSerializer 的位置参数顺序并不相同。每次调用都通过名称传递 schema_str 和 schema_registry_client，可确保不同格式的调用位置保持一致，并避免出现 TypeError: ... got multiple values for argument 'schema_registry_client'。
4. 启用幂等性的生产者。 enable.idempotence=true、acks=all。
5. 优雅关闭。 同步生产者：在 finally 中调用 flush()。异步消费者：先调用 unsubscribe()，再调用 close()，两者都要等待完成。
6. 使用实体 ID 作为键。 对任何按实体划分的事件流，传入 key=<entity_id>.encode("utf-8")，以保留分区顺序。
7. 运行前验证连接。 使用 AdminClient.list_topics() 检查 broker，并对 schema registry 的 /subjects 发起 HTTP GET 请求。

第 4 步：运行测试

使用 uv 管理环境，并通过 uv run 在环境中执行命令。uv sync 会在 pyproject.toml 同级目录创建 .venv/，解析其中声明的依赖项、生成 uv.lock，并一步安装所有依赖——无需手动运行 python -m venv 或 pip install。

cd <project-dir>
uv sync
uv run pytest tests/

如果用户的机器上没有 uv，请先提供官方安装命令（curl -LsSf https://astral.sh/uv/install.sh | sh），然后再继续——不要改用 pip。

如果有任何测试失败，请修复生成的代码（而非测试），直到所有测试通过。

第 5 步：验证关卡——串联运行 kafka-topic-audit 和 kafka-perf-review

这是此技能与一次性脚手架工具的区别所在。第 4 步完成后，自动运行两个现有技能作为质量关卡。

1. 对目标 topic 运行 kafka-topic-audit。 针对第 1 步中指定的 topic 调用该技能。直接列出所有严重问题（单 broker CE 环境下 RF=1 属于预期情况——请指出，但不要将其视为阻碍）。
2. 对生成的代码运行 kafka-perf-review。 针对生成的项目目录调用该技能。生成的 producer.py 和 consumer.py 应满足以下所有要求：启用 enable.idempotence、设置 acks=all、linger.ms > 0、设置 compression.type、max.poll.interval.ms 合理，并实现优雅关闭。如果审计发现任何问题，请在生成的代码中修复，然后重新运行审计。

第 6 步：交付

给用户的最终消息必须包含：

• 项目目录，以及对生成内容的一句话描述
• 运行生产者和消费者的准确命令（在两个终端中分别运行）
• 对验证关卡发现结果的一段摘要
• 说明 schemas/value.schema.json 中的 schema 来自实时集群，而非推断得出

成功标准

定量指标

• 对 90% 的“构建一个 Python Kafka 客户端”查询作出响应（使用 10-20 种不同措辞进行测试——参见 references/test-cases.md）
• 在总计少于 15 次 MCP 工具调用的情况下完成脚手架搭建
• 每次运行中 MCP 调用失败次数为 0（集群确实无法访问的情况除外）
• 首次运行时，生成的 pytest tests/ 通过率超过 80%，最多进行一次自行修复
• 验证关卡（第 5 步）对生成的代码未发现严重问题

定性指标

• 开始生成前，最多向用户提问一次（即第 2 步的确认），而不是进行多轮盘问
• 生成的架构文件与已注册的架构逐字节一致（不进行改写）
• 生成的代码无需进一步修改，即可针对预置的本地 Lenses CE 端到端运行
• 交付说明完整自足——用户无需再问“好的，我该如何运行？”

示例

示例 1：通过完整 MCP 发现流程从零搭建消费者（已连接 Lenses MCP）

用户说：“帮我为暂存环境 Kafka 中的 payment 事件编写一个 Python 消费者。”

操作：

1. 第 1 步：已连接 Lenses MCP，因此 Agent 使用其工具：list_environments → 选择 staging。list_datasets(search="payment") 返回三个主题。向用户展示这三个主题及其分区数和消息速率；用户选择 orders.payment.completed。get_dataset 获取已注册的 JSON Schema。get_topic_metadata 返回 12 个分区，RF=1。（使用 Confluent 的 MCP 时，对应的调用为 list_clusters → list_topics → get_schema → describe_topic——流程相同，名称不同。参见 references/mcp-discovery.md。）
2. 第 2 步：回顾并确认。用户确认。
3. 第 3 步：生成 producer.py、consumer.py、common.py、schemas/value.schema.json、tests/test_project.py、.env.example、pyproject.toml、README.md。
4. 第 4 步：测试通过。
5. 第 5 步：kafka-topic-audit 标出 RF=1（单代理 CE 上属预期情况——附注后予以接受）。kafka-perf-review 确认生产者启用了幂等性、能够优雅关闭，且 max.poll.interval.ms 设置合理。
6. 第 6 步：交付运行命令和验证摘要。

结果：这是一个贴近生产环境的项目，用户可在五分钟内端到端运行；其架构与主题上已注册的架构完全一致。

示例 2：为现有的异步 Python 服务添加生产者

用户说：“为我的 Quart 服务添加一个 Kafka 生产者，用于 orders.payment.refunded”——同一流程适用于任何异步运行时：FastAPI / Starlette / Litestar / BlackSheep / Robyn / aiohttp / Sanic / Tornado / 纯 asyncio 工作进程 / 基于 ASGI 的服务。

操作：

1. 步骤 1：MCP 发现会解析主题和架构。Agent 检测到异步框架信号（Quart），并选择 AIOProducer 作为步骤 2 的生产者样式。
2. 步骤 2：回顾说明会注明异步生产者样式，识别框架的启动/关闭钩子（Quart：@app.before_serving / @app.after_serving；FastAPI：lifespan 处理器；Starlette：lifespan 处理器；aiohttp：on_startup / on_cleanup；Sanic：@app.before_server_start / @app.before_server_stop；Litestar：on_startup / on_shutdown；纯 asyncio 脚本：在 main() 中实例化，并用 try/finally 包装），并向用户询问应用的根文件。用户确认。
3. 步骤 3：不从头搭建，而是编辑用户现有的应用：如果缺少 common.py，则添加该文件；使用框架的启动钩子实例化一次 AIOProducer；添加一个将生产者作为参数的 produce() 辅助函数；确保将 await producer.flush() 和 await producer.close() 接入框架的关闭钩子。使用 AsyncSchemaRegistryClient（而非同步版本），以免注册表调用阻塞事件循环。
4. 步骤 4：新辅助函数的测试通过。
5. 步骤 5：对新代码运行 kafka-perf-review，确认其符合异步最佳实践（异步 SR 客户端、AsyncJSONSerializer、仅使用 kwargs 构造、将优雅关闭接入框架的原生生命周期）。
6. 步骤 6：根据用户实际使用的框架，提供路由 / 处理器示例。

结果：现有服务通过其原生启动/关闭生命周期获得 Kafka 生产者，同时保留原有结构。

示例 3：未连接 Kafka MCP，退回到硬性门槛

用户说：“为 orders.payment.completed 搭建一个 Python Kafka 消费者”，但未连接 Kafka MCP 服务（或者只连接了 Atlassian / GitHub 等非 Kafka MCP）。

操作：

1. 步骤 1：Agent 枚举已连接的 MCP，发现没有任何与 Kafka 相关的 MCP。技能会注明这一限制。
2. 步骤 2：硬性门槛要求用户提供：引导服务、架构注册表 URL、架构本身（粘贴内容或提供路径）、分区数（可选，默认为 1）、消费者组 ID。
3. 步骤 3-6：使用用户提供的值正常继续。

结果：仍可完成搭建，但 Agent 会明确告知用户缺少 MCP 的影响——准确措辞为：“没有 Kafka MCP，我无法验证该架构是否确实已注册到该主题。注册此架构后，请运行 kafka-schema-review 确认其匹配。”

故障排除

主题搜索没有返回结果

原因：该集群中不存在此主题，或 MCP 尚未为其编制索引，或搜索关键词范围过窄。

解决方案：使用更宽泛的关键词重试。如果仍无结果，则使用 MCP 的完整列表调用（Lenses：list_topics；Confluent：list_topics；等），让用户从完整列表中选择。如果该主题确实不存在，请询问用户是否要通过 MCP 的主题创建工具创建该主题，或中止操作。

主题存在，但没有已注册的架构

原因：主题存在，但 <topic>-value 尚未注册任何架构。

解决方案：请用户选择：(a) 提供要注册的 JSON Schema，或 (b) 确认要搭建无架构消费者（此时改用不依赖 Schema Registry 的 JSONDeserializer——但要提醒用户这种方式不够稳健，并建议之后注册架构）。

生成的测试在 AIOProducer / AIOConsumer 模拟对象上失败

原因：未安装 pytest-asyncio，或测试固件是同步的。

解决方案：确认 pytest-asyncio 已列在 pyproject.toml 的 [dependency-groups].dev 下，重新运行 uv sync，并确保测试函数使用 @pytest.mark.asyncio 装饰。

TypeError: ... got multiple values for argument 'schema_registry_client'

原因：使用位置参数构造序列化器/反序列化器。JSON、Avro 和 Protobuf 类的位置参数签名各不相同。

解决方案：请参阅 references/common-mistakes.md 第 8 行。始终使用关键字参数传递 schema_str 和 schema_registry_client。

验证门禁（步骤 5）报告生成的代码存在严重问题

原因：该技能的版本与 kafka-perf-review 中的审计阈值存在偏差。

解决方案：修复生成的代码，使其通过审计（在此代码库中，审计是判断“何为良好”的标准），然后重新运行验证门禁。如果审计阈值本身似乎有误，请针对 kafka-perf-review/references/ 提交 PR。

输出格式

在步骤 6 返回结果时，请使用以下结构：

## Project scaffolded

Generated at: <project-dir>/

### Files written
- producer.py
- consumer.py
- common.py
- schemas/value.schema.json   (pulled from registry — matches live schema)
- tests/test_project.py
- .env.example
- pyproject.toml              (uv-managed; uv.lock generated by `uv sync`)
- README.md

### Run it

# One-off setup (creates .venv and installs everything via uv)
cd <project-dir>
uv sync

# Terminal 1 — producer
cd <project-dir>
uv run python producer.py

# Terminal 2 — consumer
cd <project-dir>
uv run python consumer.py

### Validation gate

kafka-topic-audit findings:
- [topic-name] <severity> <finding>

kafka-perf-review findings:
- [file:line] <severity> <finding>

### Notes
- Schema in `schemas/value.schema.json` was pulled from `<topic>-value` in the registry, not inferred.
- Consumer group ID: `<group-id>` (unique, no collision).
- Idempotent producer enabled, graceful shutdown implemented.