Automate Customer Context with CRM Data Retrieval

1. 1
   Set Up Your Environment

   To perform CRM Data Retrieval, you need a client to connect to your CRM's API. For Salesforce, a popular choice is the `simple-salesforce` Python library. Install it using pip. You will also need API credentials from your Salesforce instance.
2. 2
   Authenticate and Connect

   Securely connect to the Salesforce API. Use environment variables to store your username, password, security token, and instance URL. This avoids hardcoding sensitive credentials in your code. The skill will use these to establish a session.
3. 3
   Implement Contact Resolution

   The first step in CRM Data Retrieval is finding the right person. Create a function to query for a contact using a unique identifier like an email address. This function forms the basis of the skill, resolving an email to a specific CRM record.
4. 4
   Retrieve Related Records

   Once you have a contact, expand the CRM Data Retrieval skill to fetch related data. Query for the associated Account to get company-level details and any related Case objects to understand their support history. This provides a 360-degree customer view.
5. 5
   Assemble the Customer Context

   Combine the retrieved data into a structured object. An AI agent can parse this object to get the full context: who the customer is, what company they work for, and their recent support interactions. This is the final output of the CRM Data Retrieval skill.

import os
import json
from unittest.mock import MagicMock

# This starter uses a mock Salesforce client to be runnable without credentials.
# In a real application, you would replace the mock with a real Salesforce connection.

class CrmDataRetrievalSkill:
    def __init__(self, sf_client):
        """Initializes the skill with a Salesforce client (real or mock)."""
        self.sf = sf_client

    def get_customer_context(self, email):
        """Performs CRM Data Retrieval to build a customer context profile."""
        print(f"Attempting to retrieve data for: {email}")
        context = {
            'email': email,
            'contact': None,
            'account': None,
            'open_cases': []
        }
        try:
            # 1. Contact Resolution
            contact_res = self.sf.query(f"SELECT Id, Name, AccountId FROM Contact WHERE Email = '{email}' LIMIT 1")
            if contact_res['totalSize'] == 0:
                print("Contact not found.")
                return context
            
            contact_record = contact_res['records'][0]
            context['contact'] = contact_record
            account_id = contact_record['AccountId']

            # 2. Account Retrieval
            account_res = self.sf.query(f"SELECT Id, Name, Industry FROM Account WHERE Id = '{account_id}' LIMIT 1")
            if account_res['totalSize'] > 0:
                context['account'] = account_res['records'][0]

            # 3. Ticket History Retrieval
            cases_res = self.sf.query(f"SELECT CaseNumber, Subject, Status, Priority FROM Case WHERE AccountId = '{account_id}' AND IsClosed = false")
            if cases_res['totalSize'] > 0:
                context['open_cases'] = cases_res['records']

            return context
        except Exception as e:
            print(f"An error occurred: {e}")
            return None

# --- Runnable Example with Mock Data ---
def setup_mock_salesforce_client():
    """Creates a mock Salesforce client that returns predefined data."""
    mock_sf = MagicMock()

    # Mock responses for different SOQL queries
    def query_side_effect(query_string):
        if "Contact WHERE Email = 'jane.doe@acme.com'" in query_string:
            return {'totalSize': 1, 'records': [{'Id': '003xx00000ABCD', 'Name': 'Jane Doe', 'AccountId': '001xx00000EFGHI'}]}
        if "Account WHERE Id = '001xx00000EFGHI'" in query_string:
            return {'totalSize': 1, 'records': [{'Id': '001xx00000EFGHI', 'Name': 'Acme Corporation', 'Industry': 'Technology'}]}
        if "Case WHERE AccountId = '001xx00000EFGHI'" in query_string:
            return {'totalSize': 1, 'records': [{'CaseNumber': '00012345', 'Subject': 'Billing issue', 'Status': 'New', 'Priority': 'High'}]}
        if "Contact WHERE Email = 'john.smith@widget.co'" in query_string:
            return {'totalSize': 0, 'records': []} # Simulate contact not found
        return {'totalSize': 0, 'records': []}

    mock_sf.query.side_effect = query_side_effect
    return mock_sf

if __name__ == "__main__":
    mock_client = setup_mock_salesforce_client()
    retrieval_skill = CrmDataRetrievalSkill(mock_client)

    print("--- Scenario 1: Customer Found ---")
    customer_context = retrieval_skill.get_customer_context('jane.doe@acme.com')
    print(json.dumps(customer_context, indent=2))

    print("\n--- Scenario 2: Customer Not Found ---")
    customer_context_not_found = retrieval_skill.get_customer_context('john.smith@widget.co')
    print(json.dumps(customer_context_not_found, indent=2))