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_analysis | Indicates a new ai analysis has been initiated | After transcription completes |
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)3. Processing call_analysis Events
The call_analysis event is sent when AI analysis is initiated for a call. This event is sent after transcription completes and tracks the analysis generation process.
Event Flow
The call_analysis event can be sent with three different status values:
started: Sent after transcription completes, before summary generation. Bothsummaryandkeypointswill be empty (summary: "",keypoints: []).done: Sent when analysis is complete with status "Completed". Includes actualsummaryandkeypointsif available.failed: Sent if analysis fails. Bothsummaryandkeypointswill be empty (summary: "",keypoints: []).
Payload Example
{
"eventName": "call_analysis",
"timestamp": "2025-04-14T15:04:23.161Z",
"data": {
"direction": "inbound",
"dn": "101",
"analysis_link": "https://api.jasconnect.com/v2/api/analysis/123",
"status": "done",
"recording_url": "101/[John Doe]_101-5148888888_20250603153720(26).wav",
"call_id": "26",
"start_time": "2025-04-14T15:00:00.000Z",
"end_time": "2025-04-14T15:05:00.000Z",
"summary": "Customer called to inquire about product availability. Discussed pricing and delivery options.",
"keypoints": [
"Product inquiry",
"Pricing discussion",
"Delivery timeline"
]
}
}Payload Structure
| Field | Type | Description |
|---|---|---|
direction | string | Call direction: "inbound" or "outbound" |
dn | string | Directory number |
analysis_link | string | Link to call analysis |
status | string | Analysis status: "started", "done", or "failed" |
recording_url | string | URL path to the call recording |
call_id | string | Unique identifier for the call |
start_time | string | ISO 8601 timestamp when the call began |
end_time | string | ISO 8601 timestamp when the call ended |
summary | string | Call summary text (always present, empty string if not available) |
keypoints | string[] | Array of key points from the call (always present, empty array if not available) |
Important: Both summary and keypoints are always present in the payload, even if empty. When status is "started" or "failed", they will be empty (summary: "", keypoints: []).
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()