How to Implement a Phone Number Lookup Using GetExternalContactAction in Genesys Cloud Architect

How to Implement a Phone Number Lookup Using GetExternalContactAction in Genesys Cloud Architect

What You Will Build

  • You will build a Python script that uses the Genesys Cloud CX REST API to create an Architect flow containing a GetExternalContactAction step that retrieves customer data based on the caller’s phone number.
  • This tutorial uses the Genesys Cloud CX Python SDK (genesyscloud) to interact with the /api/v2/architect/flows endpoint.
  • The implementation is demonstrated in Python 3.9+, leveraging httpx for robust HTTP handling where the SDK falls short, and the official genesyscloud SDK for flow definition.

Prerequisites

  • OAuth Client: A Genesys Cloud CX Application Integration with the following scopes:
    • architect:flow:write (to create/update flows)
    • architect:flow:read (to list existing flows, optional for this tutorial)
    • architect:externalcontact:write (to create the external contact source definition)
    • architect:externalcontact:read
  • SDK Version: genesyscloud >= 110.0.0 (or the latest available version).
  • Runtime: Python 3.9 or higher.
  • Dependencies:
    • genesyscloud (Official SDK)
    • httpx (For low-level API calls if SDK gaps exist, though primarily we will use the SDK)
    • python-dotenv (For secure credential management)

Install dependencies:

pip install genesyscloud httpx python-dotenv

Authentication Setup

Genesys Cloud CX uses OAuth 2.0 for authentication. For server-to-server integration, you must use the Client Credentials Grant flow. The official SDK handles token acquisition and refresh automatically when configured correctly.

Create a .env file in your project root:

GENESYS_CLOUD_REGION=us-east-1
GENESYS_CLOUD_CLIENT_ID=your_client_id
GENESYS_CLOUD_CLIENT_SECRET=your_client_secret

Initialize the SDK client in your Python script:

import os
from dotenv import load_dotenv
from purecloud_platform_client import (
    Configuration,
    ApiClient,
    ArchitectApi,
    PlatformClient
)

# Load environment variables
load_dotenv()

def get_architect_api():
    """
    Initializes and returns an authenticated ArchitectApi instance.
    """
    config = Configuration(
        client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET"),
        region=os.getenv("GENESYS_CLOUD_REGION")
    )
    
    # The PlatformClient manages the token lifecycle
    platform_client = PlatformClient(config)
    
    # Instantiate the Architect API client
    architect_api = ArchitectApi(ApiClient(config))
    
    return architect_api, platform_client

Implementation

Step 1: Define the External Contact Source

Before you can use GetExternalContactAction, you must define an External Contact Source in Genesys Cloud. This source tells the system where to look up the data (e.g., an HTTP endpoint, a database connector, or a static JSON map).

For this tutorial, we will simulate a simple HTTP-based lookup. In a production environment, this source would point to your CRM or Customer Data Platform (CDP).

The GetExternalContactAction requires a sourceId. We will create a simple “HTTP” type external contact source. Note that creating the source is a separate API call from creating the flow.

from purecloud_platform_client.models import (
    PostExternalcontactsourceRequestBody,
    ExternalcontactsourceHttpConfig
)

def create_external_contact_source(architect_api: ArchitectApi) -> str:
    """
    Creates a simple HTTP-based External Contact Source.
    Returns the source ID.
    """
    # Define the HTTP configuration for the lookup
    http_config = ExternalcontactsourceHttpConfig(
        url="https://your-cdn-or-api.com/customer/lookup",
        method="GET",
        # The 'key' is the value passed from the flow. 
        # Here we expect the phone number to be passed as a query parameter or header.
        # We will define how the flow maps the phone number to this request in Step 2.
        timeout=3000
    )

    # Construct the request body
    request_body = PostExternalcontactsourceRequestBody(
        name="Customer Phone Lookup Source",
        type="HTTP",
        http_config=http_config
    )

    try:
        # Create the source
        response = architect_api.post_architect_external_contacts_sources(body=request_body)
        print(f"Created External Contact Source with ID: {response.id}")
        return response.id
    except Exception as e:
        print(f"Failed to create external contact source: {e}")
        raise

