Spaces:
Runtime error
Runtime error
| """ | |
| Session utilities for Pipecat server. | |
| Provides shared session management logic used across server and runner. | |
| """ | |
| import os | |
| from typing import Any, Dict, Optional | |
| from loguru import logger | |
| DEFAULT_MAX_SESSION_DURATION = 3540 # 59 minutes | |
| def get_max_session_duration( | |
| bot_data: Optional[Dict[str, Any]] = None, | |
| env_var: str = "MAX_SESSION_DURATION_SECONDS", | |
| ) -> int: | |
| """ | |
| Get max session duration with priority: API > ENV > Default. | |
| Priority: | |
| 1. API: bot_data.get("max_session_duration") or bot_data.metadata.max_duration | |
| 2. ENV: MAX_SESSION_DURATION_SECONDS environment variable | |
| 3. Default: DEFAULT_MAX_SESSION_DURATION (3540 seconds = 59 minutes) | |
| Args: | |
| bot_data: Bot data dictionary or None | |
| env_var: Name of environment variable to check | |
| Returns: | |
| Max session duration in seconds | |
| """ | |
| # Priority 1: Check if provided in API request (dict format) | |
| if bot_data: | |
| # Check direct max_session_duration field | |
| if "max_session_duration" in bot_data: | |
| duration = bot_data["max_session_duration"] | |
| if duration and duration > 0: | |
| logger.debug(f"Using max session duration from API: {duration}s") | |
| return duration | |
| # Check nested metadata.max_duration field | |
| metadata = bot_data.get("metadata", {}) | |
| if metadata and "max_duration" in metadata: | |
| duration = metadata["max_duration"] | |
| if duration and duration > 0: | |
| logger.debug(f"Using max session duration from metadata: {duration}s") | |
| return duration | |
| # Priority 2: Check environment variable | |
| env_duration = os.getenv(env_var) | |
| if env_duration: | |
| try: | |
| duration = int(env_duration) | |
| if duration > 0: | |
| logger.debug(f"Using max session duration from ENV: {duration}s") | |
| return duration | |
| except ValueError: | |
| logger.warning(f"Invalid {env_var} in ENV: {env_duration}, using default") | |
| # Priority 3: Use default | |
| logger.debug(f"Using default max session duration: {DEFAULT_MAX_SESSION_DURATION}s") | |
| return DEFAULT_MAX_SESSION_DURATION | |