CRM集成开发指南：Close、HubSpot、Salesforce API对接与双向同步实战

crm-integration by scientiacapital/skills

147 周安装量

7 GitHub Stars

GitHub

安装命令

npx skills add https://github.com/scientiacapital/skills --skill crm-integration

自动化销售 API

🇨🇳中文介绍

Close CRM - 中小型企业销售的日常首选（最简单的 API，最佳性价比）
HubSpot - 营销与销售协同，拥有丰富的生态系统
Salesforce - 满足企业需求和复杂工作流
跨 CRM 同步 - 具备冲突解决功能的双向同步

关键交付成果：

配置具有适当身份验证的 API 客户端
针对线索、联系人、交易、活动的 CRUD 操作
用于实时同步的 Webhook 处理器
管道自动化和报告

<quick_start> Close CRM (API 密钥认证):

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")

广告位招租

在这里展示您的产品或服务

触达数万 AI 开发者，精准高效

联系我们

相关 Skills

FlyClaw：零登录航班聚合查询工具，Python实现多源航班信息与价格搜索

4,000,000 周安装

Azure RBAC 权限管理工具：查找最小角色、创建自定义角色与自动化分配

135,700 周安装

GitHub Actions 官方文档查询助手 - 精准解答 CI/CD 工作流问题

43,400 周安装

通过 LiteLLM 代理让 Claude Code 对接 GitHub Copilot 运行 | 高级变通方案指南

43,100 周安装

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"]

功能	Close	HubSpot	Salesforce
认证	API 密钥	OAuth 2.0 / 私有应用	JWT Bearer
速率限制	100 次请求/10秒	100-200 次请求/10秒（按套餐）	10万次请求/天
最适合	中小型企业销售，简单性	Tim 的主要 CRM（通过 Epiphan CRM MCP）	企业
起始价格	$49/用户/月	免费（有限制）	$25/用户/月
API 访问	所有套餐	Starter+ ($45+)	所有套餐
Webhooks	所有套餐	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>

# 创建包含联系人的线索
lead = close.create_lead({
    "name": "ABC Mechanical",
    "url": "https://abcmech.com",
    "contacts": [{
        "name": "John Smith",
        "title": "Owner",
        "emails": [{"email": "john@abcmech.com", "type": "office"}],
        "phones": [{"phone": "555-1234", "type": "office"}]
    }],
    "custom.cf_tier": "Gold",
    "custom.cf_source": "sales-agent"
})

# 创建机会
opp = close._request("POST", "/opportunity/", json={
    "lead_id": lead["id"],
    "value": 50000,
    "confidence": 50,
    "status_id": "stat_xxx"  # Pipeline stage
})

# 记录活动
close._request("POST", "/activity/note/", json={
    "lead_id": lead["id"],
    "note": "Initial discovery call - interested in demo"
})

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"])

# 创建关联交易
deal = client.crm.deals.basic_api.create(
    SimplePublicObjectInputForCreate(properties={
        "dealname": "Enterprise Deal",
        "amount": "50000",
        "dealstage": "appointmentscheduled",
        "pipeline": "default"
    })
)

# 按邮箱域名搜索联系人
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

