Real-Time FreeSWITCH Event Handling (ESL) Without Third-Party Packages

FreeSWITCH is undeniably one of the most robust and flexible telephony engines available today. But to truly unlock its potential, you need to control it programmatically.

While many tutorials rely on third-party libraries (like modesl for Node.js or ESL-python) to interface with FreeSWITCH, there is a cleaner, more lightweight, and dependency-free approach: building your own Event Socket Layer (ESL) client.

In this guide, you’ll learn how to connect FreeSWITCH with pure Python and Node.js implementations using standard TCP sockets. We will listen to ESL events in real-time, print them out, and even show you how to execute operations—all without installing bloated libraries!

---

🔹 What is ESL in FreeSWITCH?

ESL (Event Socket Layer) allows external applications to interact directly with FreeSWITCH.

Using ESL, you can:

• Listen to FreeSWITCH events in real-time (calls starting, answering, hanging up).
• Control live calls dynamically (e.g., streaming audio, bridging).
• Execute FreeSWITCH API commands programmatically.

When you build your own custom ESL wrapper, you remove the abstraction layer. This gives you absolute control, lower latency, and zero dependency issues when deploying to production.

---

✅ What We’ll Build

• Configure FreeSWITCH to accept external ESL connections.
• Create native TCP clients in both Python and Node.js to connect.
• Authenticate securely.
• Capture and parse FreeSWITCH events in JSON format.
• Subscribing to specific custom events.
• Send API commands directly to the engine to originate, answer, and hangup calls.

---

⚙️ Step 1: Enable ESL in FreeSWITCH

🔓 Allow ESL Connections from External Sources

By default, FreeSWITCH listens for ESL connections only on 127.0.0.1 (localhost). To allow an external backend to connect, we need to adjust the listen IP.

Open your FreeSWITCH Event Socket configuration file: nano /etc/freeswitch/autoload_configs/event_socket.conf.xml

Ensure the configuration looks something like this (change listen-ip to 0.0.0.0 if your script is hosted on another server, but secure it with a firewall!):

<configuration name="event_socket.conf" description="Event Socket">
  <settings>
    <param name="listen-ip" value="127.0.0.1"/>
    <param name="listen-port" value="8021"/>
    <param name="password" value="ClueCon"/>
  </settings>
</configuration>

🔄 Reload the Event Socket Module Apply the changes without restarting FreeSWITCH by running: fs_cli -x "reload mod_event_socket"

---

💻 Step 2: The Core Concept of an ESL Packet

Before writing code, let's understand how FreeSWITCH talks. It uses a simple text protocol over TCP. An ESL message typically consists of:

1. Headers: Key-value pairs (like Content-Type: and Content-Length:), separated by newlines.
2. Double Newline (\n\n): Marks the end of the headers.
3. Body: The actual payload (e.g., JSON event data), depending on the Content-Length.

---

🐍 Step 3.A: Native Python ESL Client

Here is a pure-Python script that uses standard libraries (socket, json) to handle ESL communication.

import socket
import json
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("esl_custom")

# FreeSWITCH ESL Details
FS_HOST = '127.0.0.1'
FS_PORT = 8021
FS_PASSWORD = 'ClueCon'

