Leveraging Amazon Connect for Real-Time Incident Response Calls

Amazon Connect is a cloud-based contact center.

This use case was to automate incident response calls: when an alert is triggered, it sends the alert to SNS, which triggers a Lambda function. This Lambda function calls the first escalation point. If no one answers, it escalates to the second contact.

Here’s how I built it:

Ensure that you are in the Correct Region — I’m using ap-southeast-1 as I am located in Singapore.

1. Setting up AWS Identity and Access Management (IAM)

Step 1: In the AWS Console, search for IAM.

Step 2: Under Access management , click on Roles , then select Create role.

Step 3: For the trusted entity type, choose AWS Service , and under Use Case , select Lambda.

Step 4: Use the following AWS Managed Policies :

• AmazonConnect_FullAccess
• AmazonDynamoDBFullAccess
• AmazonSNSFullAccess
• AWSLambdaBasicExecutionRole

Step 5: Finally, create a role name, then click Create role. Make sure to take note of your Role Name for future use.

2. Setting Up Amazon Simple Notification Service (SNS) (Optional)

Step 1: In the AWS Console, search for SNS.

Step 2: Navigate to Topics and click on Create topic.

Step 3: Choose Standard for the topic type and give it a name (e.g., “AmazonConnectSNS”). After the topic is created, make sure to note down the ARN for later use.

3. Setting Up Amazon Lambda

Step 1: In the AWS Console, search for Lambda.

Step 2: Click on Create Function.

Step 3: On the Create Function page, provide a Function Name (e.g., “AmazonConnectLambda”), select Runtime as Python 3.xx, and choose the existing role you created in the IAM setup section.

Step 4: Add an SNS Trigger to the Lambda function.

Step 5: Go to Configuration , then navigate to Generation Configuration , and set the Timeout to 1 minute.

Step 6: Paste the following code into the lambda_function.py file. I will explain the code in the last section, where we’ll also revisit it to modify variables:

import boto3
import time
from datetime import datetime

# Initialize the DynamoDB resource
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('CallLogs')

# Create a Connect client
connect_client = boto3.client('connect')

#Amazon Connect Instance ID, Contact Flow ID and Claim number
CONNECT_INSTANCE_ID = 'xxxxxxxxx'
CONTACT_FLOW_ID = 'xxxxxxxxx'
SOURCE_PHONE_NUMBER = 'xxxxxxxxx' 

# Phone numbers for the contacts
FIRST_CONTACT_PHONE = "+65xxxxxx"
SECOND_CONTACT_PHONE = "+65xxxxxx"

def log_event_to_dynamodb(contact_id, event, phone_number, escalation_stage):
    """
    Log the event (CONNECTED, DISCONNECTED) into DynamoDB with relevant details.
    """
    try:
        table.put_item(
            Item={
                'ContactId': contact_id,
                'Event': event,
                'Timestamp': datetime.utcnow().isoformat(),
                'PhoneNumber': phone_number,
                'EscalationStage': escalation_stage
            }
        )
        print(f"Logged {event} event for ContactId {contact_id} in DynamoDB")
    except Exception as e:
        print(f"Error logging event to DynamoDB: {e}")

def start_outbound_call(phone_number, escalation_stage):
    """
    Start an outbound call using Amazon Connect and return the ContactId.
    """
    try:
        response = connect_client.start_outbound_voice_contact(
            DestinationPhoneNumber=phone_number,
            ContactFlowId=CONTACT_FLOW_ID,
            InstanceId=CONNECT_INSTANCE_ID,
            SourcePhoneNumber=SOURCE_PHONE_NUMBER,
            Attributes={
                'Escalation': escalation_stage
            }
        )
        contact_id = response['ContactId']
        print(f"Started outbound call to {phone_number} with ContactId: {contact_id}")
        return contact_id
    except Exception as e:
        print(f"Failed to start outbound call: {e}")
        raise e

def check_call_status(contact_id):
    """
    Check the status of the outbound call using the ContactId.
    """
    try:
        response = connect_client.describe_contact(
            InstanceId=CONNECT_INSTANCE_ID,
            ContactId=contact_id
        )

        if 'Contact' in response and 'DisconnectTimestamp' in response['Contact']:
            print(f"Call with ContactId {contact_id} was disconnected at {response['Contact']['DisconnectTimestamp']}")
            return 'DISCONNECTED'

        elif 'Contact' in response and 'ConnectedToSystemTimestamp' in response['Contact']:
            print(f"Call with ContactId {contact_id} was connected at {response['Contact']['ConnectedToSystemTimestamp']}")
            return 'CONNECTED'

    except Exception as e:
        print(f"Failed to check call status: {e}")
        raise e

