crm-integration

営業自動化ワークフローのために、CRM プラットフォームと連携します。

1. Close CRM - SMB 営業の日常利用に最適（最もシンプルな API、最高のコストパフォーマンス）
2. HubSpot - 充実したエコシステムによるマーケティングと営業の連携
3. Salesforce - エンタープライズ要件と複雑なワークフロー
4. CRM 間の同期 - 競合解決機能を備えた双方向同期

主な成果物：

• 適切な認証を使用した API クライアントのセットアップ
• リード、コンタクト、商談、アクティビティの CRUD 操作
• リアルタイム同期のための Webhook ハンドラー
• パイプラインの自動化とレポート作成

ワークスペースのコンテキスト

質問する前に、ブートストラップコンテキストを確認してください。ブランド、対象者、オファー、チャネル、ツール、制約、指標については strategy/brand.md を、個人の文体については about/me.md を、コンテンツ計画については content/ideas.md と content/calendar.md を確認してください。レガシーのプロダクトマーケティング用コンテキストファイルは、フォールバックとしてのみ使用してください。生成した下書きは content/<platform>/drafts/YYYY-MM-DD_short-topic-slug.md に保存し、長期的に活用できる学びは strategy/brand.md、about/me.md、または content/ideas.md に反映してください。

運用上の契約

このスキルは、フロントマターで定義された範囲について自己完結しています。ローカルの指示、参照資料、スクリプト、アセットを手順書として使用してください。不足しているタスク固有の入力についてのみ質問し、範囲を拡大せずに関連スキルへ引き継ぎ、実行可能な成果物、判断、計画、下書き、または診断結果を返してください。

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

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 ベアラー）：

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 キー	OAuth 2.0 / プライベートアプリ	JWT ベアラー
レート制限	100 req/10s	プランにより100-200 req/10s	100k req/day
最適な用途	SMB 営業、シンプルさ	マーケティングと営業	エンタープライズ
開始価格	$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（日常利用）

クエリ言語（スマートビュー用）

# 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": "[email protected]", "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"
})

レート制限ヘッダー（RFC 準拠）

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

クエリ言語、スマートビュー、シーケンス、レポートについては、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)

関連付けタイプ

From	To	Type 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)

</salesforce_patterns>

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、Bulk API については、reference/salesforce-patterns.md を参照してください。 </salesforce_patterns>

<webhook_patterns>

Webhook ハンドラー

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 イベント

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>

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>

<file_locations>

参照ファイル

CRM 固有：

• reference/close-deep-dive.md - クエリ言語、Smart Views、シーケンス、レポート
• reference/hubspot-patterns.md - SDK パターン、一括操作、ワークフロー
• reference/salesforce-patterns.md - JWT 認証、SOQL、Platform Events、Bulk 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 連携を希望している場合： → どの CRM か確認します（シンプルさを重視する場合は Close を推奨） → 認証の設定と基本的な CRUD を提示します

ユーザーが Close CRM を希望している場合： → API キーの設定とクエリ言語を提示します → 参照先：reference/close-deep-dive.md

ユーザーが HubSpot を希望している場合： → SDK の設定と検索パターンを提示します → 参照先：reference/hubspot-patterns.md

ユーザーが Salesforce を希望している場合： → JWT 認証と SOQL パターンを提示します → 参照先：reference/salesforce-patterns.md

ユーザーが CRM 間の同期を希望している場合： → 同期アーキテクチャと競合解決を提示します → 参照先：reference/sync-patterns.md

ユーザーが Webhook を希望している場合： → 指定された CRM のハンドラーパターンを提示します → 署名検証を含めます

<env_setup>

環境変数

# Close CRM
export CLOSE_API_KEY="api_xxx"
export CLOSE_WEBHOOK_SECRET="whsec_xxx"

# HubSpot
export HUBSPOT_ACCESS_TOKEN="pat-xxx"

# Salesforce
export SF_CLIENT_ID="xxx"
export SF_USERNAME="[email protected]"
export SF_PRIVATE_KEY_PATH="./salesforce.key"
export SF_INSTANCE_URL="https://yourorg.my.salesforce.com"

# Dependencies: httpx pyjwt hubspot-api-client python-dotenv

</env_setup>

<example_session>

セッション例

ユーザー：「営業エージェントから、エンリッチメント済みリードを Close CRM に送信したいです」

Claude：

1. エンリッチメント済みデータを Close に送信するパターンは次のとおりです。

async def push_to_close(close_client, enriched_data: dict) -> str:
    lead_data = {
        "name": enriched_data["company_name"],
        "url": enriched_data.get("website"),
        "custom.cf_tier": enriched_data["tier"],
        "custom.cf_source": "sales-agent",
        "contacts": [{
            "name": c["name"],
            "title": c.get("title"),
            "emails": [{"email": c["email"]}] if c.get("email") else []
        } for c in enriched_data.get("contacts", [])]
    }

    result = close_client.create_lead(lead_data)
    return result["id"]

2. Close に次のカスタムフィールドがあることを確認してください。

   • cf_tier（選択肢：Gold、Silver、Bronze）
   • cf_source（選択肢：sales-agent、inbound、referral）

3. レート制限は 10 秒あたり 100 リクエストです。一括インポートでは、リクエストの間に asyncio.sleep(0.1) を追加してください。 </example_session>