pipecat-livekit / utils /twilio.py
peacepeacecreation
Deploy Pipecat LiveKit server
24cdea4
Raw History Blame Contribute Delete
2.05 kB
"""
Twilio utility functions for call management.
Provides functions for:
- Terminating calls via REST API
- Managing call state
"""
import os
from typing import Optional
from loguru import logger
async def terminate_call(call_sid: str) -> bool:
"""
Terminate a Twilio call via REST API.
Args:
call_sid: The Call SID to terminate
Returns:
True if successful, False otherwise
"""
account_sid = os.environ.get("TWILIO_ACCOUNT_SID")
auth_token = os.environ.get("TWILIO_AUTH_TOKEN")
if not account_sid or not auth_token:
logger.warning("Twilio credentials not configured, cannot terminate call via API")
return False
if not call_sid:
logger.warning("No call_sid provided, cannot terminate call")
return False
try:
from twilio.rest import Client
client = Client(account_sid, auth_token)
client.calls(call_sid).update(status="completed")
logger.info(f"Successfully terminated Twilio call: {call_sid}")
return True
except ImportError:
logger.error("Twilio SDK not installed, cannot terminate call")
return False
except Exception as e:
logger.error(f"Failed to terminate Twilio call {call_sid}: {e}")
return False
# Global storage for call context during active calls
# This allows handlers to access call information
_active_call_context: Optional[dict] = None
def set_call_context(call_sid: str, stream_sid: str) -> None:
"""Store the current call context for handlers to access."""
global _active_call_context
_active_call_context = {
"call_sid": call_sid,
"stream_sid": stream_sid,
}
logger.debug(f"Set call context: call_sid={call_sid}")
def get_call_context() -> Optional[dict]:
"""Get the current call context."""
return _active_call_context
def clear_call_context() -> None:
"""Clear the current call context."""
global _active_call_context
_active_call_context = None
logger.debug("Cleared call context")