Claude Code×HubSpot・Salesforce API【2026】営業データの分析とCRM運用をAIエージェントで

npm install -g @salesforce/cli
# Sandbox にブラウザでログイン（本番は login.salesforce.com / My Domain のURL）
sf org login web --alias sbx --instance-url https://test.salesforce.com
sf org list

sales-ops/
├── CLAUDE.md                 # 項目定義・ステージ定義・ルール
├── .env                      # HubSpotトークン（Git・Claudeの対象外）
├── .claude/settings.json     # permissions / hooks
├── .claude/skills/weekly-pipeline/SKILL.md
├── config/stage_rules.json   # ステージごとの「停滞」日数・必須項目
├── scripts/hubspot/          # client.py, fetch_deals.py, fetch_contacts.py
├── scripts/salesforce/       # sf_query.py（sf CLI のラッパー）
├── scripts/analyze/          # stale_deals.py, find_duplicates.py, check_required.py
├── scripts/write/            # register_activity.py（承認済みの下書きだけを登録。ask）
├── reports/                  # 取得・分析結果（Git管理外）
├── notes/inbox/              # 商談メモ（人が置く）
├── drafts/activities/        # 活動記録の下書き（Claudeが書く）
├── approved/                 # 人が確認して移した下書き（Claudeは編集不可）
└── logs/audit.jsonl          # 書き込みの監査ログ

"""HubSpot CRM API v3 の薄いクライアント（読み取り用トークン）"""
import os
import time

import requests
from dotenv import load_dotenv

load_dotenv()
BASE = "https://api.hubapi.com"
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['HUBSPOT_READ_TOKEN']}"


def _request(method: str, path: str, **kwargs) -> dict:
    for attempt in range(5):
        resp = session.request(method, BASE + path, timeout=30, **kwargs)
        if resp.status_code == 429 or resp.status_code >= 500:
            time.sleep(2 ** attempt)  # レート制限・一時エラーは待って再試行
            continue
        resp.raise_for_status()
        return resp.json()
    raise RuntimeError(f"HubSpot API: retries exhausted ({path})")


def list_objects(obj: str, properties: list[str]):
    """GET /crm/v3/objects/{obj} を paging.next.after でたどって全件返す"""
    params = {"limit": 100, "properties": ",".join(properties)}
    while True:
        data = _request("GET", f"/crm/v3/objects/{obj}", params=params)
        yield from data["results"]
        after = data.get("paging", {}).get("next", {}).get("after")
        if not after:
            return
        params["after"] = after


def search(obj: str, filters: list[dict], properties: list[str]):
    """POST /crm/v3/objects/{obj}/search（1クエリで取れる件数に上限があるので範囲を絞って使う）"""
    body = {"filterGroups": [{"filters": filters}], "properties": properties, "limit": 100}
    while True:
        data = _request("POST", f"/crm/v3/objects/{obj}/search", json=body)
        yield from data["results"]
        after = data.get("paging", {}).get("next", {}).get("after")
        if not after:
            return
        body["after"] = after
        time.sleep(0.3)  # 検索APIは通常のAPIより制限が厳しい

"""商談とパイプライン定義を取得して reports/ に保存する"""
import csv
import json

from client import _request, list_objects

PROPS = ["dealname", "amount", "dealstage", "pipeline", "closedate",
         "hubspot_owner_id", "notes_last_updated", "hs_lastmodifieddate"]

# ステージIDとラベル、クローズ済みかどうか（metadata.isClosed）を保存
pipelines = _request("GET", "/crm/v3/pipelines/deals")["results"]
stages = {s["id"]: {"label": s["label"], "closed": s["metadata"].get("isClosed") == "true"}
          for p in pipelines for s in p["stages"]}
json.dump(stages, open("reports/stages.json", "w", encoding="utf-8"), ensure_ascii=False, indent=2)

