software

组织层面的电子邮件营销活动不知何故仍是未解决的问题

发布: (2025年12月29日 GMT+8 09:06)
9 min read
原文: Dev.to

Source: Dev.to

组织层级电子邮件活动的封面图片

定义目标

我们最近在 Mintlify 推出了新的 社区 Discord 和 Slack 机器人，需要进行一些客户营销，让拥有大型社区的用户了解它们。

我们的目标是向每个组织中我们识别出的能够从该功能中获益的约 3‑10 人发送一系列邮件。要理解的是，这款产品的特性是，只要组织中的任意一位用户启用了机器人，任务即告完成，我们不想继续给该组织的其他人发送邮件。

我原以为使用任何现有的邮件营销工具（如 Resend 或 Loops）都会很容易实现，但事实并非如此。所有可用的工具令人惊讶地只能在用户层面处理活动，而不是在组织层面。

它们都没有“当任何成员采取行动时停止给该组发送邮件”的概念。这些工具都是为个人滴灌活动设计的，而不是组织级的外展。

工具选择

Claude Code 让我相当懒惰，因为我尝试用它完成所有能做的事，而不是手动自己做。因此，我在选择解决上述问题的工具时的首要标准是 大型 API 表面，Claude 能够使用的。

Instantly 成为我的最终选择，因为它的 API 最为强大且文档齐全。它让 Claude 完全访问活动、潜在客户和序列，同时还能在回复和退订事件上发送 webhook。

有点随意的旁注，但 JAMstack 架构模式可能会随着 AI 的兴起而卷土重来。我认为像 trpc 这样的工具相较于 OpenAPI 驱动的模式会逐渐失宠，因为 AI 代理更容易理解后者。UI 与业务逻辑的分离是前进的方向，如果我们希望我们的应用能被 AI 访问。

解决方案架构

Instantly 活动持有多邮件序列，并配置为在收到回复时停止。一个潜客上传脚本读取联系人 CSV，按公司分组，并使用 companyName 自定义变量上传到 Instantly。

随后，Webhook 服务器监听回复事件，查找同一公司的所有潜客，并将其标记为 “无兴趣”，以停止它们的序列。

CSV (company, email)
    → Upload Script
    → Instantly (leads with companyName)

Reply received
    → Instantly webhook
    → Webhook server
    → Find leads by companyName
    → Update lead status
    → Sequence stops for whole company

创建活动

活动本身很简单。将 stop_on_reply 设置为 true，这样 Instantly 会在收到回复的对象上停止序列，然后定义电子邮件步骤并在它们之间设置延迟。

curl -X POST https://api.instantly.ai/api/v2/campaigns \
  -H "Authorization: Bearer $INSTANTLY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Feature Launch Outreach",
    "stop_on_reply": true,
    "stop_on_auto_reply": true,
    "sequences": [{
      "steps": [
        {"type": "email", "delay": 0, "variants": [{"subject": "...", "body": "..."}]},
        {"type": "email", "delay": 3, "variants": [{"subject": "...", "body": "..."}]}
      ]
    }]
  }'

上传潜在客户

关键是为每个潜在客户包含一个 companyName 自定义变量。我们的 webhook 服务器使用它来查找同一公司的所有潜在客户，并在相关事件时取消订阅它们。

import csv
import requests

API_KEY = "your-api-key"
CAMPAIGN_ID = "your-campaign-id"

with open("contacts.csv") as f:
    reader = csv.DictReader(f)
    for row in reader:
        emails = row["emails"].split(";")
        company = row["company_name"]

        for email in emails:
            requests.post(
                "https://api.instantly.ai/api/v2/leads",
                headers={"Authorization": f"Bearer {API_KEY}"},
                json={
                    "email": email.strip(),
                    "company_name": company,
                    "custom_variables": {"companyName": company},
                    "campaign": CAMPAIGN_ID
                }
            )

注册 Webhook

告诉 Instantly 在有人回复时 POST 到你的服务器。你需要之前的活动 ID。

curl -X POST https://api.instantly.ai/api/v2/webhooks \
  -H "Authorization: Bearer $INSTANTLY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "target_hook_url": "https://your-server.com/webhook/reply",
    "event_type": "reply_received",
    "campaign_id": "your-campaign-id"
  }'

Webhook 服务器

这是使整个流程运作的部分。当任何人回复时，Webhook 服务器会提取公司名称，查询 Instantly 中所有具有该公司名称的线索，并将每个线索的状态更新为停止其序列。

// Example using Node.js + Express
const express = require('express');
const fetch = require('node-fetch');

const app = express();
app.use(express.json());

const INSTANTLY_API_KEY = process.env.INSTANTLY_API_KEY;

