atlas-stream-processing

MongoDB Atlas Streams

MongoDB MCP Server の 4 つの MCP ツールを使用して、Atlas Stream Processing（ASP）のパイプラインを構築、運用、デバッグします。

前提条件

このスキルを使用するには、以下に接続された MongoDB MCP Server が必要です。

• Atlas API の認証情報（apiClientId と apiClientSecret）

4 つのツール：atlas-streams-discover、atlas-streams-build、atlas-streams-manage、atlas-streams-teardown。

すべての操作で Atlas プロジェクト ID が必要です。 不明な場合は、最初に atlas-list-projects を呼び出してプロジェクト ID を確認してください。

MCP ツールを利用できない場合

MongoDB MCP Server に接続されていない場合、または Streams ツールが見つからない場合は、診断手順と代替手段について references/mcp-troubleshooting.md を参照してください。

ツール選択マトリックス

atlas-streams-discover — すべての読み取り操作

アクション	使用する場面
list-workspaces	プロジェクト内のすべてのワークスペースを表示する
inspect-workspace	ワークスペースの設定、状態、リージョンを確認する
list-connections	ワークスペース内のすべての接続を表示する
inspect-connection	接続の状態、設定、健全性を確認する
list-processors	ワークスペース内のすべてのプロセッサーを表示する
inspect-processor	プロセッサーの状態、パイプライン、設定を確認する
diagnose-processor	完全な健全性レポート：状態、統計情報、エラー
get-networking	PrivateLink と VPC ピアリングの詳細。PrivateLink の設定に必要な Atlas アカウントの詳細を取得するには、オプションで cloudProvider と region を指定します

ページネーション（すべての一覧アクション）：limit（1〜100、デフォルトは 20）、pageNum（デフォルトは 1）。 応答形式：responseFormat — "concise"（一覧アクションのデフォルト）または "detailed"（inspect／diagnose のデフォルト）。

atlas-streams-build — すべての作成操作

リソース	主なパラメーター
workspace	cloudProvider、region、tier（デフォルトは SP10）、includeSampleData
connection	connectionName、connectionType（Kafka／Cluster／S3／Https／Kinesis／Lambda／SchemaRegistry／Sample）、connectionConfig
processor	processorName、pipeline（$source で始まり、$merge／$emit で終わる必要があります）、dlq、autoStart
privatelink	privateLinkConfig（プロジェクトレベル。特定のワークスペースには紐づきません）

フィールドのマッピング — 選択したリソースタイプのフィールドのみを入力してください：

• resource = "workspace"： 入力：projectId、workspaceName、cloudProvider、region、tier、includeSampleData。空のままにする：接続とプロセッサーのすべてのフィールド。
• resource = "connection"： 入力：projectId、workspaceName、connectionName、connectionType、connectionConfig。空のままにする：ワークスペースとプロセッサーのすべてのフィールド。（タイプ別のスキーマについては、references/connection-configs.md を参照してください。）
• resource = "processor"： 入力：projectId、workspaceName、processorName、pipeline、dlq（推奨）、autoStart（任意）。空のままにする：ワークスペースと接続のすべてのフィールド。（パイプラインの例については、references/pipeline-patterns.md を参照してください。）
• resource = "privatelink"： 入力：projectId、privateLinkConfig。注：PrivateLink はプロジェクトレベルであり、ワークスペースレベルではありません。workspaceName は不要なため、省略してください。空のままにする：接続とプロセッサーのすべてのフィールド。

atlas-streams-manage — すべての更新／状態操作

アクション	備考
start-processor	課金を開始します。tier による上書き、resumeFromCheckpoint は任意です
stop-processor	課金を停止します。状態は45日間保持されます
modify-processor	先にプロセッサを停止する必要があります。パイプライン、DLQ、または名前を変更します
update-workspace	ティアまたはリージョンを変更します
update-connection	設定を更新します（ネットワーク設定は変更不可のため、削除して再作成する必要があります）
accept-peering / reject-peering	VPC ピアリングの管理

フィールドの対応付け — 常に projectId、workspaceName を入力し、その後、アクションに応じて次の値を入力します。

