Architecting a Cross-Platform Unified Customer ID Resolution Engine using the External Contacts API
What This Guide Covers
You are building a Customer Identity Resolution (CIR) engine that maintains a canonical customer record in Genesys Cloud’s External Contacts API - the single source of truth that links a customer’s phone number, email, mobile number, CRM account ID, loyalty program ID, and social handle into one unified profile. When a customer calls from a new number they’ve never used before but mentions their email, the IVR resolves their full profile in under 500ms, populates the agent screen pop with their complete history, and routes to the queue aligned to their customer tier - all before the agent answers.
Prerequisites, Roles & Licensing
- Genesys Cloud: CX 2 or CX 3 (External Contacts API is available on all tiers)
- Permissions required (service account):
External Contacts > External Contact > AllExternal Contacts > External Organization > AllRouting > Queue > View
- External systems: Your CRM (Salesforce, Microsoft Dynamics, ServiceNow) as the master record source; the CIR engine acts as the Genesys Cloud-side synchronization layer
- Infrastructure: A low-latency lookup service (Lambda + DynamoDB or Redis) that can resolve customer identifiers in <200ms for use inside live Architect flows
The Implementation Deep-Dive
1. The Identity Resolution Problem Space
In a multi-channel contact center, a single customer may contact you through:
- Voice from their mobile (+1-917-555-0142)
- Voice from their office landline (+1-212-555-0188)
- Email (personal: john.smith@gmail.com; work: j.smith@acmecorp.com)
- Web messaging (anonymous session)
- Twitter DM (@jsmith_tweets)
- WhatsApp (+44-7700-900123 - UK number when traveling)
Without identity resolution, each contact creates a separate conversation history with no continuity. An agent answering the office landline call sees zero history even though the customer called from their mobile yesterday.
The Genesys Cloud External Contacts API solves this by maintaining a contact record with multiple address fields that each function as lookup keys:
{
"id": "external-contact-uuid",
"firstName": "John",
"lastName": "Smith",
"phoneNumber": {
"e164": "+19175550142" // Primary mobile
},
"otherPhones": [
{ "phoneNumber": { "e164": "+12125550188" }, "label": "Work" }
],
"emailAddress": "john.smith@gmail.com",
"otherEmails": [
{ "email": "j.smith@acmecorp.com", "label": "Work" }
],
"externalIds": [
{ "externalIdCaseInsensitive": "SFDC-001234567", "externalIdDefinitionId": "salesforce-account-id" },
{ "externalIdCaseInsensitive": "LOYALTY-GLD-98765", "externalIdDefinitionId": "loyalty-program-id" }
],
"title": "VP Engineering",
"customFields": {
"customerTier": "Enterprise",
"preferredLanguage": "en-US",
"lastPurchaseDate": "2025-04-15"
}
}
2. Designing the External ID Definition Schema
External ID definitions (the metadata for custom identifiers like CRM IDs) must be configured before you can store them on contacts:
import requests
def create_external_id_definition(
name: str,
definition_key: str,
trust_level: str, # "Trusted" or "Unverified"
access_token: str,
base_url: str
) -> str:
"""
Create an external ID definition. Returns the definition ID.
"""
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
resp = requests.post(
f"{base_url}/api/v2/externalcontacts/identifiers/schemas",
headers=headers,
json={
"name": name,
"key": definition_key,
"trustLevel": trust_level
}
)
resp.raise_for_status()
return resp.json()["id"]
# One-time setup: create your identifier schemas
def setup_id_schemas(access_token: str, base_url: str) -> dict:
schemas = {}
schemas["salesforce"] = create_external_id_definition(
"Salesforce Account ID", "salesforce-account-id", "Trusted", access_token, base_url
)
schemas["loyalty"] = create_external_id_definition(
"Loyalty Program ID", "loyalty-program-id", "Trusted", access_token, base_url
)
schemas["crm_ticket"] = create_external_id_definition(
"CRM Ticket Reference", "crm-ticket-id", "Unverified", access_token, base_url
)
return schemas
3. The Identity Resolution Lookup Engine
The lookup engine must resolve any incoming identifier (ANI, email, CRM ID) to the canonical External Contact record in under 200ms for use inside live IVR flows.
Multi-key lookup with fallback chain:
from functools import lru_cache
import time
class CustomerIdentityResolver:
def __init__(self, access_token: str, base_url: str, redis_client=None):
self.access_token = access_token
self.base_url = base_url
self.headers = {"Authorization": f"Bearer {access_token}"}
self.redis = redis_client # Optional cache layer
def resolve(self, **identifiers) -> dict | None:
"""
Resolve a customer by any combination of identifiers.
Tries each identifier in priority order.
Usage:
resolver.resolve(phone="+19175550142")
resolver.resolve(email="john@gmail.com")
resolver.resolve(external_id="SFDC-001234567", schema_id="salesforce-schema-uuid")
resolver.resolve(phone="+19175550142", email="john@gmail.com") # Merge check
"""
cache_key = f"cir:{sorted(identifiers.items())}"
# Check Redis cache first (TTL: 5 minutes)
if self.redis:
cached = self.redis.get(cache_key)
if cached:
return json.loads(cached)
contact = None
# Try phone lookup (fastest - indexed)
if "phone" in identifiers:
contact = self._lookup_by_phone(identifiers["phone"])
# Try email lookup if phone didn't resolve
if not contact and "email" in identifiers:
contact = self._lookup_by_email(identifiers["email"])
# Try external ID lookup (CRM ID, loyalty number)
if not contact and "external_id" in identifiers:
contact = self._lookup_by_external_id(
identifiers["external_id"],
identifiers.get("schema_id", "")
)
# Cache successful resolution
if contact and self.redis:
self.redis.setex(cache_key, 300, json.dumps(contact))
return contact
def _lookup_by_phone(self, phone_e164: str) -> dict | None:
resp = requests.get(
f"{self.base_url}/api/v2/externalcontacts/contacts",
headers=self.headers,
params={"q": phone_e164, "fields": "phoneNumber,otherPhones"}
)
resp.raise_for_status()
contacts = resp.json().get("entities", [])
return contacts[0] if contacts else None
def _lookup_by_email(self, email: str) -> dict | None:
resp = requests.get(
f"{self.base_url}/api/v2/externalcontacts/contacts",
headers=self.headers,
params={"q": email, "fields": "emailAddress,otherEmails"}
)
resp.raise_for_status()
contacts = resp.json().get("entities", [])
return contacts[0] if contacts else None
def _lookup_by_external_id(self, external_id: str, schema_id: str) -> dict | None:
resp = requests.get(
f"{self.base_url}/api/v2/externalcontacts/contacts",
headers=self.headers,
params={
"externalId": external_id,
"externalIdDefinitionId": schema_id
}
)
resp.raise_for_status()
contacts = resp.json().get("entities", [])
return contacts[0] if contacts else None
The Trap - doing real-time External Contacts API lookups inside a synchronous Architect Data Action with a 2-second timeout: The External Contacts API lookup can take 300-800ms. Add Redis caching: on first call from a given ANI, the lookup goes to Genesys Cloud’s API and the result is cached for 5 minutes. Subsequent calls within the same shift (a customer calling back 3 minutes later) return the cached result in <5ms. This is the difference between a reliable screen pop and a frustrated agent watching a spinner.
4. Creating and Merging Contact Records
Creating a new External Contact from CRM data:
def create_or_update_contact(
crm_account: dict,
access_token: str,
base_url: str,
schema_ids: dict
) -> str:
"""
Creates a new External Contact or updates an existing one.
Returns the External Contact ID.
"""
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
# Check if contact already exists by CRM ID
existing = resolver.resolve(
external_id=crm_account["id"],
schema_id=schema_ids["salesforce"]
)
payload = {
"firstName": crm_account.get("FirstName", ""),
"lastName": crm_account.get("LastName", ""),
"title": crm_account.get("Title", ""),
"phoneNumber": {
"e164": normalize_phone(crm_account.get("Phone", ""))
} if crm_account.get("Phone") else None,
"emailAddress": crm_account.get("Email", ""),
"externalIds": [
{
"externalIdCaseInsensitive": crm_account["id"],
"externalIdDefinitionId": schema_ids["salesforce"]
}
],
"customFields": {
"customerTier": crm_account.get("Account_Tier__c", "Standard"),
"accountStatus": crm_account.get("Status__c", "Active"),
"assignedQueueId": map_tier_to_queue(crm_account.get("Account_Tier__c"))
}
}
if existing:
# Update existing contact
resp = requests.patch(
f"{base_url}/api/v2/externalcontacts/contacts/{existing['id']}",
headers=headers,
json=payload
)
else:
# Create new contact
resp = requests.post(
f"{base_url}/api/v2/externalcontacts/contacts",
headers=headers,
json=payload
)
resp.raise_for_status()
return resp.json()["id"]
Merging duplicate contacts:
When identity resolution discovers that two External Contact records represent the same person (same name, overlapping phone numbers), merge them:
def merge_duplicate_contacts(
primary_contact_id: str,
duplicate_contact_id: str,
access_token: str,
base_url: str
):
"""
Merge duplicate_contact_id INTO primary_contact_id.
All interactions associated with the duplicate are reassociated.
"""
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
resp = requests.post(
f"{base_url}/api/v2/externalcontacts/contacts/{primary_contact_id}/merge",
headers=headers,
json={
"sourceId": duplicate_contact_id
}
)
resp.raise_for_status()
return resp.json()
5. Architect Integration: IVR Identity Resolution Flow
The IVR calls your identity resolution Lambda at call start, attaches the resolved profile to the conversation, and uses it for routing decisions:
Architect Data Action - Identity Lookup:
[Inbound Call]
→ [Set Variable: inboundANI = Call.Ani]
→ [Action: Call Data Action "Resolve Customer Identity"]
Input: { "ani": inboundANI }
Output: contactId, customerTier, preferredLanguage, assignedQueueId, displayName
[Decision: contactId != "NOT_FOUND"]
YES → [Set Participant Data:
customerTier = Output.customerTier,
externalContactId = Output.contactId,
customerName = Output.displayName]
→ [Play]: "Welcome back, {Output.displayName}."
→ [Route to Output.assignedQueueId]
NO → [Standard new customer flow]
→ [Collect email or account number via speech/DTMF]
→ [Action: Resolve by Email]
→ [Create new External Contact if still not found]
Lambda handler for the Data Action:
def lambda_handler(event, context):
ani = event.get("ani", "")
email = event.get("email", "")
resolver = CustomerIdentityResolver(
access_token=get_cached_token(),
base_url="https://api.mypurecloud.com",
redis_client=get_redis()
)
contact = resolver.resolve(phone=ani, email=email if email else None)
if not contact:
return {
"contactId": "NOT_FOUND",
"customerTier": "Standard",
"assignedQueueId": DEFAULT_QUEUE_ID,
"displayName": "Valued Customer"
}
return {
"contactId": contact["id"],
"customerTier": contact.get("customFields", {}).get("customerTier", "Standard"),
"assignedQueueId": contact.get("customFields", {}).get("assignedQueueId", DEFAULT_QUEUE_ID),
"displayName": f"{contact.get('firstName', '')} {contact.get('lastName', '')}".strip(),
"preferredLanguage": contact.get("customFields", {}).get("preferredLanguage", "en-US")
}
Validation, Edge Cases & Troubleshooting
Edge Case 1: ANI Spoofing Causing Incorrect Identity Resolution
Caller ID spoofing is trivial and common in fraud scenarios. An attacker who knows a VIP customer’s phone number can call from that number and receive their tier routing. For sensitive operations (accessing financial accounts, changing passwords), treat ANI-only resolution as Unverified. Require a secondary verification factor (PIN, OTP to registered email) before granting access to sensitive functions - even for known ANIs.
Edge Case 2: Bulk CRM Synchronization Performance
Synchronizing 500,000 CRM accounts into External Contacts via the API is a batch operation that could take 24+ hours at standard API rate limits. Use the Genesys Cloud External Contacts bulk import feature (CSV import via the UI or bulk API endpoint) for initial population. Reserve the real-time API for individual record creates, updates, and merges triggered by CRM webhooks.
Edge Case 3: External Contact Record Growth and Stale Data
External Contact records accumulate data over years. A customer’s customFields.customerTier that was set 3 years ago may no longer reflect their current status. Implement a webhook from your CRM that fires on significant field changes and updates the External Contact in real time. Additionally, run a nightly reconciliation job that compares External Contact custom fields against the CRM master record and patches any discrepancies.
Edge Case 4: Privacy - External Contacts Containing Sensitive Fields
External Contacts store customer data within Genesys Cloud’s systems. All data stored here is subject to Genesys Cloud’s data processing terms and your GDPR obligations. Do not store data fields in External Contacts that don’t serve a routing or personalization purpose - avoid storing financial account numbers, health diagnoses, or other sensitive data. Store the CRM ID (a pseudonymous reference) instead, and let your CRM API serve the sensitive data in real time only when needed.