crm-integration by scientiacapital/skills
npx skills add https://github.com/scientiacapital/skills --skill crm-integration关键交付成果:
<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 开发者,精准高效
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> 当满足以下条件时,CRM 集成被视为成功:
<crm_comparison>
| 功能 | 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> |
<close_patterns>
注意: Tim 的主要 CRM 是通过 Epiphan CRM MCP 使用的 HubSpot。保留 Close CRM 模式供参考,但并非当前活跃的工作流。
# 30 天内无活动的线索
'sort:date_updated asc date_updated < "30 days ago"'
# 高价值机会
'opportunities.value >= 50000 opportunities.status_type:active'
# 自定义字段筛选
'custom.cf_industry = "MEP Contractor"'
# 多种行业类型(您的理想客户画像)
'custom.cf_trades:HVAC OR custom.cf_trades:Electrical'
# 创建包含联系人的线索
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"
})
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 95
X-RateLimit-Reset: 1704067200
有关查询语言、智能视图、序列和报告,请参见
reference/close-deep-dive.md。 </close_patterns>
<hubspot_patterns>
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 |
有关批量操作、自定义属性和工作流,请参见
reference/hubspot-patterns.md。 </hubspot_patterns>
<salesforce_patterns>
-- 父子关系(账户的联系人)
SELECT Id, Name, (SELECT LastName, Email FROM Contacts)
FROM Account WHERE Industry = 'Technology'
-- 子父关系
SELECT Id, FirstName, Account.Name, Account.Industry
FROM Contact WHERE Account.Industry = 'Technology'
-- 半连接(有关联未关闭机会的账户)
SELECT Id, Name FROM Account
WHERE Id IN (SELECT AccountId FROM Opportunity WHERE IsClosed = false)
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}
)
有关 JWT 设置、平台事件和批量 API,请参见
reference/salesforce-patterns.md。 </salesforce_patterns>
<webhook_patterns>
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"}
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>
┌─────────────┐ ┌──────────────┐ ┌─────────────┐
│ Close │────▶│ 同步层 │◀────│ HubSpot │
│ (主要) │◀────│ (Postgres) │────▶│ (营销) │
└─────────────┘ └──────────────┘ └─────────────┘
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 作为事实来源
LAST_WRITE_WINS = "lww" # 最近的更新获胜
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>
| 工具 | 用途 |
|---|---|
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 | 按注册邮箱查找设备 |
模式: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-existing-search | 轮询获取丰富结果(检查 state: completed) |
ask-question-about-accounts | 对 Salesforce 账户数据进行 AI 分析 |
get-my-accounts | 按筛选条件搜索 Salesforce 账户 |
get-task | 按 taskId 检索任务状态和结果 |
成本模型:
| 工具 | 用途 |
|---|---|
ask_agent | 用于复杂 CRM/分析查询的 AI 代理 — 活动历史、互动时间线、交易智能 |
</integration_points>
<file_locations>
CRM 特定:
reference/close-deep-dive.md - 查询语言、智能视图、序列、报告reference/hubspot-patterns.md - SDK 模式、批量操作、工作流reference/salesforce-patterns.md - JWT 认证、SOQL、平台事件、批量 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-existing-search 获取结果 → 备用方案: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
用户需要 Webhooks: → 为指定的 CRM 提供处理器模式 → 包含签名验证
用户需要电话验证 / 瀑布式数据丰富: → 在 Apollo 之后使用 Clay MCP:find-and-enrich-contacts-at-company → add-contact-data-points 用于邮箱/电话 → 轮询结果 → Clay 聚合了 50 多个供应商,在电话和邮箱方面具有高匹配率 → 完整实现请参见 phone-verification-waterfall-skill
<clay_mcp_pattern> 有关 Clay MCP 瀑布式数据丰富工作流、工具前缀参考、成本考虑、环境设置和示例会话,请参见 reference/clay-enrichment-patterns.md。 </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]"}
每周安装次数
127
仓库
GitHub 星标
6
首次出现
2026年1月23日
安全审计
安装于
gemini-cli120
opencode118
codex116
github-copilot110
cursor109
amp105
Key deliverables:
<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:
<crm_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+) |
| Concept | Close | HubSpot | Salesforce |
|---|---|---|---|
| Company | lead | company | Account |
| Person | contact | contact | Contact / Lead |
| Deal |
| Stage | Close | HubSpot | Salesforce |
|---|---|---|---|
| New | Lead | appointmentscheduled | Prospecting |
| Qualified | Contacted | qualifiedtobuy | Qualification |
| Demo | Opportunity |
<close_patterns>
Note: Tim's primary CRM is HubSpot via Epiphan CRM MCP. Close CRM patterns are retained for reference but are not the active workflow.
# 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'
# 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"
})
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 95
X-RateLimit-Reset: 1704067200
See
reference/close-deep-dive.mdfor query language, Smart Views, sequences, and reporting. </close_patterns>
<hubspot_patterns>
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)
| From | To | Type ID |
|---|---|---|
| Contact | Company | 1 |
| Contact | Deal | 4 |
| Company | Deal | 6 |
| Deal | Contact | 3 |
See
reference/hubspot-patterns.mdfor batch operations, custom properties, and workflows. </hubspot_patterns>
<salesforce_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)
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.mdfor JWT setup, Platform Events, and bulk API. </salesforce_patterns>
<webhook_patterns>
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"}
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>
┌─────────────┐ ┌──────────────┐ ┌─────────────┐
│ 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
See
reference/sync-patterns.mdfor deduplication, migration scripts, and bulk sync. </sync_architecture>
<integration_points>
| 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 |
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:
| Tool | Purpose |
|---|---|
ask_agent | AI agent for complex CRM/analytics queries — activity history, engagement timelines, deal intelligence |
</integration_points>
<file_locations>
CRM-Specific:
reference/close-deep-dive.md - Query language, Smart Views, sequences, reportingreference/hubspot-patterns.md - SDK patterns, batch operations, workflowsreference/salesforce-patterns.md - JWT auth, SOQL, Platform Events, bulk APIOperations:
reference/sync-patterns.md - Cross-CRM sync, deduplication, migrationreference/automation.md - Webhook setup, sequences, workflowsTemplates:
templates/close-client.py - Full Close API clienttemplates/hubspot-client.py - HubSpot SDK wrappertemplates/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>
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
GitHub Stars
6
First Seen
Jan 23, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykWarn
Installed on
gemini-cli120
opencode118
codex116
github-copilot110
cursor109
amp105
Skills CLI 使用指南:AI Agent 技能包管理器安装与管理教程
43,100 周安装
App Store Connect ID 解析器 - 快速映射应用、构建、版本等ID,提升开发效率
1,500 周安装
代币集成分析器 - 智能合约安全审计工具,检测ERC20/ERC721漏洞与奇怪代币风险
1,500 周安装
智能合约规范代码合规性检查工具 | 区块链审计与安全验证
1,600 周安装
Web性能优化指南:提升网站加载速度与核心Web指标,改善SEO排名和用户体验
1,500 周安装
App Store Connect TestFlight 编排工具 - 自动化管理测试员、群组和构建分发
1,600 周安装
SARIF 文件解析指南:静态分析结果处理、去重与CI/CD集成
1,500 周安装
| All plans |
opportunity |
deal |
Opportunity |
| Activity | activity | engagement | Task / Event |
| Custom Field | custom.cf_xxx | properties | Field__c |
presentationscheduled |
Needs Analysis |
| Proposal | Proposal | decisionmakerboughtin | Proposal/Price Quote |
| Won | Won | closedwon | Closed Won |
| Lost | Lost | closedlost | Closed Lost |
| </crm_comparison> |
| Fetch deal details by HubSpot ID |
crm_search_customers | Search customers (fuzzy-match company names) |
crm_get_customer | Get customer details by CRM ID |
crm_search_customers | Search customers by company name or email |
crm_get_order | Get order details by order ID |
crm_get_customer_orders | Get recent orders for a customer |
analytics_get_device | Get device details by serial number |
analytics_search_by_email | Find devices registered by email |
Poll for enrichment results (check state: completed) |
ask-question-about-accounts | AI analysis of Salesforce account data |
get-my-accounts | Search Salesforce accounts by filters |
get-task | Retrieve task status and results by taskId |