• "start-processor" → resourceName。任意：tier、resumeFromCheckpoint、startAtOperationTime（特定の時点から再開するための ISO 8601 タイムスタンプ）
• "stop-processor" → resourceName
• "modify-processor" → resourceName。次のいずれかを1つ以上指定します：pipeline、dlq、newName
• "update-workspace" → newRegion または newTier
• "update-connection" → resourceName、connectionConfig。例外：ネットワーク設定（例：PrivateLink）は作成後に変更できません — 削除して再作成してください。
• "accept-peering" → peeringId、requesterAccountId、requesterVpcId
• "reject-peering" → peeringId

状態の事前チェック：

• start-processor → プロセッサがすでに STARTED の場合はエラーになります
• stop-processor → すでに STOPPED または CREATED の場合は何も実行しません（エラーではありません）
• modify-processor → プロセッサが STARTED の場合はエラーになります（先に停止する必要があります）

プロセッサの状態： CREATED → STARTED（start 経由）→ STOPPED（stop 経由）。実行時エラーが発生すると FAILED に移行することもあります。変更には STOPPED または CREATED の状態が必要です。

削除時の安全チェック：

• プロセッサの削除 → 削除前に自動停止します（事前に手動で停止する必要はありません）
• 接続の削除 → 実行中のプロセッサが参照している場合はブロックされます。先に参照しているプロセッサを停止または削除してください。
• ワークスペースの削除 → 詳細なワークフローは以下を参照してください（108～111行目）。

atlas-streams-teardown — すべての削除操作

リソース	安全動作
processor	削除前に自動停止します
connection	実行中のプロセッサから参照されている場合はブロックされます
workspace	すべての接続とプロセッサをカスケード削除します
privatelink / peering	ネットワークリソースを削除します

フィールドの対応付け — 常に projectId、resource を入力し、その後、次の値を入力します。

• resource: "workspace" → workspaceName
• resource: "connection" または "processor" → workspaceName、resourceName
• resource: "privatelink" または "peering" → resourceName（ID）。これらはプロジェクトレベルのリソースであり、特定のワークスペースには紐付けられていません。

ワークスペースを削除する前に、まず内容を確認してください。

1. atlas-streams-discover → inspect-workspace — 接続数とプロセッサ数を取得します
2. ユーザーに提示します：「ワークスペース X には接続が N 個、プロセッサが M 個含まれています。削除すると、すべてが完全に削除されます。続行しますか？」
3. atlas-streams-teardown を呼び出す前に、確認を待ちます

重要：プロセッサーを作成する前に検証してください

プロセッサーパイプラインを構成する前に、必ず search-knowledge を呼び出してください。 これは必須です。

• フィールドの検証： シンク／ソースの種類を指定してクエリします。例：「Atlas Stream Processing $emit S3 fields」または「Atlas Stream Processing Kafka $source configuration」。これにより、S3 の $emit で prefix と path を取り違えるといったエラーを防げます。
• パターンの例： 動作するパイプラインの例を取得するには、dataSources: [{"name": "devcenter"}] を指定してクエリします。例：「Atlas Stream Processing tumbling window example」。

また、複雑なプロセッサーを構築する際は、公式 ASP の例リポジトリ https://github.com/mongodb/ASP_example から例を取得してください（クイックスタート、プロセッサーの例、Terraform の例）。パターンの全カタログは、まず example_processors/README.md を確認してください。

主なクイックスタート：

クイックスタート	パターン
00_hello_world.json	$match を使うインラインの $source.documents（インフラ不要、一時的）
01_changestream_basic.json	Change Stream → タンブリングウィンドウ → Atlas への $merge
03_kafka_to_mongo.json	Kafka ソース → タンブリングウィンドウでの集計 → Atlas への $merge
04_mongo_to_mongo.json	プロセッサーの連結：集計 → 別コレクションへのアーカイブ
05_kafka_tail.json	Kafka トピックのリアルタイム監視（シンクなし、tail -f のような動作）

パイプラインのルールと警告

