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。 如果不知道项目 ID，请先调用 atlas-list-projects 查找。

如果 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 对等互连详情。可选：使用 cloudProvider 和 region 获取 PrivateLink 设置所需的 Atlas 账户详情

分页（所有列表操作）：limit（1-100，默认值为 20）、pageNum（默认值为 1）。 响应格式：responseFormat — "concise"（列表操作的默认值）或 "detailed"（检查/诊断操作的默认值）。

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。至少填写以下一项：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（通过启动）→ STOPPED（通过停止）。运行时出错时也可能进入 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”。这可以避免 $emit 到 S3 时将 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	内联 $source.documents 搭配 $match（无需基础设施、临时）
01_changestream_basic.json	变更流 → 滚动窗口 → $merge 到 Atlas
03_kafka_to_mongo.json	Kafka 来源 → 滚动窗口汇总 → $merge 到 Atlas
04_mongo_to_mongo.json	链式处理器：汇总 → 归档到单独的集合
05_kafka_tail.json	实时监控 Kafka 主题（无接收端，类似 tail -f）

管道规则与警告

无效结构 — 以下结构在流处理管道中均无效：

• $$NOW、$$ROOT、$$CURRENT — 流处理中不可用。切勿使用。事件时间应使用文档自身的时间戳字段或 _stream_meta 元数据，而不是 $$NOW。
• 将 HTTPS 连接用作 $source — HTTPS 仅用于 $https 增强或接收端，不可用作数据来源
• Kafka $source 未配置 topic — 必须提供 topic 字段
• 没有接收端的管道 — 已部署的处理器必须包含终端阶段（$merge、$emit、$https 或异步 $externalFunction）；无接收端的管道只能通过 sp.process() 使用
• 将 Lambda 用作 $emit 目标 — Lambda 使用 $externalFunction（管道中间的增强阶段），而不是 $emit
• $validate 使用 validationAction: "error" — 会导致处理器崩溃；请改用 "dlq"

各阶段的必需字段：

• $source（变更流）：包含 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 和 validationAction: "dlq" 的 validator
• $lookup：包含 parallelism 设置（例如 parallelism: 2），以实现并发 I/O
• AWS 连接（S3、Kinesis、Lambda）：必须先通过 Atlas Cloud Provider Access 注册 IAM 角色 ARN。务必向用户确认这一点。详情请参见 references/connection-configs.md。

各阶段字段的 JSON 语法示例请参见 references/pipeline-patterns.md。

SchemaRegistry 连接： connectionType 必须为 "SchemaRegistry"（而非 "Kafka"）。架构类型值区分大小写（使用小写 avro，而不是 AVRO）。必需字段和身份验证类型请参见 references/connection-configs.md。

MCP 工具行为

信息收集： 创建连接时，构建工具会通过 MCP 信息收集自动收集缺少的敏感字段（密码、引导服务器）。不要向用户询问这些信息，让工具自行收集。

自动规范化：

• bootstrapServers 数组 → 自动转换为逗号分隔的字符串
• schemaRegistryUrls 字符串 → 自动包装为数组
• dbRoleToExecute → 对集群连接默认为 {role: "readWriteAnyDatabase", type: "BUILT_IN"}

工作区创建： includeSampleData 默认为 true，会自动创建 sample_stream_solar 连接。

区域命名： region 字段使用 Atlas 专属名称，各云服务提供商的名称格式不同。使用错误的格式会返回难以理解的 dataProcessRegion 错误。

Provider	Cloud Region	Streams region Value
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。

连接能力——源／接收端参考

创建管道前，请了解每种连接类型的用途：

Connection Type	As Source ($source)	As Sink ($merge / $emit)	Mid-Pipeline	Notes
Cluster	✅ 变更流	✅ 使用 $merge 写入集合	✅ $lookup	变更流会监控插入／更新／删除／替换操作
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 架构

应避免的常见连接用法错误：

• ❌ 将 $externalFunction 用作接收端并设置 execution: "sync" → 接收端阶段必须使用 execution: "async"
• ❌ 忽略变更流的存在 → 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 — 一次性运行状况报告。务必先调用此项。
2. 明确判断具体根本原因。 将症状与诊断模式匹配：
   • 错误 419 +“no partitions found” → Kafka 主题不存在或拼写错误
   • State: FAILED + 多次重启 → 连接级错误（绕过 DLQ），检查连接配置
   • State: STARTED + 零输出 + 窗口化管道 → Kafka 分区可能处于空闲状态，导致窗口无法关闭；在 Kafka $source 中添加 partitionIdleTimeout（例如，{"size": 30, "unit": "second"}）
   • State: STARTED + 零输出 + 非窗口化管道 → 检查源是否有数据；检查 Kafka 偏移量滞后情况
   • memoryUsageBytes 较高且接近层级限制 → 有 OOM 风险；建议使用更高层级
   • DLQ 计数持续增加 → 存在逐文档错误；使用 MongoDB find 查询 DLQ 集合 完整的模式表请参阅 references/output-diagnostics.md。
3. 在解读输出量之前，先对处理器类型进行分类（告警、转换或筛选）。
4. 根据已诊断出的根本原因，提供具体且有顺序的修复步骤。不要列出假设情形。
5. 如果需要详细日志，请引导用户前往 Atlas UI：Atlas → Stream Processing → Workspace → Processor → Logs 标签页。

链式处理器（多目标模式）

重要：单个管道只能有一个终端接收器（$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	选择层级、调整并行度或优化成本时