class PureESLClient:
    def __init__(self, host, port, password):
        self.host = host
        self.port = port
        self.password = password
        self.sock = None
        self._buffer = b""

    def connect(self):
        """Establish TCP connection and authenticate."""
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((self.host, self.port))
        logger.info(f"Connected to FreeSWITCH ESL at {self.host}:{self.port}")

        # Wait for the initial auth/request header
        self._read_packet() 
        
        # Authenticate
        self._send(f"auth {self.password}")
        auth_response = self._read_packet()
        if "+OK" not in auth_response.get("headers", {}).get("Reply-Text", ""):
            raise ConnectionError("Authentication failed!")
            
        logger.info("Authentication Successful! 🚀")

    def _send(self, command):
        """Helper to send a command via socket."""
        self.sock.sendall(f"{command}\\n\\n".encode("utf-8"))

    def _read_packet(self):
        """Reads headers and optional body based on Content-Length."""
        while b"\\n\\n" not in self._buffer:
            chunk = self.sock.recv(4096)
            if not chunk:
                raise ConnectionError("Socket closed by FreeSWITCH")
            self._buffer += chunk

        # Split headers from body
        header_end = self._buffer.index(b"\\n\\n")
        header_block = self._buffer[:header_end].decode("utf-8", errors="replace")
        self._buffer = self._buffer[header_end + 2:]

        # Parse Headers
        headers = {}
        for line in header_block.split("\\n"):
            if ": " in line:
                key, value = line.split(": ", 1)
                headers[key.strip()] = value.strip()

        result = {"headers": headers}

        # Parse Body if Content-Length exists
        content_length = int(headers.get("Content-Length", 0))
        if content_length > 0:
            while len(self._buffer) < content_length:
                chunk = self.sock.recv(4096)
                self._buffer += chunk

            body = self._buffer[:content_length].decode("utf-8", errors="replace")
            self._buffer = self._buffer[content_length:]
            result["body"] = body

        return result

    def listen(self):
        """Continuous loop to receive and process events."""
        try:
            while True:
                packet = self._read_packet()
                content_type = packet.get("headers", {}).get("Content-Type", "")

                if content_type == "text/event-json":
                    body = packet.get("body", "{}")
                    try:
                        event = json.loads(body)
                        event_name = event.get("Event-Name", "UNKNOWN")
                        call_uuid = event.get("Unique-ID", "N/A")
                        print(f"📞 Event: {event_name} | UUID: {call_uuid}")
                    except json.JSONDecodeError:
                        pass
        except KeyboardInterrupt:
            logger.info("Shutting down...")
        finally:
            self.sock.close()

---

🟢 Step 3.B: Native Node.js ESL Client

For Node.js, you don't need modesl. The built-in net module provides exactly what you need to create a robust socket:

const net = require('net');

// FreeSWITCH ESL Details
const FS_HOST = '127.0.0.1';
const FS_PORT = 8021;
const FS_PASSWORD = 'ClueCon';

const client = new net.Socket();
let buffer = '';

client.connect(FS_PORT, FS_HOST, () => {
    console.log(`Connected to FreeSWITCH ESL at ${FS_HOST}:${FS_PORT}`);
});

client.on('data', (data) => {
    buffer += data.toString('utf8');

    // Simple continuous parser for ESL packets
    while (buffer.includes('\\n\\n')) {
        const headerEnd = buffer.indexOf('\\n\\n');
        const headerBlock = buffer.substring(0, headerEnd);
        
        let contentLength = 0;
        const headers = {};
        
        // Parse headers
        headerBlock.split('\\n').forEach(line => {
            if (line.includes(': ')) {
                const [key, val] = line.split(': ');
                headers[key.trim()] = val.trim();
                if (key.trim() === 'Content-Length') {
                    contentLength = parseInt(val.trim(), 10);
                }
            }
        });

        // Wait until we've received the entire body
        const totalLength = headerEnd + 2 + contentLength;
        if (buffer.length < totalLength) {
            break; 
        }

        const body = buffer.substring(headerEnd + 2, totalLength);
        buffer = buffer.substring(totalLength); // Remove from buffer

        // Route actions based on packet type
        if (headers['Content-Type'] === 'auth/request') {
            client.write(`auth ${FS_PASSWORD}\\n\\n`);
        } else if (headers['Command-Reply'] && headers['Reply-Text']?.includes('+OK accepted')) {
            console.log("Authentication Successful! 🚀");
        } else if (headers['Content-Type'] === 'text/event-json') {
            try {
                const event = JSON.parse(body);
                console.log(`📞 Event: ${event['Event-Name']} | UUID: ${event['Unique-ID'] || 'N/A'}`);
            } catch (e) {
                console.warn("Could not parse JSON body");
            }
        }
    }
});

---

📡 Step 4: Beyond Connection – Real Communication