無効な構成 — これらはストリーミングパイプラインでは無効です：

• $$NOW、$$ROOT、$$CURRENT — ストリーム処理では利用できません。絶対に使用しないでください。イベント時刻には $$NOW ではなく、ドキュメント自身のタイムスタンプフィールドまたは _stream_meta メタデータを使用してください。
• $source としての HTTPS 接続 — HTTPS は $https によるエンリッチメントまたはシンク専用であり、データソースとしては使えません
• topic のない Kafka の $source — topic フィールドは必須です
• シンクのないパイプライン — デプロイするプロセッサーには終端ステージ（$merge、$emit、$https、または非同期の $externalFunction）が必要です（シンクなしで動作するのは sp.process() 経由の場合のみです）
• $emit のターゲットとしての Lambda — Lambda には $emit ではなく、$externalFunction（パイプライン途中のエンリッチメント）を使用します
• validationAction: "error" を指定した $validate — プロセッサーがクラッシュします。代わりに "dlq" を使用してください

ステージごとの必須フィールド：

• $source（Change Stream）：ドキュメントの全内容を取得するには fullDocument: "updateLookup" を含めます
• $source（Kinesis）：stream を使用します（streamName や topic ではありません）
• $emit（Kinesis）：partitionKey を必ず含めます
• $emit（S3）：path を使用します（prefix ではありません）
• $https：connectionName、path、method、as、onError: "dlq" を必ず含めます
• $externalFunction：connectionName、functionName、execution、as、onError: "dlq" を必ず含めます
• $validate：$jsonSchema を含む validator と validationAction: "dlq" を必ず含めます
• $lookup：並行 I/O を行うため、parallelism の設定（例：parallelism: 2）を含めます
• AWS 接続（S3、Kinesis、Lambda）：IAM ロール ARN は、先に Atlas Cloud Provider Access に登録する必要があります。必ずユーザーに確認してください。詳細は references/connection-configs.md を参照してください。

ステージのフィールド例と JSON 構文については、references/pipeline-patterns.md を参照してください。

SchemaRegistry 接続： connectionType は "Kafka" ではなく "SchemaRegistry" にする必要があります。スキーマの種類の値は大文字と小文字を区別します（AVRO ではなく小文字の avro を使用してください）。必須フィールドと認証の種類については、references/connection-configs.md を参照してください。

MCP ツールの動作

情報の要求： 接続の作成時、ビルドツールは MCP の情報要求を通じて、不足している機密フィールド（パスワード、ブートストラップサーバー）を自動収集します。これらをユーザーに尋ねず、ツールに収集させてください。

自動正規化：

• bootstrapServers 配列 → カンマ区切りの文字列に自動変換
• schemaRegistryUrls 文字列 → 配列に自動変換
• dbRoleToExecute → Cluster 接続では {role: "readWriteAnyDatabase", type: "BUILT_IN"} がデフォルトになります

ワークスペースの作成： includeSampleData のデフォルト値は true で、sample_stream_solar 接続が自動作成されます。

リージョン名： region フィールドでは、クラウドプロバイダーごとに異なる Atlas 固有の名前を使用します。形式を間違えると、わかりにくい dataProcessRegion エラーが返されます。

プロバイダー	クラウドリージョン	Streams の region 値
AWS	us-east-1	VIRGINIA_USA
AWS	us-east-2	OHIO_USA
AWS	eu-west-1	DUBLIN_IRL
GCP	us-central1	US_CENTRAL1
GCP	europe-west1	EUROPE_WEST1
Azure	eastus	eastus
Azure	westeurope	westeurope

リージョンの完全な対応表については、references/connection-configs.md を参照してください。不明な場合は、atlas-streams-discover → inspect-workspace で既存のワークスペースを調べ、dataProcessRegion.region を確認してください。

接続機能 — ソース／シンクのリファレンス

パイプラインを作成する前に、各接続タイプでできることを確認してください。