# Execute Step 1
architect_api, platform_client = get_architect_api()
SOURCE_ID = create_external_contact_source(architect_api)

Important Note on External Contact Sources:
The GetExternalContactAction does not make the HTTP call directly. It queries the External Contact Service. The service uses the sourceId to determine how to fetch the data. If you are using a custom HTTP source, you must ensure the URL is accessible from Genesys Cloud’s infrastructure. For testing, you can use a service like webhook.site or a simple AWS Lambda endpoint.

Step 2: Construct the Flow with GetExternalContactAction

The core of this tutorial is defining the flow structure. A Genesys Cloud Flow is a JSON object. We will use the SDK models to build this JSON safely.

The GetExternalContactAction step requires:

  1. sourceId: The ID of the external contact source created in Step 1.
  2. key: The data field to look up (e.g., the phone number).
  3. fields: The specific fields to retrieve (optional, defaults to all).
  4. onSuccess and onFailure transitions.

We will create a flow that:

  1. Starts with a Connect step (to handle the incoming call).
  2. Moves to a GetExternalContactAction step.
  3. Uses the caller’s from number as the lookup key.
  4. Stores the result in the contact object.
from purecloud_platform_client.models import (
    PostFlowRequestBody,
    FlowStep,
    GetExternalContactAction,
    ConnectStep,
    EndStep,
    Transition,
    Condition
)

def build_flow_definition(source_id: str) -> dict:
    """
    Builds the JSON payload for the Architect Flow.
    """
    
    # 1. Define the GetExternalContactAction step
    lookup_action = GetExternalContactAction(
        source_id=source_id,
        # The key is the phone number. 
        # We use a contact attribute reference. 
        # 'from' is the caller's phone number.
        key="{contact.from}",
        # Optional: Specify fields to retrieve. 
        # If omitted, all fields from the source are retrieved.
        fields=["name", "email", "loyalty_tier"]
    )
    
    lookup_step = FlowStep(
        id="lookup_customer",
        type="GetExternalContactAction",
        action=lookup_action,
        # Define transitions
        on_success=Transition(next_step_id="set_variables"),
        on_failure=Transition(next_step_id="end_error")
    )

    # 2. Define a Connect step (Required for voice flows)
    connect_step = FlowStep(
        id="connect",
        type="Connect",
        action=ConnectStep()
    )

    # 3. Define a SetVariable step to process the result
    # After the lookup, the data is available in {contact.externalContact}
    # We can map it to user-friendly variables.
    from purecloud_platform_client.models import SetVariableAction
    
    set_var_action = SetVariableAction(
        variables=[
            {
                "name": "customer_name",
                "value": "{contact.externalContact.name}"
            },
            {
                "name": "customer_tier",
                "value": "{contact.externalContact.loyalty_tier}"
            }
        ]
    )
    
    set_var_step = FlowStep(
        id="set_variables",
        type="SetVariableAction",
        action=set_var_action,
        on_success=Transition(next_step_id="end_success")
    )

    # 4. Define End steps
    end_success = FlowStep(
        id="end_success",
        type="End",
        action=EndStep()
    )
    
    end_error = FlowStep(
        id="end_error",
        type="End",
        action=EndStep()
    )

    # 5. Assemble the full flow definition
    flow_body = PostFlowRequestBody(
        name="Phone Number Lookup Flow",
        description="Demonstrates GetExternalContactAction using phone number.",
        type="Voice",
        settings={
            "timeout": 3600,
            "max_attempts": 1
        },
        start_step_id="connect",
        steps=[
            connect_step,
            lookup_step,
            set_var_step,
            end_success,
            end_error
        ]
    )
    
    # Convert the SDK object to a dictionary for API submission
    # The SDK's to_dict() method handles the serialization
    return flow_body.to_dict()

