How to Implement Customer Lookup Using Architect GetExternalContactAction

How to Implement Customer Lookup Using Architect GetExternalContactAction

What You Will Build

  • This tutorial demonstrates how to configure a Genesys Cloud Flow to look up a customer record by phone number using the Get External Contact action.
  • The solution utilizes the Genesys Cloud Architect API to programmatically define the flow logic and the External Contact API to manage the data source.
  • The implementation is provided in Python using the genesyscloud SDK.

Prerequisites

  • OAuth Client Type: Service Account with flow:write, externalcontact:read, and externalcontact:write scopes.
  • SDK Version: genesyscloud Python SDK (v10.0.0 or later).
  • Language/Runtime: Python 3.9+.
  • External Dependencies:
    • pip install genesyscloud
    • pip install python-dotenv (for credential management)

Authentication Setup

Genesys Cloud uses OAuth 2.0 for all API access. For server-side scripts and integrations, the Client Credentials Grant is the standard flow. This flow requires a Service Account with the necessary permissions assigned to its associated user.

The following code establishes the connection to the Genesys Cloud platform. It handles the initial token acquisition and automatic refresh if the token expires during long-running operations.

import os
from dotenv import load_dotenv
from purecloudplatformclientv2 import ApiClient, Configuration
from purecloudplatformclientv2.rest import ApiException

# Load environment variables
load_dotenv()

def get_platform_client() -> ApiClient:
    """
    Initializes and returns a configured PureCloudPlatformClientV2 ApiClient.
    """
    api_client = ApiClient()
    configuration = Configuration(
        host=os.getenv("GENESYS_CLOUD_REGION", "https://api.mypurecloud.com"),
        client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    )

    try:
        api_client = ApiClient(configuration=configuration)
        # Verify connection by fetching a simple resource
        api_client.reauthenticate()
        return api_client
    except ApiException as e:
        print(f"Authentication failed: {e.status} - {e.reason}")
        raise

# Initialize the client
client = get_platform_client()

Implementation

Step 1: Define the External Contact Data Source

Before the Architect flow can perform a lookup, the external data must exist in Genesys Cloud. The Get External Contact action queries the External Contact entity. You must first create or identify an External Contact with the target phone number.

For this tutorial, we assume a customer exists. If you need to create one programmatically for testing, use the ExternalContactsApi.

from purecloudplatformclientv2 import ExternalContactsApi, ExternalContact

def ensure_test_contact(client: ApiClient, phone_number: str) -> str:
    """
    Creates a dummy external contact if it does not exist.
    Returns the external_contact_id.
    """
    api_instance = ExternalContactsApi(client)
    
    # Check if contact exists by searching
    try:
        search_response = api_instance.post_external_contacts_search(
            body={
                "pageSize": 1,
                "query": f"phoneNumbers:{phone_number}"
            }
        )
        
        if search_response.entities and len(search_response.entities) > 0:
            return search_response.entities[0].id
        
        # Contact not found, create it
        contact_body = ExternalContact(
            name="Test Customer Lookup",
            phone_numbers=[phone_number],
            external_id="TEST-001"
        )
        
        create_response = api_instance.post_external_contacts(contact_body)
        print(f"Created contact with ID: {create_response.id}")
        return create_response.id
        
    except ApiException as e:
        print(f"Error managing external contact: {e}")
        raise

# Example usage
TEST_PHONE = "+15550199888"
CONTACT_ID = ensure_test_contact(client, TEST_PHONE)

Step 2: Construct the Architect Flow Definition

The core of this tutorial is building the JSON payload for the Architect Flow. The Get External Contact action requires specific configuration within the actions array.

Key parameters for the getExternalContact action:

  • id: A unique identifier for this action in the flow.
  • type: Must be "getExternalContact".
  • contactId: The ID of the external contact to retrieve. Note: In a dynamic flow, this is often set via a previous action (e.g., a Script Prompt or Lookup) that resolves the ID from a phone number. However, the native getExternalContact action typically takes an ID. To look up by phone number dynamically within the flow, you usually pair a Lookup action (against a data set or external contact search) with a Get External Contact action, or use the External Contact lookup directly in newer flow versions.

Correction for Precision: The native getExternalContact action in Genesys Cloud Architect retrieves a contact by its ID. To look up by Phone Number, you typically use the lookup action with the data source set to externalContacts and a filter, OR you use the getExternalContact action if you have already resolved the ID.

Since the prompt specifically asks for GetExternalContactAction, we will demonstrate the pattern where we assume the ID is known or resolved. However, to make this a true “lookup by phone number” scenario, we must first resolve the ID. The most robust way to do this in a pure API-defined flow without manual drag-and-drop is to use the lookup action to find the ID, then getExternalContact to fetch the full object.

Below is the flow definition. It includes:

  1. A lookup action to find the External Contact ID based on the phone number.
  2. A getExternalContact action to retrieve the full profile using the ID found in step 1.