接続タイプ	ソースとして（$source）	シンクとして（$merge / $emit）	パイプラインの中間処理	備考
Cluster	✅ Change streams	✅ コレクションへの $merge	✅ $lookup	Change streams は insert/update/delete/replace 操作を監視します
Kafka	✅ トピックコンシューマー	✅ トピックへの $emit	❌	ソースには topic フィールドが必須です
Sample Stream	✅ サンプルデータ	❌ 無効	❌	テスト／デモ専用
S3	❌ 無効	✅ バケットへの $emit	❌	シンク専用です。path、format、compression を使用します。AWS PrivateLink をサポートしています。
Https	❌ 無効	✅ シンクとしての $https	✅ エンリッチメント用の $https	エンリッチメント用のパイプライン中間処理、または最終シンクステージとして使用できます
AWSLambda	❌ 無効	✅ $externalFunction（非同期のみ）	✅ $externalFunction（同期または非同期）	シンク： execution: "async" が必須です。パイプライン中間処理： execution: "sync" または "async"
AWS Kinesis	✅ ストリームコンシューマー	✅ ストリームへの $emit	❌	Kafka と同様のパターンです
SchemaRegistry	❌ 無効	❌ 無効	✅ スキーマ解決	メタデータ専用 — Kafka 接続で Avro スキーマに使用します

避けるべき接続のよくある使用ミス：

• ❌ シンクとして execution: "sync" で $externalFunction を使用する → シンクステージでは execution: "async" を使用する必要があります
• ❌ Change streams の存在を忘れる → Atlas Cluster は強力なソースであり、シンクだけではありません
• ❌ Kafka で $merge を使用する → Kafka のシンクには $emit を使用してください

タイプ別の接続設定スキーマの詳細については、references/connection-configs.md を参照してください。

基本的なワークフロー

最初からセットアップする

1. atlas-streams-discover → list-workspaces（既存のものを確認）
2. atlas-streams-build → resource: "workspace"（データに近いリージョン、開発用に SP10）
3. atlas-streams-build → resource: "connection"（各ソース／シンク／エンリッチメント用）
4. 接続を検証する： atlas-streams-discover → list-connections と各接続の inspect-connection — 名前が対象と一致することを確認し、概要をユーザーに提示
5. search-knowledge を呼び出してフィールド名を検証します。https://github.com/mongodb/ASP_example から関連する例を取得します
6. atlas-streams-build → resource: "processor"（DLQ を設定）
7. atlas-streams-manage → start-processor（課金について警告）

ワークフローパターン

段階的なパイプライン開発（推奨）： 5 フェーズのライフサイクル全体については、references/development-workflow.md を参照してください。

1. 基本的な $source → $merge パイプラインから始めます（接続を検証します）
2. $match ステージを追加します（フィルタリングを検証します）
3. $addFields / $project 変換を追加します（データの形状変更を検証します）
4. ウィンドウ処理またはエンリッチメントを追加します（集計ロジックを検証します）
5. エラーハンドリング / DLQ 設定を追加します

プロセッサーパイプラインの変更：

1. atlas-streams-manage → action: "stop-processor" — 最初にプロセッサーを停止する必要があります
2. atlas-streams-manage → action: "modify-processor" — 新しいパイプラインを指定します
3. atlas-streams-manage → action: "start-processor" — 再起動します

失敗しているプロセッサーのデバッグ：

1. atlas-streams-discover → diagnose-processor — 1 回限りのヘルスレポートを取得します。必ず最初に呼び出してください。
2. 具体的な根本原因を特定します。 症状を診断パターンに照らし合わせます。
   • Error 419 + "no partitions found" → Kafka トピックが存在しないか、スペルが間違っています
   • State: FAILED + 複数回の再起動 → 接続レベルのエラー（DLQ を経由しません）。接続設定を確認してください
   • State: STARTED + 出力ゼロ + ウィンドウ処理のパイプライン → Kafka パーティションのアイドル状態によりウィンドウの閉鎖が妨げられている可能性があります。Kafka $source に partitionIdleTimeout を追加してください（例：{"size": 30, "unit": "second"}）
   • State: STARTED + 出力ゼロ + ウィンドウ処理なし → ソースにデータがあるか確認し、Kafka のオフセットラグを調べてください
   • memoryUsageBytes が大きく、階層の上限に近づいている → OOM のリスクがあります。より上位の階層を推奨してください
   • DLQ 件数が増加している → ドキュメント単位のエラーです。DLQ コレクションに対して MongoDB の find を使用してください 完全なパターン表については、references/output-diagnostics.md を参照してください。
