用 MongoDB Atlas Stream…

在已连接 MongoDB MCP Server 且有 Atlas API 凭据的前提下，通过四个 MCP 工具完成 Atlas Stream Processing 工作区、连接、处理器的创建、管理与删除，并规避常见管道错误。

适用场景

解决思路

1. 确认前置条件：MongoDB MCP Server 已连接并配置 Atlas API 凭据（apiClientId 和 apiClientSecret），且四个工具 atlas-streams-discover、atlas-streams-build、atlas-streams-manage、atlas-streams-teardown 可用；若工具缺失，参考 references/mcp-troubleshooting.md 排查。
2. 所有操作都需要 Atlas project ID；若未知，先调用 atlas-list-projects 获取。
3. 读操作全部用 atlas-streams-discover：list-workspaces、inspect-workspace、list-connections、inspect-connection、list-processors、inspect-processor、diagnose-processor、get-networking；list 类操作支持 limit（1-100，默认 20）和 pageNum（默认 1），responseFormat 可选 concise（list 默认）或 detailed（inspect/diagnose 默认）。
4. 创建操作全部用 atlas-streams-build，按 resource 类型只填对应字段：workspace 填 projectId、workspaceName、cloudProvider、region、tier（默认 SP10）、includeSampleData；connection 填 projectId、workspaceName、connectionName、connectionType（Kafka/Cluster/S3/Https/Kinesis/Lambda/SchemaRegistry/Sample）、connectionConfig；processor 填 projectId、workspaceName、processorName、pipeline（必须以 $source 开头、以 $merge/$emit 结尾）、dlq、autoStart；privatelink 填 projectId、privateLinkConfig（项目级，不需要 workspaceName）。
5. 更新与状态操作全部用 atlas-streams-manage：start-processor（开始计费，可选 tier、resumeFromCheckpoint、startAtOperationTime）、stop-processor（停止计费，状态保留 45 天）、modify-processor（必须先停止，可改 pipeline、dlq、newName）、update-workspace（改 newRegion 或 newTier）、update-connection（改 connectionConfig，但网络配置不可改，需删除重建）、accept-peering（需 peeringId、requesterAccountId、requesterVpcId）、reject-peering（需 peeringId）。
6. 注意状态预检查：start-processor 对已 STARTED 的处理器报错；stop-processor 对已 STOPPED 或 CREATED 的处理器为 no-op；modify-processor 对 STARTED 的处理器报错，必须先停止。处理器状态流转为 CREATED → STARTED → STOPPED，运行时错误会进入 FAILED。
7. 删除操作全部用 atlas-streams-teardown：processor 删除前自动停止；connection 若被运行中的处理器引用会被阻止，需先停止/删除引用它的处理器；workspace 会级联删除所有连接和处理器；privatelink/peering 为项目级资源，只需 resourceName。
8. 删除 workspace 前必须先 inspect-workspace 获取连接和处理器数量，向用户展示“Workspace X 包含 N 个连接和 M 个处理器，删除将永久移除全部”，等待确认后再调用 teardown。
9. 创建任何 processor 管道前必须调用 search-knowledge 校验字段名（如查询 "Atlas Stream Processing $emit S3 fields" 或 "Atlas Stream Processing Kafka $source configuration"），并用 dataSources: [{"name": "devcenter"}] 查询可用管道示例。
10. 构建非平凡处理器时从官方示例仓库 https://github.com/mongodb/ASP_example 获取示例，先看 example_processors/README.md；关键 quickstart 包括 00_hello_world.json（内联 $source.documents + $match）、01_changestream_basic.json（change stream → tumbling window → $merge）、03_kafka_to_mongo.json（Kafka → tumbling window rollup → $merge）、04_mongo_to_mongo.json（链式处理器）、05_kafka_tail.json（实时 Kafka 监控）。
11. 避免无效构造：不要使用 $$NOW、$$ROOT、$$CURRENT；HTTPS 不能作为 $source（只能用于 $https 富化或 sink）；Kafka $source 必须包含 topic；部署的处理器必须有终止阶段（$merge、$emit、$https 或异步 $externalFunction）；Lambda 用 $externalFunction 而非 $emit；$validate 的 validationAction 用 "dlq" 而非 "error"。
12. 按阶段填写必填字段：change stream $source 需 fullDocument: "updateLookup"；Kinesis $source 用 stream（不是 streamName 或 topic）；Kinesis $emit 必须含 partitionKey；S3 $emit 用 path（不是 prefix）；$https 必须含 connectionName、path、method、as、onError: "dlq"；$externalFunction 必须含 connectionName、functionName、execution、as、onError: "dlq"；$validate 必须含 validator（$jsonSchema）和 validationAction: "dlq"；$lookup 加 parallelism（如 parallelism: 2）。
13. AWS 连接（S3、Kinesis、Lambda）的 IAM role ARN 必须先通过 Atlas Cloud Provider Access 注册，并与用户确认。
14. SchemaRegistry 连接的 connectionType 必须是 "SchemaRegistry"（不是 "Kafka"），schema 类型值区分大小写（用小写 avro，不是 AVRO）。
15. 利用 MCP 工具行为：创建连接时工具会通过 elicitation 自动收集缺失的敏感字段（密码、bootstrap servers），不要向用户索要；bootstrapServers 数组会自动转为逗号分隔字符串；schemaRegistryUrls 字符串会自动包成数组；Cluster 连接的 dbRoleToExecute 默认 {role: "readWriteAnyDatabase", type: "BUILT_IN"}；workspace 的 includeSampleData 默认 true，会自动创建 sample_stream_solar 连接。
16. region 字段使用 Atlas 专有名称：AWS us-east-1 → VIRGINIA_USA、us-east-2 → OHIO_USA、eu-west-1 → DUBLIN_IRL；GCP us-central1 → US_CENTRAL1、europe-west1 → EUROPE_WEST1；Azure eastus → eastus、westeurope → westeurope。不确定时用 inspect-workspace 查看 dataProcessRegion.region。
17. 按连接能力选择源/汇：Cluster 可作 change stream 源、$merge 汇、$lookup 中间阶段；Kafka 可作源（必须含 topic）和 $emit 汇；Sample Stream 仅作源；S3 仅作 $emit 汇；Https 可作 $https 富化或最终汇；AWSLambda 作汇时 execution 必须为 async，中间阶段可 sync 或 async；Kinesis 可作源和 $emit 汇；SchemaRegistry 仅用于 schema 解析。
18. 从零搭建流程：list-workspaces 检查已有 → 创建 workspace（region 靠近数据，开发用 SP10）→ 为每个源/汇/富化创建 connection → 用 list-connections + inspect-connection 校验连接名称与目标匹配并向用户展示摘要 → 调用 search-knowledge 校验字段名并获取示例 → 创建 processor（配置 DLQ）→ start-processor（提醒计费）。
19. 采用增量管道开发：先做基础 $source → $merge 验证连通性，再加 $match 验证过滤，再加 $addFields/$project 验证重塑，再加窗口或富化验证聚合逻辑，最后加错误处理/DLQ。
20. 修改处理器管道：先 stop-processor（必须先停止），再 modify-processor 提供新管道，最后 start-processor 重启。
21. 调试失败处理器：先调用 diagnose-processor 获取一次性健康报告；然后根据症状定位根因——Error 419 + "no partitions found" 表示 Kafka topic 不存在或拼写错误；State: FAILED + 多次重启表示连接级错误（绕过 DLQ），检查连接配置；State: STARTED + 零输出 + 窗口管道表示 Kafka 分区空闲阻塞窗口关闭，给 Kafka $source 加 partitionIdleTimeout（如 {"size": 30, "unit": "second"}）；State: STARTED + 零输出 + 非窗口管道检查源是否有数据及 Kafka offset lag；memoryUsageBytes 接近 tier 上限有 OOM 风险，建议升 tier；DLQ 计数增加表示逐文档错误，用 MongoDB find 查 DLQ 集合。
22. 调试时先分类处理器类型（告警 vs 转换 vs 过滤）再解读输出量；给出具体有序的修复步骤，不要罗列假设场景；需要详细日志时引导用户到 Atlas UI：Atlas → Stream Processing → Workspace → Processor → Logs tab。
23. 处理多汇需求时注意单管道只能有一个终止汇（$merge 或 $emit）；用户要求多个输出目标时，说明单汇约束并用中间目标提出链式处理器方案。
24. 计费注意：Atlas Stream Processing 无免费层，处理器运行期间按小时计费（按秒计算）；stop-processor 停止计费，停止后状态保留 45 天免费；无计费原型验证用 mongosh 的 sp.process() 临时运行管道。
25. 遵守安全规则：teardown 和 manage 需要用户确认，不得绕过；删除 workspace 前必须先 inspect 并展示连接/处理器数量；创建 processor 前必须按 Pre-Deployment Validation 校验所有连接；删除 workspace 会永久移除所有连接和处理器；resumeFromCheckpoint: false 会丢弃所有窗口状态，需先警告用户；不支持在工作区间移动处理器（必须重建）；不支持 dry-run/模拟，需说明将要做什么并请求确认；启动处理器前始终提醒计费；API 凭据存在连接设置中，绝不硬编码进处理器管道。

适用边界 · 注意事项

• 不适用于通用 MongoDB 查询或 Atlas 集群管理（本技能仅针对 Atlas Stream Processing）。
• 未连接 MongoDB MCP Server 或缺少 Atlas API 凭据时不可用。
• 没有 Atlas 项目 ID 且无法通过 atlas-list-projects 获取时无法执行操作。