crm-integration

与 CRM 平台集成，实现销售自动化工作流：

1. Close CRM - SMB 销售的日常主力工具（API 最简单，性价比最高）
2. HubSpot - 借助丰富的生态系统，实现营销与销售协同
3. Salesforce - 满足企业需求和复杂工作流
4. 跨 CRM 同步 - 支持冲突解决的双向同步

主要交付内容：

• 配置 API 客户端并使用适当的身份验证
• 对潜在客户、联系人、交易和活动执行 CRUD 操作
• 用于实时同步的 Webhook 处理程序
• 流水线自动化和报告

<quick_start> Close CRM（API Key 身份验证）：

import httpx

class CloseClient:
    BASE_URL = "https://api.close.com/api/v1"

    def __init__(self, api_key: str):
        self.client = httpx.Client(
            base_url=self.BASE_URL,
            auth=(api_key, ""),  # Basic auth, password empty
            timeout=30.0,
        )

    def create_lead(self, data: dict) -> dict:
        response = self.client.post("/lead/", json=data)
        response.raise_for_status()
        return response.json()

    def search_leads(self, query: str) -> list:
        response = self.client.post("/data/search/", json={
            "query": {"type": "query_string", "value": query},
            "results_limit": 100
        })
        return response.json()["data"]

# Usage
close = CloseClient(os.environ["CLOSE_API_KEY"])
leads = close.search_leads("company:Coperniq")

HubSpot（Python SDK）：

from hubspot import HubSpot
from hubspot.crm.contacts import SimplePublicObjectInputForCreate

client = HubSpot(access_token=os.environ["HUBSPOT_ACCESS_TOKEN"])

# Create contact
contact = client.crm.contacts.basic_api.create(
    SimplePublicObjectInputForCreate(properties={
        "email": "[email protected]",
        "firstname": "Jane",
        "lastname": "Smith"
    })
)
print(f"Created: {contact.id}")

Salesforce（JWT Bearer）：

import jwt
from datetime import datetime, timedelta

class SalesforceClient:
    def __init__(self, client_id: str, username: str, private_key: str):
        self.auth_url = "https://login.salesforce.com"
        self._authenticate(client_id, username, private_key)

    def _authenticate(self, client_id, username, private_key):
        payload = {
            "iss": client_id,
            "sub": username,
            "aud": self.auth_url,
            "exp": int((datetime.utcnow() + timedelta(minutes=3)).timestamp())
        }
        assertion = jwt.encode(payload, private_key, algorithm="RS256")

        response = httpx.post(f"{self.auth_url}/services/oauth2/token", data={
            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
            "assertion": assertion
        })
        self.access_token = response.json()["access_token"]
        self.instance_url = response.json()["instance_url"]

</quick_start>

<success_criteria> CRM 集成在满足以下条件时视为成功：

• API 身份验证成功，无错误
• 所有实体类型均可完成 CRUD 操作
• 遵守速率限制（Close：100 req/10s，HubSpot：因套餐等级而异）
• Webhook 能够正确触发和处理
• 数据能够双向同步，且不会产生重复项 </success_criteria>

<crm_comparison>

平台比较

功能	Close	HubSpot	Salesforce
身份验证	API Key	OAuth 2.0 / Private App	JWT Bearer
速率限制	100 req/10s	按套餐等级为 100-200 req/10s	100k req/day
适用场景	SMB 销售、简洁易用	Tim 的主要 CRM（通过 Epiphan CRM MCP）	企业
起始价格	$49/user/mo	免费（功能有限）	$25/user/mo
API 访问权限	所有套餐	Starter+（$45+）	所有套餐
Webhook	所有套餐	Pro+（$800+）	所有套餐

实体映射

概念	Close	HubSpot	Salesforce
公司	lead	company	Account
人员	contact	contact	Contact / Lead
交易	opportunity	deal	Opportunity
活动	activity	engagement	Task / Event
自定义字段	custom.cf_xxx	properties	Field__c

流水线阶段映射

阶段	Close	HubSpot	Salesforce
新建	Lead	appointmentscheduled	Prospecting
已确认资格	Contacted	qualifiedtobuy	Qualification
演示	Opportunity	presentationscheduled	Needs Analysis
提案	Proposal	decisionmakerboughtin	Proposal/Price Quote
成交	Won	closedwon	Closed Won
未成交	Lost	closedlost	Closed Lost
</crm_comparison>

<close_patterns>

Close CRM（日常主力工具）

注意： Tim 的主要 CRM 是通过 Epiphan CRM MCP 接入的 HubSpot。此处保留 Close CRM 模式供参考，但它并非当前工作流。

查询语言示例、核心操作（潜在客户、商机、活动）和速率限制标头，请参阅 reference/close-crm-examples.md。

查询语言、Smart Views、序列和报告，请参阅 reference/close-deep-dive.md。 </close_patterns>

<hubspot_patterns>

HubSpot 集成

Python SDK 模式

