vllm.connections

Classes:

• HTTPConnection –

  Helper class to send HTTP requests.

• HTTPResponseSizeExceededError –

  Raised when an HTTP response exceeds a caller-supplied byte ceiling.

• MediaDownloadSizeExceededError –

  Raised when a remote media download exceeds the configured byte limit.

• MediaRequestDownloadBudget –

  Thread-safe total of HTTP media bytes received for one request.

• MediaRequestDownloadSizeExceededError –

  Raised when all HTTP media downloads in one request exceed the limit.

Attributes:

• global_http_connection –

  The global HTTPConnection instance used

global_http_connection = HTTPConnection() module-attribute

The global HTTPConnection instance used by vLLM.

HTTPConnection

Helper class to send HTTP requests.

Source code in vllm/connections.py

class HTTPConnection:
    """Helper class to send HTTP requests."""

    def __init__(self, *, reuse_client: bool = True) -> None:
        super().__init__()

        self.reuse_client = reuse_client

        self._sync_client: requests.Session | None = None
        self._async_client: aiohttp.ClientSession | None = None

    def get_sync_client(self) -> requests.Session:
        if self._sync_client is None or not self.reuse_client:
            self._sync_client = requests.Session()

        return self._sync_client

    # NOTE: We intentionally use an async function even though it is not
    # required, so that the client is only accessible inside async event loop
    async def get_async_client(self) -> aiohttp.ClientSession:
        if self._async_client is None or not self.reuse_client:
            self._async_client = aiohttp.ClientSession(trust_env=True)

        return self._async_client

    def _validate_http_url(self, url: str):
        parsed_url = parse_url(url)

        if parsed_url.scheme not in ("http", "https"):
            raise ValueError(
                "Invalid HTTP URL: A valid HTTP URL must have scheme 'http' or 'https'."
            )

    def _headers(self, **extras: str) -> MutableMapping[str, str]:
        return {"User-Agent": f"vLLM/{VLLM_VERSION}", **extras}

    def get_response(
        self,
        url: str,
        *,
        stream: bool = False,
        timeout: float | None = None,
        extra_headers: Mapping[str, str] | None = None,
        allow_redirects: bool = True,
    ):
        self._validate_http_url(url)

        client = self.get_sync_client()
        extra_headers = extra_headers or {}

        return client.get(
            url,
            headers=self._headers(**extra_headers),
            stream=stream,
            timeout=timeout,
            allow_redirects=allow_redirects,
        )

    async def get_async_response(
        self,
        url: str,
        *,
        timeout: float | None = None,
        extra_headers: Mapping[str, str] | None = None,
        allow_redirects: bool = True,
    ):
        self._validate_http_url(url)

        client = await self.get_async_client()
        extra_headers = extra_headers or {}

        return client.get(
            url,
            headers=self._headers(**extra_headers),
            timeout=timeout,
            allow_redirects=allow_redirects,
        )

    @_sync_retry
    def get_bytes(
        self,
        url: str,
        *,
        timeout: float | None = None,
        allow_redirects: bool = True,
        max_bytes: int | None = None,
        request_budget: MediaRequestDownloadBudget | None = None,
    ) -> bytes:
        with self.get_response(
            url,
            stream=True,
            timeout=timeout,
            allow_redirects=allow_redirects,
        ) as r:
            r.raise_for_status()

            return _read_response_bytes(r, max_bytes, request_budget)

    @_async_retry
    async def async_get_bytes(
        self,
        url: str,
        *,
        timeout: float | None = None,
        allow_redirects: bool = True,
        max_bytes: int | None = None,
        request_budget: MediaRequestDownloadBudget | None = None,
    ) -> bytes:
        async with await self.get_async_response(
            url,
            timeout=timeout,
            allow_redirects=allow_redirects,
        ) as r:
            r.raise_for_status()

            return await _async_read_response_bytes(r, max_bytes, request_budget)

    def get_text(self, url: str, *, timeout: float | None = None) -> str:
        with self.get_response(url, timeout=timeout) as r:
            r.raise_for_status()

            return r.text

    async def async_get_text(
        self,
        url: str,
        *,
        timeout: float | None = None,
    ) -> str:
        async with await self.get_async_response(url, timeout=timeout) as r:
            r.raise_for_status()

            return await r.text()

    def get_json(self, url: str, *, timeout: float | None = None) -> str:
        with self.get_response(url, timeout=timeout) as r:
            r.raise_for_status()

            return r.json()

    async def async_get_json(
        self,
        url: str,
        *,
        timeout: float | None = None,
    ) -> str:
        async with await self.get_async_response(url, timeout=timeout) as r:
            r.raise_for_status()

            return await r.json()

    @_sync_retry
    def download_file(
        self,
        url: str,
        save_path: Path,
        *,
        timeout: float | None = None,
        chunk_size: int = 128,
    ) -> Path:
        try:
            with self.get_response(url, timeout=timeout) as r:
                r.raise_for_status()

                with save_path.open("wb") as f:
                    for chunk in r.iter_content(chunk_size):
                        f.write(chunk)

            return save_path
        except Exception:
            # Clean up partial downloads before retrying or propagating
            if save_path.exists():
                save_path.unlink()
            raise

    @_async_retry
    async def async_download_file(
        self,
        url: str,
        save_path: Path,
        *,
        timeout: float | None = None,
        chunk_size: int = 128,
    ) -> Path:
        try:
            async with await self.get_async_response(
                url,
                timeout=timeout,
            ) as r:
                r.raise_for_status()

                with save_path.open("wb") as f:
                    async for chunk in r.content.iter_chunked(chunk_size):
                        f.write(chunk)

            return save_path
        except Exception:
            # Clean up partial downloads before retrying or propagating
            if save_path.exists():
                save_path.unlink()
            raise

