kafka-python-client

Kafka Python Client Scaffold

confluent-kafka-python を使用して Kafka トピックにメッセージを生成・消費する、本番環境対応の Python プロジェクトを生成します。Schema Registry（JSON Schema）、グレースフルシャットダウン、冪等プロデューサー、ヘッダーベースのスキーマ識別、テストに対応しています。エージェントは、ユーザーに質問する前に、接続されている Kafka MCP サーバー（Lenses MCP、Confluent、Aiven、その他のいずれか）を通じて、対象トピックの情報（名前、パーティション、登録済みスキーマ）をすべて検出してください。Kafka MCP が接続されていない場合、または検出結果が何もない場合に限り、質問に切り替えてください。

対象環境とキーワード：$ARGUMENTS

最初の返信は、次の文で始めてください："Running the kafka-python-client skill to scaffold this project."

ワークフロー

このチェックリストをコピーして、進捗を記録してください。

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. 生成した tests/ に対して pytest を実行し、失敗を修正します（テストではなく、コードを修正してください）。
5. 検証ゲートを実行する：対象トピックに対して kafka-topic-audit を、生成した producer.py と consumer.py に対して kafka-perf-review を実行します。
6. 実行手順と検証ゲートの所見を本文に含めて引き渡す

Step 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. 選択したトピックの値サブジェクトに登録されたスキーマを取得します（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 が接続されていない場合、または検出結果が何もない場合は、Step 2 の必須ゲートに進み、不足している情報を直接ユーザーに尋ねてください。失われた情報について明確に伝えてください：「Kafka MCP がないため、スキーマがトピックに登録されているものと一致するか検証できません。」

Step 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. ステップ 1 で見つかった値を箇条書きで振り返ります。
2. 調査で判明しなかった未解決の質問を列挙します（例：プロジェクトディレクトリ、ユーザーに強い希望がある場合の同期／非同期プロデューサースタイル）。
3. ユーザーに確認または訂正を明示的に依頼します。

その後、停止してユーザーの返信を待ちます。この振り返りは、ファイルを書き込む前にユーザーが読み違いに気付けるようにするためのものであり、プロンプトの内容が完全に指定されているように見える場合でも必須です。

このゲートを省略できるのは、この会話の前の時点ですでにユーザーが振り返りを確認している場合だけです。

確認後に適用するデフォルト：

• プロデューサースタイル： ユーザーが非同期ランタイムを指定していない限り、confluent_kafka の同期 Producer を使用します。その場合は AIOProducer を提案します。非同期を示すものには、asyncio の明示的な言及、非同期優先の Web フレームワーク（FastAPI、Starlette、Quart、Litestar、BlackSheep、Robyn、aiohttp、Sanic、Tornado）、一般的な ASGI アプリケーション、非同期タスクシステム（ARQ、Dramatiq、非同期タスクを使用する Celery）があります。Django が言及されている場合は、非同期 ASGI が明示されない限り、同期 Producer をデフォルトにします。不明な場合は質問します。
• コンシューマースタイル： 常に非同期の 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

このプロジェクトでは、Python 環境と依存関係の管理に uv を使用します。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 に使用します

schemas/value.schema.json は、ステップ 1 で取得したスキーマから直接作成してください。再生成したり、言い換えたりしないでください。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 の dataset／cluster-metadata 呼び出しで返された値に置き換え、SASL 認証情報のプレースホルダーを追加します。

pyproject.toml

uv が依存関係を解決してロックできるよう、PEP 621 の [project] テーブルで依存関係を宣言します。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 ブロックを指定すると、「unable to determine which files to ship」というエラーが発生します。