3. 出力件数を解釈する前に、プロセッサーの種類（アラート、変換、フィルター）を分類します。
4. 診断した根本原因に応じた具体的な修正手順を、順序立てて提示します。仮定上のシナリオを並べてはいけません。
5. 詳細なログが必要な場合は、ユーザーを Atlas UI の Atlas → Stream Processing → Workspace → Processor → Logs タブに案内します。

プロセッサーの連結（複数シンクパターン）

重要：1 つのパイプラインに設定できる終端シンクは 1 つだけです（$merge または $emit）。ユーザーが複数の出力先（例：「Atlas に書き込み、さらに Kafka に出力」）を求めた場合は、単一シンクの制約を必ず伝え、中間の出力先を使う連結プロセッサーを提案してください。例を含むパターン全体については、references/pipeline-patterns.md を参照してください。

デプロイ前・デプロイ後のチェックリスト

接続の検証やパイプラインの検証を含む、デプロイ前の品質チェックリスト全体と、デプロイ後の検証ワークフローについては、references/development-workflow.md を参照してください。

階層のサイジングとパフォーマンス

階層の仕様、並列度の計算式、複雑さのスコアリング、パフォーマンス最適化戦略については、references/sizing-and-parallelism.md を参照してください。

トラブルシューティング

プロセッサーの障害、API エラー、設定の問題、パフォーマンスの問題を網羅したトラブルシューティング表全体については、references/development-workflow.md を参照してください。

課金とコスト

Atlas Stream Processing に無料枠はありません。 デプロイ済みのプロセッサーは、稼働中は継続的に課金されます。

• 課金は時間単位で、プロセッサーの稼働中のみ秒単位で計算されます
• stop-processor を実行すると課金が停止します。停止したプロセッサーの状態は、料金なしで 45 日間保持されます
• 課金なしでプロトタイプを作成するには： mongosh で sp.process() を使用します。プロセッサーをデプロイせずに、一時的にパイプラインを実行できます
• 階層ごとの料金とコスト最適化戦略については、references/sizing-and-parallelism.md を参照してください

安全ルール

• atlas-streams-teardown と atlas-streams-manage にはユーザーの確認が必要です。回避しないでください
• ワークスペースに対して atlas-streams-teardown を呼び出す前に、必ず最初に atlas-streams-discover でワークスペースを調査し、接続数とプロセッサー数を確認してから、確認を求める前にこの情報をユーザーに提示してください
• プロセッサーを作成する前に、必ず references/development-workflow.md の「デプロイ前の検証」セクションに従って、すべての接続を検証してください
• ワークスペースを削除すると、すべての接続とプロセッサーが完全に削除されます
• プロセッサーを停止した後、状態は45日間保持されます。その後、チェックポイントは破棄されます
• resumeFromCheckpoint: false にすると、すべてのウィンドウ状態が失われます。事前にユーザーに警告してください
• ワークスペース間でプロセッサーを移動することはできません（再作成が必要です）
• ドライラン／シミュレーションはサポートされていません。実行する内容を説明し、確認を求めてください
• プロセッサーを開始する前に、必ず料金についてユーザーに警告してください
• API 認証情報は接続設定に保存し、プロセッサーパイプラインにハードコードしないでください

参照ファイル

ファイル	参照するタイミング...
references/pipeline-patterns.md	プロセッサーパイプラインの構築または変更時
references/connection-configs.md	接続の作成時（種類ごとのスキーマ）
references/development-workflow.md	ライフサイクル管理に従うとき、またはデバッグの判断ツリーを使うとき
references/output-diagnostics.md	プロセッサーの出力が想定外の場合（ゼロ、少ない、または誤っている）
references/sizing-and-parallelism.md	ティアの選択、並列性の調整、またはコストの最適化時