"""
Voice API Client
HTTP wrapper for the outbound call API
"""
import json
import urllib.request
import urllib.error
from typing import Optional, Dict, Any
# ============================================================
# CONFIGURATION - Update these for your setup
# ============================================================
API_BASE_URL = "http://YOUR_SERVER:3000"
DEFAULT_CALLER_ID = "9000"
TIMEOUT_SECONDS = 30
# Contact Directory - Map names/aliases to phone numbers
CONTACTS = {
"me": "YOUR_EXTENSION",
"myself": "YOUR_EXTENSION",
# Add your contacts here:
# "wife": "+15551234567",
# "office": "5000",
}
# Device Registry - AI personalities
DEVICES = {
"morpheus": {
"name": "Morpheus",
"extension": "9000",
"description": "Principal AI assistant"
},
"cephanie": {
"name": "Cephanie",
"extension": "9002",
"description": "Storage server (sassy personality)"
},
}
DEFAULT_DEVICE = "Morpheus"
# ============================================================
# EXCEPTIONS
# ============================================================
class VoiceAPIError(Exception):
"""Base exception for Voice API errors"""
def __init__(self, error: str, message: str, data: Optional[Dict] = None):
self.error = error
self.message = message
self.data = data or {}
super().__init__(message)
class ContactNotFoundError(VoiceAPIError):
pass
class ServiceUnavailableError(VoiceAPIError):
pass
class CallFailedError(VoiceAPIError):
pass
class DeviceNotFoundError(VoiceAPIError):
pass
# ============================================================
# CONTACT RESOLUTION
# ============================================================
def resolve_contact(target: str) -> str:
"""Resolve a contact name/alias to a phone number."""
target_lower = target.lower().strip()
# Check contact directory
if target_lower in CONTACTS:
return CONTACTS[target_lower]
# Check if it's already a valid phone number
cleaned = target.replace("-", "").replace(" ", "").replace("(", "").replace(")", "")
# E.164 format: +[digits]
if cleaned.startswith("+") and cleaned[1:].isdigit() and len(cleaned) > 2:
return cleaned
# Extension or dial string
if cleaned.isdigit() and 1 <= len(cleaned) <= 15:
return cleaned
raise ContactNotFoundError(
error="contact_not_found",
message=f"Unknown contact: {target}"
)
def resolve_device(device: Optional[str]) -> str:
"""Resolve device name to canonical form."""
if device is None or device.strip() == "":
return DEFAULT_DEVICE
device_lower = device.lower().strip()
if device_lower in DEVICES:
return DEVICES[device_lower]["name"]
available = ', '.join(d['name'] for d in DEVICES.values())
raise DeviceNotFoundError(
error="device_not_found",
message=f"Unknown device: {device}. Available: {available}"
)
# ============================================================
# MESSAGE SANITIZATION
# ============================================================
def sanitize_message(message: str, max_words: int = 200) -> str:
"""Sanitize a message for TTS delivery."""
import re
# Remove code blocks
message = re.sub(r'```[\s\S]*?```', '[code omitted]', message)
message = re.sub(r'`[^`]+`', '', message)
# Remove URLs
message = re.sub(r'https?://\S+', '[link omitted]', message)
# Remove special characters
message = re.sub(r'[{}\[\]<>|\\^~]', '', message)
# Normalize whitespace
message = ' '.join(message.split())
# Truncate
words = message.split()
if len(words) > max_words:
message = ' '.join(words[:max_words]) + '...'
return message.strip()
# ============================================================
# API CALLS
# ============================================================
def initiate_call(to: str, message: str, caller_id: str = DEFAULT_CALLER_ID,
mode: str = "announce", device: Optional[str] = None) -> Dict[str, Any]:
"""Initiate an outbound call via the Voice API."""
url = f"{API_BASE_URL}/api/outbound-call"
payload = {
"to": to,
"message": sanitize_message(message),
"callerId": caller_id,
"mode": mode,
"device": resolve_device(device)
}
try:
request_body = json.dumps(payload).encode('utf-8')
req = urllib.request.Request(
url,
data=request_body,
headers={'Content-Type': 'application/json'},
method='POST'
)
with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as response:
result = json.loads(response.read().decode('utf-8'))
if result.get('success'):
return {
"callId": result.get('callId'),
"to": to,
"device": payload['device'],
"mode": mode,
"status": result.get('status', 'initiated'),
}
else:
raise CallFailedError(
error=result.get('error', 'call_failed'),
message=result.get('message', 'Call initiation failed')
)
except urllib.error.URLError as e:
raise ServiceUnavailableError(
error="service_unavailable",
message=f"Voice server not reachable: {str(e.reason)}"
)
def get_call_status(call_id: str) -> Dict[str, Any]:
"""Get the status of an existing call."""
url = f"{API_BASE_URL}/api/call/{call_id}"
req = urllib.request.Request(url, method='GET')
with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as response:
return json.loads(response.read().decode('utf-8'))
def list_calls() -> Dict[str, Any]:
"""List all active calls."""
url = f"{API_BASE_URL}/api/calls"
req = urllib.request.Request(url, method='GET')
with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as response:
return json.loads(response.read().decode('utf-8'))