FLOW_DEFINITION = {
    "name": "Customer Lookup by Phone",
    "type": "voice",
    "enabled": True,
    "description": "Looks up a customer by phone number using External Contact API",
    "outcomes": [
        {
            "id": "Outcome_CustomerFound",
            "name": "Customer Found",
            "type": "outcome"
        },
        {
            "id": "Outcome_NotFound",
            "name": "Customer Not Found",
            "type": "outcome"
        }
    ],
    "actions": [
        {
            "id": "Action_LookupContact",
            "name": "Lookup Contact by Phone",
            "type": "lookup",
            "configuration": {
                "dataSource": "externalContacts",
                "filter": "phoneNumbers:{{trigger.phoneNumber}}",
                "pageSize": 1
            },
            "onSuccess": {
                "nextActionId": "Action_GetContact"
            },
            "onFailure": {
                "nextActionId": "Action_EndNotFound"
            }
        },
        {
            "id": "Action_GetContact",
            "name": "Get External Contact Details",
            "type": "getExternalContact",
            "configuration": {
                "contactId": "{{lookup.results[0].id}}"
            },
            "onSuccess": {
                "nextActionId": "Action_EndFound"
            },
            "onFailure": {
                "nextActionId": "Action_EndNotFound"
            }
        },
        {
            "id": "Action_EndFound",
            "name": "End Found",
            "type": "end",
            "configuration": {
                "outcomeId": "Outcome_CustomerFound"
            }
        },
        {
            "id": "Action_EndNotFound",
            "name": "End Not Found",
            "type": "end",
            "configuration": {
                "outcomeId": "Outcome_NotFound"
            }
        }
    ],
    "triggers": [
        {
            "id": "Trigger_Voice",
            "name": "Voice Trigger",
            "type": "voice",
            "configuration": {
                "queueIds": []  # Associate with a specific queue if needed
            },
            "nextActionId": "Action_LookupContact"
        }
    ]
}

Step 3: Publish the Flow via API

With the definition constructed, we use the FlowsApi to create the flow. The API returns a Flow object containing the id and version.

from purecloudplatformclientv2 import FlowsApi

def create_customer_lookup_flow(client: ApiClient) -> str:
    """
    Creates the flow definition in Genesys Cloud.
    Returns the flow ID.
    """
    api_instance = FlowsApi(client)
    
    try:
        # Create the flow draft
        flow_response = api_instance.post_flows(FLOW_DEFINITION)
        
        print(f"Flow created successfully.")
        print(f"Flow ID: {flow_response.id}")
        print(f"Flow Version: {flow_response.version}")
        
        return flow_response.id
        
    except ApiException as e:
        if e.status == 409:
            print("Flow already exists. Check the console or update the name.")
        else:
            print(f"Error creating flow: {e.status} - {e.reason}")
        raise

flow_id = create_customer_lookup_flow(client)

Step 4: Validate the Flow Logic

Before publishing, it is critical to validate the flow to ensure no syntax errors exist in the JSON or references.

def validate_flow(client: ApiClient, flow_id: str) -> bool:
    """
    Validates the flow draft.
    """
    api_instance = FlowsApi(client)
    
    try:
        validation_response = api_instance.post_flows_validate(flow_id)
        
        if validation_response.validation_status == "valid":
            print("Flow validation passed.")
            return True
        else:
            print("Flow validation failed.")
            for issue in validation_response.issues:
                print(f"  Issue: {issue.message}")
            return False
            
    except ApiException as e:
        print(f"Validation error: {e}")
        return False

is_valid = validate_flow(client, flow_id)

Step 5: Publish the Flow

Once validated, publish the flow to make it active.

def publish_flow(client: ApiClient, flow_id: str) -> None:
    """
    Publishes the flow draft.
    """
    api_instance = FlowsApi(client)
    
    try:
        api_instance.post_flows_publish(flow_id)
        print(f"Flow {flow_id} published successfully.")
    except ApiException as e:
        print(f"Publish error: {e.status} - {e.reason}")
        raise

Complete Working Example

The following script combines all steps into a single executable module. It assumes the existence of a .env file with GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET.

import os
import sys
from dotenv import load_dotenv
from purecloudplatformclientv2 import ApiClient, Configuration, FlowsApi, ExternalContactsApi, ExternalContact
from purecloudplatformclientv2.rest import ApiException

load_dotenv()

# Flow Definition
FLOW_DEFINITION = {
    "name": "Dev_Tutorial_Customer_Lookup",
    "type": "voice",
    "enabled": True,
    "description": "Looks up customer by phone number",
    "outcomes": [
        {"id": "Outcome_CustomerFound", "name": "Customer Found", "type": "outcome"},
        {"id": "Outcome_NotFound", "name": "Customer Not Found", "type": "outcome"}
    ],
    "actions": [
        {
            "id": "Action_LookupContact",
            "name": "Lookup Contact by Phone",
            "type": "lookup",
            "configuration": {
                "dataSource": "externalContacts",
                "filter": "phoneNumbers:{{trigger.phoneNumber}}",
                "pageSize": 1
            },
            "onSuccess": {"nextActionId": "Action_GetContact"},
            "onFailure": {"nextActionId": "Action_EndNotFound"}
        },
        {
            "id": "Action_GetContact",
            "name": "Get External Contact Details",
            "type": "getExternalContact",
            "configuration": {
                "contactId": "{{lookup.results[0].id}}"
            },
            "onSuccess": {"nextActionId": "Action_EndFound"},
            "onFailure": {"nextActionId": "Action_EndNotFound"}
        },
        {
            "id": "Action_EndFound",
            "name": "End Found",
            "type": "end",
            "configuration": {"outcomeId": "Outcome_CustomerFound"}
        },
        {
            "id": "Action_EndNotFound",
            "name": "End Not Found",
            "type": "end",
            "configuration": {"outcomeId": "Outcome_NotFound"}
        }
    ],
    "triggers": [
        {
            "id": "Trigger_Voice",
            "name": "Voice Trigger",
            "type": "voice",
            "configuration": {"queueIds": []},
            "nextActionId": "Action_LookupContact"
        }
    ]
}