app.post('/webhook/reply', async (req, res) => {
  const { lead_id, campaign_id } = req.body;

  // 1️⃣ Get lead details (including custom variable `companyName`)
  const leadResp = await fetch(`https://api.instantly.ai/api/v2/leads/${lead_id}`, {
    headers: { Authorization: `Bearer ${INSTANTLY_API_KEY}` },
  });
  const leadData = await leadResp.json();
  const companyName = leadData.custom_variables?.companyName;

  if (!companyName) {
    return res.status(400).send('No companyName found on lead');
  }

  // 2️⃣ Find all leads in the same campaign with that companyName
  const searchResp = await fetch(
    `https://api.instantly.ai/api/v2/leads?campaign_id=${campaign_id}&custom_variables.companyName=${encodeURIComponent(companyName)}`,
    { headers: { Authorization: `Bearer ${INSTANTLY_API_KEY}` } }
  );
  const leads = await searchResp.json();

  // 3️⃣ Update each lead to “not interested” (or any status that stops the sequence)
  await Promise.all(
    leads.map(l =>
      fetch(`https://api.instantly.ai/api/v2/leads/${l.id}`, {
        method: 'PATCH',
        headers: {
          Authorization: `Bearer ${INSTANTLY_API_KEY}`,
          'Content-Type': 'application/json',
        },
        body: JSON.stringify({ status: 'not_interested' }),
      })
    )
  );

  res.sendStatus(200);
});

app.listen(3000, () => console.log('Webhook server listening on :3000'));

Source: …

TL;DR

  1. 上传线索 时使用 companyName 自定义变量。
  2. 创建活动 时设置 stop_on_reply: true。
  3. 注册 webhook，监听 reply_received 事件。
  4. Webhook 服务器 查找所有拥有相同 companyName 的线索，并将它们标记为 “不感兴趣”，从而停止整个组织的序列。
express = require('express');
const app = express();
app.use(express.json());

const API_KEY = process.env.INSTANTLY_API_KEY;
const CAMPAIGN_ID = process.env.CAMPAIGN_ID;

app.post('/webhook/reply', async (req, res) => {
  const event = req.body;

  if (event.event_type !== 'reply_received') {
    return res.json({ status: 'ignored' });
  }

  const leadEmail = event.lead_email;
  const companyName = event.lead?.company_name;

  if (!companyName) {
    return res.json({ status: 'ignored', reason: 'no company' });
  }

  // Find all leads from this company
  const leads = await findLeadsByCompany(companyName);

  // Stop all of them (except the one who replied, they're already stopped)
  for (const lead of leads) {
    if (lead.email !== leadEmail) {
      await stopLead(lead.email);
    }
  }

  res.json({ status: 'ok', stopped: leads.length - 1 });
});

// Instantly's API doesn't let you filter by company_name server‑side.
// The search param only works on name/email. So we fetch all leads
// and filter client‑side. For large campaigns, you'd want to cache
// this or build your own company → leads index.
async function findLeadsByCompany(companyName) {
  const leads = [];
  let cursor = null;

  do {
    const resp = await fetch('https://api.instantly.ai/api/v2/leads/list', {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${API_KEY}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({ campaign: CAMPAIGN_ID, starting_after: cursor })
    });
    const data = await resp.json();
    leads.push(...data.items);
    cursor = data.next_starting_after;
  } while (cursor);

  return leads.filter(lead => lead.company_name === companyName);
}

async function stopLead(email) {
  await fetch('https://api.instantly.ai/api/v2/leads/update-interest-status', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${API_KEY}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({ lead_email: email, interest_value: -1 })
  });
}

app.listen(3000);

你可以将此代码部署在任何能够接收 HTTP 请求的地方。我使用了带有 Caddy 进行 SSL 的廉价 VPS + Docker。其他可选方案包括 Cloudflare Workers、Railway、Render、AWS Lambda 或 Vercel。服务器是无状态的，只要能运行 Node.js 的托管环境都可以。

请帮我实现这个

这个方案可以工作，但它比实际需要的要复杂得多。为了实现组织级别的行为，我不得不搭建一个 webhook 服务器，这简直荒唐。每个 B2B 邮件工具都应该提供一个复选框来实现这一功能。

我真正想要的是按公司划分联系人组，在活动设置中切换 “回复时停止组”，并让邮件工具在没有 webhook 的情况下自行处理。

如果你们正在构建邮件工具，请加入此功能。

Back to Blog

相关文章

阅读更多 »

版本控制入门：U盘类比

封面图片：Version Control for Beginners：Pendrive 类比 https://media2.dev.to/dynamic/image/width=1000,height=420,fit=cover,gravity=auto,format=auto

Go的秘密生活：错误处理

第12章：碎玻璃的声音 星期一的早晨，厚重的灰色雾气笼罩着整座城市。档案馆内部，寂静至极，随后被打破……