def create_opportunity(self, data: dict) -> dict:
    """必需字段：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()

# 复合 API（批量最多 200 条记录）
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}
    )

from fastapi import FastAPI, Request, HTTPException
import hmac, hashlib

app = FastAPI()

@app.post("/webhooks/close")
async def close_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("Close-Sig")

    expected = hmac.new(
        CLOSE_WEBHOOK_SECRET.encode(), body, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        raise HTTPException(401, "Invalid signature")

    data = await request.json()
    event_type = data["event"]["event_type"]

    handlers = {
        "lead.created": handle_lead_created,
        "opportunity.status_changed": handle_opp_stage_change,
    }

    if handler := handlers.get(event_type):
        await handler(data["event"]["data"])

    return {"status": "ok"}

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│   Close     │────▶│  同步层      │◀────│  HubSpot    │
│  (主要)     │◀────│  (Postgres)  │────▶│  (营销)     │
└─────────────┘     └──────────────┘     └─────────────┘

工具	用途
`hubspot_search_companies`	按名称或域名查找公司
`hubspot_search_contacts`	按邮箱或姓名查找联系人
`hubspot_search_deals`	按名称或采购订单号查找交易
`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`	按注册邮箱查找设备

工具	用途
`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-existing-search`	轮询获取丰富结果（检查 `state: completed`）
`ask-question-about-accounts`	对 Salesforce 账户数据进行 AI 分析
`get-my-accounts`	按筛选条件搜索 Salesforce 账户
`get-task`	按 taskId 检索任务状态和结果

工具	用途
`ask_agent`	用于复杂 CRM/分析查询的 AI 代理 — 活动历史、互动时间线、交易智能

用户需要数据丰富 / 联系人数据： → 使用 Clay MCP 瀑布模式（首选瀑布模式，积分可用） → 工作流：find-and-enrich-contacts-at-company → add-contact-data-points → 轮询 get-existing-search 获取结果 → 备用方案：Apollo MCP 用于快速免费的数据丰富（无需轮询） → 参考：参见上方“集成点”中的“Clay MCP 数据丰富” → 成本：Apollo 免费，Clay 约 $150-300/月

用户需要 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

用户需要电话验证 / 瀑布式数据丰富： → 在 Apollo 之后使用 Clay MCP：find-and-enrich-contacts-at-company → add-contact-data-points 用于邮箱/电话 → 轮询结果 → Clay 聚合了 50 多个供应商，在电话和邮箱方面具有高匹配率 → 完整实现请参见 phone-verification-waterfall-skill

🇺🇸English

Close CRM - Daily driver for SMB sales (simplest API, best value)
HubSpot - Marketing + Sales alignment with rich ecosystem
Salesforce - Enterprise requirements and complex workflows
Cross-CRM Sync - Bidirectional sync with conflict resolution

Key deliverables:

API client setup with proper authentication
CRUD operations for leads, contacts, deals, activities
Webhook handlers for real-time sync
Pipeline automation and reporting

<quick_start> Close CRM (API Key Auth):

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": "user@example.com",
        "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> A CRM integration is successful when:

API authentication works without errors
CRUD operations complete for all entity types
Rate limits are respected (Close: 100 req/10s, HubSpot: varies by tier)
Webhooks fire and process correctly
Data syncs bidirectionally without duplicates </success_criteria>

<crm_comparison>

Platform Comparison

Feature	Close	HubSpot	Salesforce
Auth	API Key	OAuth 2.0 / Private App	JWT Bearer
Rate Limit	100 req/10s	100-200 req/10s by tier	100k req/day
Best For	SMB sales, simplicity	Tim's primary CRM (via Epiphan CRM MCP)	Enterprise
Starting Price	$49/user/mo	Free (limited)	$25/user/mo
API Access	All plans	Starter+ ($45+)	All plans
Webhooks	All plans	Pro+ ($800+)

Entity Mapping

Concept	Close	HubSpot	Salesforce
Company	`lead`	`company`	`Account`
Person	`contact`	`contact`	`Contact` / `Lead`
Deal

Pipeline Stage Mapping

Stage	Close	HubSpot	Salesforce
New	`Lead`	`appointmentscheduled`	`Prospecting`
Qualified	`Contacted`	`qualifiedtobuy`	`Qualification`
Demo	`Opportunity`

<close_patterns>

Close CRM (Daily Driver)

Note: Tim's primary CRM is HubSpot via Epiphan CRM MCP. Close CRM patterns are retained for reference but are not the active workflow.

Query Language (for Smart Views)

# Leads with no activity in 30 days
'sort:date_updated asc date_updated < "30 days ago"'

# High-value opportunities
'opportunities.value >= 50000 opportunities.status_type:active'

# Custom field filtering
'custom.cf_industry = "MEP Contractor"'

# Multiple trade types (your ICP)
'custom.cf_trades:HVAC OR custom.cf_trades:Electrical'

Core Operations

# Create lead with contacts
lead = close.create_lead({
    "name": "ABC Mechanical",
    "url": "https://abcmech.com",
    "contacts": [{
        "name": "John Smith",
        "title": "Owner",
        "emails": [{"email": "john@abcmech.com", "type": "office"}],
        "phones": [{"phone": "555-1234", "type": "office"}]
    }],
    "custom.cf_tier": "Gold",
    "custom.cf_source": "sales-agent"
})

# Create opportunity
opp = close._request("POST", "/opportunity/", json={
    "lead_id": lead["id"],
    "value": 50000,
    "confidence": 50,
    "status_id": "stat_xxx"  # Pipeline stage
})

# Log activity
close._request("POST", "/activity/note/", json={
    "lead_id": lead["id"],
    "note": "Initial discovery call - interested in demo"
})

Rate Limit Headers (RFC-compliant)

X-RateLimit-Limit: 100
X-RateLimit-Remaining: 95
X-RateLimit-Reset: 1704067200

See reference/close-deep-dive.md for query language, Smart Views, sequences, and reporting. </close_patterns>

<hubspot_patterns>

HubSpot Integration

Python SDK Pattern

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)

Association Types

From	To	Type ID
Contact	Company	1
Contact	Deal	4
Company	Deal	6
Deal	Contact	3

See reference/hubspot-patterns.md for batch operations, custom properties, and workflows. </hubspot_patterns>

<salesforce_patterns>

Salesforce Integration

SOQL Query Patterns

-- 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}
    )

See reference/salesforce-patterns.md for JWT setup, Platform Events, and bulk API. </salesforce_patterns>

<webhook_patterns>

Webhook Handlers

Close Webhook (FastAPI)

from fastapi import FastAPI, Request, HTTPException
import hmac, hashlib

app = FastAPI()

@app.post("/webhooks/close")
async def close_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("Close-Sig")

    expected = hmac.new(
        CLOSE_WEBHOOK_SECRET.encode(), body, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        raise HTTPException(401, "Invalid signature")

    data = await request.json()
    event_type = data["event"]["event_type"]

    handlers = {
        "lead.created": handle_lead_created,
        "opportunity.status_changed": handle_opp_stage_change,
    }

    if handler := handlers.get(event_type):
        await handler(data["event"]["data"])

    return {"status": "ok"}

Close Webhook Events

lead.created, lead.updated, lead.deleted, lead.status_changed
contact.created, contact.updated
opportunity.created, opportunity.status_changed
activity.note.created, activity.call.created, activity.email.created
unsubscribed_email.created

</webhook_patterns>

<sync_architecture>

Cross-CRM Sync

Architecture

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

Sync Record Schema

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);

Conflict Resolution

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

See reference/sync-patterns.md for deduplication, migration scripts, and bulk sync. </sync_architecture>

<integration_points>

Integration Points (MCP Tools Available)

HubSpot / Epiphan CRM

Tool	Purpose
`hubspot_search_companies`	Find companies by name or domain
`hubspot_search_contacts`	Find contacts by email or name
`hubspot_search_deals`	Find deals by name or PO number
`hubspot_get_company`	Fetch company details by HubSpot ID
`hubspot_get_contact`	Fetch contact details by HubSpot ID
`hubspot_get_deal`

Clay MCP Enrichment (mcp__claude_ai_Clay__)

Pattern: HubSpot → Apollo → Clay → HubSpot sync

Tool	Purpose
`find-and-enrich-company`	Find and enrich company by domain or LinkedIn URL
`find-and-enrich-contacts-at-company`	Find contacts by role/title/location at a company
`find-and-enrich-list-of-contacts`	Find specific named contacts at their companies
`add-contact-data-points`	Queue contact enrichment (Email, Phone, Work History, Thought Leadership)
`add-company-data-points`	Queue company enrichment (Tech Stack, Funding, Headcount, Competitors, etc.)
`get-existing-search`

Cost Model:

Apollo: Free (but rate-limited)
Clay: Credits-based (~$150-300/month typical usage for BDR teams)
Waterfall strategy: Try Apollo first (fast, free), fallback to Clay for missing data (phones, emails, work history)

Call Data & Intelligence

Tool	Purpose
`ask_agent`	AI agent for complex CRM/analytics queries — activity history, engagement timelines, deal intelligence

</integration_points>

<file_locations>

Reference Files

CRM-Specific:

reference/close-deep-dive.md - Query language, Smart Views, sequences, reporting
reference/hubspot-patterns.md - SDK patterns, batch operations, workflows
reference/salesforce-patterns.md - JWT auth, SOQL, Platform Events, bulk API

Operations:

reference/sync-patterns.md - Cross-CRM sync, deduplication, migration
reference/automation.md - Webhook setup, sequences, workflows

Templates:

templates/close-client.py - Full Close API client
templates/hubspot-client.py - HubSpot SDK wrapper
templates/sync-service.py - Cross-CRM sync service </file_locations>

User wants CRM integration: → Default to HubSpot (Epiphan CRM MCP) for Tim's BDR workflow → Provide auth setup + basic CRUD

User wants enrichment / contact data: → Use Clay MCP waterfall pattern (preferred for waterfall, credits OK) → Workflow: find-and-enrich-contacts-at-company → add-contact-data-points → poll get-existing-search for results → Fallback: Apollo MCP for quick free enrichment (no polling needed) → Reference: See "Clay MCP Enrichment" in Integration Points above → Cost: Apollo free, Clay $150-300/mo estimate

User wants Close CRM: → Provide API key setup, query language → Reference: reference/close-deep-dive.md

User wants HubSpot: → Use Epiphan CRM MCP tools for direct integration → Available tools: hubspot_search_companies, hubspot_search_contacts, hubspot_search_deals, hubspot_get_company, hubspot_get_contact, hubspot_get_deal → Company identification: crm_search_customers (fuzzy matching) → Activity data: ask_agent (CRM/analytics AI queries) → Enrichment: Clay MCP for waterfall enrichment before writeback to HubSpot → Reference: reference/hubspot-patterns.md

User wants Salesforce: → Provide JWT auth, SOQL patterns → Reference: reference/salesforce-patterns.md

User wants sync between CRMs: → Provide sync architecture, conflict resolution → Reference: reference/sync-patterns.md

User wants webhooks: → Provide handler pattern for specified CRM → Include signature verification

User wants phone verification / waterfall enrichment: → Use Clay MCP after Apollo: find-and-enrich-contacts-at-company → add-contact-data-points for Email/Phone → poll results → Clay aggregates 50+ providers for high match rates on phones and emails → See phone-verification-waterfall-skill for full implementation

<clay_mcp_pattern> See reference/clay-enrichment-patterns.md for Clay MCP waterfall enrichment workflow, tool prefix reference, cost considerations, env setup, and example session. </clay_mcp_pattern>

Emit Outcome Sidecar

Write to ~/.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]"}

Weekly Installs

127

Repository

scientiacapital/skills

GitHub Stars

First Seen

Jan 23, 2026

Security Audits

Gen Agent Trust HubPass SocketPass SnykWarn

Installed on

gemini-cli120

opencode118

codex116

github-copilot110

cursor109

amp105

CRM集成开发指南：Close、HubSpot、Salesforce API对接与双向同步实战

🇨🇳中文介绍

相关 Skills

平台对比

实体映射

管道阶段映射

Close CRM (日常首选)

查询语言（用于智能视图）

核心操作

速率限制头部信息（符合 RFC 标准）

HubSpot 集成

Python SDK 模式

关联类型

Salesforce 集成

SOQL 查询模式

REST API v59.0

Webhook 处理器

Close Webhook (FastAPI)

Close Webhook 事件

跨 CRM 同步

架构

同步记录模式

冲突解决

集成点（可用的 MCP 工具）

HubSpot / Epiphan CRM

Clay MCP 数据丰富 (mcp__claude_ai_Clay__)

通话数据与智能

参考文件

发出结果 Sidecar

🇺🇸English

Platform Comparison

Entity Mapping

Pipeline Stage Mapping

Close CRM (Daily Driver)

Query Language (for Smart Views)

Core Operations

Rate Limit Headers (RFC-compliant)

HubSpot Integration

Python SDK Pattern

Association Types

Salesforce Integration

SOQL Query Patterns

REST API v59.0

Webhook Handlers

Close Webhook (FastAPI)

Close Webhook Events

Cross-CRM Sync

Architecture

Sync Record Schema

Conflict Resolution

Integration Points (MCP Tools Available)

HubSpot / Epiphan CRM

Clay MCP Enrichment (mcp__claude_ai_Clay__)

Call Data & Intelligence

Reference Files

Emit Outcome Sidecar

最新 Skills