python/src/mcp_server/utils/timeout_config.py

"""
Centralized timeout configuration for MCP Server.

Provides consistent timeout values across all tools.
"""

import os

import httpx


def get_default_timeout() -> httpx.Timeout:
    """
    Get default timeout configuration from environment or defaults.

    Environment variables:
    - MCP_REQUEST_TIMEOUT: Total request timeout in seconds (default: 30)
    - MCP_CONNECT_TIMEOUT: Connection timeout in seconds (default: 5)
    - MCP_READ_TIMEOUT: Read timeout in seconds (default: 20)
    - MCP_WRITE_TIMEOUT: Write timeout in seconds (default: 10)

    Returns:
        Configured httpx.Timeout object
    """
    return httpx.Timeout(
        timeout=float(os.getenv("MCP_REQUEST_TIMEOUT", "30.0")),
        connect=float(os.getenv("MCP_CONNECT_TIMEOUT", "5.0")),
        read=float(os.getenv("MCP_READ_TIMEOUT", "20.0")),
        write=float(os.getenv("MCP_WRITE_TIMEOUT", "10.0")),
    )


def get_polling_timeout() -> httpx.Timeout:
    """
    Get timeout configuration for polling operations.

    Polling operations may need longer timeouts.

    Returns:
        Configured httpx.Timeout object for polling
    """
    return httpx.Timeout(
        timeout=float(os.getenv("MCP_POLLING_TIMEOUT", "60.0")),
        connect=float(os.getenv("MCP_CONNECT_TIMEOUT", "5.0")),
        read=float(os.getenv("MCP_POLLING_READ_TIMEOUT", "30.0")),
        write=float(os.getenv("MCP_WRITE_TIMEOUT", "10.0")),
    )


def get_max_polling_attempts() -> int:
    """
    Get maximum number of polling attempts.

    Returns:
        Maximum polling attempts (default: 30)
    """
    try:
        return int(os.getenv("MCP_MAX_POLLING_ATTEMPTS", "30"))
    except ValueError:
        # Fall back to default if env var is not a valid integer
        return 30


def get_polling_interval(attempt: int) -> float:
    """
    Get polling interval with exponential backoff.

    Args:
        attempt: Current attempt number (0-based)

    Returns:
        Sleep interval in seconds
    """
    base_interval = float(os.getenv("MCP_POLLING_BASE_INTERVAL", "1.0"))
    max_interval = float(os.getenv("MCP_POLLING_MAX_INTERVAL", "5.0"))

    # Exponential backoff: 1s, 2s, 4s, 5s, 5s, ...
    interval = min(base_interval * (2**attempt), max_interval)
    return float(interval)