owners = {o["id"]: f"{o.get('lastName', '')} {o.get('firstName', '')}".strip()
          for o in _request("GET", "/crm/v3/owners", params={"limit": 500})["results"]}

with open("reports/deals.csv", "w", newline="", encoding="utf-8") as f:
    w = csv.writer(f)
    w.writerow(["id", *PROPS, "stage_label", "is_closed", "owner_name"])
    for d in list_objects("deals", PROPS):
        p = d["properties"]
        stage = stages.get(p.get("dealstage"), {"label": "(不明)", "closed": False})
        w.writerow([d["id"], *[p.get(k) or "" for k in PROPS],
                    stage["label"], stage["closed"], owners.get(p.get("hubspot_owner_id"), "")])

{
  "_comment": "キーは HubSpot のステージID（reports/stages.json で確認）。値は営業部門と合意した定義",
  "appointmentscheduled": { "stale_days": 7,  "required": ["amount", "closedate", "hubspot_owner_id"] },
  "qualifiedtobuy":       { "stale_days": 10, "required": ["amount", "closedate", "hubspot_owner_id"] },
  "presentationscheduled":{ "stale_days": 14, "required": ["amount", "closedate", "hubspot_owner_id"] },
  "contractsent":         { "stale_days": 7,  "required": ["amount", "closedate", "hubspot_owner_id"] }
}

"""停滞商談と必須項目の空欄を、config/stage_rules.json の定義で抽出する"""
import csv
import json
from datetime import datetime, timezone

rules = json.load(open("config/stage_rules.json", encoding="utf-8"))
now = datetime.now(timezone.utc)


def days_since(value: str) -> int | None:
    if not value:
        return None
    return (now - datetime.fromisoformat(value.replace("Z", "+00:00"))).days


stale, missing = [], []
for d in csv.DictReader(open("reports/deals.csv", encoding="utf-8")):
    rule = rules.get(d["dealstage"])
    if d["is_closed"] == "True" or rule is None:
        continue
    # 最終活動日（notes_last_updated）を優先。なければ最終更新日で代用し、その旨を残す
    basis = "last_activity" if d["notes_last_updated"] else "last_modified"
    idle = days_since(d["notes_last_updated"] or d["hs_lastmodifieddate"])
    if idle is not None and idle >= rule["stale_days"]:
        stale.append([d["id"], d["dealname"], d["stage_label"], d["owner_name"], d["amount"], idle, basis])
    empty = [k for k in rule["required"] if not d.get(k)]
    if empty:
        missing.append([d["id"], d["dealname"], d["stage_label"], d["owner_name"], "|".join(empty)])

with open("reports/stale_deals.csv", "w", newline="", encoding="utf-8") as f:
    csv.writer(f).writerows([["id", "deal", "stage", "owner", "amount", "idle_days", "basis"], *stale])
with open("reports/missing_fields.csv", "w", newline="", encoding="utf-8") as f:
    csv.writer(f).writerows([["id", "deal", "stage", "owner", "missing"], *missing])
print(f"stale={len(stale)} missing={len(missing)}")

"""重複コンタクトの「候補」を出す。メールはマスクして出力し、統合はしない"""
import csv
import re
from collections import defaultdict


def norm_email(e: str) -> str:
    return (e or "").strip().lower()


def norm_phone(p: str) -> str:
    digits = re.sub(r"\D", "", p or "")
    return digits[-10:] if len(digits) >= 10 else ""  # 国番号・ハイフンの違いを吸収


def mask(e: str) -> str:
    local, _, domain = e.partition("@")
    return f"{local[:1]}***@{domain}" if domain else ""


groups = defaultdict(set)
# fetch_contacts.py が保存した id, email, phone, firstname, lastname, company, createdate の列
contacts = {c["id"]: c for c in csv.DictReader(open("reports/contacts.csv", encoding="utf-8"))}
for cid, c in contacts.items():
    if norm_email(c["email"]):
        groups[("email", norm_email(c["email"]))].add(cid)
    if norm_phone(c["phone"]):
        groups[("phone", norm_phone(c["phone"]))].add(cid)
    name = (c["lastname"] + c["firstname"]).replace(" ", "").replace("　", "")
    if name and c["company"]:
        groups[("name+company", name + "/" + c["company"].strip())].add(cid)

