Integracja z CRM: Salesforce i HubSpot
Pobieranie historii klienta, rejestrowanie interakcji i aktualizowanie rekordów CRM.
Integracja z CRM: Salesforce i HubSpot to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 2 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.
Dlaczego integracja z CRM ma znaczenie
Agent obsługi klienta, który nie widzi historii klienta, działa po omacku. Systemy CRM, takie jak Salesforce i HubSpot, przechowują pełne dane klienta: wcześniejsze zgłoszenia, zakupy, poziom subskrypcji oraz notatki dotyczące konta.
Integracja z CRM pozwala agentowi personalizować odpowiedzi i szybciej rozwiązywać problemy.
Łączenie z Salesforce za pomocą simple-salesforce
simple-salesforce to najpopularniejsza biblioteka Python do obsługi interfejsu Salesforce REST API. Połączenie można nawiązać za pomocą nazwy użytkownika, hasła i tokenu bezpieczeństwa albo w ramach przepływu aplikacji połączonej 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)Zapytania SOQL: pobieranie danych klientów
Salesforce Object Query Language (SOQL) przypomina SQL, ale działa na obiektach Salesforce. Należy używać go do wyszukiwania kontaktu na podstawie adresu e-mail i pobierania danych jego konta.
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']}")
Pobieranie historii zgłoszeń z Salesforce
Należy odpytać obiekt Case, aby pobrać wcześniejsze zgłoszenia klienta do pomocy technicznej. Warto uwzględnić status zgłoszenia i sposób rozwiązania, aby agent nie zadawał ponownie tych samych pytań.
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']})")
Tworzenie nowego zgłoszenia w Salesforce
Gdy agent tworzy nowe zgłoszenie do pomocy technicznej na podstawie rozmowy, należy wstawić nowy rekord Case. Należy powiązać go z kontaktem i kontem, aby pojawił się na osi czasu klienta.
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')
Łączenie z HubSpot
Biblioteka hubspot-api-client opakowuje interfejs HubSpot REST API. Uwierzytelnianie odbywa się za pomocą tokenu dostępu prywatnej aplikacji (w integracjach serwer-serwer OAuth nie jest potrzebny).
from hubspot import HubSpot
from hubspot.crm.contacts import SimplePublicObjectInput
client = HubSpot(access_token='YOUR_HUBSPOT_PRIVATE_APP_TOKEN')
print('HubSpot client ready')Wyszukiwanie kontaktu w HubSpot
Należy użyć interfejsu wyszukiwania kontaktów, aby znaleźć klienta na podstawie adresu e-mail. Odpowiedź zawiera wszystkie właściwości kontaktu, w tym etap cyklu życia, powiązania z transakcjami oraz pola niestandardowe.
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}Pobieranie historii zgłoszeń z HubSpot
HubSpot przechowuje zgłoszenia do pomocy technicznej w obiekcie tickets. Powiązane zgłoszenia kontaktu należy pobierać za pomocą interfejsu associations 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']}]")
Tworzenie zgłoszenia w HubSpot
Należy wstawić nowe zgłoszenie i powiązać je z kontaktem w ramach jednego przepływu pracy. Dzięki temu rekord klienta w CRM jest automatycznie aktualizowany podczas obsługi jego próśb przez agenta.
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.idAktualizowanie rekordu kontaktu
Po pomyślnym rozwiązaniu problemu należy zaktualizować rekord kontaktu, dodając notatki i zmiany etapu cyklu życia. Dzięki temu CRM pozostaje aktualny bez konieczności ręcznej pracy agenta.
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.'
})Ujednolicony kontekst CRM dla agenta
Należy utworzyć jedną funkcję pobierającą kontekst klienta z używanego systemu CRM. Agent wywołuje ją na początku każdej rozmowy, aby personalizować odpowiedzi.
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'}Jakiego języka zapytań Salesforce używa się do pobierania rekordów za pośrednictwem biblioteki simple-salesforce?
Salesforce używa własnego języka zapytań do pobierania rekordów. Znajomość tego języka jest niezbędna do tworzenia narzędzi agenta wyszukujących historię klienta.
Podsumowanie integracji z CRM
Integracja z Salesforce (simple-salesforce + SOQL) lub HubSpot (hubspot-api-client) zapewnia agentowi obsługi klienta pełny kontekst klienta: historię, sprawy, zgłoszenia i dane konta.
Należy zawsze utworzyć ujednoliconą funkcję get_customer_context(), aby wzbogacić prompt agenta o istotną historię przed wygenerowaniem odpowiedzi.
Często zadawane pytania
Czy lekcja „Integracja z CRM: Salesforce i HubSpot” jest bezpłatna?
Tak — pełny tekst „Integracja z CRM: Salesforce i HubSpot” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.
Co nauczysz się w „Integracja z CRM: Salesforce i HubSpot”?
Pobieranie historii klienta, rejestrowanie interakcji i aktualizowanie rekordów CRM. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.
Czy potrzebuję doświadczenia, aby zacząć AI Agents?
Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 2 z 4.
Ile czasu zajmuje lekcja „Integracja z CRM: Salesforce i HubSpot”?
Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.
Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?
Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.
Wszystkie lekcje w tym kursie
- Kierowanie zgłoszeń i logika eskalacji
- Integracja z CRM: Salesforce i HubSpot
- Protokoły przekazywania obsługi człowiekowi
- Zarządzanie kontekstem i historią klienta