Connecting is only half the battle. Now that we have our TCP socket established, we need to instruct FreeSWITCH on what to do. The FreeSWITCH ESL provides two main ways to communicate: Commands and API Calls.

1. Subscribing to Specific Events While you could use event json all to subscribe to everything, in a production environment, you typically only want to listen to specific events to avoid cluttering your network traffic and buffer.

You can pass a space-separated list of events. If you have custom modules (like mod_audio_pipe), you can listen to them using the CUSTOM keyword!

Python Example:

def subscribe_events(self):
    # Subscribe to Answer, Hangup, and a custom module event
    events = "CHANNEL_ANSWER CHANNEL_HANGUP CUSTOM mod_audio_pipe::connect"
    self._send(f"event json {events}")
    self._read_packet() # Consume the reply-text header
    print("✅ Subscribed to targeted events.")

Node.js Example:

// Once Authenticated Successfully:
client.write("event json CHANNEL_ANSWER CHANNEL_HANGUP CUSTOM mod_audio_pipe::connect\\n\\n");

2. Executing API Commands (Originate, Answer, Hangup) To execute built-in FreeSWITCH API commands programmatically via ESL, prefix your operation with api . The response from FreeSWITCH will return the result of the command execution in the body of the subsequent ESL packet.

Here's how you can wrap common functionalities in your custom client:

Python Action Examples:

def send_api_command(self, command):
    """Sends an API command and reads the immediate response."""
    self.sock.sendall(f"api {command}\\n\\n".encode("utf-8"))
    
    # Wait for the API response packet
    response = self._read_packet()
    return response.get("body", "").strip()

# Examples of usage in application logic:

# 1. Originate a new outbound call to an extension
client.send_api_command("originate sofia/internal/1000@192.168.1.10 &echo")

# 2. Answer an incoming call dynamically based on UUID
client.send_api_command("uuid_answer 1234-abcd-5678")

# 3. Set Multiple Channel Variables (Crucial for routing)
client.send_api_command("uuid_setvar_multi 1234-abcd-5678 PIPE_BIDIRECTIONAL=true;is_bot_call=true")

# 4. Attach an Audio Pipe for AI voicebot integrations
client.send_api_command("uuid_audio_pipe 1234-abcd-5678 start wss://my-ai-bot.local")

# 5. Terminate / Hang up a call with a specific cause code
client.send_api_command("uuid_kill 1234-abcd-5678 NORMAL_CLEARING")

Node.js Action Examples:

function sendApiCommand(client, command) {
    // Fire and forget, or modify your listener to catch 'Content-Type: api/response'
    client.write(`api ${command}\\n\\n`);
}

// 1. Originate a call
sendApiCommand(client, 'originate sofia/internal/1000@192.168.1.10 &echo');

// 2. Answer a call
sendApiCommand(client, 'uuid_answer 1234-abcd-5678');

// 3. Hang up a call 
sendApiCommand(client, 'uuid_kill 1234-abcd-5678 NORMAL_CLEARING');

By adding an api prefix alongside your native socket writes, your script transforms from a simple passive event listener into a reactive dialplan controller capable of routing and managing live media sessions dynamically!

---

💡 Why Build Your Own Wrapper?

1. Zero Dependencies: You aren't tied to the maintenance lifecycle of npm packages (modesl) or external PIP libraries. It uses built-in features (net in Node.js and socket in Python).
2. Speed & Efficiency: By handling the buffer directly, you avoid the overhead of heavy wrappers (we've actually built our own lightweight, custom TCP client for our product's microservices for this exact reason!).
3. Total Customization: You can map custom variables, handle specific payloads like mod_audio_pipe events natively, and seamlessly integrate ESL directly into your preferred web frameworks (FastAPI, Express, NestJS, Django) or webhook loops.

🔚 Final Thoughts

FreeSWITCH API + Native Sockets = Maximum Performance and Control 🚀

By speaking the ESL text protocol directly, you gain a massive advantage in understanding how FreeSWITCH behaves under the hood. You can expand this core logic to build everything from dynamic IVRs to AI voice-bot integrations!