CRM-Integration: Salesforce und HubSpot
Kundenhistorie abrufen, Interaktionen protokollieren und CRM-Datensätze aktualisieren.
CRM-Integration: Salesforce und HubSpot ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 2 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.
Warum die CRM-Integration wichtig ist
Ein Kundenservice-Agent, der die Historie des Kunden nicht sehen kann, arbeitet im Blindflug. CRM-Systeme wie Salesforce und HubSpot enthalten den vollständigen Kundendatensatz: frühere Tickets, Käufe, Abonnementstufe und Kontonotizen.
Durch die Integration mit dem CRM kann der Agent Antworten personalisieren und Probleme schneller lösen.
Verbindung zu Salesforce mit simple-salesforce
simple-salesforce ist die beliebteste Python-Bibliothek für die Salesforce REST API. Stellen Sie die Verbindung mit username/password/security-token oder über einen OAuth-Connected-App-Flow her.
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-Abfragen: Kundendatensätze abrufen
Salesforce Object Query Language (SOQL) ähnelt SQL, arbeitet aber mit Salesforce-Objekten. Verwenden Sie SOQL, um einen Kontakt anhand seiner E-Mail-Adresse zu suchen und die zugehörigen Kontodaten abzurufen.
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']}")
Supportfallhistorie aus Salesforce abrufen
Fragen Sie das Case-Objekt ab, um die früheren Supporttickets des Kunden abzurufen. Berücksichtigen Sie den Status und die Lösung des Falls, damit der Agent nicht zweimal dieselben Fragen stellen muss.
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']})")
Einen neuen Case in Salesforce erstellen
Wenn der Agent aus einem Gespräch ein neues Supportticket erstellt, fügen Sie einen neuen Case-Datensatz ein. Verknüpfen Sie ihn mit dem Kontakt und dem Konto, damit er in der Zeitleiste des Kunden erscheint.
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')
Verbindung zu HubSpot herstellen
Die Bibliothek hubspot-api-client kapselt die REST API von HubSpot. Authentifizieren Sie sich mit einem Zugriffstoken einer privaten App (für Server-zu-Server-Integrationen ist kein OAuth erforderlich).
from hubspot import HubSpot
from hubspot.crm.contacts import SimplePublicObjectInput
client = HubSpot(access_token='YOUR_HUBSPOT_PRIVATE_APP_TOKEN')
print('HubSpot client ready')Einen HubSpot-Kontakt suchen
Verwenden Sie die Contacts Search API, um einen Kunden anhand seiner E-Mail-Adresse zu finden. Die Antwort enthält alle Kontakteigenschaften, einschließlich Lifecycle-Phase, Deal-Verknüpfungen und benutzerdefinierter Felder.
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}Ticketverlauf aus HubSpot abrufen
HubSpot speichert Supporttickets in seinem tickets-Objekt. Rufen Sie die mit einem Kontakt verknüpften Tickets über die Associations API ab.
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']}]")
Ein HubSpot-Ticket erstellen
Fügen Sie ein neues Ticket ein und verknüpfen Sie es innerhalb eines einzigen Workflows mit dem Kontakt. So bleibt der CRM-Datensatz des Kunden automatisch aktuell, während der Agent Anfragen bearbeitet.
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.idEinen Kontaktdatensatz aktualisieren
Aktualisieren Sie den Datensatz des Kontakts nach einer erfolgreichen Lösung mit Notizen und Änderungen an der Lifecycle-Phase. So bleibt das CRM ohne manuellen Aufwand für den Agenten aktuell.
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.'
})Einheitlicher CRM-Kontext für den Agenten
Erstellen Sie eine einzelne Funktion, die den Kundenkontext aus dem jeweils konfigurierten CRM abruft. Der Agent ruft diese Funktion zu Beginn jedes Gesprächs auf, um seine Antworten zu personalisieren.
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'}Welche Salesforce-Abfragesprache wird verwendet, um Datensätze über die simple-salesforce-Bibliothek abzurufen?
Salesforce verwendet eine eigene Abfragesprache, um Datensätze abzurufen. Dieses Wissen ist entscheidend für die Entwicklung von Agent-Tools, die die Kundenhistorie durchsuchen.
Zusammenfassung der CRM-Integration
Die Integration mit Salesforce (simple-salesforce + SOQL) oder HubSpot (hubspot-api-client) gibt Ihrem Kundenservice-Agenten den vollständigen Kundenkontext: Historie, Fälle, Tickets und Kontodaten.
Erstellen Sie immer eine einheitliche Funktion get_customer_context(), damit der Agenten-Prompt vor der Antwortgenerierung mit relevanter Historie angereichert wird.
Häufig gestellte Fragen
Ist die Lektion „CRM-Integration: Salesforce und HubSpot“ kostenlos?
Ja — der vollständige Text von „CRM-Integration: Salesforce und HubSpot“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.
Was lerne ich in „CRM-Integration: Salesforce und HubSpot“?
Kundenhistorie abrufen, Interaktionen protokollieren und CRM-Datensätze aktualisieren. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.
Brauche ich Erfahrung, um AI Agents zu starten?
Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 2 von 4.
Wie lange dauert die Lektion „CRM-Integration: Salesforce und HubSpot“?
Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.
Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?
Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.
Alle Lektionen in diesem Kurs
- Ticket-Routing und Eskalationslogik
- CRM-Integration: Salesforce und HubSpot
- Protokolle zur Übergabe an Menschen
- Kundenkontext und -historie verwalten