# Execute Step 2
flow_payload = build_flow_definition(SOURCE_ID)
print("Flow Payload Generated:")
import json
print(json.dumps(flow_payload, indent=2))

Key Parameter Explanation:

  • key="{contact.from}": This is a contact attribute reference. When the flow executes, Genesys Cloud replaces {contact.from} with the actual phone number of the caller (e.g., +15550199999). This value is sent to the External Contact Source as the lookup identifier.
  • source_id: Must match the ID returned from Step 1.
  • fields: If your external source returns a large JSON object, specifying fields reduces payload size and latency.

Step 3: Create the Flow in Genesys Cloud

Now that we have the payload, we will send it to the Genesys Cloud API to create the flow.

def create_flow(architect_api: ArchitectApi, flow_payload: dict):
    """
    Creates the flow in Genesys Cloud using the generated payload.
    """
    try:
        # The SDK expects a PostFlowRequestBody object, but we can often pass 
        # a dict if the SDK model supports it, or we reconstruct the object.
        # For robustness, we will pass the dict directly to the API call 
        # if the SDK allows, or convert back to model.
        
        # Reconstructing the model from dict is safer for strict typing
        flow_body = PostFlowRequestBody.from_dict(flow_payload)
        
        response = architect_api.post_architect_flows(body=flow_body)
        print(f"Successfully created Flow ID: {response.id}")
        print(f"Flow Name: {response.name}")
        return response.id
    except Exception as e:
        # Handle API errors
        print(f"Error creating flow: {e}")
        if hasattr(e, 'body'):
            print(f"Response Body: {e.body}")
        raise

# Execute Step 3
FLOW_ID = create_flow(architect_api, flow_payload)

Complete Working Example

Below is the complete, consolidated Python script. It handles authentication, source creation, flow definition, and flow creation.

import os
import json
from dotenv import load_dotenv
from purecloud_platform_client import (
    Configuration,
    ApiClient,
    ArchitectApi,
    PlatformClient,
    PostExternalcontactsourceRequestBody,
    ExternalcontactsourceHttpConfig,
    PostFlowRequestBody,
    FlowStep,
    GetExternalContactAction,
    ConnectStep,
    EndStep,
    Transition,
    SetVariableAction
)

# Load environment variables
load_dotenv()

def get_architect_api():
    """
    Initializes and returns an authenticated ArchitectApi instance.
    """
    config = Configuration(
        client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET"),
        region=os.getenv("GENESYS_CLOUD_REGION")
    )
    
    platform_client = PlatformClient(config)
    architect_api = ArchitectApi(ApiClient(config))
    
    return architect_api, platform_client

def create_external_contact_source(architect_api: ArchitectApi) -> str:
    """
    Creates a simple HTTP-based External Contact Source.
    """
    http_config = ExternalcontactsourceHttpConfig(
        url="https://webhook.site/your-unique-id", # Replace with your actual endpoint
        method="GET",
        timeout=3000
    )

    request_body = PostExternalcontactsourceRequestBody(
        name="Dev Test Customer Lookup",
        type="HTTP",
        http_config=http_config
    )

    try:
        response = architect_api.post_architect_external_contacts_sources(body=request_body)
        print(f"Created External Contact Source with ID: {response.id}")
        return response.id
    except Exception as e:
        print(f"Failed to create external contact source: {e}")
        raise

