TypeScript与Zapier SDK构建智能HubSpot公司信息补全工作流

1. 项目概述：告别OAuth烦恼，用TypeScript、Zapier SDK与AI构建智能HubSpot公司信息补全工作流

如果你曾经尝试过将外部数据源与HubSpot CRM进行深度集成，那么对OAuth授权流程、令牌刷新、权限范围管理以及紧随API版本变更而来的SDK升级维护，一定不会陌生。这套复杂的“标准流程”常常让一个本应轻量、敏捷的数据同步任务，变得笨重且脆弱。今天，我想分享一个我们团队在实际业务中打磨出来的解决方案：一个完全绕开传统HubSpot OAuth集成痛点，利用TypeScript、Zapier SDK和轻量级AI模型（Claude Haiku）构建的自动化公司信息补全工作流。

这个工作流的核心任务非常明确：输入一个公司的网站域名，自动从Apollo数据平台获取丰富的公司画像（包括行业、员工规模、融资阶段等），经过智能语义匹配，将自由文本的行业分类精准映射到HubSpot预设的枚举值中，最后将完整信息创建或更新到HubSpot的公司记录里。整个过程仅需四步，其中两步可以并行执行，最关键的是，全程无需编写和维护任何HubSpot的OAuth认证代码。这不仅仅是技术栈的替换，更是一种集成思路的转变——将最擅长的工具用于最合适的环节：用直接的REST API获取深度数据，用Zapier SDK处理广泛的、标准化的CRM写入操作，再用AI作为两者之间的“语义粘合剂”。

2. 架构设计与核心思路拆解

2.1 为何选择“去OAuth”架构？

传统的HubSpot集成模式要求开发者在其平台注册应用，处理完整的OAuth 2.0授权流程，管理访问令牌和刷新令牌，并为应用申请精确的API权限范围。这对于一个长期运行、需要高权限的成熟应用是必要的。但对于我们这种单向、定时或触发的数据补全任务来说，就显得过于沉重了。每次HubSpot API更新，都可能需要检查SDK兼容性；令牌管理增加了系统的复杂性和故障点；更重要的是，这限制了工作流的可移植性——如果你想为Salesforce或Pipedrive实现同样的功能，几乎需要重写整个集成层。

我们的新架构巧妙地规避了这些问题。其核心在于将“数据获取”与“系统写入”解耦：

1. 数据获取层：直接调用Apollo等数据提供商的REST API。这部分是纯粹的数据消费，认证简单（通常是API Key），响应速度快，数据维度丰富。
2. 系统写入层：完全交由Zapier SDK处理。Zapier已经与包括HubSpot在内的9000多个应用完成了深度集成，并替我们管理了所有繁琐的OAuth流程。我们只需要一个预先在Zapier上配置好的“连接”（Connection），就可以通过SDK以该用户身份执行操作。
3. 语义映射层：这是本项目的创新点。Apollo返回的行业信息是自由文本（如"Internet Software & Services"），而HubSpot的行业字段是一个预设的下拉枚举。手动维护一个映射表极其脆弱——HubSpot可能增减选项，Apollo的行业分类体系也可能变化。因此，我们引入一个轻量级LLM（Claude Haiku），在运行时动态获取HubSpot的最新枚举列表，并让其进行智能语义匹配。

注意：这种架构的本质是“借道”Zapier。这意味着你对HubSpot数据的操作权限，完全取决于在Zapier上配置那个“连接”的账户权限。这通常更安全，也简化了权限管理，因为业务人员可以直接用他们的HubSpot账户来授权Zapier，而无需开发者介入。

2.2 技术栈选型与考量