HTTPResponseSizeExceededError

Bases: VLLMValidationError

Raised when an HTTP response exceeds a caller-supplied byte ceiling.

Source code in vllm/connections.py

class HTTPResponseSizeExceededError(VLLMValidationError):
    """Raised when an HTTP response exceeds a caller-supplied byte ceiling."""

    def __init__(self, max_bytes: int, received_bytes: int) -> None:
        self.max_bytes = max_bytes
        self.received_bytes = received_bytes
        super().__init__(
            "Maximum file size exceeded: HTTP response exceeded maximum size "
            f"of {max_bytes} bytes (received at least {received_bytes} bytes)",
            parameter="audio_filesize_mb",
            value=received_bytes / MiB_bytes,
        )

MediaDownloadSizeExceededError

Bases: ValueError

Raised when a remote media download exceeds the configured byte limit.

Source code in vllm/connections.py

class MediaDownloadSizeExceededError(ValueError):
    """Raised when a remote media download exceeds the configured byte limit."""

    def __init__(self, max_bytes: int, received_bytes: int | None = None) -> None:
        self.max_bytes = max_bytes
        self.received_bytes = received_bytes
        max_size_mb = max_bytes / MiB_bytes
        message = (
            "Remote media download exceeds "
            f"VLLM_MAX_MEDIA_DOWNLOAD_SIZE_MB={max_size_mb:g}"
        )
        if received_bytes is not None:
            message += f" (received at least {received_bytes} bytes)"
        super().__init__(message)

MediaRequestDownloadBudget

Thread-safe total of HTTP media bytes received for one request.

Source code in vllm/connections.py

class MediaRequestDownloadBudget:
    """Thread-safe total of HTTP media bytes received for one request."""

    def __init__(self) -> None:
        self.used = 0
        self._lock = threading.Lock()

    def charge(self, size: int) -> None:
        limit_mb = envs.VLLM_MAX_MEDIA_REQUEST_DOWNLOAD_SIZE_MB
        if limit_mb <= 0:
            return
        limit = limit_mb * MiB_bytes
        with self._lock:
            received = self.used + size
            if received > limit:
                raise MediaRequestDownloadSizeExceededError(
                    "Request media downloads exceed "
                    f"VLLM_MAX_MEDIA_REQUEST_DOWNLOAD_SIZE_MB={limit_mb:g} "
                    f"({received} bytes received)."
                )
            self.used = received

MediaRequestDownloadSizeExceededError

Bases: ValueError

Raised when all HTTP media downloads in one request exceed the limit.

Source code in vllm/connections.py

class MediaRequestDownloadSizeExceededError(ValueError):
    """Raised when all HTTP media downloads in one request exceed the limit."""

_async_retry(fn)

Add retry logic with exponential backoff to an async method.

The decorated method must accept timeout as a keyword argument. The decorator replaces it with a per-attempt timeout that grows by _RETRY_BACKOFF_FACTOR on each retry so transient slowness on busy hosts is absorbed.

Source code in vllm/connections.py

