Use HubSpot CRM records as evidence for sales, pipeline, and customer analysis. — Claude Skill

name: hubspot-crm description: 연락처나 목록을 HubSpot CRM에 동기화할 때 사용합니다. 환경에서 HUBSPOT_API_TOKEN을 자동으로 사용합니다.

HubSpot CRM 연동

REST API를 사용해 연락처와 목록을 HubSpot에 동기화합니다.

환경 변수

HUBSPOT_API_TOKEN="pat-na1-..."  # HubSpot의 비공개 앱 토큰

빠른 시작

import os
import json
from urllib.request import Request, urlopen
from urllib.error import HTTPError

class HubSpotClient:
    """간단한 HubSpot API 클라이언트."""

    def __init__(self):
        self.token = os.environ.get('HUBSPOT_API_TOKEN')
        if not self.token:
            raise ValueError("HUBSPOT_API_TOKEN 환경 변수가 설정되지 않았습니다")
        self.base_url = "https://api.hubapi.com"

    def _request(self, method: str, endpoint: str, data: dict = None) -> dict:
        url = f"{self.base_url}{endpoint}"
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json"
        }
        body = json.dumps(data).encode('utf-8') if data else None
        request = Request(url, data=body, headers=headers, method=method)

        with urlopen(request, timeout=30) as response:
            return json.loads(response.read().decode('utf-8'))

정적 목록 생성

def create_static_list(client, name: str) -> str:
    """연락처용 정적 목록을 생성합니다. 목록 ID를 반환합니다."""
    payload = {
        "name": name,
        "objectTypeId": "0-1",  # 필수: 0-1 = 연락처
        "processingType": "MANUAL"
    }

    result = client._request("POST", "/crm/v3/lists", payload)
    # 응답은 중첩 구조입니다: {"list": {"listId": "..."}}
    list_data = result.get("list", result)
    list_id = list_data.get("listId")

    print(f"✅ 목록 생성됨: {name} (ID: {list_id})")
    return list_id

이메일로 연락처 검색

def search_contact(client, email: str) -> str | None:
    """이메일로 연락처를 찾습니다. 연락처 ID 또는 None을 반환합니다."""
    payload = {
        "filterGroups": [{
            "filters": [{
                "propertyName": "email",
                "operator": "EQ",
                "value": email
            }]
        }],
        "properties": ["email"],
        "limit": 1
    }

    result = client._request("POST", "/crm/v3/objects/contacts/search", payload)
    results = result.get("results", [])
    return results[0]["id"] if results else None

연락처 생성

def create_contact(client, email: str, firstname: str = None, lastname: str = None) -> str:
    """이메일만으로 새 연락처를 생성합니다(기본 업로드).

    대상 계정에 사용자 지정 속성이 없어서 생기는 오류를 피하려고
    표준 HubSpot 속성(email, firstname, lastname)만 사용합니다.
    """
    properties = {"email": email}
    if firstname:
        properties["firstname"] = firstname
    if lastname:
        properties["lastname"] = lastname

    result = client._request("POST", "/crm/v3/objects/contacts", {"properties": properties})
    return result["id"]

중요: 임의의 CSV 열을 속성으로 전달하지 마세요. HubSpot은 대상 계정에 존재하지 않는 속성 이름을 거부합니다. 사용자 지정 속성이 존재한다고 확인하지 않았다면 표준 필드 (email, firstname, lastname)만 사용합니다.

목록에 연락처 추가

def add_to_list(client, list_id: str, contact_ids: list[str]):
    """정적 목록에 연락처를 추가합니다. 100개씩 묶어 처리합니다."""
    endpoint = f"/crm/v3/lists/{list_id}/memberships/add"

    batch_size = 100
    total_added = 0

    for i in range(0, len(contact_ids), batch_size):
        batch = contact_ids[i:i + batch_size]
        # 중요: payload는 {"recordIdsToAdd": [...]}가 아니라 단순 배열입니다.
        result = client._request("PUT", endpoint, batch)
        added = len(result.get("recordsIdsAdded", []))
        total_added += added
        print(f"  배치 추가됨: 연락처 {added}개")

    print(f"✅ 목록에 총 연락처 {total_added}개 추가됨")

전체 업로드 흐름

def upload_users_to_hubspot(emails: list[str], list_name: str) -> str:
    """이메일 주소 목록을 HubSpot에 업로드합니다."""
    client = HubSpotClient()

    # 목록 생성
    list_id = create_static_list(client, list_name)

    # 연락처 찾기 또는 생성
    contact_ids = []
    for email in emails:
        contact_id = search_contact(client, email)
        if not contact_id:
            contact_id = create_contact(client, email)
        contact_ids.append(contact_id)

    # 목록에 추가
    add_to_list(client, list_id, contact_ids)

    print(f"\n✅ 완료!")
    print(f"   List: https://app.hubspot.com/contacts/lists/{list_id}")

    return list_id

사용 예시

# 분석에서 나온 이탈 위험 사용자를 업로드
at_risk_emails = [
    "[email protected]",
    "[email protected]",
    "[email protected]"
]

list_id = upload_users_to_hubspot(
    emails=at_risk_emails,
    list_name="이탈 위험 체험판 사용자 - 2024년 12월"
)

연동 스크립트 사용

CSV 파일에는 제공된 스크립트를 사용합니다:

.venv/bin/python demos/04-trial-to-paid/scripts/hubspot_integration.py \
  path/to/users.csv \
  "여기에 목록 이름"

CSV에는 email 열이 있어야 합니다.

API 주의 사항

1. 목록 생성에는 objectTypeId가 필요: 연락처에는 항상 "objectTypeId": "0-1"을 포함합니다.
2. 응답은 중첩 구조: 목록 ID는 result["list"]["listId"]에 있으며 result["listId"]가 아닙니다.
3. 목록 추가 payload 형식: {"recordIdsToAdd": [...]}가 아니라 단순 배열 ["id1", "id2"]를 사용합니다.
4. 연락처 ID는 문자열: 숫자처럼 보이더라도 문자열로 다룹니다.
5. 배치 제한: 연락처는 최대 100개씩 묶어 추가합니다.

오류 처리

from urllib.error import HTTPError

try:
    result = client._request("POST", endpoint, payload)
except HTTPError as e:
    error_body = e.read().decode('utf-8')
    print(f"HubSpot API error: {e.code}")
    print(f"  {error_body}")

HubSpot 비공개 앱 토큰 가져오기

1. HubSpot에서 Settings(기어 아이콘)로 이동합니다.
2. Integrations > Private Apps로 이동합니다.
3. Create a private app을 클릭합니다.
4. 이름을 지정합니다(예: "AI 에이전트 연동").
5. Scopes 아래에서 다음을 활성화합니다:
   • crm.lists.read
   • crm.lists.write
   • crm.objects.contacts.read
   • crm.objects.contacts.write
6. Create app을 클릭하고 토큰을 복사합니다.
7. HUBSPOT_API_TOKEN 환경 변수로 설정합니다.