with open("reports/duplicate_candidates.csv", "w", newline="", encoding="utf-8") as f:
    w = csv.writer(f)
    w.writerow(["reason", "contact_ids", "masked_emails", "created_dates"])
    for (reason, _), ids in groups.items():
        if len(ids) > 1:
            ids = sorted(ids)
            w.writerow([reason, "|".join(ids),
                        "|".join(mask(norm_email(contacts[i]["email"])) for i in ids),
                        "|".join(contacts[i]["createdate"][:10] for i in ids)])

# 停滞している商談（オープンで、14日以上活動がない）
sf data query --target-org sbx --result-format csv --query "
  SELECT Id, Name, StageName, Amount, CloseDate, Owner.Name, LastActivityDate
  FROM Opportunity
  WHERE IsClosed = false
    AND (LastActivityDate = null OR LastActivityDate < LAST_N_DAYS:14)
  ORDER BY Amount DESC NULLS LAST" > reports/sf_stale_opps.csv

# パイプラインの週次サマリー（ステージ別の件数と金額）
sf data query --target-org sbx --query "
  SELECT StageName, COUNT(Id) cnt, SUM(Amount) total
  FROM Opportunity WHERE IsClosed = false GROUP BY StageName"

# 必須項目の空欄チェック
sf data query --target-org sbx --result-format csv --query "
  SELECT Id, Name, StageName, Owner.Name FROM Opportunity
  WHERE IsClosed = false AND (Amount = null OR CloseDate = null OR NextStep = null)" > reports/sf_missing.csv

# 同じメールアドレスのコンタクト（重複候補）
sf data query --target-org sbx --query "
  SELECT Email, COUNT(Id) c FROM Contact
  WHERE Email != null GROUP BY Email HAVING COUNT(Id) > 1"

"""sf CLI で SOQL を実行し、JSON の records を CSV に保存する（認証は sf CLI に任せる）"""
import csv
import json
import subprocess
import sys

ORG = "sbx"  # 本番に切り替えるときは人が明示的に変更する


def run_soql(soql: str) -> list[dict]:
    out = subprocess.run(
        ["sf", "data", "query", "--target-org", ORG, "--query", soql, "--json"],
        capture_output=True, text=True, check=True,
    )
    records = json.loads(out.stdout)["result"]["records"]
    for r in records:
        r.pop("attributes", None)
        owner = r.pop("Owner", None)  # リレーション項目はネストで返るので平らにする
        if owner:
            r["OwnerName"] = owner.get("Name")
    return records


if __name__ == "__main__":
    soql_file, out_csv = sys.argv[1], sys.argv[2]  # 例: queries/stale.soql reports/sf_stale.csv
    rows = run_soql(open(soql_file, encoding="utf-8").read())
    with open(out_csv, "w", newline="", encoding="utf-8") as f:
        if rows:
            w = csv.DictWriter(f, fieldnames=list(rows[0]))
            w.writeheader()
            w.writerows(rows)
    print(f"{len(rows)} records -> {out_csv}")

{
  "deal_id": "1234567890",
  "timestamp": "2026-09-25T06:00:00Z",
  "body": "【打ち合わせ】先方の情シス部門と導入範囲を確認。\n・決裁は10月の定例会議\n・懸念点：既存システムとのデータ連携\n【次アクション】連携方式の資料を10/2までに送付（担当：営業）",
  "next_step": "連携方式の資料を送付",
  "next_step_due": "2026-10-02",
  "source": "notes/inbox/2026-09-25_acme.md"
}

"""approved/ に人が移した活動記録（JSON）だけを HubSpot のメモとして登録し、監査ログを残す"""
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

