Voice - Tutorial: Record and retrieve a call
NOTE:
To add this product to your account, contact a Telesign expert. This product is available for full-service accounts only.
This tutorial describes how to use Telesign Voice to make, record, and retrieve a call.
Sample code
import base64
import enum
import json
import requests
import time
from functools import partial
class RecordMethod(enum.Enum):
START = "record_start"
STOP = "record_stop"
PAUSE = "record_pause"
RESUME = "record_resume"
class DialAction(object):
def __init__(self, to, caller_id, external_id = None): #external_id is optional parameter
self.method = 'dial'
self.parameters = {
'to': to,
'caller_id_number': caller_id,
'external_id': external_id
}
class HangupAction(object):
def __init__(self, reference_id):
self.method = 'hangup'
self.parameters = {
'reference_id': reference_id,
}
class RecordAction(object):
def __init__(self, reference_id, method: RecordMethod):
self.method = method.value
self.parameters = {
'reference_id': reference_id,
}
telesign_base_url = 'https://rest-ww.telesign.com'
# Configure this to be the virtual number received from Telesign.
virtual_number = '15559876543'
# Telesign authentication credentials
customer_id = 'Your Customer ID'
secret_key = 'Your API key'
def telesign_headers():
auth_token = base64.b64encode(bytes(':'.join([customer_id, secret_key]), 'utf-8')).decode('utf-8')
return {
'Content-Type': 'application/json',
'Authorization': 'Basic {}'.format(auth_token),
}
def send_action(action):
body = {
'jsonrpc': '2.0',
'method': action.method,
'params': action.parameters,
}
rpc_endpoint = telesign_base_url + '/v2/voice'
print("POST {} {} {}".format(rpc_endpoint, telesign_headers(), body))
return requests.post(
rpc_endpoint,
headers=telesign_headers(),
data=json.dumps(body))
def call_object(reference_id):
return requests.get(
telesign_base_url + '/call/{}'.format(reference_id),
headers=telesign_headers(),
).json()
def poll(fn, checker, max_timeout=60):
start_time = time.time()
result = fn()
while time.time() < start_time + max_timeout:
if checker(result):
return result
time.sleep(1)
result = fn()
raise RuntimeError('Polling Timed out after {} second(s)'.format(max_timeout))
def call_event(reference_id, event_name):
call = requests.get(telesign_base_url + '/call/{}'.format(reference_id), headers=telesign_headers()).json()
for event in call['events']:
if event['event'] == event_name:
print(event)
return event
else:
return None
def call_recording(reference_id):
call_recordings = requests.get(telesign_base_url + '/call_recording/{}'.format(reference_id), headers=telesign_headers()).json()
for recording in call_recordings:
if recording['status'] == 'available':
# There can be multiple call_recordings, we're just going to poll for the first available recording
print(recording['url'])
return recording
else:
return None
def poll_call_for_event(reference_id, event_name):
return poll(partial(call_event, reference_id, event_name), lambda event: event, max_timeout=60)
def poll_call_recording(reference_id):
return poll(partial(call_recording, reference_id), lambda recording: recording, max_timeout=60)
def outbound_record_flow():
target_number = '15551234567'
external_id = '4142b474-d65d-4265-a5fa-a581b1fe6101'
dial_response = send_action(DialAction(caller_id=virtual_number, to=target_number, external_id=external_id))
dial_response_data = dial_response.json()
print(dial_response_data)
reference_id = dial_response_data['reference_id']
# Dialing has connected to the target end user
dial_completed_event = poll_call_for_event(reference_id, 'dial_completed')
# Start recording for 2 seconds
record_start_response = send_action(RecordAction(reference_id, RecordMethod.START))
time.sleep(2)
# Pause recording for 2 seconds
record_pause_response = send_action(RecordAction(reference_id, RecordMethod.PAUSE))
time.sleep(2)
# Resume recording for 2 seconds
record_resume_response = send_action(RecordAction(reference_id, RecordMethod.RESUME))
time.sleep(2)
# Stop recording for 2 seconds
record_stop_response = send_action(RecordAction(reference_id, RecordMethod.STOP))
time.sleep(2)
# Hang up call
hangup_response = send_action(HangupAction(reference_id))
call_completed_event = poll_call_for_event(reference_id, 'call_completed')
# Query for call recording
call_recording = poll_call_recording(reference_id)
# Delete call recording
requests.delete(telesign_base_url + '/call_recording/{}'.format(reference_id), headers=telesign_headers())
outbound_record_flow()
Updated about 1 month ago
What’s Next