CRM 통합: Salesforce와 HubSpot
고객 이력을 가져오고 상호작용을 기록하며 CRM 기록을 업데이트합니다.
CRM 통합: Salesforce와 HubSpot은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 2번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
CRM 통합이 중요한 이유
고객의 이력을 볼 수 없는 고객 서비스 agent는 아무런 정보 없이 업무를 수행하는 것과 같습니다. Salesforce와 HubSpot 같은 CRM 시스템에는 과거 티켓, 구매 내역, 구독 등급, 계정 메모 등 고객의 전체 기록이 저장됩니다.
CRM을 통합하면 agent가 응답을 개인화하고 문제를 더 빠르게 해결할 수 있습니다.
simple-salesforce로 Salesforce에 연결
simple-salesforce는 Salesforce REST API에 가장 널리 사용되는 Python 라이브러리입니다. 사용자 이름/비밀번호/보안 토큰을 사용하거나 OAuth 연결 앱 흐름을 통해 연결할 수 있습니다.
from simple_salesforce import Salesforce
sf = Salesforce(
username='agent@company.com',
password='YOUR_SF_PASSWORD',
security_token='YOUR_SF_TOKEN',
domain='login' # use 'test' for sandbox
)
print('Connected to Salesforce org:', sf.base_url)SOQL 쿼리: 고객 기록 가져오기
Salesforce Object Query Language(SOQL)는 SQL과 비슷하지만 Salesforce 객체를 대상으로 동작합니다. 이를 사용해 이메일로 연락처를 조회하고 계정 세부 정보를 가져올 수 있습니다.
def get_customer_by_email(sf, email: str) -> dict | None:
query = (
f"SELECT Id, FirstName, LastName, Email, AccountId, "
f"Account.Name, Account.Type "
f"FROM Contact "
f"WHERE Email = '{email}' "
f"LIMIT 1"
)
result = sf.query(query)
if result['totalSize'] == 0:
return None
record = result['records'][0]
return {
'contact_id': record['Id'],
'name': f"{record['FirstName']} {record['LastName']}",
'account_id': record['AccountId'],
'account_name': record['Account']['Name']
}
if __name__ == '__main__':
class FakeSF:
def query(self, query):
return {
'totalSize': 1,
'records': [{
'Id': '003ABC', 'FirstName': 'Jane', 'LastName': 'Doe',
'AccountId': '001XYZ',
'Account': {'Name': 'Acme Corp', 'Type': 'Customer'},
}],
}
customer = get_customer_by_email(FakeSF(), 'jane@acme.com')
print(f"Found customer: {customer['name']} at {customer['account_name']}")
Salesforce에서 지원 사례 이력 가져오기
Case 객체를 조회해 고객의 이전 지원 티켓을 가져옵니다. 사례 상태와 해결 내용을 포함하면 agent가 같은 질문을 두 번 하는 일을 피할 수 있습니다.
def get_case_history(sf, contact_id: str, limit: int = 10) -> list:
query = (
f"SELECT Id, CaseNumber, Subject, Status, "
f"CreatedDate, ClosedDate, Resolution__c "
f"FROM Case "
f"WHERE ContactId = '{contact_id}' "
f"ORDER BY CreatedDate DESC "
f"LIMIT {limit}"
)
result = sf.query(query)
return [
{
'case_number': r['CaseNumber'],
'subject': r['Subject'],
'status': r['Status'],
'created': r['CreatedDate'][:10],
'resolution': r.get('Resolution__c')
}
for r in result['records']
]
if __name__ == '__main__':
class FakeSF:
def query(self, query):
return {'records': [{
'Id': '500A', 'CaseNumber': '00001234', 'Subject': 'Login issue',
'Status': 'Closed', 'CreatedDate': '2026-07-01T10:00:00Z',
'ClosedDate': '2026-07-02T09:00:00Z', 'Resolution__c': 'Reset password',
}]}
for case in get_case_history(FakeSF(), '003ABC'):
print(f"Case {case['case_number']}: {case['subject']} ({case['status']})")
Salesforce에서 새 사례 만들기
agent가 대화에서 새 지원 티켓을 만들 때 새 Case 기록을 삽입합니다. 연락처와 계정에 연결하면 고객의 타임라인에 표시됩니다.
def create_case(sf, contact_id: str, account_id: str,
subject: str, description: str, priority: str = 'Medium') -> str:
result = sf.Case.create({
'ContactId': contact_id,
'AccountId': account_id,
'Subject': subject,
'Description': description,
'Priority': priority,
'Status': 'New',
'Origin': 'AI Agent'
})
case_id = result['id']
print(f'Created Case: {case_id}')
return case_id
if __name__ == '__main__':
class FakeCase:
def create(self, data):
return {'id': '500NEW01'}
class FakeSF:
Case = FakeCase()
create_case(FakeSF(), '003ABC', '001XYZ', 'Cannot reset password', 'Customer locked out after 3 attempts')
HubSpot에 연결
hubspot-api-client 라이브러리는 HubSpot REST API를 감쌉니다. 비공개 앱 액세스 토큰으로 인증합니다(서버 간 통합에는 OAuth가 필요하지 않습니다).
from hubspot import HubSpot
from hubspot.crm.contacts import SimplePublicObjectInput
client = HubSpot(access_token='YOUR_HUBSPOT_PRIVATE_APP_TOKEN')
print('HubSpot client ready')HubSpot 연락처 검색
연락처 검색 API를 사용해 이메일로 고객을 찾습니다. 응답에는 수명 주기 단계, 거래 연결, 사용자 지정 필드를 비롯한 모든 연락처 속성이 포함됩니다.
from hubspot.crm.contacts import PublicObjectSearchRequest
def get_hubspot_contact(client, email: str) -> dict | None:
search_request = PublicObjectSearchRequest(
filter_groups=[{
'filters': [{
'propertyName': 'email',
'operator': 'EQ',
'value': email
}]
}],
properties=['firstname', 'lastname', 'email',
'lifecyclestage', 'hs_object_id']
)
response = client.crm.contacts.search_api.do_search(
public_object_search_request=search_request
)
if response.total == 0:
return None
contact = response.results[0]
return {'id': contact.id, **contact.properties}HubSpot 티켓 이력 가져오기
HubSpot은 tickets 객체에 지원 티켓을 저장합니다. 연결 API를 사용해 연락처와 연결된 티켓을 가져옵니다.
def get_hubspot_tickets(client, contact_id: str) -> list:
assoc = client.crm.contacts.associations_api.get_all(
contact_id,
to_object_type='tickets'
)
ticket_ids = [r.id for r in assoc.results]
if not ticket_ids:
return []
tickets = []
for tid in ticket_ids[:10]: # limit to 10 most recent
ticket = client.crm.tickets.basic_api.get_by_id(
tid,
properties=['subject', 'hs_ticket_priority',
'hs_pipeline_stage', 'createdate']
)
tickets.append({
'id': tid,
'subject': ticket.properties.get('subject'),
'priority': ticket.properties.get('hs_ticket_priority'),
'stage': ticket.properties.get('hs_pipeline_stage')
})
return tickets
if __name__ == '__main__':
from types import SimpleNamespace
class AssocApi:
def get_all(self, contact_id, to_object_type):
return SimpleNamespace(results=[SimpleNamespace(id='t1'), SimpleNamespace(id='t2')])
class TicketsBasicApi:
_data = {
't1': {'subject': 'Cannot login', 'hs_ticket_priority': 'HIGH', 'hs_pipeline_stage': '1'},
't2': {'subject': 'Refund request', 'hs_ticket_priority': 'MEDIUM', 'hs_pipeline_stage': '2'},
}
def get_by_id(self, tid, properties):
return SimpleNamespace(properties=self._data[tid])
fake_client = SimpleNamespace(crm=SimpleNamespace(
contacts=SimpleNamespace(associations_api=AssocApi()),
tickets=SimpleNamespace(basic_api=TicketsBasicApi()),
))
for t in get_hubspot_tickets(fake_client, 'contact_1'):
print(f"Ticket {t['id']}: {t['subject']} [{t['priority']}]")
HubSpot 티켓 만들기
하나의 작업 흐름에서 새 티켓을 삽입하고 연락처와 연결합니다. 이렇게 하면 agent가 요청을 처리하는 동안 고객의 CRM 기록이 자동으로 최신 상태로 유지됩니다.
from hubspot.crm.tickets import SimplePublicObjectInputForCreate
from hubspot.crm.associations.v4.models import AssociationSpec
def create_hubspot_ticket(client, contact_id: str,
subject: str, description: str) -> str:
ticket_input = SimplePublicObjectInputForCreate(
properties={
'subject': subject,
'content': description,
'hs_ticket_priority': 'MEDIUM',
'hs_pipeline': 'support_pipeline',
'hs_pipeline_stage': 'new'
},
associations=[{
'types': [AssociationSpec(association_category='HUBSPOT_DEFINED',
association_type_id=16)],
'to': {'id': contact_id}
}]
)
ticket = client.crm.tickets.basic_api.create(
simple_public_object_input_for_create=ticket_input
)
return ticket.id연락처 기록 업데이트
문제를 성공적으로 해결한 후 메모와 수명 주기 단계 변경 사항으로 연락처 기록을 업데이트합니다. 이를 통해 agent가 수동으로 작업하지 않아도 CRM을 최신 상태로 유지할 수 있습니다.
def update_hubspot_contact(client, contact_id: str, updates: dict) -> None:
client.crm.contacts.basic_api.update(
contact_id=contact_id,
simple_public_object_input=SimplePublicObjectInput(
properties=updates
)
)
# After resolving a billing issue:
update_hubspot_contact(client, 'CONTACT_123', {
'hs_lead_status': 'CONNECTED',
'notes_last_contacted': 'Resolved billing duplicate charge issue via AI agent.'
})Agent를 위한 통합 CRM 컨텍스트
구성된 CRM에 관계없이 고객 컨텍스트를 가져오는 하나의 함수를 만듭니다. agent는 각 대화 시작 시 이 함수를 호출해 응답을 개인화합니다.
def get_customer_context(email: str, crm: str = 'salesforce') -> dict:
if crm == 'salesforce':
customer = get_customer_by_email(sf, email)
if not customer:
return {'found': False}
cases = get_case_history(sf, customer['contact_id'])
return {'found': True, 'customer': customer, 'history': cases}
elif crm == 'hubspot':
customer = get_hubspot_contact(client, email)
if not customer:
return {'found': False}
tickets = get_hubspot_tickets(client, customer['id'])
return {'found': True, 'customer': customer, 'history': tickets}
return {'found': False, 'error': 'Unknown CRM'}simple-salesforce 라이브러리를 통해 기록을 가져올 때 사용하는 Salesforce 쿼리 언어는 무엇인가요?
Salesforce는 기록을 가져올 때 자체 쿼리 언어를 사용합니다. 고객 이력을 조회하는 agent 도구를 구축하려면 이 언어를 아는 것이 필수적입니다.
CRM 통합 요약
Salesforce(simple-salesforce + SOQL) 또는 HubSpot(hubspot-api-client)과 통합하면 고객 서비스 agent가 이력, 사례, 티켓, 계정 데이터를 비롯한 고객의 전체 컨텍스트를 얻을 수 있습니다.
항상 통합된 get_customer_context() 함수를 만들어 응답을 생성하기 전에 관련 이력으로 agent 프롬프트를 보강해야 합니다.
자주 묻는 질문
“CRM 통합: Salesforce와 HubSpot” 강의는 무료인가요?
네 — “CRM 통합: Salesforce와 HubSpot” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.
“CRM 통합: Salesforce와 HubSpot”에서 뭘 배우나요?
고객 이력을 가져오고 상호작용을 기록하며 CRM 기록을 업데이트합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Agents을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 2번째 강의입니다.
“CRM 통합: Salesforce와 HubSpot” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 티켓 라우팅 및 에스컬레이션 로직
- CRM 통합: Salesforce와 HubSpot
- 사람에게 인계하는 프로토콜
- 고객 맥락 및 이력 관리