Call Events Integration Guide
Overview
This guide explains how to integrate with JAS Connect's webhook system to monitor call events, determine participants and call direction, and use call recordings after completion.
You will:
- Use
call_statusevents to track when a call status events during the call. - Use
new_recordingevents to detect when a call ends. - Make further requests to get the recording and other information.
Webhook Event Types
JAS Connect emits two main webhook events:
| Event Name | Description | Timing |
|---|---|---|
call_status | Indicates a call is in progress | During the call |
call_summary | Indicates a new ai summary has been initiated | After a call summary is initiated |
new_recording | Indicates a new recording has been created | After the call ends |
Event Handling Workflow
1. Processing call_status Events
Payload Example
{
"eventName": "call_status",
"timestamp": "2025-04-14T15:04:23.161Z",
"data": {
"event": {
"entity": "/callcontrol/101/participants/172",
"event_type": 0
},
"sequence": 518
}
}The event_type will be 0 if the participant is inserted, 1 if the participant is removed.
Steps
-
Extract the extension and participant ID:
From the
entitystring:/callcontrol/{dn}/participants/{participant_id}For example:
const entity = "/callcontrol/101/participants/172" const parts = entity.split("/") const dn = parts[2] // "101" const participantId = parts[4] // "172" if (event.event_type === 0) { // fetch call status } else { // participant is removed (call leg ended) } -
Fetch participant details:
Call the participant info API to get call metadata:
GET /api/callcontrol/{dn}/participants/{participantId}Response example:
{ "id": 36, "status": "Dialing", "dn": "101", "party_caller_name": "", "party_dn": "10000", "party_caller_id": "5148888888", "party_did": "", "device_id": "sip:101@127.0.0.1:5063", "party_dn_type": "Wexternalline", "direct_control": false, "originated_by_dn": "", "originated_by_type": "None", "referred_by_dn": "", "referred_by_type": "None", "on_behalf_of_dn": "", "on_behalf_of_type": "None", "callid": 26, "legid": 1 } -
Interpret the data:
The callid attribute uniquely identifies the call across call_status events. The id attribute uniquely identifies the call participant accross call_status events.
- Store active call:
const dnCalls = activeCalls.get(dn)
dnCalls.set(participantId, status)2. Processing new_recording Events
Payload Example
{
"eventName": "new_recording",
"timestamp": "2025-04-14T15:04:23.161Z",
"data": {
"call_id": "26", // from call control
"start_time": "2025-04-14T15:04:23.161Z",
"end_time": "2025-04-14T15:04:23.161Z",
"recording_url": "101/[John Doe]_101-5148888888_20250603153720(26).wav",
"id_recording": 1,
"cdr_id": "61f450de-8b46-4326-a306-32eb6ad5e534"
}
}Fetch the Recording
GET /api/recordings/:urlimport requests
recording_url = "101/[John Doe]_101-5148888888_20250603153720(26).wav"
url = f"https://api.jasconnect.com/v2/api/recordings/{recording_url}"
headers = {
"Authorization": "Bearer TOKEN"
}
response = requests.get(url, headers=headers)
response.raise_for_status()
with open("example.mp3", "wb") as f:
f.write(response.content)Example Application
This project is a simple HTTP server in Python that handles incoming call_status and new_recording events via a Jas Connect Webhooks. The server registers itself with the Jas Connect API to receive call status updates and logs relevant call information upon receiving those updates.
Prerequisites
- Python 3.x
- Required Python packages:
requests
You can install the required dependencies using:
pip install requestsOverview
The server does the following:
- Registers itself to receive
call_statusevents from the Jas Connect API. - Listens for incoming HTTP
POSTrequests that contain call status updates. - Upon receiving an update, logs relevant call details by calling the call control API endpoint that provides information about the call participants.
- Unregisters the webhook when the server is stopped.
Configuration
You need to replace the following variables with your specific configuration:
TOKEN: The API token for authenticating requests (recieved from/auth/authenticate).CALLBACK_URL: The public URL where the webhook will send events (An easy way test this is usingngrok).
TOKEN = "your_token"
CALLBACK_URL = "your_callback_url"How to Run
- Update the configuration at the top of the script with your values.
- Run the Python script:
python main.py-
The server will start listening on the specified port (
9999by default) and will register forcall_statusevents with the Jas Connect API. -
Example application output
Successfully registered for call status event
Serving on http://0.0.0.0:9999
Recieved event #256 call_status at 2025-03-27T22:02:01.758811044-04:00
Status Dialing | 103 <-> 5148888888
Recieved event #257 call_status at 2025-03-27T22:02:01.770012536-04:00
Status Dialing | 103 <-> 5148888888
Recieved event #259 call_status at 2025-03-27T22:02:07.60556135-04:00
Status Connected | 103 <-> 5148888888
Recieved event #260 call_status at 2025-03-27T22:02:07.608772016-04:00
Status Connected | 103 <-> 5148888888
Recieved event #258 call_status at 2025-03-27T22:02:07.605558344-04:00
Status Connected | 103 <-> 5148888888
Recieved event #261 call_status at 2025-03-27T22:02:11.192685568-04:00
Recieved event #262 call_status at 2025-03-27T22:02:11.197793332-04:00
^CSuccessfully unregistered call status eventUsage
- When the server receives a
call_statusevent, it will parse the event, extract call information, and log it by making a GET request to the Jas Connect API. - The server will continue running and listening for events until manually stopped.
Application
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
import hmac
import hashlib
import requests
TOKEN = ""
CALLBACK_URL = ""
WEBHOOK_SECRET = ""
PORT = 9999
HOST = "0.0.0.0"
JAS_CONNECT = "https://api.jasconnect.com/v2"
active_calls: dict[str, dict[str, dict]] = {}
def register_call_status(url: str):
headers = {
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json",
}
for event in ["call_status", "new_recording"]:
data = {
"event": event,
"url": url,
}
response = requests.post(f"{JAS_CONNECT}/webhooks", json=data, headers=headers)
if response.status_code == 200:
print(f"Successfully registered for {event} event")
else:
print(
f"Failed to register for events: {response.status_code}, {response.text}"
)
exit(1)
def unregister_call_status():
headers = {
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json",
}
for event in ["call_status", "new_recording"]:
response = requests.delete(
f"{JAS_CONNECT}/webhooks?event={event}", json=None, headers=headers
)
if response.status_code == 202:
print(f"Successfully unregistered {event} event")
def handle_call_status(data, timestamp):
sequence = data["data"]["sequence"]
entity: str = data["data"]["event"]["entity"]
parts = entity.split("/")
dn, id = parts[2], parts[4]
print(f"Recieved event #{sequence} call_status at {timestamp}")
if data["data"]["event"]["event_type"] > 0:
del active_calls[dn][id]
print(f"Call {id} ended")
return
headers = {
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json",
}
response = requests.get(
f"{JAS_CONNECT}/api/callcontrol/{dn}/participants/{id}",
json=None,
headers=headers,
)
if response.status_code != 200:
if response.status_code != 403:
print(f"Recieved status {response.status_code} from call control api")
exit(1)
return
data = response.json()
if not active_calls.get(dn):
active_calls[dn] = {}
active_calls[dn][id] = data
def handle_new_recording(data, timestamp):
print(
f"Recieved event for dn {data['data']['dn']}, call #{data['data']['call_id']} new_recording at {timestamp} url: {data['data']['recording_url']}"
)
# updated last call with dn and call_id
pass
def verify_webhook(secret: str, message: str, received_signature: str) -> bool:
expected_signature = hmac.new(
secret.encode(), message.encode(), hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected_signature, received_signature)
class CallHandler(BaseHTTPRequestHandler):
def log_message(self, format: str, *args) -> None:
pass
def do_POST(self):
content_length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_length).decode("utf-8")
data: dict = json.loads(body)
sig = self.headers.get("X-SIG")
if not sig:
print("No signature found in request")
return
if not verify_webhook(WEBHOOK_SECRET, body, sig):
print("Webhook signature is invalid")
return
timestamp = data["timestamp"]
event_name = data["eventName"]
if event_name == "call_status":
handle_call_status(data, timestamp)
self.send_response(200)
elif event_name == "new_recording":
handle_new_recording(data, timestamp)
self.send_response(200)
print(active_calls)
print()
def main():
unregister_call_status()
register_call_status(CALLBACK_URL)
try:
with HTTPServer((HOST, PORT), CallHandler) as server:
print(f"Serving on http://{HOST}:{PORT}")
server.serve_forever()
except KeyboardInterrupt:
unregister_call_status()
if __name__ == "__main__":
main()