• TypeScript：作为主要开发语言，其强大的类型系统对于构建涉及多个外部API（Apollo, Zapier SDK, LLM）的复杂工作流至关重要。它能提前在编译阶段捕获大量的数据格式错误，比如Apollo返回的某个字段可能为null，而HubSpot期望的是空字符串""。我们使用Zod库进行运行时数据验证，与TypeScript静态类型检查形成双重保障。
• Zapier SDK (Beta)：这是本项目的基石。它不是一个封装了HubSpot API的客户端，而是一个通往整个Zapier生态的程序化接口。通过它，我们不仅可以执行search_or_write（查找或创建）这样的操作，还能调用listInputFieldChoices这样的元数据方法，动态获取下拉字段的可选值。这避免了硬编码，实现了“配置即代码”的灵活性。
• Claude Haiku (via Output.ai)：选择Haiku而非更强大的Claude 3.5 Sonnet或Opus，是出于成本与任务复杂度的平衡。行业映射是一个典型的“受限词汇分类”任务：输入是一个字符串和一份固定的候选列表，输出必须是列表中的一项。Haiku模型对此类任务精度足够，且速度极快、成本极低。我们通过Output.ai平台调用，它提供了良好的工作流编排、步骤追踪和缓存能力。
• Apollo API：作为数据源，它提供了高质量的公司画像数据。当然，这个位置可以替换为Clearbit、Hunter.io或其他任何提供RESTful接口的数据服务。

2.3 工作流执行流程解析

整个工作流被设计为一个有向无环图（DAG），清晰地分为四个步骤，并充分利用了并行化以提高效率。

开始 │ ├───────────────┬───────────────┐ │ │ │ ▼ ▼ ▼ 步骤1: 步骤2: (输入) Apollo数据 获取HubSpot 公司网站 补全 行业枚举列表 │ │ └───────────────┼───────────────┘ │ ▼ 步骤3: (条件执行) AI语义映射行业 │ ▼ 步骤4: Zapier SDK写入HubSpot │ ▼ 结束

并行化的妙用：步骤1（调用Apollo API）和步骤2（通过Zapier SDK获取HubSpot行业列表）之间没有任何数据依赖。Apollo不需要知道HubSpot的行业列表，反之亦然。因此，使用Promise.all()让它们同时执行，可以将整体耗时缩短近一半。这是提升工作流响应速度的关键优化点。

条件执行逻辑：步骤3（AI映射）是一个条件步骤。只有当Apollo返回了有效的industry字段时，才会触发AI进行映射。如果Apollo没有找到行业信息，则直接跳过，将undefined传递给后续步骤。这避免了不必要的LLM调用，节省了成本。

3. 核心模块实现与代码深度解析

3.1 项目结构与职责划分

项目采用了一个清晰的功能模块化结构，这对于维护和测试非常有利。

zapier_hubspot_company_enrichment/ ├── workflow.ts # 工作流编排中枢：定义步骤顺序与并行逻辑 ├── steps.ts # 四个核心步骤的具体实现 ├── types.ts # 所有输入输出的Zod模式定义与TypeScript类型 ├── prompts/ # 存放LLM提示词模板 │ └── map_hubspot_industry@v1.prompt └── scenarios/ # 测试场景与用例数据 └── stripe.json

3.2 工作流编排核心 (workflow.ts)

这个文件是工作流的大脑，它定义了执行的骨架和逻辑流。

