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
GetExternalContactActionstep 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/flowsendpoint. - The implementation is demonstrated in Python 3.9+, leveraging
httpxfor robust HTTP handling where the SDK falls short, and the officialgenesyscloudSDK 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:
sourceId: The ID of the external contact source created in Step 1.key: The data field to look up (e.g., the phone number).fields: The specific fields to retrieve (optional, defaults to all).onSuccessandonFailuretransitions.
We will create a flow that:
- Starts with a
Connectstep (to handle the incoming call). - Moves to a
GetExternalContactActionstep. - Uses the caller’s
fromnumber as the lookup key. - Stores the result in the
contactobject.
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
.envfile has validGENESYS_CLOUD_CLIENT_IDandGENESYS_CLOUD_CLIENT_SECRET. Verify the Application Integration in Genesys Cloud hasarchitect:flow:writeandarchitect:externalcontact:writescopes enabled. - Code Check: The
PlatformClientin 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
keyparameter inGetExternalContactActionis 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
timeoutsetting in theExternalcontactsourceHttpConfig. Increase it if necessary (max is usually 30 seconds). Ensure your endpoint returns a valid JSON response with a200 OKstatus.
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
stepsarray in thePostFlowRequestBody. Ensure everyon_successandon_failuretransition points to a valididin thestepsarray. Ensure thestart_step_idmatches one of the step IDs.