def lambda_handler(event, context):
    try:
        acknowledgment = event.get('Details', {}).get('Parameters', {}).get('acknowledgment', 'False')
        escalation_stage = event.get('Details', {}).get('Parameters', {}).get('EscalationStage', 'FirstAttempt')
        print(f"Acknowledgment: {acknowledgment}, Escalation Stage: {escalation_stage}")
    except KeyError:
        print("Expected keys not found in event")
        acknowledgment = 'False'

    if acknowledgment == 'False' and escalation_stage == 'FirstAttempt':
        print("Calling the first contact...")
        contact_id = start_outbound_call(FIRST_CONTACT_PHONE, 'FirstAttempt')

        for attempt in range(10):
            time.sleep(5) 
            status = check_call_status(contact_id)

            if status == 'CONNECTED':
                log_event_to_dynamodb(contact_id, 'CONNECTED', FIRST_CONTACT_PHONE, 'FirstAttempt')
                print("First contact answered the call. No need to escalate.")
                return {
                    'statusCode': 200,
                    'body': 'First contact answered the call.'
                }

            elif status == 'DISCONNECTED':
                print(f"First contact did not pick up or disconnected. Escalating to second contact.")
                break

        print("Escalating to second contact...")
        contact_id = start_outbound_call(SECOND_CONTACT_PHONE, 'SecondAttempt')
        log_event_to_dynamodb(contact_id, 'ESCALATION', SECOND_CONTACT_PHONE, 'SecondAttempt')
        return {
            'statusCode': 200,
            'body': 'Second contact attempt made.'
        }

    elif acknowledgment == 'False' and escalation_stage == 'SecondAttempt':
        print("Calling the second contact...")
        contact_id = start_outbound_call(SECOND_CONTACT_PHONE, 'SecondAttempt')
        log_event_to_dynamodb(contact_id, 'ESCALATION', SECOND_CONTACT_PHONE, 'SecondAttempt')
        return {
            'statusCode': 200,
            'body': 'Second contact attempt made.'
        }

    elif acknowledgment == 'True':
        print("First contact acknowledged the call.")
        return {
            'statusCode': 200,
            'body': 'Call acknowledged by the first contact.'
        }

    else:
        print("Unexpected scenario encountered.")
        return {
            'statusCode': 500,
            'body': 'Unexpected scenario.'
        }

Step 6 : Click on Deploy

4. Setting up DynamoDB

Step 1: In the AWS Console, search for DynamoDB.

Step 2: In the navigation pane, click on Tables , then select Create Table.

Step 3: In the Create Table form, follow these steps:

• Set the Table name to CallLogs.
• For the Partition key , use ContactId.
• For the Sort key , use Timestamp.

Finally, click Create table at the bottom.

5. Setting Up Amazon Connect

Step 1 : In the AWS Console, search for Amazon Connect.

Step 2: Create a new instance by following the on-screen instructions. Make sure to enable both Inbound and Outbound calls during setup.

Step 3: After the instance is created, navigate to Flows in the side menu. In the flow editor, under AWS Lambda , add the Lambda function that was created earlier.

Step 4: Log in to your Amazon Connect instance with admin privileges. Once logged in, navigate to the sidebar and select Phone numbers.

Step 5: Click on Claim a number and follow the recommended steps to complete the process.

Step 6: Test both incoming and outgoing calls:

• To test incoming calls, use a mobile phone to dial the number you claimed.
• For outgoing calls, use the Contact Control Panel in Amazon Connect.

Step 7: In the Amazon Connect navigation bar, click on Flows and create a new flow.

Step 8: Add a title to your flow, design the flow, and once done, click Publish. You can follow this basic structure for your flow:

Entry -> Set Logging Behavior block -> Play Prompt -> Get Customer Input

• If the user presses 1, proceed to the Play Prompt block-> Disconnect block
• If not, invoke the Lambda function block -> Disconnect block

In the Invoke AWS Lambda block, you’ll need to set up the function input parameters. Follow this structure for the input parameters:

6. Source Code Explanation

• Calling the First Contact : Once the Lambda function is triggered, it initiates an outbound call to the FIRST_CONTACT_PHONE. The function then monitors the call status (whether it’s connected or disconnected) and logs the event accordingly.
• Escalating to the Second Contact : If the first contact doesn’t respond, the function escalates the process by making an outbound call to the SECOND_CONTACT_PHONE.
• Handling Call Acknowledgment : If an acknowledgment is received (set to ‘True’), the function stops any further escalation.

7. Wrapping Things Up

When revisiting the Lambda function, remember to update the following variables:

• CONNECT_INSTANCE_ID: This is the unique Amazon Connect instance ARN (UUID) associated with your instance.
• CONTACT_FLOW_ID: This is the UUID of your contact flow. You can find it in the URL of your flow in Amazon Connect, located immediately after /contact-flow/.
• SOURCE_PHONE_NUMBER: This is the phone number you claimed in Amazon Connect.
• FIRST_CONTACT_PHONE: The phone number of the first person to receive the call.
• SECOND_CONTACT_PHONE: The phone number of the second person to receive the call if the first contact doesn’t answer.

In this setup, SNS is used to receive alerts from an AWS CloudWatch alarm, which triggers the SNS topic. You can simulate an alert by going to SNS and publishing a message manually.

However, for simplicity, we can skip that and go straight to Lambda and trigger the function directly to test the flow.

If everything works as expected, you should receive a call, and you’ll be able to view the call logs in DynamoDB under the CallLogs table.

While in DynamoDB, you will be able to see the CallLogs.

I hope you found this use case helpful.

Overall, I found it enjoyable to experiment with Aamzon Connect.

You can connect with me on linkedin: https://www.linkedin.com/in/marcuschanjh/