import requests
from dotenv import load_dotenv

load_dotenv()
draft_path, approver = Path(sys.argv[1]).resolve(), sys.argv[2]  # 例: approved/2026-09-25_acme.json yamada
if Path("approved").resolve() not in draft_path.parents:
    raise SystemExit("approved/ 配下のファイル（人が承認して移したもの）だけ登録できます")
draft = json.load(open(draft_path, encoding="utf-8"))

body = {
    "properties": {"hs_timestamp": draft["timestamp"], "hs_note_body": draft["body"]},
    "associations": [{
        "to": {"id": draft["deal_id"]},
        "types": [{"associationCategory": "HUBSPOT_DEFINED", "associationTypeId": 214}],  # メモ→商談（公式で確認）
    }],
}
resp = requests.post(
    "https://api.hubapi.com/crm/v3/objects/notes",
    headers={"Authorization": f"Bearer {os.environ['HUBSPOT_WRITE_TOKEN']}"},
    json=body, timeout=30,
)
resp.raise_for_status()

with open("logs/audit.jsonl", "a", encoding="utf-8") as log:
    log.write(json.dumps({
        "at": datetime.now(timezone.utc).isoformat(), "action": "create_note",
        "deal_id": draft["deal_id"], "note_id": resp.json()["id"],
        "file": draft_path.name, "approved_by": approver,
    }, ensure_ascii=False) + "\n")
print("registered:", resp.json()["id"])

# CLAUDE.md（営業データ分析・CRM運用）

## 目的
- HubSpot / Salesforce のデータを読み、停滞商談・週次サマリー・重複候補・データ品質の点検と、
  活動記録の下書きを作る。CRMのデータ更新は、人の承認後に scripts/write/ 経由でのみ行う