from hubspot import HubSpot
from hubspot.crm.deals import SimplePublicObjectInputForCreate
from hubspot.crm.contacts import PublicObjectSearchRequest

client = HubSpot(access_token=os.environ["HUBSPOT_ACCESS_TOKEN"])

# Create deal with association
deal = client.crm.deals.basic_api.create(
    SimplePublicObjectInputForCreate(properties={
        "dealname": "Enterprise Deal",
        "amount": "50000",
        "dealstage": "appointmentscheduled",
        "pipeline": "default"
    })
)

# Search contacts by email domain
search = PublicObjectSearchRequest(
    filter_groups=[{
        "filters": [{
            "propertyName": "email",
            "operator": "CONTAINS",
            "value": "@example.com"
        }]
    }],
    properties=["email", "firstname", "lastname"],
    limit=50
)
results = client.crm.contacts.search_api.do_search(search)

关联类型

起点	终点	类型 ID
联系人	公司	1
联系人	交易	4
公司	交易	6
交易	联系人	3

批量操作、自定义属性和工作流，请参阅 reference/hubspot-patterns.md。 </hubspot_patterns>

<salesforce_patterns>

Salesforce 集成

SOQL 查询模式

-- Parent-child relationship (Contacts of Account)
SELECT Id, Name, (SELECT LastName, Email FROM Contacts)
FROM Account WHERE Industry = 'Technology'

-- Child-parent relationship
SELECT Id, FirstName, Account.Name, Account.Industry
FROM Contact WHERE Account.Industry = 'Technology'

-- Semi-join (Accounts with open Opportunities)
SELECT Id, Name FROM Account
WHERE Id IN (SELECT AccountId FROM Opportunity WHERE IsClosed = false)

REST API v59.0

def create_opportunity(self, data: dict) -> dict:
    """Required: Name, StageName, CloseDate."""
    response = self.client.post(
        f"{self.instance_url}/services/data/v59.0/sobjects/Opportunity/",
        headers={"Authorization": f"Bearer {self.access_token}"},
        json=data
    )
    return response.json()

# Composite API (batch up to 200 records)
def composite_create(self, records: list) -> dict:
    return self.client.post(
        f"{self.instance_url}/services/data/v59.0/composite/sobjects",
        json={"allOrNone": False, "records": records}
    )

JWT 配置、Platform Events 和批量 API，请参阅 reference/salesforce-patterns.md。 </salesforce_patterns>

<webhook_patterns>

Webhook 处理程序

Close Webhook FastAPI 处理程序（含签名验证）以及完整的 Webhook 事件列表，请参阅 reference/webhook-handlers.md。 </webhook_patterns>

<sync_architecture>

跨 CRM 同步

架构

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│   Close     │────▶│  Sync Layer  │◀────│  HubSpot    │
│  (Primary)  │◀────│  (Postgres)  │────▶│  (Marketing)│
└─────────────┘     └──────────────┘     └─────────────┘

同步记录架构

CREATE TABLE crm_sync_records (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    entity_type VARCHAR(50) NOT NULL,
    close_id VARCHAR(100) UNIQUE,
    hubspot_id VARCHAR(100) UNIQUE,
    salesforce_id VARCHAR(100) UNIQUE,
    email VARCHAR(255),
    company_name VARCHAR(255),
    last_synced_at TIMESTAMPTZ,
    sync_source VARCHAR(50),
    sync_hash VARCHAR(64)
);

CREATE INDEX idx_sync_email ON crm_sync_records(email);

冲突解决

from enum import Enum

class ConflictStrategy(Enum):
    CLOSE_WINS = "close"      # Close is source of truth
    LAST_WRITE_WINS = "lww"   # Most recent update wins

def resolve_conflict(close_record, hubspot_record, strategy):
    if strategy == ConflictStrategy.CLOSE_WINS:
        merged = close_record.copy()
        for key, value in hubspot_record.items():
            if key not in merged or not merged[key]:
                merged[key] = value
        return merged

去重、迁移脚本和批量同步，请参阅 reference/sync-patterns.md。 </sync_architecture>

<integration_points>

集成点（可用的 MCP 工具）

HubSpot / Epiphan CRM

工具	用途
hubspot_search_companies	按名称或域名查找公司
hubspot_search_contacts	按电子邮件或姓名查找联系人
hubspot_search_deals	按名称或 PO 编号查找交易
hubspot_get_company	按 HubSpot ID 获取公司详情
hubspot_get_contact	按 HubSpot ID 获取联系人详情
hubspot_get_deal	按 HubSpot ID 获取交易详情
crm_search_customers	搜索客户（模糊匹配公司名称）
crm_get_customer	按 CRM ID 获取客户详情
crm_search_customers	按公司名称或电子邮件搜索客户
crm_get_order	按订单 ID 获取订单详情
crm_get_customer_orders	获取客户的近期订单
analytics_get_device	按序列号获取设备详情
analytics_search_by_email	查找使用该电子邮件注册的设备