def build_flow_definition(source_id: str) -> dict:
    """
    Builds the JSON payload for the Architect Flow.
    """
    lookup_action = GetExternalContactAction(
        source_id=source_id,
        key="{contact.from}",
        fields=["name", "email", "loyalty_tier"]
    )
    
    lookup_step = FlowStep(
        id="lookup_customer",
        type="GetExternalContactAction",
        action=lookup_action,
        on_success=Transition(next_step_id="set_variables"),
        on_failure=Transition(next_step_id="end_error")
    )

    connect_step = FlowStep(
        id="connect",
        type="Connect",
        action=ConnectStep()
    )

    set_var_action = SetVariableAction(
        variables=[
            {
                "name": "customer_name",
                "value": "{contact.externalContact.name}"
            },
            {
                "name": "customer_tier",
                "value": "{contact.externalContact.loyalty_tier}"
            }
        ]
    )
    
    set_var_step = FlowStep(
        id="set_variables",
        type="SetVariableAction",
        action=set_var_action,
        on_success=Transition(next_step_id="end_success")
    )

    end_success = FlowStep(
        id="end_success",
        type="End",
        action=EndStep()
    )
    
    end_error = FlowStep(
        id="end_error",
        type="End",
        action=EndStep()
    )

    flow_body = PostFlowRequestBody(
        name="Phone Lookup Demo",
        description="Uses GetExternalContactAction to look up customer by phone.",
        type="Voice",
        settings={
            "timeout": 3600,
            "max_attempts": 1
        },
        start_step_id="connect",
        steps=[
            connect_step,
            lookup_step,
            set_var_step,
            end_success,
            end_error
        ]
    )
    
    return flow_body.to_dict()

def create_flow(architect_api: ArchitectApi, flow_payload: dict):
    """
    Creates the flow in Genesys Cloud.
    """
    try:
        flow_body = PostFlowRequestBody.from_dict(flow_payload)
        response = architect_api.post_architect_flows(body=flow_body)
        print(f"Successfully created Flow ID: {response.id}")
        return response.id
    except Exception as e:
        print(f"Error creating flow: {e}")
        if hasattr(e, 'body'):
            print(f"Response Body: {e.body}")
        raise

if __name__ == "__main__":
    # Step 1: Authenticate
    architect_api, _ = get_architect_api()
    
    # Step 2: Create External Contact Source
    source_id = create_external_contact_source(architect_api)
    
    # Step 3: Build Flow Definition
    flow_payload = build_flow_definition(source_id)
    
    # Step 4: Create Flow
    flow_id = create_flow(architect_api, flow_payload)
    
    print(f"Final Flow ID: {flow_id}")

Common Errors & Debugging

Error: 401 Unauthorized or 403 Forbidden

  • Cause: The OAuth token is expired, or the client credentials lack the required scopes.
  • Fix: Ensure your .env file has valid GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET. Verify the Application Integration in Genesys Cloud has architect:flow:write and architect:externalcontact:write scopes enabled.
  • Code Check: The PlatformClient in the SDK automatically refreshes tokens. If you are making raw HTTP calls, ensure you implement token refresh logic.

Error: 400 Bad Request - “Invalid key format”

  • Cause: The key parameter in GetExternalContactAction is malformed.
  • Fix: Ensure the key uses valid contact attribute references. For phone numbers, {contact.from} is correct. If you are looking up by a custom variable, use {contact.variable_name}. Do not include curly braces in the string literal if the SDK model expects a raw string; however, in Genesys Cloud Architect, the braces are part of the syntax. The SDK usually handles this correctly, but if you are sending raw JSON, ensure the braces are escaped or included as required by the API version.

Error: 500 Internal Server Error - “External Contact Source Timeout”

  • Cause: The HTTP endpoint specified in the External Contact Source is not reachable from Genesys Cloud’s servers, or it takes too long to respond.
  • Fix: Verify the URL is public and accessible. Check the timeout setting in the ExternalcontactsourceHttpConfig. Increase it if necessary (max is usually 30 seconds). Ensure your endpoint returns a valid JSON response with a 200 OK status.

Error: 422 Unprocessable Entity - “Flow validation failed”

  • Cause: The flow definition has structural errors, such as missing steps, disconnected transitions, or invalid step IDs.
  • Fix: Review the steps array in the PostFlowRequestBody. Ensure every on_success and on_failure transition points to a valid id in the steps array. Ensure the start_step_id matches one of the step IDs.

Official References