## データの取り方
- HubSpot: python scripts/hubspot/fetch_deals.py / fetch_contacts.py → scripts/analyze/*.py
- Salesforce: sf data query（--target-org sbx）または scripts/salesforce/sf_query.py
- 件数・金額の集計は reports/ のCSVを使う。暗算で再集計しない

## 項目定義（HubSpot 内部名 / Salesforce API参照名）
- 商談金額: amount / Amount（税抜。見込み金額であり受注額ではない）
- 完了予定日: closedate / CloseDate
- 最終活動日: notes_last_updated / LastActivityDate（メール・通話・会議の記録日）
- 次のステップ: next_step（カスタム）/ NextStep

## ステージ定義（停滞判定は config/stage_rules.json が正本）
- 初回商談設定 → 要件確認 → 提案 → 契約書送付 → 受注 / 失注
- 「停滞」= 最終活動日からの経過日数がステージごとの stale_days 以上
- ステージIDとラベルの対応は reports/stages.json を参照（IDは表示名と異なる）

## 下書きのルール
- 活動記録: notes/inbox/ のメモから drafts/activities/*.json を作る。approved/ へ移すのは人
- メモにない事実（金額・日付・決裁者）を補わない。不明点は「要確認」と書く
- 次アクション案は「誰が・何を・いつまでに」の形で、最大3つ

## 禁止
- レコードの更新・削除・統合（merge）、オーナー変更、一括インポート
- .env の参照、reports/ の個人情報を外部サービスやチャットに貼ること

{
  "permissions": {
    "allow": [
      "Bash(python scripts/hubspot/fetch_deals.py:*)",
      "Bash(python scripts/hubspot/fetch_contacts.py:*)",
      "Bash(python scripts/analyze/stale_deals.py:*)",
      "Bash(python scripts/analyze/find_duplicates.py:*)",
      "Bash(python scripts/salesforce/sf_query.py:*)",
      "Bash(sf data query:*)",
      "Read(./reports/**)",
      "Read(./notes/**)",
      "Edit(./drafts/**)"
    ],
    "ask": [
      "Bash(python scripts/write/register_activity.py:*)",
      "Edit(./scripts/**)",
      "Edit(./config/**)"
    ],
    "deny": [
      "Read(./.env)",
      "Bash(curl:*)",
      "Bash(sf data update:*)",
      "Bash(sf data delete:*)",
      "Bash(sf data upsert:*)",
      "Bash(sf data import:*)",
      "Bash(sf data create:*)",
      "Bash(sf apex:*)",
      "Bash(sf project deploy:*)",
      "Bash(sf org display:*)",
      "Edit(./approved/**)",
      "Edit(./logs/**)"
    ]
  }
}

> 今週のパイプラインのサマリーを作って。停滞している商談は
  担当者別に並べて、それぞれ次のアクション案も。

● Bash(python scripts/hubspot/fetch_deals.py)
● Bash(python scripts/analyze/stale_deals.py)
  ⎿ stale=14 missing=9
● Read(reports/deals.csv) / Read(reports/stale_deals.csv) / Read(config/stage_rules.json)

  ## パイプライン週次サマリー（数値は reports/ から転記）
  | ステージ | 件数 | 金額合計 | 前週比 | ...

  ## 停滞商談（14件）担当者別
  ■ 佐藤さん（5件）
   - A社 導入案件（提案／活動なし21日）
     次アクション案：提案書への質問期限を確認する連絡を、今週中に
   - B社（契約書送付／活動なし9日）
     次アクション案：法務確認の進捗を先方担当に確認
  ...
  ※ 3件は最終活動日が空欄のため、最終更新日で判定しています（basis=last_modified）
  ※ 必須項目の空欄が9件あります → reports/missing_fields.csv

> notes/inbox/2026-09-25_acme.md から活動記録の下書きを作って。

● Read(notes/inbox/2026-09-25_acme.md)
● Read(reports/deals.csv)       ← 商談名から deal_id を特定
● Write(drafts/activities/2026-09-25_acme.json)

  下書きを作成しました。メモに決裁者の役職がなかったため「要確認」と
  しています。内容を確認し、approved/ に移してから登録を指示してください。

> 1行直して approved/ に移した。登録して。承認者は yamada。

● Bash(python scripts/write/register_activity.py approved/2026-09-25_acme.json yamada)
  ← ask 設定のため、実行前に確認を求められる
  ⎿ registered: 98765432101

# Salesforce DX MCP Server（執筆時点で公開。オプションは公式READMEで確認）
claude mcp add salesforce -- npx -y @salesforce/mcp --orgs sbx
# HubSpot のMCPサーバー（公開形態・URL・認証方式は公式ドキュメントで確認）
claude mcp add --transport http hubspot <公式ドキュメントに記載のURL>

---
name: weekly-pipeline
description: HubSpot（またはSalesforce）の商談データから、パイプラインの週次サマリーと停滞商談の一覧・次アクション案を作る。週次の営業会議資料、パイプライン確認、停滞商談の洗い出しを頼まれたときに使う。
---
1. fetch_deals.py → stale_deals.py を実行する
2. reports/weekly_YYYYMMDD.md に次の順で書く
   - ステージ別の件数・金額（表）と前週からの変化（前週のファイルがあれば比較）
   - 今週クローズ予定の商談 / 停滞商談（担当者別、金額の大きい順）
   - 必須項目の空欄（担当者別の件数のみ。個別の指摘は担当者に直接）
3. 次アクション案は「誰が・何を・いつまでに」で1商談1〜2個。事実とAIの提案を分けて書く
4. 個人の連絡先（メール・電話）はレポートに書かない

#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")/.."
claude -p "weekly-pipeline の手順で今週のサマリーを作成して" \
  --allowedTools "Bash(python scripts/hubspot/fetch_deals.py:*)" \
                 "Bash(python scripts/analyze/stale_deals.py:*)" \
                 "Read" "Write(./reports/**)" \
  --output-format json > "logs/weekly_$(date +%Y%m%d).json"