AI Agents · 课时

CRM 集成：Salesforce 与 HubSpot

Q: 「CRM 集成：Salesforce 与 HubSpot」这节课中我会学到什么？

获取客户历史记录、记录互动并更新 CRM 记录 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

获取客户历史记录、记录互动并更新 CRM 记录

CRM 集成：Salesforce 与 HubSpot 是 CoddyKit 上的免费 AI Agents 课时。这是第 2 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 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 对象查询语言（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 联系人

使用联系人搜索 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'}

通过 simple-salesforce 库获取记录时使用哪种 Salesforce 查询语言？

Salesforce 使用自己的查询语言来获取记录。了解这一点对于构建可查询客户历史记录的代理工具至关重要。

CRM 集成回顾

与 Salesforce（simple-salesforce + SOQL）或 HubSpot（hubspot-api-client）集成后，您的客服代理就能获得完整的客户上下文：历史记录、案例、工单和账户数据。

请始终构建统一的 get_customer_context() 函数，让代理提示词在生成响应前就加入相关历史记录。

常见问题解答

「CRM 集成：Salesforce 与 HubSpot」课时是免费的吗？

是的 — 「CRM 集成：Salesforce 与 HubSpot」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「CRM 集成：Salesforce 与 HubSpot」这节课中我会学到什么？

获取客户历史记录、记录互动并更新 CRM 记录你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 2 节课，共 4 节。

「CRM 集成：Salesforce 与 HubSpot」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Agents