Spaces:
Runtime error
Runtime error
Download utils/twilio.py from Smartwaylabs/pipecat-livekit: direct link, hf CLI and curl.
- Browser
- Download file 2.05 kB
-
https://huggingface.co/spaces/Smartwaylabs/pipecat-livekit/resolve/main/utils/twilio.py
- Command line
-
hf download hf://spaces/Smartwaylabs/pipecat-livekit/utils/twilio.py
-
curl -L -o twilio.py https://huggingface.co/spaces/Smartwaylabs/pipecat-livekit/resolve/main/utils/twilio.py
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") | |