Kinesis Quickstart

Create a Kinesis stream for your Account

We have prepared a script for you that sets up a Kinesis stream with a shard count. You can get the script here. Copy the script and save it to a file named create_kinesis_stream.sh. Make it executable with chmod 755. The script will set up the policies and roles needed to allow Twilio to write to your stream. The only thing you need to do is run the script and pass the stream name and shard count as arguments.

Use the name twilio-events for the Kinesis stream. You can use any name, but if you use a different one than twilio-events, you will need to change the permission instructions later on with your custom name.

Set the number of shards to 1. This makes initial validation easier. It can be modified after the Sink has been set up.

Here’s an example:

./create_kinesis_stream.sh twilio-events 1 | jq . > twilio-sink.json 

The script uses the AWS CLI to create the Kinesis stream with the specific shard count. It will return a JSON payload with the details you need to create a Sink. In the above example, the received data will be written to a file named twilio-sink.json.

You can update the shard count of the Kinesis stream after it has been validated by using this script:

if [ $# -ne 2 ]; then
  echo
  echo "usage: $0 <stream_name> <shard_count> <scaling_type>"
  echo "<scaling_type> possible values UNIFORM_SCALING”
  echo
  exit 1
fi

# update stream
STREAM_NAME=$1
SHARD_COUNT=$2
SCALING_TYPE=$3

aws kinesis update-shard-count --stream-name $STREAM_NAME --scaling-type $SCALING_TYPE --target-shard-count $SHARD_COUNT

Here is an example showing how the Kinesis stream shard count is increased to 6:

./update_kinesis_shard_count.sh twilio-events 6 UNIFORM_SCALING

Create a Sink instance

To create a new Kinesis Sink, run this command:

twilio api:events:v1:sinks:create --description <add description here> \
  --sink-configuration '{"arn":"${your kinesis arn}","role_arn":"${your role arn created above}","external_id":"${external id created above}"}' \
  --sink-type kinesis

Validate the Sink

Kinesis Sinks need to be validated before events can be delivered to them. To validate a Sink, you need to tell the Sink that you want to test it. You do this by sending a test message to your Sink using this Twilio CLI command:

twilio api:events:v1:sinks:test:create --sid ${sid from above}

If you are validating a Kinesis Sink, you can use our script to retrieve the test message from your account’s Kinesis stream.

Now download this script as cat_kinesis.sh and make it executable with chmod 755. Again, you will need to ensure your terminal has valid AWS credentials before executing this command:

./cat_kinesis.sh twilio-events

The script will output a test event like this:

{
  "specversion": "1.0",
  "type": "com.twilio.eventstreams.test-event",
  "source": "Sink",
  "id": "AC1234567890123456789012",
  "dataschema": "https://events-schemas.twilio.com/EventStreams.TestSink/1.json",
  "datacontenttype": "application/json",
  "time": "2020-08-18T21:10:02.113Z",
  "data": "{\"test_id\":\"95001c1e-5b8e-45f5-830d-f9cbf1d16420\"}"
}

The data key in the test event includes the test ID: the value of the test_id key. Once you have it, hit Control-C to exit the script.

Now execute the following command, using the URL from the Sink creation response above and the test ID returned just now:

twilio api:events:v1:sinks:validate:create --sid ${sid from above} \
  --test-id ${test id from payload}

This should return the result valid.

Subscribe to Twilio events

Now that you have a valid Sink, you can subscribe to one or more events. Below are some of the available events. You can also view a full list of available events.

You can subscribe to any of these events by making an API call. This is done with the following command. The new subscription is configured to read the event-types from the --types argument — you’ll need the event type and schema version from the table above. The event-types you specify in the--types argument will be sent to the Sink specified by the --sink-sid argument. Use the sink SID of the sink you created above.

twilio api:events:v1:subscriptions:create --description <description> \
  --sink-sid <sink id validated above DGxxx> \
  --types '{"type": "<event_type>","schema_version": <version>}'

For instance, to subscribe to all call summary events, you would run:

twilio api:events:v1:subscriptions:create \
  --description "Subscription on 3 call_summary events" \
  --sink-sid <sink id validated above DGxxx> \
  --types '{"type":"com.twilio.voice.insights.call-summary.partial","schema_version":1}' \
  --types '{"type":"com.twilio.voice.insights.call-summary.predicted-complete","schema_version":1}' \
  --types '{"type":"com.twilio.voice.insights.call-summary.complete","schema_version":1}'

Read from Kinesis

You can quickly tail the events coming out of Kinesis with the cat_kinesis.sh script from above. Again, you will need to ensure your terminal has valid AWS credentials before executing the script:

./cat_kinesis.sh twilio-events

The script iterates over the specified shards in a stream and requests records from them. Any data that is received is Base64 encoded, so the script decodes it and makes it human readable using jq.

Parse the data

Data is sent to Kinesis in the CloudEvents format. For Kinesis the Data is automatically base64 encoded/decoded by the SDK. Therefore depending on how you are reading the data from kinesis you may have to base64 decode the data.

For example, here’s a Voice Insights Call Summary Predicted Complete event:

{
"specversion": "1.0",
"type": "com.twilio.voice.insights.call-summary.predicted-complete",
"source": "/v1/Voice/CA00000000000000000000000000000000/Summary",
"id": "EZ00000000000000000000000000000000",
"dataschema": "https://events-schemas.twilio.com/VoiceInsights.CallSummary/1",
"datacontenttype": "application/json",
"time": "2020-07-23T22:56:33.000Z",
"data": "{\"call_sid\":\"CA00000000000000000000000000000000\",\"account_sid\":\"AC00000000000000000000000000000000\",\"parent_call_sid\":\"\",\"parent_account_sid\":\"\",\"start_time\":\"2020-07-23T22:56:28Z\",\"end_time\":\"2020-07-23T22:56:33Z\",\"duration\":0,\"connect_duration\":0,\"call_type\":\"client\",\"call_state\":\"canceled\",\"from\":{\"caller\":\"+55555555\",\"callee\":\"\",\"carrier\":\"MEGA:CORP\",\"connection\":\"landline\",\"number_prefix\":\"5555\",\"location\":{\"lat\":0.0,\"lon\":-0.0},\"city\":\"\",\"country_code\":\"US\",\"country_subdivision\":\"\",\"ip_address\":\"\",\"sdk\":null},\"to\":{\"caller\":\"\",\"callee\":\"client:SOMECLIENT\",\"carrier\":\"\",\"connection\":\"twilio_sdk\",\"number_prefix\":\"\",\"location\":null,\"city\":\"\",\"country_code\":\"\",\"country_subdivision\":\"\",\"ip_address\":\"\",\"sdk\":null},\"processing_state\":\"partial\",\"processing_version\":1,\"sip_edge\":null,\"carrier_edge\":null,\"sdk_edge\":null,\"client_edge\":null,\"tags\":[],\"attributes\":null,\"properties\":{\"q850_cause\":0,\"last_sip_response_num\":0,\"pdd_ms\":0,\"route_id\":\"\",\"media_region\":\"unknown_realm\",\"signaling_region\":\"unknown_realm\",\"twilio_media_ip\":\"\",\"twilio_signaling_ip\":\"\",\"external_media_ip\":\"\",\"external_signaling_ip\":\"\",\"sip_call_id\":\"\",\"user_agent\":\"\",\"selected_region\":\"unknown_realm\",\"region\":\"unknown_realm\",\"trunk_sid\":\"\",\"disconnected_by\":\"unknown_disconnected_by\",\"direction\":\"outbound_api\",\"settings\":null}}"
}

The data field can be expected to meet the Voice Insights Call Summary schema.