import { workflow } from '@outputai/core'; import { enrichCompanyWithApollo, fetchHubspotIndustries, mapHubspotIndustry, upsertHubspotCompany, } from './steps.js'; import { workflowInputSchema, workflowOutputSchema } from './types.js'; export default workflow({ name: 'zapier_company_enrichment', description: 'Enriches a company profile using Apollo via REST API and upserts the result into HubSpot via Zapier SDK', inputSchema: workflowInputSchema, outputSchema: workflowOutputSchema, fn: async (input) => { // 步骤1 & 2：并行执行，互不依赖 const [apolloData, { industries }] = await Promise.all([ enrichCompanyWithApollo({ website: input.website }), fetchHubspotIndustries(), ]); // 步骤3：条件性AI语义映射 const hubspotIndustry = apolloData.industry ? (await mapHubspotIndustry({ industry: apolloData.industry, hubspotIndustries: industries, })).hubspotIndustry : undefined; // 步骤4：执行最终的创建或更新操作 const { hubspotCompanyId, action } = await upsertHubspotCompany({ ...apolloData, hubspotIndustry, }); return { companyName: apolloData.name, website: input.website, hubspotCompanyId, apolloData, action, // 明确返回是‘created’还是‘updated’，便于下游触发 }; }, });

关键设计点：

1. Promise.all的使用是性能优化的核心。确保这两个异步函数没有副作用和共享状态冲突。
2. 对apolloData.industry的检查是条件执行的关键。这里使用了三元运算符进行简洁的条件判断和异步调用。
3. 最终输出包含action字段。这是一个非常重要的设计，它明确告知调用方本次操作是创建了新记录还是更新了已有记录。基于这个信息，可以轻松触发后续动作，例如：只有当action === 'created'时，才向销售团队发送Slack通知。

3.3 步骤实现详解 (steps.ts)

每一步都被封装为一个独立的、可追踪的“步骤”，这得益于Output.ai的step()函数。它提供了自动重试、缓存和分布式追踪的能力。

3.3.1 步骤1：Apollo数据补全

export const enrichCompanyWithApollo = step({ name: 'enrich_company_with_apollo', description: 'Enriches company data using Apollo REST API directly', inputSchema: enrichCompanyInputSchema, outputSchema: apolloCompanySchema, fn: async ({ website }) => { const domain = extractDomain(website); // 关键：从URL提取纯净域名 const org = await enrichOrganization(domain); // 调用封装的Apollo客户端 if (!org?.name) { throw new Error(`Apollo returned no data for domain: ${domain}`); } // 数据转换与清洗：将Apollo的响应映射到我们的内部格式 return { name: org.name, website: org.website_url ?? website, // 优先使用Apollo返回的URL domain: org.primary_domain ?? domain, industry: org.industry ?? undefined, // 可能为null employeeCount: org.estimated_num_employees ?? undefined, // ... 其他字段映射 }; }, });

实操心得：extractDomain函数至关重要。用户可能输入https://www.stripe.com、stripe.com甚至http://stripe.com。统一提取出stripe.com能确保API查询的准确性。此外，Apollo API返回的字段非常丰富，我们只选取了对CRM最有价值的字段。用??（空值合并运算符）和?.（可选链）操作符可以优雅地处理API返回中可能缺失的字段。

3.3.2 步骤2：动态获取HubSpot行业枚举

这是体现Zapier SDK价值的关键一步。

export const fetchHubspotIndustries = step({ name: 'fetch_hubspot_industries', description: 'Fetches available HubSpot industry field choices via Zapier SDK', outputSchema: fetchHubspotIndustriesOutputSchema, fn: async () => { const zapier = createZapierClient(); const industries: string[] = []; // 使用异步迭代器遍历分页结果 for await (const item of zapier.listInputFieldChoices({ appKey: 'hubspot', actionType: 'search_or_write', actionKey: 'company_crmSearch', inputFieldKey: 'industry', // 指定要获取的字段 connectionId: HUBSPOT_CONNECTION_ID, // 指定哪个HubSpot账户 }).items()) { const value = item.value ?? item.key ?? item.label; if (value) industries.push(value); } return { industries }; }, });

listInputFieldChoices的威力：这个方法直接查询Zapier平台，获取指定应用、指定动作下，某个输入字段当前所有可接受的值。这意味着：

• 永远最新：如果HubSpot管理员在后台增加了新的行业选项（如“Web3协议”），下次工作流运行时就会自动包含它，无需修改代码或部署。
• 环境隔离：如果你有开发、测试、生产三个HubSpot门户，它们的行业列表可能不同。通过切换connectionId，工作流会自动获取对应门户的列表。
• 通用性：这个方法适用于HubSpot的任何下拉列表字段，如lifecyclestage、lead_source等，也适用于Zapier支持的任何其他应用的类似字段。

3.3.3 步骤3：AI语义映射

我们将提示词工程从代码中分离出来，放在独立的prompt文件中，这提高了可维护性。

提示词文件 (map_hubspot_industry@v1.prompt):

--- provider: anthropic model: claude-haiku-4-5 temperature: 0 maxTokens: 256 --- <system> You are an expert at mapping company industry strings to HubSpot's predefined industry ENUM values. Given an industry category return the single best-matching HubSpot industry value. Valid HubSpot industry values: {{ hubspotIndustries }} Rules: - Return EXACTLY one value from the list above - Pick the closest semantic match even if it's not an exact string match - If no reasonable match exists, return the closest category </system> <user> Map this industry to a HubSpot industry value: Industry: {{ industry }} </user>

步骤调用代码：

export const mapHubspotIndustry = step({ name: 'map_hubspot_industry', description: 'Maps a raw industry string to a valid HubSpot industry enum value using an LLM', inputSchema: mapHubspotIndustryInputSchema, outputSchema: mapHubspotIndustryOutputSchema, fn: async ({ industry, hubspotIndustries }) => { const { output } = await generateText({ prompt: 'map_hubspot_industry@v1', // 引用提示词模板 variables: { industry, hubspotIndustries: hubspotIndustries.join(', '), // 将数组转换为逗号分隔的字符串 }, output: Output.object({ schema: mapHubspotIndustryOutputSchema }), // 强制结构化输出 }); return output; }, });

关键设计点：

1. temperature: 0：设置为0确保对于相同的输入，AI总是产生相同的输出。这对于数据管道的一致性至关重要。
2. 系统提示词（System Prompt）：明确界定了AI的角色、任务和规则。最关键的一行是Valid HubSpot industry values: {{ hubspotIndustries }}，它在运行时将动态获取的列表注入，严格限制了AI的输出范围，杜绝了“幻觉”产生无效值的可能。
3. 结构化输出（Output.object）：我们要求AI返回一个符合mapHubspotIndustryOutputSchema的JSON对象，而不是纯文本。这使代码能类型安全地解析结果。

3.3.4 步骤4：通过Zapier SDK写入HubSpot

这是将一切成果落地的最后一步。

export const upsertHubspotCompany = step({ name: 'upsert_hubspot_company', description: 'Creates or updates a HubSpot company record using enriched Apollo data via Zapier SDK', inputSchema: hubspotUpsertInputSchema, outputSchema: hubspotUpsertOutputSchema, fn: async (input) => { const zapier = createZapierClient(); const domain = input.domain ?? extractDomain(input.website ?? ''); // 准备Zapier Action所需的输入参数 const inputs = { first_search_property_name: 'name', first_search_property_value: input.name, name: input.name, domain: domain ?? '', website: input.website ?? '', city: input.city ?? '', country: input.country ?? '', industry: input.hubspotIndustry ?? '', // 使用AI映射后的值 numberofemployees: input.employeeCount ? String(input.employeeCount) : '', description: input.description ?? '', linkedin_company_page: input.linkedinUrl ?? '', total_money_raised: input.totalFunding ? String(input.totalFunding) : '', }; // 调用Zapier SDK执行search_or_write操作 const { data: result } = await zapier.apps.hubspot.search_or_write.company_crmSearch({ inputs, connectionId: HUBSPOT_CONNECTION_ID, }); const [record] = zapierHubspotResponseSchema.parse(result); return { hubspotCompanyId: record.id, action: record.isNew ? 'created' : 'updated', }; }, });

search_or_write操作解析：这是Zapier SDK中一个非常强大的动作。它首先根据first_search_property_name和first_search_property_value（这里我们用name和公司名）在HubSpot中查找是否存在匹配的公司。

• 如果找到：则用inputs中的数据更新该条记录。
• 如果没找到：则用inputs中的数据创建一条新记录。 这实现了一个“幂等”的写入操作，无论运行多少次，对于同一家公司，HubSpot里只会存在一条记录。

注意事项：Zapier SDK要求的输入字段名，有时与HubSpot API本身的字段名略有不同（例如linkedin_company_pagevslinkedinUrl）。你需要查阅Zapier对于HubSpot App的特定文档，或者利用SDK的TypeScript类型提示来找到正确的字段名。将数字类型（如employeeCount）转换为字符串是必要的，因为许多CRM系统的自定义字段接口以字符串形式接收数据。

3.4 类型安全与数据验证 (types.ts)

使用Zod构建从输入到输出完整的数据验证链条，是保证工作流健壮性的基石。

import { z } from '@outputai/core'; // 工作流输入：最简形式，只需公司名和网站 export const workflowInputSchema = z.object({ companyName: z.string().describe('The name of the company to enrich'), website: z.string().url().describe('The company website URL (e.g. https://acme.com)'), }); // Apollo返回数据的模式：大部分字段可选，适应API响应 export const apolloCompanySchema = z.object({ name: z.string(), website: z.string().optional(), domain: z.string().optional(), industry: z.string().optional(), // 可能为空 employeeCount: z.number().optional(), // ... 其他可选字段 }); // 写入HubSpot的输入模式：在Apollo数据基础上增加映射后的行业字段 export const hubspotUpsertInputSchema = apolloCompanySchema.extend({ hubspotIndustry: z.string().optional(), // AI映射后的结果 }); // 工作流最终输出 export const workflowOutputSchema = z.object({ companyName: z.string(), website: z.string(), hubspotCompanyId: z.string(), apolloData: apolloCompanySchema, action: z.enum(['created', 'updated']), // 明确的操作类型 });

类型设计哲学：我们严格区分了“外部数据”（Apollo，字段可选）、“内部传递数据”（增加映射字段）和“最终输出数据”。使用.optional()和.extend()可以优雅地构建这些模式。action字段使用z.enum限定，比简单的字符串更安全。

4. Zapier SDK深度解析与配置指南

4.1 Zapier SDK的核心概念

理解以下几个概念对于正确使用SDK至关重要：

4.2 初始化与认证配置

你需要在Zapier开发者平台创建一个“CLI App”来获取客户端凭证。

// shared/clients/zapier.ts import { createZapierSdk } from '@zapier/zapier-sdk'; export function createZapierClient() { // 从环境变量或安全的密钥管理服务中读取凭证 return createZapierSdk({ credentials: { clientId: process.env.ZAPIER_CLIENT_ID!, clientSecret: process.env.ZAPIER_CLIENT_SECRET!, }, }); }

安全警告：绝对不要将clientId和clientSecret硬编码在代码中或提交到版本控制系统。务必使用环境变量或专业的密钥管理服务（如AWS Secrets Manager, HashiCorp Vault）。

4.3 如何获取Connection ID

这是配置中最容易卡住的一步。Connection ID不是凭空生成的，它代表一个已经授权给Zapier的HubSpot账户。

1. 手动创建（测试用）：在Zapier.com上，用你的HubSpot账户创建一个简单的Zap（例如，由Google Sheets新行触发，在HubSpot创建公司）。在创建过程中，Zapier会引导你授权HubSpot账户。授权后，这个“连接”就存在了。
2. 获取Connection ID：你有几种方式获取这个连接的ID：
   • Zapier CLI：安装Zapier CLI后，使用命令zapier connections:list来列出所有连接。
   • Zapier Platform UI：在开发者平台的App管理界面，通常也有地方可以查看和管理连接。
   • 通过SDK以编程方式获取：你可以编写一个简单的脚本，使用SDK列出当前用户的所有连接。

// 示例：获取连接列表的脚本 const zapier = createZapierClient(); const connections = await zapier.connections.list(); console.log(connections.data.map(conn => ({ id: conn.id, app: conn.appKey, label: conn.label })));

找到对应HubSpot App的那个连接，其id就是你要用在代码中的HUBSPOT_CONNECTION_ID。

4.4 探索可用的Actions

Zapier SDK的优势在于其庞大的动作库。如何知道hubspot这个App下有哪些actionKey可用？

1. 查阅官方文档：Zapier为热门应用提供了详细的文档。
2. 使用SDK的自动补全：在TypeScript项目中，输入zapier.apps.hubspot.后，你的IDE会给出search、write、search_or_write等提示。继续输入search_or_write.，可能会提示出company_crmSearch、contact_crmSearch等。
3. 使用元数据方法：SDK提供了listActions等方法，可以在运行时动态查询。

// 探索HubSpot App支持的所有Actions const actions = await zapier.apps.hubspot.listActions(); console.log(actions.data.map(a => ({ key: a.key, type: a.type, description: a.description })));

5. 部署、测试与扩展实践

5.1 环境配置与部署

本项目基于Output.ai平台运行，部署流程相对简单。

1. 安装依赖：npm install
2. 配置环境变量：创建.env文件，填入必要的密钥。

   ZAPIER_CLIENT_ID=your_client_id_here ZAPIER_CLIENT_SECRET=your_client_secret_here APOLLO_API_KEY=your_apollo_api_key_here OUTPUTAI_API_KEY=your_outputai_api_key_here ANTHROPIC_API_KEY=your_anthropic_api_key_here # 如果直接调用Claude

3. 链接Output.ai项目：在Output.ai平台创建新项目，并按照指引将本地代码库与其关联。
4. 部署：使用Output.ai CLI命令output deploy将工作流部署到云端。部署后，你会获得一个可以HTTP触发的端点URL，或者可以在平台的工作流画布中手动触发测试。

5.2 编写与执行测试场景

Output.ai支持使用JSON文件定义测试场景，这对于复杂工作流的调试至关重要。

// scenarios/stripe.json { "input": { "companyName": "Stripe", "website": "https://stripe.com" }, "expectedOutput": { // 你可以定义一些断言，但非强制 "companyName": "Stripe", "action": { "$oneOf": ["created", "updated"] } } }

在Output.ai平台的测试界面导入这个场景并运行，可以直观地看到每一步的执行结果、耗时和中间数据，方便定位问题。

5.3 常见问题排查与调试技巧

在实际运行中，你可能会遇到以下问题：

调试黄金法则：充分利用Output.ai平台提供的分布式追踪功能。执行一次工作流后，你可以清晰地看到一个甘特图，展示每个步骤的开始、结束时间、输入、输出和任何错误信息。这是定位性能瓶颈和逻辑错误的最有效工具。

5.4 模式扩展与变体

这个“直接API + Zapier SDK + LLM粘合”的模式具有强大的可扩展性。

1. 更换数据源：将Apollo替换为Clearbit、Hunter.io或你的内部数据库，只需重写步骤1的客户端和数据处理逻辑。
2. 更换目标CRM：将appKey从hubspot改为salesforce、pipedrive或zoho_crm，并调整对应的actionKey和字段映射，即可将数据写入另一个CRM系统。行业映射的逻辑完全通用。
3. 增加后续动作：基于action字段，可以轻松扩展工作流。例如，在upsertHubspotCompany之后增加一个条件步骤：

   if (result.action === 'created') { await sendSlackNotification({ channel: '#sales-leads', text: `新公司已创建: ${result.companyName} (${result.website})`, }); }

4. 处理更多字段映射：不仅仅是行业，你还可以用同样的listInputFieldChoices + LLM模式处理“国家/地区”、“融资阶段”、“客户生命周期阶段”等任何需要从自由文本映射到预设枚举值的字段。
5. 批量处理：当前工作流处理单条记录。你可以很容易地将其包装在一个循环中，从一个CSV文件或Google Sheets读取一批网站域名进行批量补全。

这个工作流展示了一种现代、松散耦合、智能化的集成方式。它用最小的开发维护成本，解决了数据丰富化与系统集成中的经典难题。希望这个详细的拆解能为你构建自己的自动化流程提供扎实的参考。