在 Celigo 集成中配置…

当需要在 Celigo 流程中把记录写入外部系统（REST API、数据库、ERP、文件服务器或 AI 模型）时，按此流程选择适配器类型并配置 import。

适用场景

解决思路

1. 先确定目标应用：要写入哪个系统，这决定 adaptorType、连接类型和配置形态。
2. 构建前先查已有模式：用 `celigo account search "<keyword>"` 搜索账户，用 `celigo account dependencies import <id>` 查看已有 import 的连接与引用它的流程，用 `celigo account lint` 找孤立 import，用 `celigo templates marketplace` 搜模板，用 `celigo templates preview <id> --model Import` 和 `--summary` 提取模板中的 import；账户索引超过 4 小时会自动刷新，可用 `celigo account snapshot` 强制刷新。
3. 写任何 HTTP 配置前必须先查预置连接器：用 `celigo http-connectors list | grep -i "<application-name>"` 搜索，用 `celigo http-connectors get <id> --full` 查看端点、资源与认证，用 `celigo http-connectors catalog <id> --resource-type import --published-only` 和 `celigo http-connectors endpoint-detail <id> --resource-type import --resource-id <rid> --endpoint-id <epid>` 深入 import 端点，用 `celigo tp-connectors list` 搜 EDI/AS2 连接器；若存在连接器，则从连接器创建连接（`http._httpConnectorId`），并从连接器端点元数据取 `relativeURI`、method、body/response 形状，而不是照公开 API 文档手写。
4. 对 NetSuite、Salesforce、RDBMS 连接查询元数据：用 `celigo metadata types <connectionId>` 列记录类型/sObject/表，用 `celigo metadata fields <connectionId> <type>` 列字段；NetSuite 返回字段 ID、类型和分组（子列表名如 `item`、`addressbook` 作为分组，对应 `mapping.lists[].generate`，查找字段 ID 用作 `netsuite_da.lookups[]` 的 `searchField`/`resultField`）；Salesforce 返回字段 API 名、类型和关系信息；RDBMS 返回列名和类型，用于写 SQL 并校验 `bulkInsert.tableName` 或 `bulkLoad.tableName`。
5. 判断 import 类别：是记录型（向 API/数据库/ERP 提交记录）、文件型（向存储写文件）还是 AI 型（调用模型）。
6. 按适配器决策矩阵选择 adaptorType：REST/GraphQL 用 `HTTPImport`；NetSuite 用 `NetSuiteDistributedImport`；Salesforce 用 `SalesforceImport`；SQL 数据库用 `RDBMSImport`；MongoDB 用 `MongodbImport`；DynamoDB 用 `DynamodbImport`；JDBC 用 `JDBCImport`；HTTP 传文件（Google Drive、Box、Dropbox、Azure Blob）用 `HTTPImport` 且 `http.type: "file"`；FTP/SFTP 用 `FTPImport`；S3 用 `S3Import`；AS2 EDI 用 `AS2Import`；本地文件系统用 `FileSystemImport`；OpenAI/Gemini 用 `AiAgentImport`；PII 检测/内容审核用 `GuardrailImport`；Celigo Tool 用 `ToolImport`；预置栈连接器用 `WrapperImport`。选择顺序为：优先原生适配器，其次预置 HTTP 连接器，最后才手写 HTTP 配置。注意 `adaptorType` 大小写敏感。
7. 按 Schema 索引构建 import JSON：始终读 `references/schemas/request.yml`（基础字段），加上决策矩阵对应的适配器文件，文件型还要读 `file.yml`，克隆时读 `clone-request.yml` 和 `clone-response.yml`。
8. 配置字段映射：默认用 Mapper 2.0（`mappings[]` 数组），NetSuite 和 Salesforce 只支持 Mapper 1.0（`mapping.fields[]` / `mapping.lists[]`），包含静态映射和查找表的值解析。
9. 确定操作逻辑：create、update、upsert、delete、attach/detach。
10. 按需配置钩子：在流水线各阶段用 JavaScript 做 preMap、postMap、postSubmit 处理；从记录生成文件的文件型 import 还支持 postAggregate。
11. 处理一对多：当一条源记录应产生多个 import 操作时，设 `oneToMany: true` 并把 `pathToMany` 设为子数组路径（如 `"lineItems"`）。
12. 规划响应映射：在流程的 `pageProcessors[]` 条目上配置（不在 import 资源上），但构建 import 时就要规划；响应通过 `_json`（原始 API 响应）和 `errors` 获取，用 `_json.fieldName` 提取（如 `_json.id` 取新建记录 ID，`_json.output.1.content.0.text` 取 OpenAI 响应）；响应映射使用 Transformation 1.0 语法（extract/generate 对），不是新的表达式转换。
13. 规划 postResponseMap 钩子：在流程的 `pageProcessors[]` 条目上配置，但构建 import 时就要规划；用于在响应映射把响应合并回记录后、下游步骤前转换或丰富记录。
14. 理解 import 执行流水线顺序：1) 输入过滤器（可选，在流程 `pageProcessors[]` 条目上以表达式配置，在任何处理前丢弃记录）；2) preMap 钩子（可选）；3) 字段映射；4) postMap 钩子（可选）；5) 提交到目标；6) postSubmit 钩子（可选）；7) 响应映射（可选，在流程 `pageProcessors[]` 上配置）；8) postResponseMap 钩子（可选）。
15. 选择匹配行为（create/update/upsert）：五种行为——始终创建（不匹配不检查）、仅缺失时创建（先匹配检查，未找到则创建，找到则静默跳过）、仅存在时更新（先匹配检查，找到则更新，未找到则静默跳过）、仅存在时更新且缺失则失败（严格变体，未匹配记录报错）、upsert（未匹配则创建，匹配则更新，用户意图模糊时的默认选择）。
16. 当需要匹配时决定四件事：匹配行为、匹配键字段（`email`、`external_id`、`customer.id`，除始终创建外都必填）、匹配时动作（update/skip/fail）、未匹配时动作（create/skip/fail）。
17. 优先使用 import 级匹配而非单独的 lookup 步骤：import 原生支持写前检查，单个 import 同时做匹配和写入可减少步骤、往返和胶水逻辑；只有当查得数据有写入之外的消费者（如基于非存在性分支的路由器、AI agent 推理、多个下游步骤读不同字段）时，独立 lookup 才值得。
18. 当 HTTP 目标没有原生 upsert 但有独立的 create 和 update 端点时，使用复合 import：在单个 import 节点上固定两个端点并标记 create/update 角色，运行时按记录通过匹配键检查选择，保持流程为一步；只有当 create 和 update 路径在端点选择之外还需分叉（不同映射、不同下游消费者或不同钩子链）时才用两个独立 import。
19. 处理 HTTP 文件上传（multipart/form-data）：1) 字节来源——import 本身不携带文件，由前置 blob export 或 blob lookup 把文件拉入 blob 存储，其响应映射把引用放到记录上（惯用 `{"extract": "data.0.blobKey", "generate": "blobKey"}`），记录携带 `blobKey`（存储指针而非内容）；2) 媒体类型——连接媒体类型（或 import 的请求媒体类型覆盖）为 `multipart/form-data`，成功/错误响应媒体类型通常覆盖为 JSON；3) 请求体——不是原始 MIME 也不是普通 handlebars 载荷，而是 JSON 部件数组，每项 `{name, value, type}` 加可选 `filename`（含扩展名）和 `mime-headers`，文件部件为 `"type": "attachment"` 且 `"value": "{{blob}}"`（`{{blob}}` 是附件唯一可接受值，其他值会 422），API 需要的伴随字段作为 `"type": "inline"` 部件，值为 JSON 对象的 inline 部件必须序列化，每个 import 只引用一个文件；4) `blobKeyPath`（高级设置）——记录中 blobKey 所在的 JSON 路径（`blobKey`，或嵌套时 `file.blobKey`），发送时平台据此进入 blob 存储并把真实字节流式写入附件部件。平台自行组装最终 MIME 体（生成 `boundary`，绝不硬编码；写各部件 `Content-Disposition`；用原始文件字节替换附件部件）。注意有些 multipart API 期望 `multipart/related`（如 Google Drive 的 `/upload/drive/v3/files?uploadType=multipart`），此时部件数组机制不适用，请求体是字面 MIME 文档（显式 boundary、JSON 元数据部件、引用 `{{blob}}` 的内容部件），构建前先确认目标 API 文档。
20. 对异步目标（submit、poll、confirm）：当目标 API 只确认写入（HTTP 202、job ticket）并在后台完成时，给 import 附加异步助手（`http._asyncHelperId`），使该步骤提交、轮询状态 export 直到外部工作完成才 resolve；机制与约束（带 done/error 值列表和轮询间隔的状态 export；异步配置步骤上无 transform、输出过滤器或钩子；助手不能嵌套）与 export 侧相同；仅当 API 确实无法同步确认写入时才添加。
21. 用 CLI 命令操作：CRUD 用 `celigo imports list`、`get <id>`、`create < import.json`、`update <id> < import.json`、`set <id> key=value`、`delete <id> [-y]`；测试提交用 `echo '[{"name":"test"}]' | celigo imports invoke <id>`；克隆与连接管理用 `echo '{"connectionMap":{"oldConnId":"newConnId"}}' | celigo imports clone <id>` 和 `celigo imports replace-connection <id> <newConnectionId>`；发现用 `celigo templates marketplace`、`celigo http-connectors list`、`celigo http-connectors catalog <id> --resource-type import --published-only`、`celigo http-connectors endpoint-detail <id> --resource-type import --resource-id <rid> --endpoint-id <epid>`、`celigo tp-connectors list`、`celigo metadata types <connectionId>`、`celigo metadata fields <connectionId> <entityType>`；调试用 `celigo imports enable-debug <id> [--duration <minutes>]` 和 `celigo imports disable-debug <id>`。
22. 提交前检查必填项：`name` 已设；`adaptorType` 大小写与连接类型完全匹配；`_connectionId` 引用有效在线连接（无 BYOK 的 AI import 跳过）；适配器配置块名与 adaptorType 匹配（HTTPImport 用 `http{}`，NetSuiteDistributedImport 用 `netsuite_da{}` 等）。
23. 提交前检查适配器特定项：HTTP 已查预置连接器且 `http.method` 与 `http.relativeURI` 已设；NetSuite 的 `netsuite_da.operation` 与 `netsuite_da.recordType` 已设；RDBMS 的 `rdbms.queryType` 为 `per_record` 或 `bulk_insert`（不是遗留的 `insert`/`update`）；Salesforce 的 `salesforce.sObjectType` 与 `salesforce.operation` 已设。
24. 提交前检查跨资源一致性：连接 `type` 与 import 的 `adaptorType` 匹配；如需响应映射，配置在流程 `pageProcessors[]` 条目而非 import 本身；如一对多，`oneToMany: true` 且 `pathToMany` 已设；如用 Mapper 1.0（NetSuite/Salesforce），用 `mapping.fields[]` / `mapping.lists[]` 而非 `mappings[]`。
25. 避免常见陷阱：PUT 会抹掉未包含字段，总是先 GET 再改再 PUT（`set` 命令会处理）；包含 `rest:` 块会创建遗留 RESTImport，新 import 只用 `http:`；输入过滤器跳过的是该 import 步骤而非记录，被过滤记录继续走后续步骤，可用 job 的 `numIgnore` 检查；multipart 文件部件只接受 `{{blob}}`，绝不硬编码 MIME `boundary`；日志中的 `bodyKey`/`blobKey` 是内部存储指针而非载荷字段，若目标真的收到字面字符串 `bodyKey` 或 `blobKey`，说明文件部件配置错误（该用 attachment 却用了 inline，或 `blobKeyPath` 未解析）。
26. 按错误表修复常见错误：422 `adaptorType invalid` 是大小写错误，用决策矩阵中的精确大小写；422 `_connectionId required` 是缺连接，设有效连接 ID；422 `queryType invalid` 是遗留 Snowflake 值，用 `per_record` 或 `bulk_insert`；422 `distributed required` 是缺 NetSuite 标志，用 `NetSuiteDistributedImport` 且连接上 `distributed: true`；422 `mapping invalid` 是映射器版本错误，NetSuite/Salesforce 用 Mapper 1.0；422 attachment `value` invalid 是 multipart 文件部件不是 `{{blob}}`，设为 `"type": "attachment"`、`"value": "{{blob}}"` 并让平台生成 `boundary`。

适用边界 · 注意事项

• 不适用于配置 export（数据源）或连接本身，那些属于 configuring-exports 和 configuring-connections 技能。
• 不适用于编写字段映射表达式、Handlebars 模板或 JavaScript 钩子的具体语法，那些属于 writing-mappings、writing-handlebars、writing-scripts 技能。
• 不适用于排查已上线流程的运行失败，那属于 troubleshooting-flows 技能。