producer.py、consumer.py、common.py がインポートするサードパーティーパッケージは、すべて [project].dependencies に含める必要があります。テスト専用パッケージ（pytest、pytest-asyncio）は dev 依存関係グループに含めます。これらは uv sync で自動的にインストールされ、本番環境へのインストール時には uv sync --no-dev で除外されます。コンシューマーは常に非同期なので、pytest-asyncio を含めてください。非同期の Schema Registry クライアントは、confluent-kafka の依存関係として宣言されていなくても、実行時に httpx と authlib をインポートするため、これらも明示的に指定します。

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 を作成してすべてをインストール）、スキーマがまだ登録されていない場合の登録方法（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. 1 プロセスにつきプロデューサーは 1 つ。 main() で作成し、引数として produce() に渡します。呼び出しごとに新しいプロデューサーを作成してはいけません。
2. スキーマ登録は独立したステップとして行います。 シリアライザーを auto.register.schemas=False および use.latest.version=True で設定し、sr_client.register_schema() を直接呼び出す小さな 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() を呼び出し、どちらも await します。
6. エンティティ ID をキーにします。 エンティティ単位のイベントストリームでは、パーティション内の順序を維持するため、key=<entity_id>.encode("utf-8") を渡します。
7. 実行前に接続を確認します。 ブローカーには AdminClient.list_topics()、スキーマレジストリには /subjects への HTTP GET を使います。

Step 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 へフォールバックしてはいけません。

テストに失敗した場合は、テストではなく生成されたコードを修正し、すべて合格するまで続けてください。

Step 5: 検証ゲート — kafka-topic-audit と kafka-perf-review を連続して実行する

このステップが、このスキルを一度きりのスキャフォルダーと差別化します。Step 4 の後、品質ゲートとして既存の 2 つのスキルを自動的に実行します。

1. 対象トピックに対して kafka-topic-audit を実行します。 Step 1 で指定したトピックを対象にスキルを呼び出して実行します。重大な指摘事項があればその場で伝えます（単一ブローカーの CE で RF=1 となるのは想定内です。指摘はしますが、ブロッカーとして扱わないでください）。
2. 生成されたコードに対して kafka-perf-review を実行します。 生成されたプロジェクトディレクトリを対象に実行します。生成された producer.py と consumer.py は、enable.idempotence、acks=all、linger.ms が 0 より大きいこと、compression.type が設定されていること、max.poll.interval.ms が妥当であること、正常なシャットダウンが実装されていることのすべてを満たす必要があります。監査で問題が見つかった場合は、生成されたコードを修正して再実行します。

Step 6: 結果を引き渡す

ユーザーへの最終メッセージには、次の内容を含める必要があります。

• プロジェクトディレクトリと、生成したものの簡潔な説明
• プロデューサーとコンシューマーを実行する正確なコマンド（2 つのターミナルで実行）
• 検証ゲートでの指摘事項の概要（1 段落）
• schemas/value.schema.json のスキーマは推測ではなく、稼働中のクラスターから取得したものであることを示す注記

成功基準

定量的な基準

• 「Python Kafka クライアントを作成して」という問い合わせの 90% で起動します（10～20 種類の異なる言い回しでテストしてください。references/test-cases.md を参照）
• MCP ツールの呼び出しが合計 15 回未満でスキャフォールドを完成させます
• 1 回の実行につき MCP 呼び出しの失敗は 0 回です（実際にクラスターへ接続できない場合を除く）
• 生成された pytest tests/ は、最大 1 回の自己修正後、80% 超の割合で初回実行時に合格します
• 検証ゲート（ステップ 5）で、生成コードに重大な指摘が 0 件返されます

定性的な基準

• 生成開始前にユーザーへ尋ねる質問は最大 1 つです（ステップ 2 の確認）。複数ターンにわたる聞き取りは行いません
• 生成されたスキーマファイルは、登録済みスキーマとバイト単位で一致します（言い換えはありません）
• 生成されたコードは、追加の編集をせずに、シード済みのローカル Lenses CE に対してエンドツーエンドで実行できます
• 引き継ぎメッセージだけで必要な情報がそろっており、ユーザーが「では、どうやって実行すればよいですか？」と尋ねる必要はありません

例

例 1：MCP による完全な検出を伴う新規コンシューマー（Lenses MCP を接続）

ユーザーの発言：「ステージング Kafka の決済イベントを処理する Python コンシューマーを作ってください」

アクション：

1. ステップ 1：Lenses MCP が接続されているため、エージェントはそのツールを使用します。list_environments → staging を選択します。list_datasets(search="payment") は 3 つのトピックを返します。パーティション数とメッセージレートを添えて 3 つすべてをユーザーに提示し、ユーザーは 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：実行コマンドと検証結果の概要を添えて引き継ぎます。