def _async_retry(
    fn: Callable[_P, Coroutine[Any, Any, _T]],
) -> Callable[_P, Coroutine[Any, Any, _T]]:
    """Add retry logic with exponential backoff to an async method.

    The decorated method must accept ``timeout`` as a keyword argument.
    The decorator replaces it with a per-attempt timeout that grows by
    ``_RETRY_BACKOFF_FACTOR`` on each retry so transient slowness on busy
    hosts is absorbed.
    """

    @functools.wraps(fn)
    async def wrapper(*args: Any, **kwargs: Any) -> _T:
        base_timeout: float | None = kwargs.get("timeout")
        max_retries = max(envs.VLLM_MEDIA_FETCH_MAX_RETRIES, 1)

        for attempt in range(max_retries):
            attempt_timeout = (
                base_timeout * (_RETRY_BACKOFF_FACTOR**attempt)
                if base_timeout is not None
                else None
            )
            kwargs["timeout"] = attempt_timeout
            try:
                return await fn(*args, **kwargs)
            except Exception as e:
                if not _is_retryable(e) or attempt + 1 >= max_retries:
                    raise
                backoff = _RETRY_BACKOFF_FACTOR**attempt
                _log_retry(
                    args,
                    kwargs,
                    attempt,
                    max_retries,
                    attempt_timeout,
                    e,
                    backoff,
                    base_timeout,
                )
                await asyncio.sleep(backoff)

        raise AssertionError("unreachable")

    return wrapper  # type: ignore[return-value]

_is_retryable(exc)

Return True for transient errors that are worth retrying.

Retryable

• Timeouts (aiohttp, requests, stdlib)
• Connection-level failures (refused, reset, DNS)
• Server errors (5xx) -- includes S3 503 SlowDown

Not retryable: - Client errors (4xx) -- bad URL, auth, not-found - Programming errors (ValueError, TypeError, ...)

Source code in vllm/connections.py

def _is_retryable(exc: Exception) -> bool:
    """Return True for transient errors that are worth retrying.

    Retryable:
      - Timeouts (aiohttp, requests, stdlib)
      - Connection-level failures (refused, reset, DNS)
      - Server errors (5xx) -- includes S3 503 SlowDown
    Not retryable:
      - Client errors (4xx) -- bad URL, auth, not-found
      - Programming errors (ValueError, TypeError, ...)
    """
    # Timeouts
    if isinstance(
        exc,
        (
            TimeoutError,
            asyncio.TimeoutError,
            requests.exceptions.Timeout,
            aiohttp.ServerTimeoutError,
        ),
    ):
        return True
    # Connection-level failures
    if isinstance(
        exc,
        (
            ConnectionError,
            aiohttp.ClientConnectionError,
            requests.exceptions.ConnectionError,
        ),
    ):
        return True
    # aiohttp server-side disconnects
    if isinstance(exc, aiohttp.ServerDisconnectedError):
        return True
    # requests 5xx -- raise_for_status() throws HTTPError
    if (
        isinstance(exc, requests.exceptions.HTTPError)
        and exc.response is not None
        and exc.response.status_code >= 500
    ):
        return True
    # aiohttp 5xx -- raise_for_status() throws ClientResponseError
    return isinstance(exc, aiohttp.ClientResponseError) and exc.status >= 500

_sync_retry(fn)

Add retry logic with exponential backoff to a sync method.

The decorated method must accept timeout as a keyword argument. The decorator replaces it with a per-attempt timeout that grows by _RETRY_BACKOFF_FACTOR on each retry so transient slowness on busy hosts is absorbed.

Source code in vllm/connections.py

def _sync_retry(
    fn: Callable[_P, _T],
) -> Callable[_P, _T]:
    """Add retry logic with exponential backoff to a sync method.

    The decorated method must accept ``timeout`` as a keyword argument.
    The decorator replaces it with a per-attempt timeout that grows by
    ``_RETRY_BACKOFF_FACTOR`` on each retry so transient slowness on busy
    hosts is absorbed.
    """

    @functools.wraps(fn)
    def wrapper(*args: Any, **kwargs: Any) -> _T:
        base_timeout: float | None = kwargs.get("timeout")
        max_retries = max(envs.VLLM_MEDIA_FETCH_MAX_RETRIES, 1)

        for attempt in range(max_retries):
            attempt_timeout = (
                base_timeout * (_RETRY_BACKOFF_FACTOR**attempt)
                if base_timeout is not None
                else None
            )
            kwargs["timeout"] = attempt_timeout
            try:
                return fn(*args, **kwargs)
            except Exception as e:
                if not _is_retryable(e) or attempt + 1 >= max_retries:
                    raise
                backoff = _RETRY_BACKOFF_FACTOR**attempt
                _log_retry(
                    args,
                    kwargs,
                    attempt,
                    max_retries,
                    attempt_timeout,
                    e,
                    backoff,
                    base_timeout,
                )
                time.sleep(backoff)

        raise AssertionError("unreachable")

    return wrapper  # type: ignore[return-value]