CRM 連携：Salesforce と HubSpot
顧客履歴の取得、やり取りの記録、CRM レコードの更新を行います。
「CRM 連携：Salesforce と HubSpot」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
CRM統合が重要な理由
顧客の履歴を確認できないカスタマーサービスエージェントは、手探りで対応することになります。SalesforceやHubSpotなどのCRMシステムには、過去のチケット、購入履歴、契約プラン、アカウントのメモなど、顧客情報全体が保存されています。
CRMと統合すれば、エージェントは応答をパーソナライズし、問題をより速く解決できます。
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オブジェクトをクエリして、顧客の過去のサポートチケットを取得します。ケースのステータスと解決内容も含めることで、エージェントが同じ質問を繰り返さずに済みます。
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で新しいケースを作成する
エージェントが会話から新しいサポートチケットを作成するときは、新しい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でコンタクトを検索する
Contacts検索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チケットを作成する
1つのワークフローで新しいチケットを挿入し、コンタクトに関連付けます。これにより、エージェントがリクエストに対応するたびに、顧客の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コンタクトレコードを更新する
問題が正常に解決したら、メモとライフサイクルステージの変更内容でコンタクトのレコードを更新します。これにより、エージェントが手動で作業しなくても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.'
})エージェント向けの統合CRMコンテキスト
設定されているCRMに応じて顧客コンテキストを取得する単一の関数を作成します。エージェントは各会話の開始時にこの関数を呼び出し、応答をパーソナライズします。
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では、独自のクエリ言語を使ってレコードを取得します。顧客履歴を検索するエージェントツールを構築するには、この知識が不可欠です。
CRM統合のまとめ
Salesforce（simple-salesforce + SOQL）またはHubSpot（hubspot-api-client）と統合すると、カスタマーサービスエージェントは履歴、ケース、チケット、アカウントデータなど、顧客に関する完全なコンテキストを利用できます。
エージェントのプロンプトに応答生成前の関連する履歴を追加できるよう、必ず統合されたget_customer_context()関数を構築してください。
AI チューターと学ぶ AI Agents — 無料
ブラウザでリアルコードを書いて実行し、24/7 の AI チューターから瞬時にサポートを受け、ウェブまたはアプリで続きから学習できます。
- コース
- 60
- レッスン
- 239
よくある質問
「CRM 連携：Salesforce と HubSpot」レッスンは無料ですか？
はい。「CRM 連携：Salesforce と HubSpot」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「CRM 連携：Salesforce と HubSpot」で何を学びますか？
顧客履歴の取得、やり取りの記録、CRM レコードの更新を行います。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。
「CRM 連携：Salesforce と HubSpot」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- チケットの振り分けとエスカレーションロジック
- CRM 連携：Salesforce と HubSpot
- 有人引き継ぎのプロトコル
- 顧客コンテキストと履歴の管理