Clay MCP 数据丰富 (mcp__claude_ai_Clay__)

模式：HubSpot → Apollo → Clay → HubSpot 同步

工具	用途
find-and-enrich-company	按域名或 LinkedIn URL 查找并丰富公司信息
find-and-enrich-contacts-at-company	按公司内的职位、头衔或地点查找联系人
find-and-enrich-list-of-contacts	在联系人所在公司查找指定联系人
add-contact-data-points	将联系人信息丰富任务加入队列（电子邮件、电话、工作经历、思想领导力）
add-company-data-points	将公司信息丰富任务加入队列（技术栈、融资、员工人数、竞争对手等）
get-task	轮询信息丰富结果（检查 state: completed）
ask-question-about-accounts	使用 AI 分析 Salesforce 账户数据
get-my-accounts	按筛选条件搜索 Salesforce 账户
get-task	按 taskId 获取任务状态和结果

费用模式：

• Apollo：免费（但有速率限制）
• Clay：基于点数收费（BDR 团队的典型用量约为每月 $150-300）
• 瀑布式策略：先尝试 Apollo（快速、免费），缺少数据时（电话、电子邮件、工作经历）再使用 Clay

通话数据与情报

工具	用途
ask_agent	用于处理复杂 CRM/分析查询的 AI Agent——活动历史记录、互动时间线、交易情报

</integration_points>

<file_locations>

参考文件

CRM 专用：

• reference/close-deep-dive.md - 查询语言、Smart Views、序列、报告
• reference/close-crm-examples.md - Close 查询语言、核心操作、速率限制标头
• reference/webhook-handlers.md - Close webhook FastAPI 处理程序和事件列表
• reference/hubspot-patterns.md - SDK 模式、批量操作、工作流
• reference/salesforce-patterns.md - JWT 身份验证、SOQL、Platform Events、批量 API

运维：

• reference/sync-patterns.md - 跨 CRM 同步、去重、迁移
• reference/automation.md - Webhook 设置、序列、工作流

模板：

• templates/close-client.py - 完整的 Close API 客户端
• templates/hubspot-client.py - HubSpot SDK 封装
• templates/sync-service.py - 跨 CRM 同步服务 </file_locations>

## 请求路由

用户需要 CRM 集成： → 默认使用 HubSpot（Epiphan CRM MCP）处理 Tim 的 BDR 工作流 → 提供身份验证设置和基本 CRUD

用户需要数据补充／联系人数据： → 使用 Clay MCP 瀑布式模式（首选瀑布式方案，额度可接受） → 工作流：find-and-enrich-contacts-at-company → add-contact-data-points → 轮询 get-task 获取结果 → 备用方案：使用 Apollo MCP 进行快速免费数据补充（无需轮询） → 参考：请参阅上文“集成点”中的“Clay MCP 数据补充” → 费用：Apollo 免费，Clay 预计每月 $150-300

用户需要 Close CRM： → 提供 API 密钥设置和查询语言 → 参考：reference/close-deep-dive.md

用户需要 HubSpot： → 使用 Epiphan CRM MCP 工具进行直接集成 → 可用工具：hubspot_search_companies、hubspot_search_contacts、hubspot_search_deals、hubspot_get_company、hubspot_get_contact、hubspot_get_deal → 公司识别：crm_search_customers（模糊匹配） → 活动数据：ask_agent（CRM／分析 AI 查询） → 数据补充：写回 HubSpot 前，使用 Clay MCP 进行瀑布式数据补充 → 参考：reference/hubspot-patterns.md

用户需要 Salesforce： → 提供 JWT 身份验证和 SOQL 模式 → 参考：reference/salesforce-patterns.md

用户需要在 CRM 之间同步： → 提供同步架构和冲突解决方案 → 参考：reference/sync-patterns.md

用户需要 Webhook： → 提供指定 CRM 的处理器模式 → 包含签名验证

用户需要电话验证／瀑布式数据补充： → Apollo 之后使用 Clay MCP：find-and-enrich-contacts-at-company → add-contact-data-points 获取 Email／Phone → 轮询结果 → Clay 汇总 50 多个提供商的数据，可提高电话和电子邮件的匹配率 → 完整实现请参阅 phone-verification-waterfall-skill

<clay_mcp_pattern> 请参阅 reference/clay-enrichment-patterns.md，了解 Clay MCP 瀑布式数据补充工作流、工具前缀参考、费用注意事项、环境设置和会话示例。 </clay_mcp_pattern>

写入结果附属文件

写入 ~/.claude/skill-analytics/last-outcome-crm-integration.json： {"ts":"[UTC ISO8601]","skill":"crm-integration","version":"1.0.0","variant":"default","status":"[success|partial|error]","runtime_ms":[ms],"metrics":{"integrations_configured":[n],"records_synced":[n]},"error":null,"session_id":"[YYYY-MM-DD]"}