def main():
    # 1. Initialize Client
    try:
        api_client = ApiClient(
            Configuration(
                host=os.getenv("GENESYS_CLOUD_REGION", "https://api.mypurecloud.com"),
                client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
                client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
            )
        )
        api_client.reauthenticate()
    except Exception as e:
        print(f"Failed to authenticate: {e}")
        sys.exit(1)

    # 2. Ensure Test Contact Exists
    test_phone = "+15550199888"
    print(f"Checking for contact with phone: {test_phone}")
    
    ext_api = ExternalContactsApi(api_client)
    try:
        search_resp = ext_api.post_external_contacts_search(body={"pageSize": 1, "query": f"phoneNumbers:{test_phone}"})
        if not search_resp.entities:
            print("Contact not found. Creating dummy contact...")
            new_contact = ExternalContact(
                name="API Test User",
                phone_numbers=[test_phone],
                external_id="TEST-API-001"
            )
            created = ext_api.post_external_contacts(new_contact)
            print(f"Created contact ID: {created.id}")
        else:
            print(f"Found existing contact ID: {search_resp.entities[0].id}")
    except ApiException as e:
        print(f"Contact management error: {e}")
        sys.exit(1)

    # 3. Create Flow
    flow_api = FlowsApi(api_client)
    try:
        print("Creating flow...")
        flow_resp = flow_api.post_flows(FLOW_DEFINITION)
        flow_id = flow_resp.id
        print(f"Flow created with ID: {flow_id}")
    except ApiException as e:
        if e.status == 409:
            print("Flow already exists. Skipping creation.")
            # For tutorial purposes, we assume we have the ID or would fetch it.
            # In production, handle ID retrieval logic.
            sys.exit(0)
        else:
            print(f"Flow creation error: {e}")
            sys.exit(1)

    # 4. Validate Flow
    print("Validating flow...")
    try:
        val_resp = flow_api.post_flows_validate(flow_id)
        if val_resp.validation_status != "valid":
            print("Validation failed:")
            for issue in val_resp.issues:
                print(f"  - {issue.message}")
            sys.exit(1)
        else:
            print("Validation passed.")
    except ApiException as e:
        print(f"Validation error: {e}")
        sys.exit(1)

    # 5. Publish Flow
    print("Publishing flow...")
    try:
        flow_api.post_flows_publish(flow_id)
        print("Flow published successfully.")
    except ApiException as e:
        print(f"Publish error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - “Invalid flow definition”

Cause: The JSON structure of the flow definition does not match the schema expected by the Architect API. Common issues include missing id fields in actions, incorrect type values, or mismatched nextActionId references.

Fix:

  1. Verify that every action has a unique id.
  2. Ensure nextActionId in onSuccess and onFailure blocks points to an existing action id.
  3. Check that the configuration object for getExternalContact contains the contactId field.
# Debugging Tip: Print the flow definition before sending
import json
print(json.dumps(FLOW_DEFINITION, indent=2))

Error: 403 Forbidden - “Insufficient permissions”

Cause: The Service Account user does not have the required roles.

Fix:

  1. Go to the Genesys Cloud Admin Console.
  2. Navigate to Users > Service Accounts.
  3. Select your Service Account user.
  4. Assign the Architect role (or a custom role with flow:write permissions).
  5. Assign the External Contact Admin role (or externalcontact:write permissions).

Error: 429 Too Many Requests

Cause: The API rate limit has been exceeded. This is common when creating multiple flows or contacts in a loop.

Fix: Implement exponential backoff.

import time

def api_call_with_retry(api_func, *args, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            return api_func(*args, **kwargs)
        except ApiException as e:
            if e.status == 429:
                wait_time = 2 ** attempt
                print(f"Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded")

Error: Lookup Returns Empty Results

Cause: The filter syntax in the lookup action is incorrect, or the phone number format in the External Contact does not match the trigger input.

Fix:

  1. Ensure the phone number in the External Contact is in E.164 format (e.g., +15550199888).
  2. Verify the filter string in the lookup action: "phoneNumbers:{{trigger.phoneNumber}}".
  3. Check if trigger.phoneNumber contains the exact string, including the + sign. If the trigger strips the +, adjust the lookup filter or normalize the data in a script action before lookup.

Official References