結果：ユーザーが 5 分以内にエンドツーエンドで実行でき、トピックに登録されているものと完全に一致するスキーマを備えた、本番環境を想定したプロジェクトです。

例 2：既存の非同期 Python サービスにプロデューサーを追加する

ユーザーが「Quart サービスに orders.payment.refunded 用の Kafka プロデューサーを追加して」と言った場合、同じフローをあらゆる非同期ランタイムに適用できます：FastAPI ／ Starlette ／ Litestar ／ BlackSheep ／ Robyn ／ aiohttp ／ Sanic ／ Tornado ／通常の asyncio ワーカー ／ ASGI ベースのサービス。

アクション：

1. ステップ 1：MCP による検出でトピックとスキーマを特定します。エージェントは非同期フレームワークのシグナル（Quart）を検出し、ステップ 2 のプロデューサー形式として AIOProducer を選択します。
2. ステップ 2：要約で非同期プロデューサー形式を説明し、フレームワークの起動／終了フック（Quart：@app.before_serving ／ @app.after_serving、FastAPI：ライフスパンハンドラー、Starlette：ライフスパンハンドラー、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、キーワード引数のみの構築、フレームワーク標準のライフサイクルに組み込まれたグレースフルシャットダウン）が確認されます。
6. ステップ 6：ユーザーが実際に使用しているフレームワークでのルート／ハンドラーの例を添えて引き渡します。

結果：既存のサービスは、既存の構造を損なうことなく、ネイティブの起動／終了ライフサイクルを使う Kafka プロデューサーを備えます。

例 3：Kafka MCP が接続されていない場合、ハードゲートにフォールバックする

ユーザーが「orders.payment.completed 用の Python Kafka コンシューマーのひな形を作って」と言ったものの、Kafka MCP サーバーが接続されていない（または Atlassian ／ GitHub など Kafka 以外の MCP しか接続されていない）場合です。

アクション：

1. ステップ 1：エージェントは接続されている MCP を列挙し、Kafka に該当するものがないことを確認します。スキルがこの制約を通知します。
2. ステップ 2：ハードゲートで次の情報をユーザーに尋ねます：ブートストラップサーバー、スキーマレジストリの URL、スキーマ自体（貼り付けまたはパス）、パーティション数（任意、デフォルトは 1）、コンシューマーグループ ID。
3. ステップ 3〜6：ユーザーから提供された値を使って通常どおり進めます。

結果：ひな形の作成は引き続き可能ですが、MCP がないことで確認できない内容をエージェントが明示的にユーザーに伝えます。正確な文言は次のとおりです：「Kafka MCP がないため、スキーマがトピックに実際に登録されているものと一致するか確認できません。このスキーマを登録した後、kafka-schema-review を実行して一致することを確認してください。」

トラブルシューティング

トピック検索の結果が 0 件になる

原因：このクラスターにトピックが存在しないか、MCP がまだインデックス化していないか、検索キーワードが限定的すぎます。

解決策：より広いキーワードで再試行します。それでも結果がない場合は、MCP の全件取得呼び出し（Lenses：list_topics、Confluent：list_topics など）を使用し、全件リストからユーザーに選んでもらいます。トピックが実際に存在しない場合は、MCP のトピック作成ツールを使って作成するか、中止するかをユーザーに尋ねます。

トピックは存在するが、登録済みのスキーマがない

原因：トピックは存在しますが、<topic>-value にスキーマが登録されていません。

解決策：登録する JSON Schema を提示するか、スキーマなしのコンシューマーのひな形を作成するか、ユーザーに確認します（その場合は Schema Registry を使わず JSONDeserializer にフォールバックします。ただし、これは壊れやすいため、後でスキーマを登録するよう推奨します）。

生成されたテストが AIOProducer ／ AIOConsumer のモックで失敗する

原因：pytest-asyncio がインストールされていないか、テストフィクスチャが同期処理になっています。

解決策：pyproject.toml の [dependency-groups].dev に pytest-asyncio が含まれていることを確認し、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.