#!/usr/bin/env python3 """ Genesis Telnyx Manager Skill ============================= Skill: Telnyx Connectivity Engine Manages SIP Trunking and Call Control for the Genesis Voice Agency. PM-036: Telnyx Voice Skill Enhancement - Supports: outbound calls, IVR, recording - Logs call metadata - Complete voice integration """ import os import json import time import logging import requests from typing import Dict, Any, List, Optional from datetime import datetime logger = logging.getLogger("TelnyxManagerSkill") logging.basicConfig(level=logging.INFO) class TelnyxManagerSkill: """ Skill: Telnyx Connectivity Engine. Manages SIP Trunking and Call Control for the Genesis Voice Agency. """ def __init__(self, api_key: Optional[str] = None): self.api_key = api_key or os.getenv("TELNYX_API_KEY") self.base_url = "https://api.telnyx.com/v2" self.headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } # Call metadata storage self.call_logs: List[Dict[str, Any]] = [] logger.info(f"Telnyx Manager Skill Initialized (API Key: {'Configured' if self.api_key else 'MISSING'})") def _make_request(self, method: str, endpoint: str, data: Dict = None, params: Dict = None) -> Dict[str, Any]: """Make API request with error handling.""" url = f"{self.base_url}{endpoint}" try: if method.upper() == "GET": response = requests.get(url, headers=self.headers, params=params, timeout=30) elif method.upper() == "POST": response = requests.post(url, headers=self.headers, json=data, timeout=30) elif method.upper() == "DELETE": response = requests.delete(url, headers=self.headers, timeout=30) else: return {"error": f"Unsupported method: {method}"} if response.status_code in [200, 201]: return response.json() if response.text else {"status": "success"} elif response.status_code == 204: return {"status": "success", "message": "No content"} else: logger.error(f"API Error {response.status_code}: {response.text}") return {"error": response.text, "status_code": response.status_code} except requests.exceptions.Timeout: return {"error": "timeout", "status_code": 408} except Exception as e: logger.error(f"Request failed: {e}") return {"error": str(e)} def _log_call_metadata(self, call_data: Dict[str, Any]) -> None: """Log call metadata for tracking and analytics.""" metadata = { "timestamp": datetime.utcnow().isoformat(), "call_control_id": call_data.get("call_control_id"), "call_leg_id": call_data.get("call_leg_id"), "from_number": call_data.get("from"), "to_number": call_data.get("to"), "direction": call_data.get("direction"), "state": call_data.get("state"), "record_type": call_data.get("record_type") } self.call_logs.append(metadata) logger.info(f"Call logged: {metadata.get('call_control_id')}") # ===== SIP TRUNK OPERATIONS ===== def provision_vapi_sip_trunk(self, name: str) -> Dict[str, Any]: """ Automates the creation of a SIP trunk configured for Vapi.ai. Ref: sip.vapi.ai (Port 5060) """ logger.info(f"Provisioning Telnyx SIP Trunk for Vapi: {name}") # Create credential for authentication cred_payload = { "connection_name": f"{name}_cred", "sip_uri": "sip.vapi.ai" } # Create outbound voice profile profile_payload = { "name": name, "traffic_type": "conversational", "service_plan": "global", "usage_payment_method": "rate-deck", "whitelisted_destinations": ["US", "AU", "GB"], "concurrent_call_limit": 10 } result = self._make_request("POST", "/outbound_voice_profiles", data=profile_payload) if "data" in result: logger.info(f"SIP Trunk created: {result['data'].get('id')}") return result def list_sip_connections(self) -> List[Dict[str, Any]]: """Lists all SIP connections.""" result = self._make_request("GET", "/sip_connections") return result.get("data", []) # ===== OUTBOUND CALL OPERATIONS ===== def initiate_outbound_call(self, from_number: str, to_number: str, connection_id: str, webhook_url: str = None, answering_machine_detection: bool = False) -> Dict[str, Any]: """ Initiates an outbound call. Args: from_number: Caller ID (must be owned/verified) to_number: Destination number (E.164 format) connection_id: Telnyx connection ID webhook_url: URL for call events answering_machine_detection: Enable AMD Returns: Dict with call control ID """ logger.info(f"Initiating outbound call: {from_number} -> {to_number}") payload = { "from": from_number, "to": to_number, "connection_id": connection_id, "answering_machine_detection": "detect" if answering_machine_detection else "disabled" } if webhook_url: payload["webhook_url"] = webhook_url result = self._make_request("POST", "/calls", data=payload) if "data" in result: self._log_call_metadata(result["data"]) return result def dial_number(self, to_number: str, from_display: str = None) -> Dict[str, Any]: """ Simplified dial operation using default connection. Args: to_number: Number to call from_display: Optional caller display name Returns: Call result """ # Get default connection connections = self.list_sip_connections() if not connections: return {"error": "No SIP connections available"} connection_id = connections[0].get("id") from_number = os.getenv("TELNYX_FROM_NUMBER", "+15551234567") return self.initiate_outbound_call(from_number, to_number, connection_id) # ===== CALL CONTROL OPERATIONS ===== def execute_call_command(self, call_control_id: str, command: str, payload: Dict[str, Any] = None) -> Dict[str, Any]: """ Sends a Call Control command. Args: call_control_id: The call to control command: Command name (answer, hangup, bridge, transfer, etc.) payload: Command-specific parameters Returns: Command result """ logger.info(f"Executing Telnyx Command [{command}] on Call {call_control_id}") endpoint = f"/calls/{call_control_id}/actions/{command}" return self._make_request("POST", endpoint, data=payload or {}) def answer_call(self, call_control_id: str, webhook_url: str = None) -> Dict[str, Any]: """Answers an incoming call.""" payload = {} if webhook_url: payload["webhook_url"] = webhook_url return self.execute_call_command(call_control_id, "answer", payload) def hangup_call(self, call_control_id: str) -> Dict[str, Any]: """Ends a call.""" return self.execute_call_command(call_control_id, "hangup") def transfer_call(self, call_control_id: str, to_number: str, from_display: str = None) -> Dict[str, Any]: """Transfers a call to another number.""" payload = {"to": to_number} if from_display: payload["from_display_name"] = from_display return self.execute_call_command(call_control_id, "transfer", payload) def bridge_calls(self, call_control_id: str, target_call_control_id: str) -> Dict[str, Any]: """Bridges two calls together.""" payload = {"call_control_id": target_call_control_id} return self.execute_call_command(call_control_id, "bridge", payload) # ===== IVR OPERATIONS ===== def speak_text(self, call_control_id: str, text: str, voice: str = "female", language: str = "en-US") -> Dict[str, Any]: """ Speaks text on the call using TTS. Args: call_control_id: Target call text: Text to speak voice: Voice type (male/female) language: Language code Returns: Command result """ logger.info(f"Speaking text on call {call_control_id}") payload = { "payload": text, "voice": voice, "language": language } return self.execute_call_command(call_control_id, "speak", payload) def play_audio(self, call_control_id: str, audio_url: str, loop: int = 1) -> Dict[str, Any]: """Plays an audio file on the call.""" payload = { "audio_url": audio_url, "loop": loop } return self.execute_call_command(call_control_id, "playback_start", payload) def stop_playback(self, call_control_id: str) -> Dict[str, Any]: """Stops audio playback.""" return self.execute_call_command(call_control_id, "playback_stop") def gather_dtmf(self, call_control_id: str, prompt_text: str = None, max_digits: int = 10, timeout_secs: int = 30, terminating_digit: str = "#") -> Dict[str, Any]: """ Gathers DTMF input from the caller. Args: call_control_id: Target call prompt_text: Optional TTS prompt max_digits: Maximum digits to collect timeout_secs: Timeout in seconds terminating_digit: Digit that ends collection Returns: Command result """ payload = { "maximum_digits": max_digits, "timeout_millis": timeout_secs * 1000, "terminating_digit": terminating_digit, "valid_digits": "0123456789*#" } # Optionally speak a prompt first if prompt_text: self.speak_text(call_control_id, prompt_text) time.sleep(1) # Wait for TTS to start return self.execute_call_command(call_control_id, "gather", payload) def send_dtmf(self, call_control_id: str, digits: str) -> Dict[str, Any]: """Sends DTMF tones on the call.""" payload = {"digits": digits} return self.execute_call_command(call_control_id, "send_dtmf", payload) # ===== RECORDING OPERATIONS ===== def start_recording(self, call_control_id: str, channels: str = "dual", format: str = "mp3") -> Dict[str, Any]: """ Starts call recording. Args: call_control_id: Target call channels: Recording channels (single/dual) format: Audio format (mp3/wav) Returns: Recording result with recording ID """ logger.info(f"Starting recording on call {call_control_id}") payload = { "channels": channels, "format": format } return self.execute_call_command(call_control_id, "record_start", payload) def stop_recording(self, call_control_id: str) -> Dict[str, Any]: """Stops call recording.""" logger.info(f"Stopping recording on call {call_control_id}") return self.execute_call_command(call_control_id, "record_stop") def pause_recording(self, call_control_id: str) -> Dict[str, Any]: """Pauses call recording.""" return self.execute_call_command(call_control_id, "record_pause") def resume_recording(self, call_control_id: str) -> Dict[str, Any]: """Resumes paused recording.""" return self.execute_call_command(call_control_id, "record_resume") def get_recordings(self, call_control_id: str = None) -> List[Dict[str, Any]]: """Lists recordings, optionally filtered by call.""" params = {} if call_control_id: params["filter[call_control_id]"] = call_control_id result = self._make_request("GET", "/recordings", params=params) return result.get("data", []) # ===== MEDIA FORKING OPERATIONS ===== def toggle_media_fork(self, call_control_id: str, rx_url: str, enable: bool = True) -> Dict[str, Any]: """ Enables/Disables media forking for real-time auditing (Genesis Gate 3). Args: call_control_id: Target call rx_url: WebSocket URL to receive forked media enable: True to start, False to stop Returns: Command result """ if enable: payload = { "rx": rx_url, "tx": rx_url, # Fork both directions "stream_type": "raw" } return self.execute_call_command(call_control_id, "fork_start", payload) else: return self.execute_call_command(call_control_id, "fork_stop") # ===== MESSAGING OPERATIONS ===== def send_sms(self, from_number: str, to_number: str, text: str, messaging_profile_id: str = None) -> Dict[str, Any]: """Sends an SMS message.""" logger.info(f"Sending SMS: {from_number} -> {to_number}") payload = { "from": from_number, "to": to_number, "text": text } if messaging_profile_id: payload["messaging_profile_id"] = messaging_profile_id return self._make_request("POST", "/messages", data=payload) # ===== PHONE NUMBER OPERATIONS ===== def list_phone_numbers(self) -> List[Dict[str, Any]]: """Lists all phone numbers in the account.""" result = self._make_request("GET", "/phone_numbers") return result.get("data", []) def search_available_numbers(self, country_code: str = "US", features: List[str] = None) -> List[Dict[str, Any]]: """Searches for available phone numbers.""" params = {"filter[country_code]": country_code} if features: params["filter[features]"] = ",".join(features) result = self._make_request("GET", "/available_phone_numbers", params=params) return result.get("data", []) def order_phone_number(self, phone_number: str, connection_id: str) -> Dict[str, Any]: """Orders a phone number.""" payload = { "phone_numbers": [{"phone_number": phone_number}], "connection_id": connection_id } return self._make_request("POST", "/number_orders", data=payload) # ===== UTILITY METHODS ===== def execute(self, action: str, **kwargs) -> Dict[str, Any]: """ Generic action executor for skill interface compatibility. Args: action: Action name **kwargs: Action-specific parameters Returns: Action result """ action_map = { "dial": self.dial_number, "outbound_call": self.initiate_outbound_call, "answer": self.answer_call, "hangup": self.hangup_call, "transfer": self.transfer_call, "bridge": self.bridge_calls, "speak": self.speak_text, "play_audio": self.play_audio, "gather_dtmf": self.gather_dtmf, "send_dtmf": self.send_dtmf, "start_recording": self.start_recording, "stop_recording": self.stop_recording, "media_fork": self.toggle_media_fork, "send_sms": self.send_sms, "list_numbers": self.list_phone_numbers, "provision_sip": self.provision_vapi_sip_trunk } if action not in action_map: return {"error": f"Unknown action: {action}", "available": list(action_map.keys())} try: return action_map[action](**kwargs) except TypeError as e: return {"error": f"Invalid parameters for {action}: {e}"} except Exception as e: return {"error": str(e)} def get_call_logs(self, limit: int = 100) -> List[Dict[str, Any]]: """Returns recent call metadata logs.""" return self.call_logs[-limit:] def get_call_status(self, call_control_id: str) -> Dict[str, Any]: """Gets the current status of a call.""" return self._make_request("GET", f"/calls/{call_control_id}") if __name__ == "__main__": # Test Logic manager = TelnyxManagerSkill() if manager.api_key: # Test SIP trunk provisioning result = manager.execute("provision_sip", name="Genesis_Vapi_Bridge") print(f"[TEST] SIP Trunk: {result}") # Test media fork result = manager.execute("media_fork", call_control_id="call_123_xyz", rx_url="wss://genesis-audit.internal/stream", enable=True) print(f"[TEST] Media Fork: {result}") # Show call logs print(f"[TEST] Call Logs: {manager.get_call_logs()}") else: print("[TEST] Skipping API calls (Key missing)")