Интеграция с CRM: Salesforce и HubSpot
Получение истории клиентов, регистрация взаимодействий и обновление записей CRM.
«Интеграция с CRM: Salesforce и HubSpot» — бесплатный урок AI Agents на CoddyKit. Это урок 2 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Agents, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Agents содержит 4 уроков всего.
Почему важна интеграция с CRM
Агент службы поддержки, который не видит историю клиента, действует вслепую. CRM-системы, такие как Salesforce и HubSpot, хранят полную запись о клиенте: предыдущие обращения, покупки, уровень подписки и примечания к учётной записи.
Интеграция с CRM позволяет агенту персонализировать ответы и быстрее решать проблемы.
Подключение к Salesforce с помощью simple-salesforce
simple-salesforce — самая популярная библиотека Python для REST API Salesforce. Подключайтесь с помощью пары имя пользователя/пароль/токен безопасности или через поток 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 (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 предоставляет оболочку над REST API HubSpot. Аутентифицируйтесь с помощью токена доступа частного приложения (для интеграций сервер-сервер 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
Создайте новое обращение и свяжите его с контактом в рамках одного процесса. Благодаря этому запись клиента в 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'}Какой язык запросов Salesforce используется для получения записей через библиотеку simple-salesforce
Salesforce использует собственный язык запросов для получения записей. Знание этого языка необходимо для создания инструментов агента, которые находят историю клиента.
Повторение интеграции с CRM
Интеграция с Salesforce (simple-salesforce + SOQL) или HubSpot (hubspot-api-client) даёт агенту службы поддержки полный контекст клиента: историю, обращения, заявки и данные учётной записи.
Всегда создавайте единую функцию get_customer_context(), чтобы перед созданием ответа запрос агента обогащался релевантной историей.
Часто задаваемые вопросы
Урок «Интеграция с CRM: Salesforce и HubSpot» бесплатный?
Да — полный текст урока «Интеграция с CRM: Salesforce и HubSpot» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Agents, подпишись на CoddyKit PRO. Курс AI Agents содержит 4 уроков всего.
Чему я научусь в уроке «Интеграция с CRM: Salesforce и HubSpot»?
Получение истории клиентов, регистрация взаимодействий и обновление записей CRM. Ты практикуешь AI Agents с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.
Нужен ли мне опыт, чтобы начать AI Agents?
Предыдущий опыт не требуется. AI Agents на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 2 из 4.
Сколько времени занимает урок «Интеграция с CRM: Salesforce и HubSpot»?
Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.
Можно ли писать и запускать код в этом уроке AI Agents?
Да. Каждый урок AI Agents включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.
Все уроки этого курса
- Маршрутизация обращений и логика эскалации
- Интеграция с CRM: Salesforce и HubSpot
- Протоколы передачи оператору
- Управление контекстом и историей клиента