vllm.multimodal.media

Modules:

• audio –
• base –
• connector –
• decode –
• image –
• video –

Classes:

• AudioEmbeddingMediaIO –

  Configuration values can be user-provided either by --media-io-kwargs or

• AudioMediaIO –

  Configuration values can be user-provided either by --media-io-kwargs or

• DecodeSpec –

  Canonical snapshot of the decode settings included in a media key.

• ImageEmbeddingMediaIO –

  Image embedding MediaIO implementation.

• ImageMediaIO –

  Configuration values can be user-provided either by --media-io-kwargs or

• MediaConnector –

  Fetches media URLs down to their encoded bytes.

• MediaIO –

  Configuration values can be user-provided either by --media-io-kwargs or

• MediaRef –

  A reference to one piece of fetched media.

• VideoMediaIO –

  Configuration values can be user-provided either by --media-io-kwargs or

AudioEmbeddingMediaIO

Bases: MediaIO[Tensor]

Configuration values can be user-provided either by --media-io-kwargs or by the runtime API field "media_io_kwargs". Ensure proper validation and error handling.

Source code in vllm/multimodal/media/audio.py

class AudioEmbeddingMediaIO(MediaIO[torch.Tensor]):
    """Configuration values can be user-provided either by --media-io-kwargs or
    by the runtime API field "media_io_kwargs". Ensure proper validation and
    error handling.
    """

    def __init__(self) -> None:
        super().__init__()

    def load_bytes(self, data: bytes) -> torch.Tensor:
        buffer = BytesIO(data)
        with check_sparse_tensor_invariants_threadsafe():
            tensor = torch.load(buffer, weights_only=True)
            return safe_to_dense(tensor, parameter="audio_embeds")

    def load_base64(self, media_type: str, data: str) -> torch.Tensor:
        return self.load_bytes(pybase64.b64decode(data, validate=True))

    def load_file(self, filepath: Path) -> torch.Tensor:
        with check_sparse_tensor_invariants_threadsafe():
            tensor = torch.load(filepath, weights_only=True)
            return safe_to_dense(tensor, parameter="audio_embeds")

    def encode_base64(self, media: torch.Tensor) -> str:
        return tensor2base64(media)

AudioMediaIO

Bases: MediaIO[tuple[NDArray, float]]

Configuration values can be user-provided either by --media-io-kwargs or by the runtime API field "media_io_kwargs". Ensure proper validation and error handling.

Source code in vllm/multimodal/media/audio.py

class AudioMediaIO(MediaIO[tuple[npt.NDArray, float]]):
    """Configuration values can be user-provided either by --media-io-kwargs or
    by the runtime API field "media_io_kwargs". Ensure proper validation and
    error handling.
    """

    def __init__(self, **kwargs) -> None:
        super().__init__()

        # Allow per-server override of the audio decoding backend, e.g.:
        #   --media-io-kwargs '{"audio": {"audio_backend": "torchcodec"}}'
        # Validate eagerly so a misconfigured deployment fails at startup
        # rather than per request.
        backend = kwargs.pop("audio_backend", None)
        if backend is not None and backend not in AUDIO_BACKENDS:
            raise ValueError(
                f"Unknown audio_backend {backend!r}. "
                f"Available backends: {list(AUDIO_BACKENDS)}"
            )
        self.audio_backend = backend

        # `kwargs` contains custom arguments from
        # --media-io-kwargs for this modality, merged with
        # per-request runtime media_io_kwargs via merge_kwargs().
        # They can be passed to the underlying
        # media loaders (e.g. custom implementations)
        # for flexible control.
        self.kwargs = kwargs

    def get_max_bytes(self) -> int:
        return int(envs.VLLM_MAX_AUDIO_CLIP_FILESIZE_MB * MiB_bytes)

    def _validate_encoded_size(self, size: int) -> None:
        max_bytes = self.get_max_bytes()
        if size > max_bytes:
            raise VLLMValidationError(
                "Maximum file size exceeded",
                parameter="audio_filesize_mb",
                value=size / MiB_bytes,
            )

    def get_decode_spec(self) -> DecodeSpec:
        return DecodeSpec({**self.kwargs, "audio_backend": self.audio_backend})

    def load_bytes(self, data: bytes) -> tuple[npt.NDArray, float]:
        self._validate_encoded_size(len(data))
        return load_audio(
            BytesIO(data),
            sr=None,
            max_duration_s=envs.VLLM_MAX_AUDIO_DECODE_DURATION_S,
            max_decode_bytes=envs.VLLM_MAX_AUDIO_DECODE_BYTES,
            backend=self.audio_backend,
        )

    def load_bytes_ref(self, data: bytes) -> MediaRef[tuple[npt.NDArray, float]]:
        # The size guard stays eager so oversize errors surface at fetch time;
        # only the decode is deferred.
        self._validate_encoded_size(len(data))
        return super().load_bytes_ref(data)

    def _validate_base64_size(self, data: str) -> None:
        max_encoded_chars = 4 * ((self.get_max_bytes() + 2) // 3)
        if len(data) > max_encoded_chars:
            raise VLLMValidationError(
                "Maximum file size exceeded",
                parameter="audio_filesize_mb",
                value=(len(data) * 3 / 4) / MiB_bytes,
            )

    def load_base64(
        self,
        media_type: str,
        data: str,
    ) -> tuple[npt.NDArray, float]:
        self._validate_base64_size(data)
        return self.load_bytes(pybase64.b64decode(data, validate=True))

    def load_base64_ref(
        self,
        media_type: str,
        data: str,
    ) -> MediaRef[tuple[npt.NDArray, float]]:
        self._validate_base64_size(data)
        return super().load_base64_ref(media_type, data)

    def load_file(self, filepath: Path) -> tuple[npt.NDArray, float]:
        self._validate_encoded_size(filepath.stat().st_size)
        return load_audio(
            filepath,
            sr=None,
            max_duration_s=envs.VLLM_MAX_AUDIO_DECODE_DURATION_S,
            max_decode_bytes=envs.VLLM_MAX_AUDIO_DECODE_BYTES,
            backend=self.audio_backend,
        )

    def load_file_ref(self, filepath: Path) -> MediaRef[tuple[npt.NDArray, float]]:
        self._validate_encoded_size(filepath.stat().st_size)
        return super().load_file_ref(filepath)

    def encode_base64(
        self,
        media: tuple[npt.NDArray, int],
        *,
        audio_format: str = "WAV",
    ) -> str:
        audio, sr = media

        with BytesIO() as buffer:
            soundfile.write(buffer, audio, sr, format=audio_format)
            data = buffer.getvalue()

        return pybase64.b64encode(data).decode("utf-8")

DecodeSpec dataclass

Canonical snapshot of the decode settings included in a media key.

Methods:

• canonical_bytes –

  Return the canonical JSON bytes used for key derivation.

• extend –

  Return a copy with settings merged over the current ones.

Attributes:

• settings (Mapping[str, Any]) –

  Return a fresh copy of the settings without exposing mutable state.

Source code in vllm/multimodal/media/base.py

@dataclass(frozen=True, init=False)
class DecodeSpec:
    """Canonical snapshot of the decode settings included in a media key."""

    _canonical: bytes

    def __init__(self, settings: Mapping[str, Any] | None = None) -> None:
        canonical = json.dumps(
            dict(settings) if settings is not None else {},
            sort_keys=True,
            separators=(",", ":"),
            allow_nan=False,
        ).encode("utf-8")
        object.__setattr__(self, "_canonical", canonical)

    @property
    def settings(self) -> Mapping[str, Any]:
        """Return a fresh copy of the settings without exposing mutable state."""
        return json.loads(self._canonical)

    def extend(self, **settings: Any) -> "DecodeSpec":
        """Return a copy with `settings` merged over the current ones."""
        return DecodeSpec({**self.settings, **settings})

    def canonical_bytes(self) -> bytes:
        """Return the canonical JSON bytes used for key derivation."""
        return self._canonical

settings property

Return a fresh copy of the settings without exposing mutable state.

canonical_bytes()

Return the canonical JSON bytes used for key derivation.

Source code in vllm/multimodal/media/base.py

def canonical_bytes(self) -> bytes:
    """Return the canonical JSON bytes used for key derivation."""
    return self._canonical

extend(**settings)

Return a copy with settings merged over the current ones.

Source code in vllm/multimodal/media/base.py

def extend(self, **settings: Any) -> "DecodeSpec":
    """Return a copy with `settings` merged over the current ones."""
    return DecodeSpec({**self.settings, **settings})

ImageEmbeddingMediaIO

Bases: MediaIO[Tensor]

Image embedding MediaIO implementation.

Configuration values can be user-provided either by --media-io-kwargs or by the runtime API field "media_io_kwargs". Ensure proper validation and error handling.

Source code in vllm/multimodal/media/image.py

class ImageEmbeddingMediaIO(MediaIO[torch.Tensor]):
    """Image embedding MediaIO implementation.

    Configuration values can be user-provided either by --media-io-kwargs or
    by the runtime API field "media_io_kwargs". Ensure proper validation and
    error handling.
    """

    def __init__(self) -> None:
        super().__init__()

    def _load_pickled_torch(self, data: bytes) -> torch.Tensor:
        buffer = BytesIO(data)
        with check_sparse_tensor_invariants_threadsafe():
            tensor = torch.load(buffer, weights_only=True)
            return safe_to_dense(tensor, parameter="image_embeds")

    def _load_numpy(self, data: bytes) -> torch.Tensor:
        with BytesIO(data) as buffer:
            return torch.from_numpy(np.load(buffer))

    def load_bytes(self, data: bytes) -> torch.Tensor:
        if data[:6] == MAGIC_NUMPY_PREFIX:
            return self._load_numpy(data)

        return self._load_pickled_torch(data)

    def load_base64(self, media_type: str, data: str) -> torch.Tensor:
        return self.load_bytes(pybase64.b64decode(data, validate=True))

    def load_file(self, filepath: Path) -> torch.Tensor:
        if filepath.suffix == ".npy":
            return torch.from_numpy(np.load(filepath))

        with check_sparse_tensor_invariants_threadsafe():
            tensor = torch.load(filepath, weights_only=True)
            return safe_to_dense(tensor, parameter="image_embeds")

    def encode_base64(self, media: torch.Tensor) -> str:
        return tensor2base64(media)

ImageMediaIO

Bases: MediaIO[Image]

Configuration values can be user-provided either by --media-io-kwargs or by the runtime API field "media_io_kwargs". Ensure proper validation and error handling.

Methods:

• load_bytes_ref –

  Eager header parse + lazy pixel decode.

• open_header –

  Open only the image header, leaving the pixels undecoded.

Source code in vllm/multimodal/media/image.py

class ImageMediaIO(MediaIO[Image.Image]):
    """Configuration values can be user-provided either by --media-io-kwargs or
    by the runtime API field "media_io_kwargs". Ensure proper validation and
    error handling.
    """

    def __init__(self, image_mode: str | None = "RGB", **kwargs) -> None:
        super().__init__()

        # Target mode for loaded images; `None` keeps the original mode
        # (i.e. no conversion, alpha channel is preserved as-is).
        self.image_mode = image_mode
        # `kwargs` contains custom arguments from
        # --media-io-kwargs for this modality, merged with
        # per-request runtime media_io_kwargs via merge_kwargs().
        # They can be passed to the underlying
        # media loaders (e.g. custom implementations)
        # for flexible control.
        self.kwargs = kwargs

        # Extract RGBA background color from kwargs if provided
        # Default to white background for backward compatibility
        rgba_bg = kwargs.get("rgba_background_color", (255, 255, 255))
        # Convert list to tuple for consistency
        if isinstance(rgba_bg, list):
            rgba_bg = tuple(rgba_bg)

        # Validate rgba_background_color format
        if not (
            isinstance(rgba_bg, tuple)
            and len(rgba_bg) == 3
            and all(isinstance(c, int) and 0 <= c <= 255 for c in rgba_bg)
        ):
            raise ValueError(
                "rgba_background_color must be a list or tuple of 3 integers "
                "in the range [0, 255]."
            )
        self.rgba_background_color = rgba_bg

    def get_decode_spec(self) -> DecodeSpec:
        # The resolved values win: `rgba_background_color` also sits in
        # `kwargs`, in whatever shape the caller supplied it.
        return DecodeSpec(
            {
                **self.kwargs,
                "image_mode": self.image_mode,
                "rgba_background_color": self.rgba_background_color,
            }
        )

    def _convert_image_mode(self, image: Image.Image) -> Image.Image:
        """Convert image mode with custom background color."""
        if self.image_mode is None or image.mode == self.image_mode:
            return image
        elif image.mode == "RGBA" and self.image_mode == "RGB":
            return rgba_to_rgb(image, self.rgba_background_color)
        else:
            return convert_image_mode(
                image, self.image_mode, self.rgba_background_color
            )

    def open_header(self, data: bytes) -> Image.Image:
        """Open only the image header, leaving the pixels undecoded.

        Mode, size and EXIF are readable from the header alone, so this is
        also where the max-pixels guard and the cache key's EXIF probe run.
        """
        try:
            image = Image.open(BytesIO(data))
        except (OSError, Image.UnidentifiedImageError) as e:
            raise ValueError(f"Failed to load image: {e}") from e

        self._validate_dimensions(image)
        return image

    def _validate_dimensions(self, image: Image.Image) -> None:
        w, h = image.size
        max_pixels = envs.VLLM_MAX_IMAGE_PIXELS
        if max_pixels > 0 and w * h > max_pixels:
            raise ValueError(
                f"Image dimensions {w}x{h} ({w * h} pixels) exceed "
                f"the maximum of {max_pixels} pixels. Set "
                f"VLLM_MAX_IMAGE_PIXELS to increase this limit."
            )

    def _exif_key(self, header_image: Image.Image) -> bytes | None:
        """The EXIF `ImageID` cache key of a header-opened image, if it has one.

        Key derivation must never rasterize pixels: `PngImageFile.getexif()`
        calls `.load()` to scan for a trailing eXIf chunk when `info` has
        none, and on Pillow >= 12 `getexif()` loads unconditionally. Per the
        PNG spec eXIf must precede IDAT, so compliant PNGs carry their EXIF in
        `info` at open time; a non-compliant trailing-eXIf PNG with an
        `ImageID` is not recognized here and keys off its bytes instead.
        """
        if header_image.format == "PNG" and "exif" not in header_image.info:
            return None
        return get_image_id_bytes(header_image)

    def _decode_pixels(self, image: Image.Image) -> Image.Image:
        """Rasterize a header-opened image, then normalize and convert it."""
        try:
            image = normalize_image(image)
            image.load()
            return self._convert_image_mode(image)
        except (OSError, Image.UnidentifiedImageError) as e:
            raise ValueError(f"Failed to load image: {e}") from e

    def _decode_from_bytes(self, data: bytes) -> Image.Image:
        return self._decode_pixels(self.open_header(data))

    def load_bytes(self, data: bytes) -> Image.Image:
        return self._decode_from_bytes(data)

    def load_bytes_ref(self, data: bytes) -> MediaRef[Image.Image]:
        """Eager header parse + lazy pixel decode.

        The header is parsed eagerly so oversized images still fail at fetch
        time and so the cache key can carry the EXIF `ImageID` without
        decoding. Unparsable headers and pixel decoding are deferred to first
        access, where decode errors surface.
        """
        spec = self.get_decode_spec()
        try:
            header_image = Image.open(BytesIO(data))
        except (OSError, Image.UnidentifiedImageError):
            return MediaRef(partial(self._decode_from_bytes, data), data, spec)
        self._validate_dimensions(header_image)
        return MediaRef(
            partial(self._decode_pixels, header_image),
            data,
            spec,
            key=self._exif_key(header_image),
        )

    def load_base64(self, media_type: str, data: str) -> Image.Image:
        return self.load_bytes(pybase64.b64decode(data, validate=True))

    def load_file(self, filepath: Path) -> Image.Image:
        return self.load_bytes(filepath.read_bytes())

    def encode_base64(
        self,
        media: Image.Image,
        *,
        image_format: str = "PNG",
    ) -> str:
        image = media

        with BytesIO() as buffer:
            image = self._convert_image_mode(image)
            image.save(buffer, image_format)
            data = buffer.getvalue()

        return pybase64.b64encode(data).decode("utf-8")

_convert_image_mode(image)

Convert image mode with custom background color.

Source code in vllm/multimodal/media/image.py

def _convert_image_mode(self, image: Image.Image) -> Image.Image:
    """Convert image mode with custom background color."""
    if self.image_mode is None or image.mode == self.image_mode:
        return image
    elif image.mode == "RGBA" and self.image_mode == "RGB":
        return rgba_to_rgb(image, self.rgba_background_color)
    else:
        return convert_image_mode(
            image, self.image_mode, self.rgba_background_color
        )

_decode_pixels(image)

Rasterize a header-opened image, then normalize and convert it.

Source code in vllm/multimodal/media/image.py

def _decode_pixels(self, image: Image.Image) -> Image.Image:
    """Rasterize a header-opened image, then normalize and convert it."""
    try:
        image = normalize_image(image)
        image.load()
        return self._convert_image_mode(image)
    except (OSError, Image.UnidentifiedImageError) as e:
        raise ValueError(f"Failed to load image: {e}") from e

_exif_key(header_image)

The EXIF ImageID cache key of a header-opened image, if it has one.

Key derivation must never rasterize pixels: PngImageFile.getexif() calls .load() to scan for a trailing eXIf chunk when info has none, and on Pillow >= 12 getexif() loads unconditionally. Per the PNG spec eXIf must precede IDAT, so compliant PNGs carry their EXIF in info at open time; a non-compliant trailing-eXIf PNG with an ImageID is not recognized here and keys off its bytes instead.

Source code in vllm/multimodal/media/image.py

def _exif_key(self, header_image: Image.Image) -> bytes | None:
    """The EXIF `ImageID` cache key of a header-opened image, if it has one.

    Key derivation must never rasterize pixels: `PngImageFile.getexif()`
    calls `.load()` to scan for a trailing eXIf chunk when `info` has
    none, and on Pillow >= 12 `getexif()` loads unconditionally. Per the
    PNG spec eXIf must precede IDAT, so compliant PNGs carry their EXIF in
    `info` at open time; a non-compliant trailing-eXIf PNG with an
    `ImageID` is not recognized here and keys off its bytes instead.
    """
    if header_image.format == "PNG" and "exif" not in header_image.info:
        return None
    return get_image_id_bytes(header_image)

load_bytes_ref(data)

Eager header parse + lazy pixel decode.

The header is parsed eagerly so oversized images still fail at fetch time and so the cache key can carry the EXIF ImageID without decoding. Unparsable headers and pixel decoding are deferred to first access, where decode errors surface.

Source code in vllm/multimodal/media/image.py

def load_bytes_ref(self, data: bytes) -> MediaRef[Image.Image]:
    """Eager header parse + lazy pixel decode.

    The header is parsed eagerly so oversized images still fail at fetch
    time and so the cache key can carry the EXIF `ImageID` without
    decoding. Unparsable headers and pixel decoding are deferred to first
    access, where decode errors surface.
    """
    spec = self.get_decode_spec()
    try:
        header_image = Image.open(BytesIO(data))
    except (OSError, Image.UnidentifiedImageError):
        return MediaRef(partial(self._decode_from_bytes, data), data, spec)
    self._validate_dimensions(header_image)
    return MediaRef(
        partial(self._decode_pixels, header_image),
        data,
        spec,
        key=self._exif_key(header_image),
    )

open_header(data)

Open only the image header, leaving the pixels undecoded.

Mode, size and EXIF are readable from the header alone, so this is also where the max-pixels guard and the cache key's EXIF probe run.

Source code in vllm/multimodal/media/image.py

def open_header(self, data: bytes) -> Image.Image:
    """Open only the image header, leaving the pixels undecoded.

    Mode, size and EXIF are readable from the header alone, so this is
    also where the max-pixels guard and the cache key's EXIF probe run.
    """
    try:
        image = Image.open(BytesIO(data))
    except (OSError, Image.UnidentifiedImageError) as e:
        raise ValueError(f"Failed to load image: {e}") from e

    self._validate_dimensions(image)
    return image

MediaConnector

Fetches media URLs down to their encoded bytes.

Transport only: scheme dispatch (data: / http(s): / file:), the domain allow-list, the encoded-size cap, the on-disk download cache, the redirect policy and error normalization. How bytes decode -- and therefore what identifies the resulting item in the processor cache -- is decided by the caller-supplied MediaIO, which the multi-modal processor's info resolves from the model config.

Methods:

• __init__ –

  Args:

• fetch_audio –

  Fetch audio and pair it with a caller-supplied decoder.

• fetch_audio_async –

  Asynchronously fetch audio and pair it with a caller-supplied decoder.

• fetch_audio_embedding –

  Load audio embedding from a URL.

• fetch_audio_embedding_async –

  Asynchronously load audio embedding from a URL.

• fetch_image –

  Fetch an image and pair it with a caller-supplied decoder.

• fetch_image_async –

  Asynchronously fetch an image and pair it with a supplied decoder.

• fetch_image_embedding –

  Load image embedding from a URL.

• fetch_image_embedding_async –

  Asynchronously load image embedding from a URL.

• fetch_video –

  Fetch a video and pair it with a caller-supplied decoder.

• fetch_video_and_audio –

  Fetch video_url once and decode it as both video and audio.

• fetch_video_and_audio_async –

  Asynchronous fetch_video_and_audio.

• fetch_video_async –

  Asynchronously fetch a video and pair it with a supplied decoder.

• fetch_video_embedding –

  Load video embedding from a URL.

• fetch_video_embedding_async –

  Asynchronously load video embedding from a URL.

• load_from_url –

  Fetch url and pair it with a caller-supplied decoder.

• load_from_url_async –

  Asynchronously fetch url and pair it with a caller-supplied decoder.

Source code in vllm/multimodal/media/connector.py

@MEDIA_CONNECTOR_REGISTRY.register("http")
class MediaConnector:
    """Fetches media URLs down to their encoded bytes.

    Transport only: scheme dispatch (`data:` / `http(s):` / `file:`), the
    domain allow-list, the encoded-size cap, the on-disk download cache, the
    redirect policy and error normalization. How bytes decode -- and therefore
    what identifies the resulting item in the processor cache -- is decided by
    the caller-supplied [`MediaIO`][vllm.multimodal.media.base.MediaIO], which
    the multi-modal processor's `info` resolves from the model config.
    """

    def __init__(
        self,
        connection: HTTPConnection = global_http_connection,
        *,
        allowed_local_media_path: str = "",
        allowed_media_domains: list[str] | None = None,
    ) -> None:
        """Args:
        connection: HTTP connection client to download media contents.
        allowed_local_media_path: A local directory to load media files from.
        allowed_media_domains: If set, only media URLs that belong to this
                               domain can be used for multi-modal inputs.

        """
        super().__init__()

        self.connection = connection

        if allowed_local_media_path:
            allowed_local_media_path_ = Path(allowed_local_media_path).resolve()

            if not allowed_local_media_path_.exists():
                raise ValueError(
                    "Invalid `--allowed-local-media-path`: The path "
                    f"{allowed_local_media_path_} does not exist."
                )
            if not allowed_local_media_path_.is_dir():
                raise ValueError(
                    "Invalid `--allowed-local-media-path`: The path "
                    f"{allowed_local_media_path_} must be a directory."
                )
        else:
            allowed_local_media_path_ = None

        self.allowed_local_media_path = allowed_local_media_path_
        if allowed_media_domains is None:
            allowed_media_domains = []
        self.allowed_media_domains = allowed_media_domains

        # Media download cache (opt-in via VLLM_MEDIA_CACHE)
        self._media_cache_dir: str | None = None
        self._media_cache_max_bytes: int = 0
        self._media_cache_ttl_secs: float = 0
        media_cache = envs.VLLM_MEDIA_CACHE
        if media_cache:
            try:
                os.makedirs(media_cache, exist_ok=True)
                # Verify the directory is writable before enabling caching
                with tempfile.NamedTemporaryFile(dir=media_cache, delete=True):
                    pass
                self._media_cache_dir = media_cache
                self._media_cache_max_bytes = (
                    envs.VLLM_MEDIA_CACHE_MAX_SIZE_MB * 1024 * 1024
                )
                self._media_cache_ttl_secs = envs.VLLM_MEDIA_CACHE_TTL_HOURS * 3600
                logger.info(
                    "Media cache enabled at %s (max %d MB, TTL %s hours)",
                    media_cache,
                    envs.VLLM_MEDIA_CACHE_MAX_SIZE_MB,
                    envs.VLLM_MEDIA_CACHE_TTL_HOURS,
                )
            except OSError:
                logger.warning(
                    "VLLM_MEDIA_CACHE path %s is not writable, media caching disabled",
                    media_cache,
                )

    def _get_cached_bytes(self, url: str) -> bytes | None:
        """Return cached bytes for a URL, or None if not cached/expired."""
        if not self._media_cache_dir:
            return None
        cache_path = self._media_cache_path(url)
        # Check TTL
        try:
            age = time.time() - cache_path.stat().st_mtime
        except OSError:
            return None
        if age > self._media_cache_ttl_secs:
            cache_path.unlink(missing_ok=True)
            return None
        # Touch mtime for LRU ordering
        try:
            cache_path.touch()
            return cache_path.read_bytes()
        except OSError:
            return None

    def _put_cached_bytes(self, url: str, data: bytes) -> None:
        """Store downloaded bytes and evict if over budget."""
        if not self._media_cache_dir:
            return
        cache_path = self._media_cache_path(url)
        # Atomic write via temp file + rename
        tmp_path = None
        try:
            with tempfile.NamedTemporaryFile(
                mode="wb", dir=self._media_cache_dir, delete=False
            ) as tmp_file:
                tmp_file.write(data)
                tmp_path = tmp_file.name
            os.rename(tmp_path, str(cache_path))
        except OSError:
            # Another process beat us or disk issue
            if tmp_path is not None:
                with contextlib.suppress(OSError):
                    os.remove(tmp_path)
            return
        self._maybe_evict(exclude=cache_path)

    def _maybe_evict(self, exclude: Path | None = None) -> None:
        """Evict expired entries first, then LRU until under size limit."""
        cache_dir = Path(self._media_cache_dir)  # type: ignore[arg-type]
        entries = []
        expired = []
        total_size = 0
        now = time.time()
        for f in cache_dir.iterdir():
            if f.name.startswith("."):
                continue
            try:
                stat = f.stat()
            except OSError:
                continue
            age = now - stat.st_mtime
            if age > self._media_cache_ttl_secs:
                expired.append(f)
                continue
            total_size += stat.st_size
            # Never evict the file we just wrote
            if exclude is not None and f.name == exclude.name:
                continue
            entries.append((stat.st_mtime, stat.st_size, f))

        # Evict items according to LRU policy
        entries.sort(key=lambda e: e[0], reverse=True)
        while total_size > self._media_cache_max_bytes and entries:
            mtime, size, f = entries.pop()
            expired.append(f)
            total_size -= size

        for f in expired:
            f.unlink(missing_ok=True)

    def _media_cache_path(self, url: str) -> Path:
        url_hash = hashlib.sha256(url.encode()).hexdigest()[:20]
        ext = Path(url.split("?", 1)[0]).suffix or ""
        return Path(self._media_cache_dir) / f"{url_hash}{ext}"  # type: ignore[arg-type]

    def _resolve_allowed_file(self, url_spec: Url) -> Path:
        """Resolve a `file:` URL, checking it against the allowed media path."""
        allowed_local_media_path = self.allowed_local_media_path
        if allowed_local_media_path is None:
            raise RuntimeError(
                "Cannot load local files without `--allowed-local-media-path`."
            )

        url_spec_path = url_spec.path or ""
        url_spec_netloc = url_spec.netloc or ""
        filepath = Path(url2pathname(url_spec_netloc + url_spec_path))
        if allowed_local_media_path not in filepath.resolve().parents:
            raise ValueError(
                f"The file path {filepath} must be a subpath "
                f"of `--allowed-local-media-path {allowed_local_media_path}`."
            )

        return filepath

    def _assert_url_in_allowed_media_domains(self, url_spec: Url) -> None:
        if (
            self.allowed_media_domains
            and url_spec.hostname not in self.allowed_media_domains
        ):
            raise ValueError(
                f"The URL must be from one of the allowed domains: "
                f"{self.allowed_media_domains}. Input URL domain: "
                f"{url_spec.hostname}"
            )

    def _fetch_url_bytes(
        self,
        url: str,
        url_spec: Url,
        *,
        max_bytes: int | None = None,
        fetch_timeout: int | None = None,
    ) -> bytes:
        """Download an HTTP(S) URL down to its encoded bytes."""
        self._assert_url_in_allowed_media_domains(url_spec)

        cached = self._get_cached_bytes(url)
        if cached is not None:
            return cached

        try:
            data = self.connection.get_bytes(
                url_spec.url,
                timeout=fetch_timeout,
                allow_redirects=envs.VLLM_MEDIA_URL_ALLOW_REDIRECTS,
                max_bytes=max_bytes,
            )
        except Exception as e:
            wrapped = _wrap_media_fetch_error(url, e)
            if isinstance(wrapped, VLLMUnprocessableEntityError):
                raise wrapped from e
            raise

        self._put_cached_bytes(url, data)
        return data

    async def _fetch_url_bytes_async(
        self,
        url: str,
        url_spec: Url,
        *,
        max_bytes: int | None = None,
        fetch_timeout: int | None = None,
    ) -> bytes:
        """Asynchronous `_fetch_url_bytes`."""
        loop = asyncio.get_running_loop()

        self._assert_url_in_allowed_media_domains(url_spec)

        cached = await loop.run_in_executor(
            global_thread_pool, self._get_cached_bytes, url
        )
        if cached is not None:
            return cached

        try:
            data = await self.connection.async_get_bytes(
                url_spec.url,
                timeout=fetch_timeout,
                allow_redirects=envs.VLLM_MEDIA_URL_ALLOW_REDIRECTS,
                max_bytes=max_bytes,
            )
        except Exception as e:
            wrapped = _wrap_media_fetch_error(url, e)
            if isinstance(wrapped, VLLMUnprocessableEntityError):
                raise wrapped from e
            raise

        await loop.run_in_executor(
            global_thread_pool, self._put_cached_bytes, url, data
        )
        return data

    def _load_refs(
        self,
        url: str,
        media_ios: Sequence[MediaIO[Any]],
        *,
        fetch_timeout: int | None = None,
    ) -> list[MediaRef[Any]]:
        """Fetch `url` once and pair the payload with each of `media_ios`.

        One payload can legitimately feed several decoders -- reading the audio
        track out of a video file needs the same bytes as its frames -- and
        fetching once per decoder would repeat the download.
        """
        if _is_data_url(url):
            media_type, encoded = _parse_data_url(url)
            return [
                media_io.load_base64_ref(media_type, encoded) for media_io in media_ios
            ]

        url_spec = parse_url(url)

        if url_spec.scheme == "file":
            filepath = self._resolve_allowed_file(url_spec)
            return [media_io.load_file_ref(filepath) for media_io in media_ios]

        if url_spec.scheme and url_spec.scheme.startswith("http"):
            data = self._fetch_url_bytes(
                url,
                url_spec,
                max_bytes=_strictest_max_bytes(media_ios),
                fetch_timeout=fetch_timeout,
            )
            return _pair_bytes(media_ios, data)

        msg = "The URL must be either a HTTP, data or file URL."
        raise ValueError(msg)

    async def _load_refs_async(
        self,
        url: str,
        media_ios: Sequence[MediaIO[Any]],
        *,
        fetch_timeout: int | None = None,
    ) -> list[MediaRef[Any]]:
        """Asynchronous `_load_refs`."""
        loop = asyncio.get_running_loop()
        url_spec = None if _is_data_url(url) else parse_url(url)

        if (
            url_spec is not None
            and url_spec.scheme
            and url_spec.scheme.startswith("http")
        ):
            data = await self._fetch_url_bytes_async(
                url,
                url_spec,
                max_bytes=_strictest_max_bytes(media_ios),
                fetch_timeout=fetch_timeout,
            )
            # Building the refs digests the payload for their cache keys (and
            # parses an image header), so keep it off the event loop.
            return await loop.run_in_executor(
                global_thread_pool, _pair_bytes, media_ios, data
            )

        # `data:` and `file:` URLs never touch the network, but splitting the
        # payload, reading the file and building the refs are all linear in it.
        return await loop.run_in_executor(
            global_thread_pool, self._load_refs, url, media_ios
        )

    def load_from_url(
        self,
        url: str,
        media_io: MediaIO[_M],
        *,
        fetch_timeout: int | None = None,
    ) -> MediaRef[_M]:
        """Fetch `url` and pair it with a caller-supplied decoder.

        Returns a lazy handle: decoding happens on first access (inside the
        multi-modal processor), so decode errors surface there rather than at
        the fetch site.
        """
        (ref,) = self._load_refs(url, [media_io], fetch_timeout=fetch_timeout)
        return ref

    async def load_from_url_async(
        self,
        url: str,
        media_io: MediaIO[_M],
        *,
        fetch_timeout: int | None = None,
    ) -> MediaRef[_M]:
        """Asynchronously fetch `url` and pair it with a caller-supplied decoder."""
        (ref,) = await self._load_refs_async(
            url, [media_io], fetch_timeout=fetch_timeout
        )
        return ref

    def fetch_audio(
        self,
        audio_url: str,
        media_io: MediaIO[tuple[np.ndarray, int | float]],
    ) -> MediaRef[tuple[np.ndarray, int | float]]:
        """Fetch audio and pair it with a caller-supplied decoder.

        Returns a lazy handle: decoding happens on first access (inside the
        multi-modal processor), so decode errors surface there rather than at
        the fetch site.
        """
        return self.load_from_url(
            audio_url,
            media_io,
            fetch_timeout=envs.VLLM_AUDIO_FETCH_TIMEOUT,
        )

    async def fetch_audio_async(
        self,
        audio_url: str,
        media_io: MediaIO[tuple[np.ndarray, int | float]],
    ) -> MediaRef[tuple[np.ndarray, int | float]]:
        """Asynchronously fetch audio and pair it with a caller-supplied decoder."""
        return await self.load_from_url_async(
            audio_url,
            media_io,
            fetch_timeout=envs.VLLM_AUDIO_FETCH_TIMEOUT,
        )

    def fetch_image(
        self,
        image_url: str,
        media_io: MediaIO[Image.Image],
    ) -> MediaRef[Image.Image]:
        """Fetch an image and pair it with a caller-supplied decoder.

        Returns a lazy handle: decoding happens on first access (inside the
        multi-modal processor), so decode errors surface there rather than at
        the fetch site.
        """
        return self.load_from_url(
            image_url,
            media_io,
            fetch_timeout=envs.VLLM_IMAGE_FETCH_TIMEOUT,
        )

    async def fetch_image_async(
        self,
        image_url: str,
        media_io: MediaIO[Image.Image],
    ) -> MediaRef[Image.Image]:
        """Asynchronously fetch an image and pair it with a supplied decoder."""
        return await self.load_from_url_async(
            image_url,
            media_io,
            fetch_timeout=envs.VLLM_IMAGE_FETCH_TIMEOUT,
        )

    def fetch_video(
        self,
        video_url: str,
        media_io: MediaIO[tuple[npt.NDArray, dict[str, Any]]],
    ) -> MediaRef[tuple[npt.NDArray, dict[str, Any]]]:
        """Fetch a video and pair it with a caller-supplied decoder.

        Returns a lazy handle: decoding happens on first access (inside the
        multi-modal processor), so decode errors surface there rather than at
        the fetch site.
        """
        return self.load_from_url(
            video_url,
            media_io,
            fetch_timeout=envs.VLLM_VIDEO_FETCH_TIMEOUT,
        )

    async def fetch_video_async(
        self,
        video_url: str,
        media_io: MediaIO[tuple[npt.NDArray, dict[str, Any]]],
    ) -> MediaRef[tuple[npt.NDArray, dict[str, Any]]]:
        """Asynchronously fetch a video and pair it with a supplied decoder."""
        return await self.load_from_url_async(
            video_url,
            media_io,
            fetch_timeout=envs.VLLM_VIDEO_FETCH_TIMEOUT,
        )

    def fetch_video_and_audio(
        self,
        video_url: str,
        video_io: MediaIO[_V],
        audio_io: MediaIO[_A],
    ) -> tuple[MediaRef[_V], MediaRef[_A]]:
        """Fetch `video_url` once and decode it as both video and audio.

        `use_audio_in_video` reads the audio track out of the video payload, so
        both refs are built from one download instead of two.
        """
        video, audio = self._load_refs(
            video_url,
            [video_io, audio_io],
            fetch_timeout=envs.VLLM_VIDEO_FETCH_TIMEOUT,
        )
        return video, audio

    async def fetch_video_and_audio_async(
        self,
        video_url: str,
        video_io: MediaIO[_V],
        audio_io: MediaIO[_A],
    ) -> tuple[MediaRef[_V], MediaRef[_A]]:
        """Asynchronous `fetch_video_and_audio`."""
        video, audio = await self._load_refs_async(
            video_url,
            [video_io, audio_io],
            fetch_timeout=envs.VLLM_VIDEO_FETCH_TIMEOUT,
        )
        return video, audio

    def fetch_image_embedding(
        self,
        data: str,
    ) -> torch.Tensor:
        """Load image embedding from a URL."""
        image_embedding_io = ImageEmbeddingMediaIO()

        return image_embedding_io.load_base64("", data)

    async def fetch_image_embedding_async(
        self,
        data: str,
    ) -> torch.Tensor:
        """Asynchronously load image embedding from a URL."""
        image_embedding_io = ImageEmbeddingMediaIO()
        loop = asyncio.get_running_loop()

        return await loop.run_in_executor(
            global_thread_pool, image_embedding_io.load_base64, "", data
        )

    def fetch_audio_embedding(
        self,
        data: str,
    ) -> torch.Tensor:
        """Load audio embedding from a URL."""
        audio_embedding_io = AudioEmbeddingMediaIO()

        return audio_embedding_io.load_base64("", data)

    async def fetch_audio_embedding_async(
        self,
        data: str,
    ) -> torch.Tensor:
        """Asynchronously load audio embedding from a URL."""
        audio_embedding_io = AudioEmbeddingMediaIO()
        loop = asyncio.get_running_loop()

        return await loop.run_in_executor(
            global_thread_pool, audio_embedding_io.load_base64, "", data
        )

    def fetch_video_embedding(
        self,
        data: str,
    ) -> torch.Tensor:
        """Load video embedding from a URL."""
        video_embedding_io = VideoEmbeddingMediaIO()

        return video_embedding_io.load_base64("", data)

    async def fetch_video_embedding_async(
        self,
        data: str,
    ) -> torch.Tensor:
        """Asynchronously load video embedding from a URL."""
        video_embedding_io = VideoEmbeddingMediaIO()
        loop = asyncio.get_running_loop()

        return await loop.run_in_executor(
            global_thread_pool, video_embedding_io.load_base64, "", data
        )

__init__(connection=global_http_connection, *, allowed_local_media_path='', allowed_media_domains=None)

Args: connection: HTTP connection client to download media contents. allowed_local_media_path: A local directory to load media files from. allowed_media_domains: If set, only media URLs that belong to this domain can be used for multi-modal inputs.

Source code in vllm/multimodal/media/connector.py

def __init__(
    self,
    connection: HTTPConnection = global_http_connection,
    *,
    allowed_local_media_path: str = "",
    allowed_media_domains: list[str] | None = None,
) -> None:
    """Args:
    connection: HTTP connection client to download media contents.
    allowed_local_media_path: A local directory to load media files from.
    allowed_media_domains: If set, only media URLs that belong to this
                           domain can be used for multi-modal inputs.

    """
    super().__init__()

    self.connection = connection

    if allowed_local_media_path:
        allowed_local_media_path_ = Path(allowed_local_media_path).resolve()

        if not allowed_local_media_path_.exists():
            raise ValueError(
                "Invalid `--allowed-local-media-path`: The path "
                f"{allowed_local_media_path_} does not exist."
            )
        if not allowed_local_media_path_.is_dir():
            raise ValueError(
                "Invalid `--allowed-local-media-path`: The path "
                f"{allowed_local_media_path_} must be a directory."
            )
    else:
        allowed_local_media_path_ = None

    self.allowed_local_media_path = allowed_local_media_path_
    if allowed_media_domains is None:
        allowed_media_domains = []
    self.allowed_media_domains = allowed_media_domains

    # Media download cache (opt-in via VLLM_MEDIA_CACHE)
    self._media_cache_dir: str | None = None
    self._media_cache_max_bytes: int = 0
    self._media_cache_ttl_secs: float = 0
    media_cache = envs.VLLM_MEDIA_CACHE
    if media_cache:
        try:
            os.makedirs(media_cache, exist_ok=True)
            # Verify the directory is writable before enabling caching
            with tempfile.NamedTemporaryFile(dir=media_cache, delete=True):
                pass
            self._media_cache_dir = media_cache
            self._media_cache_max_bytes = (
                envs.VLLM_MEDIA_CACHE_MAX_SIZE_MB * 1024 * 1024
            )
            self._media_cache_ttl_secs = envs.VLLM_MEDIA_CACHE_TTL_HOURS * 3600
            logger.info(
                "Media cache enabled at %s (max %d MB, TTL %s hours)",
                media_cache,
                envs.VLLM_MEDIA_CACHE_MAX_SIZE_MB,
                envs.VLLM_MEDIA_CACHE_TTL_HOURS,
            )
        except OSError:
            logger.warning(
                "VLLM_MEDIA_CACHE path %s is not writable, media caching disabled",
                media_cache,
            )

_fetch_url_bytes(url, url_spec, *, max_bytes=None, fetch_timeout=None)

Download an HTTP(S) URL down to its encoded bytes.

Source code in vllm/multimodal/media/connector.py

def _fetch_url_bytes(
    self,
    url: str,
    url_spec: Url,
    *,
    max_bytes: int | None = None,
    fetch_timeout: int | None = None,
) -> bytes:
    """Download an HTTP(S) URL down to its encoded bytes."""
    self._assert_url_in_allowed_media_domains(url_spec)

    cached = self._get_cached_bytes(url)
    if cached is not None:
        return cached

    try:
        data = self.connection.get_bytes(
            url_spec.url,
            timeout=fetch_timeout,
            allow_redirects=envs.VLLM_MEDIA_URL_ALLOW_REDIRECTS,
            max_bytes=max_bytes,
        )
    except Exception as e:
        wrapped = _wrap_media_fetch_error(url, e)
        if isinstance(wrapped, VLLMUnprocessableEntityError):
            raise wrapped from e
        raise

    self._put_cached_bytes(url, data)
    return data

_fetch_url_bytes_async(url, url_spec, *, max_bytes=None, fetch_timeout=None) async

Asynchronous _fetch_url_bytes.

Source code in vllm/multimodal/media/connector.py

async def _fetch_url_bytes_async(
    self,
    url: str,
    url_spec: Url,
    *,
    max_bytes: int | None = None,
    fetch_timeout: int | None = None,
) -> bytes:
    """Asynchronous `_fetch_url_bytes`."""
    loop = asyncio.get_running_loop()

    self._assert_url_in_allowed_media_domains(url_spec)

    cached = await loop.run_in_executor(
        global_thread_pool, self._get_cached_bytes, url
    )
    if cached is not None:
        return cached

    try:
        data = await self.connection.async_get_bytes(
            url_spec.url,
            timeout=fetch_timeout,
            allow_redirects=envs.VLLM_MEDIA_URL_ALLOW_REDIRECTS,
            max_bytes=max_bytes,
        )
    except Exception as e:
        wrapped = _wrap_media_fetch_error(url, e)
        if isinstance(wrapped, VLLMUnprocessableEntityError):
            raise wrapped from e
        raise

    await loop.run_in_executor(
        global_thread_pool, self._put_cached_bytes, url, data
    )
    return data

_get_cached_bytes(url)

Return cached bytes for a URL, or None if not cached/expired.

Source code in vllm/multimodal/media/connector.py

def _get_cached_bytes(self, url: str) -> bytes | None:
    """Return cached bytes for a URL, or None if not cached/expired."""
    if not self._media_cache_dir:
        return None
    cache_path = self._media_cache_path(url)
    # Check TTL
    try:
        age = time.time() - cache_path.stat().st_mtime
    except OSError:
        return None
    if age > self._media_cache_ttl_secs:
        cache_path.unlink(missing_ok=True)
        return None
    # Touch mtime for LRU ordering
    try:
        cache_path.touch()
        return cache_path.read_bytes()
    except OSError:
        return None

_load_refs(url, media_ios, *, fetch_timeout=None)

Fetch url once and pair the payload with each of media_ios.

One payload can legitimately feed several decoders -- reading the audio track out of a video file needs the same bytes as its frames -- and fetching once per decoder would repeat the download.

Source code in vllm/multimodal/media/connector.py

def _load_refs(
    self,
    url: str,
    media_ios: Sequence[MediaIO[Any]],
    *,
    fetch_timeout: int | None = None,
) -> list[MediaRef[Any]]:
    """Fetch `url` once and pair the payload with each of `media_ios`.

    One payload can legitimately feed several decoders -- reading the audio
    track out of a video file needs the same bytes as its frames -- and
    fetching once per decoder would repeat the download.
    """
    if _is_data_url(url):
        media_type, encoded = _parse_data_url(url)
        return [
            media_io.load_base64_ref(media_type, encoded) for media_io in media_ios
        ]

    url_spec = parse_url(url)

    if url_spec.scheme == "file":
        filepath = self._resolve_allowed_file(url_spec)
        return [media_io.load_file_ref(filepath) for media_io in media_ios]

    if url_spec.scheme and url_spec.scheme.startswith("http"):
        data = self._fetch_url_bytes(
            url,
            url_spec,
            max_bytes=_strictest_max_bytes(media_ios),
            fetch_timeout=fetch_timeout,
        )
        return _pair_bytes(media_ios, data)

    msg = "The URL must be either a HTTP, data or file URL."
    raise ValueError(msg)

_load_refs_async(url, media_ios, *, fetch_timeout=None) async

Asynchronous _load_refs.

Source code in vllm/multimodal/media/connector.py

async def _load_refs_async(
    self,
    url: str,
    media_ios: Sequence[MediaIO[Any]],
    *,
    fetch_timeout: int | None = None,
) -> list[MediaRef[Any]]:
    """Asynchronous `_load_refs`."""
    loop = asyncio.get_running_loop()
    url_spec = None if _is_data_url(url) else parse_url(url)

    if (
        url_spec is not None
        and url_spec.scheme
        and url_spec.scheme.startswith("http")
    ):
        data = await self._fetch_url_bytes_async(
            url,
            url_spec,
            max_bytes=_strictest_max_bytes(media_ios),
            fetch_timeout=fetch_timeout,
        )
        # Building the refs digests the payload for their cache keys (and
        # parses an image header), so keep it off the event loop.
        return await loop.run_in_executor(
            global_thread_pool, _pair_bytes, media_ios, data
        )

    # `data:` and `file:` URLs never touch the network, but splitting the
    # payload, reading the file and building the refs are all linear in it.
    return await loop.run_in_executor(
        global_thread_pool, self._load_refs, url, media_ios
    )

_maybe_evict(exclude=None)

Evict expired entries first, then LRU until under size limit.

Source code in vllm/multimodal/media/connector.py

def _maybe_evict(self, exclude: Path | None = None) -> None:
    """Evict expired entries first, then LRU until under size limit."""
    cache_dir = Path(self._media_cache_dir)  # type: ignore[arg-type]
    entries = []
    expired = []
    total_size = 0
    now = time.time()
    for f in cache_dir.iterdir():
        if f.name.startswith("."):
            continue
        try:
            stat = f.stat()
        except OSError:
            continue
        age = now - stat.st_mtime
        if age > self._media_cache_ttl_secs:
            expired.append(f)
            continue
        total_size += stat.st_size
        # Never evict the file we just wrote
        if exclude is not None and f.name == exclude.name:
            continue
        entries.append((stat.st_mtime, stat.st_size, f))

    # Evict items according to LRU policy
    entries.sort(key=lambda e: e[0], reverse=True)
    while total_size > self._media_cache_max_bytes and entries:
        mtime, size, f = entries.pop()
        expired.append(f)
        total_size -= size

    for f in expired:
        f.unlink(missing_ok=True)

_put_cached_bytes(url, data)

Store downloaded bytes and evict if over budget.

Source code in vllm/multimodal/media/connector.py

def _put_cached_bytes(self, url: str, data: bytes) -> None:
    """Store downloaded bytes and evict if over budget."""
    if not self._media_cache_dir:
        return
    cache_path = self._media_cache_path(url)
    # Atomic write via temp file + rename
    tmp_path = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="wb", dir=self._media_cache_dir, delete=False
        ) as tmp_file:
            tmp_file.write(data)
            tmp_path = tmp_file.name
        os.rename(tmp_path, str(cache_path))
    except OSError:
        # Another process beat us or disk issue
        if tmp_path is not None:
            with contextlib.suppress(OSError):
                os.remove(tmp_path)
        return
    self._maybe_evict(exclude=cache_path)

_resolve_allowed_file(url_spec)

Resolve a file: URL, checking it against the allowed media path.

Source code in vllm/multimodal/media/connector.py

def _resolve_allowed_file(self, url_spec: Url) -> Path:
    """Resolve a `file:` URL, checking it against the allowed media path."""
    allowed_local_media_path = self.allowed_local_media_path
    if allowed_local_media_path is None:
        raise RuntimeError(
            "Cannot load local files without `--allowed-local-media-path`."
        )

    url_spec_path = url_spec.path or ""
    url_spec_netloc = url_spec.netloc or ""
    filepath = Path(url2pathname(url_spec_netloc + url_spec_path))
    if allowed_local_media_path not in filepath.resolve().parents:
        raise ValueError(
            f"The file path {filepath} must be a subpath "
            f"of `--allowed-local-media-path {allowed_local_media_path}`."
        )

    return filepath

fetch_audio(audio_url, media_io)

Fetch audio and pair it with a caller-supplied decoder.

Returns a lazy handle: decoding happens on first access (inside the multi-modal processor), so decode errors surface there rather than at the fetch site.

Source code in vllm/multimodal/media/connector.py

def fetch_audio(
    self,
    audio_url: str,
    media_io: MediaIO[tuple[np.ndarray, int | float]],
) -> MediaRef[tuple[np.ndarray, int | float]]:
    """Fetch audio and pair it with a caller-supplied decoder.

    Returns a lazy handle: decoding happens on first access (inside the
    multi-modal processor), so decode errors surface there rather than at
    the fetch site.
    """
    return self.load_from_url(
        audio_url,
        media_io,
        fetch_timeout=envs.VLLM_AUDIO_FETCH_TIMEOUT,
    )

fetch_audio_async(audio_url, media_io) async

Asynchronously fetch audio and pair it with a caller-supplied decoder.

Source code in vllm/multimodal/media/connector.py

async def fetch_audio_async(
    self,
    audio_url: str,
    media_io: MediaIO[tuple[np.ndarray, int | float]],
) -> MediaRef[tuple[np.ndarray, int | float]]:
    """Asynchronously fetch audio and pair it with a caller-supplied decoder."""
    return await self.load_from_url_async(
        audio_url,
        media_io,
        fetch_timeout=envs.VLLM_AUDIO_FETCH_TIMEOUT,
    )

fetch_audio_embedding(data)

Load audio embedding from a URL.

Source code in vllm/multimodal/media/connector.py

def fetch_audio_embedding(
    self,
    data: str,
) -> torch.Tensor:
    """Load audio embedding from a URL."""
    audio_embedding_io = AudioEmbeddingMediaIO()

    return audio_embedding_io.load_base64("", data)

fetch_audio_embedding_async(data) async

Asynchronously load audio embedding from a URL.

Source code in vllm/multimodal/media/connector.py

async def fetch_audio_embedding_async(
    self,
    data: str,
) -> torch.Tensor:
    """Asynchronously load audio embedding from a URL."""
    audio_embedding_io = AudioEmbeddingMediaIO()
    loop = asyncio.get_running_loop()

    return await loop.run_in_executor(
        global_thread_pool, audio_embedding_io.load_base64, "", data
    )

fetch_image(image_url, media_io)

Fetch an image and pair it with a caller-supplied decoder.

Returns a lazy handle: decoding happens on first access (inside the multi-modal processor), so decode errors surface there rather than at the fetch site.

Source code in vllm/multimodal/media/connector.py

def fetch_image(
    self,
    image_url: str,
    media_io: MediaIO[Image.Image],
) -> MediaRef[Image.Image]:
    """Fetch an image and pair it with a caller-supplied decoder.

    Returns a lazy handle: decoding happens on first access (inside the
    multi-modal processor), so decode errors surface there rather than at
    the fetch site.
    """
    return self.load_from_url(
        image_url,
        media_io,
        fetch_timeout=envs.VLLM_IMAGE_FETCH_TIMEOUT,
    )

fetch_image_async(image_url, media_io) async

Asynchronously fetch an image and pair it with a supplied decoder.

Source code in vllm/multimodal/media/connector.py

async def fetch_image_async(
    self,
    image_url: str,
    media_io: MediaIO[Image.Image],
) -> MediaRef[Image.Image]:
    """Asynchronously fetch an image and pair it with a supplied decoder."""
    return await self.load_from_url_async(
        image_url,
        media_io,
        fetch_timeout=envs.VLLM_IMAGE_FETCH_TIMEOUT,
    )

fetch_image_embedding(data)

Load image embedding from a URL.

Source code in vllm/multimodal/media/connector.py

def fetch_image_embedding(
    self,
    data: str,
) -> torch.Tensor:
    """Load image embedding from a URL."""
    image_embedding_io = ImageEmbeddingMediaIO()

    return image_embedding_io.load_base64("", data)

fetch_image_embedding_async(data) async

Asynchronously load image embedding from a URL.

Source code in vllm/multimodal/media/connector.py

async def fetch_image_embedding_async(
    self,
    data: str,
) -> torch.Tensor:
    """Asynchronously load image embedding from a URL."""
    image_embedding_io = ImageEmbeddingMediaIO()
    loop = asyncio.get_running_loop()

    return await loop.run_in_executor(
        global_thread_pool, image_embedding_io.load_base64, "", data
    )

fetch_video(video_url, media_io)

Fetch a video and pair it with a caller-supplied decoder.

Returns a lazy handle: decoding happens on first access (inside the multi-modal processor), so decode errors surface there rather than at the fetch site.

Source code in vllm/multimodal/media/connector.py

def fetch_video(
    self,
    video_url: str,
    media_io: MediaIO[tuple[npt.NDArray, dict[str, Any]]],
) -> MediaRef[tuple[npt.NDArray, dict[str, Any]]]:
    """Fetch a video and pair it with a caller-supplied decoder.

    Returns a lazy handle: decoding happens on first access (inside the
    multi-modal processor), so decode errors surface there rather than at
    the fetch site.
    """
    return self.load_from_url(
        video_url,
        media_io,
        fetch_timeout=envs.VLLM_VIDEO_FETCH_TIMEOUT,
    )

fetch_video_and_audio(video_url, video_io, audio_io)

Fetch video_url once and decode it as both video and audio.

use_audio_in_video reads the audio track out of the video payload, so both refs are built from one download instead of two.

Source code in vllm/multimodal/media/connector.py

def fetch_video_and_audio(
    self,
    video_url: str,
    video_io: MediaIO[_V],
    audio_io: MediaIO[_A],
) -> tuple[MediaRef[_V], MediaRef[_A]]:
    """Fetch `video_url` once and decode it as both video and audio.

    `use_audio_in_video` reads the audio track out of the video payload, so
    both refs are built from one download instead of two.
    """
    video, audio = self._load_refs(
        video_url,
        [video_io, audio_io],
        fetch_timeout=envs.VLLM_VIDEO_FETCH_TIMEOUT,
    )
    return video, audio

fetch_video_and_audio_async(video_url, video_io, audio_io) async

Asynchronous fetch_video_and_audio.

Source code in vllm/multimodal/media/connector.py

async def fetch_video_and_audio_async(
    self,
    video_url: str,
    video_io: MediaIO[_V],
    audio_io: MediaIO[_A],
) -> tuple[MediaRef[_V], MediaRef[_A]]:
    """Asynchronous `fetch_video_and_audio`."""
    video, audio = await self._load_refs_async(
        video_url,
        [video_io, audio_io],
        fetch_timeout=envs.VLLM_VIDEO_FETCH_TIMEOUT,
    )
    return video, audio

fetch_video_async(video_url, media_io) async

Asynchronously fetch a video and pair it with a supplied decoder.

Source code in vllm/multimodal/media/connector.py

async def fetch_video_async(
    self,
    video_url: str,
    media_io: MediaIO[tuple[npt.NDArray, dict[str, Any]]],
) -> MediaRef[tuple[npt.NDArray, dict[str, Any]]]:
    """Asynchronously fetch a video and pair it with a supplied decoder."""
    return await self.load_from_url_async(
        video_url,
        media_io,
        fetch_timeout=envs.VLLM_VIDEO_FETCH_TIMEOUT,
    )

fetch_video_embedding(data)

Load video embedding from a URL.

Source code in vllm/multimodal/media/connector.py

def fetch_video_embedding(
    self,
    data: str,
) -> torch.Tensor:
    """Load video embedding from a URL."""
    video_embedding_io = VideoEmbeddingMediaIO()

    return video_embedding_io.load_base64("", data)

fetch_video_embedding_async(data) async

Asynchronously load video embedding from a URL.

Source code in vllm/multimodal/media/connector.py

async def fetch_video_embedding_async(
    self,
    data: str,
) -> torch.Tensor:
    """Asynchronously load video embedding from a URL."""
    video_embedding_io = VideoEmbeddingMediaIO()
    loop = asyncio.get_running_loop()

    return await loop.run_in_executor(
        global_thread_pool, video_embedding_io.load_base64, "", data
    )

load_from_url(url, media_io, *, fetch_timeout=None)

Fetch url and pair it with a caller-supplied decoder.

Returns a lazy handle: decoding happens on first access (inside the multi-modal processor), so decode errors surface there rather than at the fetch site.

Source code in vllm/multimodal/media/connector.py

def load_from_url(
    self,
    url: str,
    media_io: MediaIO[_M],
    *,
    fetch_timeout: int | None = None,
) -> MediaRef[_M]:
    """Fetch `url` and pair it with a caller-supplied decoder.

    Returns a lazy handle: decoding happens on first access (inside the
    multi-modal processor), so decode errors surface there rather than at
    the fetch site.
    """
    (ref,) = self._load_refs(url, [media_io], fetch_timeout=fetch_timeout)
    return ref

load_from_url_async(url, media_io, *, fetch_timeout=None) async

Asynchronously fetch url and pair it with a caller-supplied decoder.

Source code in vllm/multimodal/media/connector.py

async def load_from_url_async(
    self,
    url: str,
    media_io: MediaIO[_M],
    *,
    fetch_timeout: int | None = None,
) -> MediaRef[_M]:
    """Asynchronously fetch `url` and pair it with a caller-supplied decoder."""
    (ref,) = await self._load_refs_async(
        url, [media_io], fetch_timeout=fetch_timeout
    )
    return ref

MediaIO

Bases: ABC, Generic[_T]

Configuration values can be user-provided either by --media-io-kwargs or by the runtime API field "media_io_kwargs". Ensure proper validation and error handling.

Methods:

• get_decode_spec –

  Return the decode configuration to fold into a ref's cache key.

• get_max_bytes –

  Return the maximum encoded payload size accepted by this media IO.

• load_base64 –

  List of media types:

• load_base64_ref –

  Ref variant of load_base64; the base64 decoding stays eager.

• load_bytes_ref –

  Return a reference that decodes data on first access.

• load_file_ref –

  Ref variant of load_file; the file read stays eager.

• merge_kwargs –

  Merge config-level kwargs and request-level kwargs.

Source code in vllm/multimodal/media/base.py

class MediaIO(ABC, Generic[_T]):
    """Configuration values can be user-provided either by --media-io-kwargs or
    by the runtime API field "media_io_kwargs". Ensure proper validation and
    error handling.
    """

    @classmethod
    def merge_kwargs(
        cls,
        default_kwargs: dict[str, Any] | None,
        runtime_kwargs: dict[str, Any] | None,
    ) -> dict[str, Any]:
        """Merge config-level kwargs and request-level kwargs.

        By default this performs a shallow merge where runtime kwargs override
        keys in default kwargs. Subclasses may override to apply modality-
        specific behavior.
        """
        merged = dict(default_kwargs or {})
        if runtime_kwargs:
            merged.update(runtime_kwargs)
        return merged

    def get_max_bytes(self) -> int | None:
        """Return the maximum encoded payload size accepted by this media IO."""
        return None

    def get_decode_spec(self) -> DecodeSpec | None:
        """Return the decode configuration to fold into a ref's cache key.

        `None` when this IO has no settings that affect the decoded media.
        """
        return None

    @abstractmethod
    def load_bytes(self, data: bytes) -> _T:
        raise NotImplementedError

    def load_bytes_ref(self, data: bytes) -> MediaRef[_T]:
        """Return a reference that decodes `data` on first access.

        Decoding errors surface at the use site (e.g. inside the
        multi-modal processor) instead of at the fetch site.
        """
        return MediaRef(partial(self.load_bytes, data), data, self.get_decode_spec())

    def load_base64_ref(self, media_type: str, data: str) -> MediaRef[_T]:
        """Ref variant of `load_base64`; the base64 decoding stays eager."""
        return self.load_bytes_ref(pybase64.b64decode(data, validate=True))

    def load_file_ref(self, filepath: Path) -> MediaRef[_T]:
        """Ref variant of `load_file`; the file read stays eager."""
        return self.load_bytes_ref(filepath.read_bytes())

    @abstractmethod
    def load_base64(self, media_type: str, data: str) -> _T:
        """List of media types:
        https://www.iana.org/assignments/media-types/media-types.xhtml
        """
        raise NotImplementedError

    @abstractmethod
    def load_file(self, filepath: Path) -> _T:
        raise NotImplementedError

get_decode_spec()

Return the decode configuration to fold into a ref's cache key.

None when this IO has no settings that affect the decoded media.

Source code in vllm/multimodal/media/base.py

def get_decode_spec(self) -> DecodeSpec | None:
    """Return the decode configuration to fold into a ref's cache key.

    `None` when this IO has no settings that affect the decoded media.
    """
    return None

get_max_bytes()

Return the maximum encoded payload size accepted by this media IO.

Source code in vllm/multimodal/media/base.py

def get_max_bytes(self) -> int | None:
    """Return the maximum encoded payload size accepted by this media IO."""
    return None

load_base64(media_type, data) abstractmethod

List of media types: https://www.iana.org/assignments/media-types/media-types.xhtml

Source code in vllm/multimodal/media/base.py

@abstractmethod
def load_base64(self, media_type: str, data: str) -> _T:
    """List of media types:
    https://www.iana.org/assignments/media-types/media-types.xhtml
    """
    raise NotImplementedError

load_base64_ref(media_type, data)

Ref variant of load_base64; the base64 decoding stays eager.

Source code in vllm/multimodal/media/base.py

def load_base64_ref(self, media_type: str, data: str) -> MediaRef[_T]:
    """Ref variant of `load_base64`; the base64 decoding stays eager."""
    return self.load_bytes_ref(pybase64.b64decode(data, validate=True))

load_bytes_ref(data)

Return a reference that decodes data on first access.

Decoding errors surface at the use site (e.g. inside the multi-modal processor) instead of at the fetch site.

Source code in vllm/multimodal/media/base.py

def load_bytes_ref(self, data: bytes) -> MediaRef[_T]:
    """Return a reference that decodes `data` on first access.

    Decoding errors surface at the use site (e.g. inside the
    multi-modal processor) instead of at the fetch site.
    """
    return MediaRef(partial(self.load_bytes, data), data, self.get_decode_spec())

load_file_ref(filepath)

Ref variant of load_file; the file read stays eager.

Source code in vllm/multimodal/media/base.py

def load_file_ref(self, filepath: Path) -> MediaRef[_T]:
    """Ref variant of `load_file`; the file read stays eager."""
    return self.load_bytes_ref(filepath.read_bytes())

merge_kwargs(default_kwargs, runtime_kwargs) classmethod

Merge config-level kwargs and request-level kwargs.

By default this performs a shallow merge where runtime kwargs override keys in default kwargs. Subclasses may override to apply modality- specific behavior.

Source code in vllm/multimodal/media/base.py

@classmethod
def merge_kwargs(
    cls,
    default_kwargs: dict[str, Any] | None,
    runtime_kwargs: dict[str, Any] | None,
) -> dict[str, Any]:
    """Merge config-level kwargs and request-level kwargs.

    By default this performs a shallow merge where runtime kwargs override
    keys in default kwargs. Subclasses may override to apply modality-
    specific behavior.
    """
    merged = dict(default_kwargs or {})
    if runtime_kwargs:
        merged.update(runtime_kwargs)
    return merged

MediaRef

Bases: Generic[_T]

A reference to one piece of fetched media.

This is the single value type for media that vLLM fetched and has not (yet) decoded. It carries three things:

• key: the cache identity, derived from the encoded bytes and spec without decoding, so hashing never pays a decode cost.
• spec: the decode configuration that parameterized or mutated the media, already folded into key.
• decode(): the decoded media, produced at most once and thread-safely.

Methods:

• decode –

  Return the decoded media, decoding at most once.

• fork –

  Borrow this ref through an independently releasable wrapper.

• map –

  Return a ref decoding to transform(self.decode()).

• release –

  Drop this ref's encoded bytes and decode closure.

Attributes:

• data (bytes) –

  The encoded bytes; b"" once release() has run.

Source code in vllm/multimodal/media/base.py

class MediaRef(Generic[_T]):
    """A reference to one piece of fetched media.

    This is the single value type for media that vLLM fetched and has not
    (yet) decoded. It carries three things:

    - `key`: the cache identity, derived from the encoded bytes and `spec`
      without decoding, so hashing never pays a decode cost.
    - `spec`: the decode configuration that parameterized or mutated the
      media, already folded into `key`.
    - `decode()`: the decoded media, produced at most once and thread-safely.
    """

    __slots__ = ("_data", "_decoder", "_lock", "_media", "_key", "_spec")

    def __init__(
        self,
        decoder: Callable[[], _T],
        data: bytes,
        spec: DecodeSpec | None = None,
        *,
        key: bytes | None = None,
    ) -> None:
        super().__init__()

        self._decoder: Callable[[], _T] | None = decoder
        self._data = data
        self._media: Any = _UNDECODED
        self._lock = threading.Lock()

        self._spec = DecodeSpec() if spec is None else spec
        self._key = derive_media_key(data, self.spec) if key is None else key

    @property
    def key(self) -> bytes:
        return self._key

    @property
    def spec(self) -> DecodeSpec:
        return self._spec

    @property
    def data(self) -> bytes:
        """The encoded bytes; `b""` once `release()` has run."""
        return self._data

    @property
    def is_decoded(self) -> bool:
        return self._media is not _UNDECODED

    def decode(self) -> _T:
        """Return the decoded media, decoding at most once.

        Thread-safe. Exceptions from the decoder propagate unchanged;
        wrapping them is the caller's job.

        Raises:
            RuntimeError: If the ref was released before being decoded.

        """
        media = self._media
        if media is _UNDECODED:
            decoder = self._decoder
            if decoder is None:
                raise RuntimeError(
                    "Cannot decode a MediaRef whose bytes were released; "
                    "release() may only run once the item is decoded or is "
                    "known to be a cache hit."
                )
            with self._lock:
                media = self._media
                if media is _UNDECODED:
                    media = self._media = decoder()
        return cast(_T, media)

    def fork(self) -> "MediaRef[_T]":
        """Borrow this ref through an independently releasable wrapper."""
        return MediaRef(self.decode, self._data, self.spec, key=self.key)

    def map(
        self,
        transform: Callable[[_T], _U],
        **settings: Any,
    ) -> "MediaRef[_U]":
        """Return a ref decoding to `transform(self.decode())`.

        Shares this ref's bytes (no copy) so the parse layer can stack
        transforms (e.g. audio resampling) onto the decode step. `settings`
        extend the spec -- and therefore the key -- so the transform's own
        parameters stay part of the cache identity. The new key is derived
        from this ref's key rather than by re-digesting the payload.

        With no settings, the key is preserved: use this only for transforms
        that preserve cache identity. Output-changing transforms must supply
        settings identifying the transform and every output-affecting parameter.
        """
        if not settings:
            spec, key = self.spec, self.key
        else:
            spec = self.spec.extend(**settings)
            key = derive_media_key(self.key, spec)

        return MediaRef(lambda: transform(self.decode()), self._data, spec, key=key)

    def release(self) -> None:
        """Drop this ref's encoded bytes and decode closure.

        Call once hashing is done and the item is either decoded or known to
        be a cache hit, after any in-flight decode has finished. Decoded media
        remains readable; an undecoded ref cannot decode after release.

        Other owners can retain the payload: a mapped ref does not release a
        separately retained parent, and decoded media may contain encoded bytes.
        """
        self._data = b""
        self._decoder = None

    def __repr__(self) -> str:
        state = "decoded" if self.is_decoded else "undecoded"
        return f"<MediaRef {state}, {len(self._data)} bytes>"

    def __reduce__(self):
        # Pickling materializes the media: the decoder is a closure over the
        # encoded bytes and the MediaIO that produced it, neither of which can
        # cross a process boundary. A ref released before being decoded has
        # nothing to materialize, so decode() raises.
        return _media_ref_from_pickle, (self.decode(), self.key, self.spec)

data property

The encoded bytes; b"" once release() has run.

decode()

Return the decoded media, decoding at most once.

Thread-safe. Exceptions from the decoder propagate unchanged; wrapping them is the caller's job.

Raises:

• RuntimeError –

  If the ref was released before being decoded.

Source code in vllm/multimodal/media/base.py

def decode(self) -> _T:
    """Return the decoded media, decoding at most once.

    Thread-safe. Exceptions from the decoder propagate unchanged;
    wrapping them is the caller's job.

    Raises:
        RuntimeError: If the ref was released before being decoded.

    """
    media = self._media
    if media is _UNDECODED:
        decoder = self._decoder
        if decoder is None:
            raise RuntimeError(
                "Cannot decode a MediaRef whose bytes were released; "
                "release() may only run once the item is decoded or is "
                "known to be a cache hit."
            )
        with self._lock:
            media = self._media
            if media is _UNDECODED:
                media = self._media = decoder()
    return cast(_T, media)

fork()

Borrow this ref through an independently releasable wrapper.

Source code in vllm/multimodal/media/base.py

def fork(self) -> "MediaRef[_T]":
    """Borrow this ref through an independently releasable wrapper."""
    return MediaRef(self.decode, self._data, self.spec, key=self.key)

map(transform, **settings)

Return a ref decoding to transform(self.decode()).

Shares this ref's bytes (no copy) so the parse layer can stack transforms (e.g. audio resampling) onto the decode step. settings extend the spec -- and therefore the key -- so the transform's own parameters stay part of the cache identity. The new key is derived from this ref's key rather than by re-digesting the payload.

With no settings, the key is preserved: use this only for transforms that preserve cache identity. Output-changing transforms must supply settings identifying the transform and every output-affecting parameter.

Source code in vllm/multimodal/media/base.py

def map(
    self,
    transform: Callable[[_T], _U],
    **settings: Any,
) -> "MediaRef[_U]":
    """Return a ref decoding to `transform(self.decode())`.

    Shares this ref's bytes (no copy) so the parse layer can stack
    transforms (e.g. audio resampling) onto the decode step. `settings`
    extend the spec -- and therefore the key -- so the transform's own
    parameters stay part of the cache identity. The new key is derived
    from this ref's key rather than by re-digesting the payload.

    With no settings, the key is preserved: use this only for transforms
    that preserve cache identity. Output-changing transforms must supply
    settings identifying the transform and every output-affecting parameter.
    """
    if not settings:
        spec, key = self.spec, self.key
    else:
        spec = self.spec.extend(**settings)
        key = derive_media_key(self.key, spec)

    return MediaRef(lambda: transform(self.decode()), self._data, spec, key=key)

release()

Drop this ref's encoded bytes and decode closure.

Call once hashing is done and the item is either decoded or known to be a cache hit, after any in-flight decode has finished. Decoded media remains readable; an undecoded ref cannot decode after release.

Other owners can retain the payload: a mapped ref does not release a separately retained parent, and decoded media may contain encoded bytes.

Source code in vllm/multimodal/media/base.py

def release(self) -> None:
    """Drop this ref's encoded bytes and decode closure.

    Call once hashing is done and the item is either decoded or known to
    be a cache hit, after any in-flight decode has finished. Decoded media
    remains readable; an undecoded ref cannot decode after release.

    Other owners can retain the payload: a mapped ref does not release a
    separately retained parent, and decoded media may contain encoded bytes.
    """
    self._data = b""
    self._decoder = None

VideoMediaIO

Bases: MediaIO[tuple[DecodedFrames, dict[str, Any]]]

Configuration values can be user-provided either by --media-io-kwargs or by the runtime API field "media_io_kwargs". Ensure proper validation and error handling.

Use --media-io-kwargs '{"video": {"backend": "torchcodec", "device": "cuda"}}' to decode with NVDEC and keep frames on the GPU, which pairs with a device-side multi-modal processor (--mm-processor-device, EPD encode-only instances).

Source code in vllm/multimodal/media/video.py

class VideoMediaIO(MediaIO[tuple[DecodedFrames, dict[str, Any]]]):
    """Configuration values can be user-provided either by --media-io-kwargs or
    by the runtime API field "media_io_kwargs". Ensure proper validation and
    error handling.

    Use --media-io-kwargs '{"video": {"backend": "torchcodec",
    "device": "cuda"}}' to decode with NVDEC and keep frames on the GPU,
    which pairs with a device-side multi-modal processor
    (--mm-processor-device, EPD encode-only instances).
    """

    @classmethod
    def merge_kwargs(
        cls,
        default_kwargs: dict[str, Any] | None,
        runtime_kwargs: dict[str, Any] | None,
    ) -> dict[str, Any]:
        if runtime_kwargs:
            # Decoder GPU memory is reserved from the startup value.
            runtime_kwargs = dict(runtime_kwargs)
            runtime_kwargs.pop("hw_decoders", None)
            runtime_kwargs.pop("pool_size", None)
            # The decode device determines which device the decoded frames
            # (and thus the processor pipeline) live on; startup-only.
            runtime_kwargs.pop("device", None)

            # Block request-level selection of GPU video backends that
            # were not configured (and VRAM-reserved) at startup.
            for key in ("video_backend", "backend"):
                requested = runtime_kwargs.get(key)
                if requested and VIDEO_LOADER_REGISTRY.backend_requires_gpu(requested):
                    static_val = (default_kwargs or {}).get(key)
                    if static_val != requested:
                        logger.warning_once(
                            "Stripping request-level %s=%r: GPU video "
                            "backend not configured at startup.",
                            key,
                            requested,
                        )
                        runtime_kwargs = {
                            k: v for k, v in runtime_kwargs.items() if k != key
                        }

        merged = super().merge_kwargs(default_kwargs, runtime_kwargs)
        # fps and num_frames interact with each other, so if either is
        # overridden at request time, wipe the other from defaults to
        # avoid unintuitive cross-field interactions.
        if runtime_kwargs:
            if "num_frames" in runtime_kwargs and "fps" not in runtime_kwargs:
                merged.pop("fps", None)
            elif "fps" in runtime_kwargs and "num_frames" not in runtime_kwargs:
                merged.pop("num_frames", None)

            requested_backend = runtime_kwargs.get("backend")
            static_backend = (default_kwargs or {}).get("backend")
            if requested_backend and requested_backend != static_backend:
                for name in incompatible_backend_options(
                    static_backend, requested_backend
                ):
                    merged.pop(name, None)
        return merged

    def __init__(
        self,
        image_io: ImageMediaIO,
        num_frames: int = 32,
        **kwargs,
    ) -> None:
        super().__init__()

        self.image_io = image_io
        self.num_frames = num_frames
        # `kwargs` contains custom arguments from
        # --media-io-kwargs for this modality, merged with
        # per-request runtime media_io_kwargs via merge_kwargs().
        # They can be passed to the underlying
        # media loaders (e.g. custom implementations)
        # for flexible control.

        # Allow per-request override of video backend via kwargs.
        # This enables users to specify a different backend than the
        # global VLLM_VIDEO_LOADER_BACKEND env var, e.g.:
        #   --media-io-kwargs '{"video": {"video_backend": "torchcodec"}}'
        video_loader_backend = (
            kwargs.pop("video_backend", None) or envs.VLLM_VIDEO_LOADER_BACKEND
        )
        self.kwargs = kwargs
        self.video_loader_backend = video_loader_backend
        self.video_loader = VIDEO_LOADER_REGISTRY.load(video_loader_backend)

    def get_decode_spec(self) -> DecodeSpec:
        return DecodeSpec(
            {
                **self.kwargs,
                "num_frames": self.num_frames,
                "video_backend": self.video_loader_backend,
            }
        )

    def load_bytes(self, data: bytes) -> tuple[DecodedFrames, dict[str, Any]]:
        return self.video_loader.load_bytes(
            data, num_frames=self.num_frames, **self.kwargs
        )

    def _prepare_jpeg_sequence(
        self, data: str
    ) -> tuple[Callable[[], npt.NDArray], dict[str, Any]]:
        """Split a base64 jpeg frame sequence and validate its arguments.

        Splitting and argument validation (num_frames/frames_indices/
        total_num_frames/duration) run eagerly; the per-frame decode and
        np.stack are deferred into the returned callable.
        """
        if self.num_frames > 0:
            frame_parts = data.split(",", self.num_frames)[: self.num_frames]
        elif self.num_frames == 0:
            raise ValueError("num_frames must be greater than 0 or -1")
        else:
            frame_parts = data.split(",")

        total = len(frame_parts)
        fps = float(self.kwargs.get("fps", 1))

        # validate and extract frames_indices
        frames_indices = self.kwargs.get("frames_indices")
        if frames_indices is not None:
            if not (
                isinstance(frames_indices, list)
                and all(isinstance(i, int) for i in frames_indices)
            ):
                raise ValueError("frames_indices must be a list of integers")
            if len(frames_indices) != total:
                raise ValueError(
                    f"frames_indices length ({len(frames_indices)}) must "
                    f"match number of frames sent ({total})"
                )
        else:
            frames_indices = list(range(total))

        # validate and extract total_num_frames
        total_num_frames = self.kwargs.get("total_num_frames", total)
        if not isinstance(total_num_frames, int) or total_num_frames < 1:
            raise ValueError("total_num_frames must be a positive integer")
        if total_num_frames < total:
            raise ValueError(
                f"total_num_frames ({total_num_frames}) must be >= "
                f"number of frames sent ({total})"
            )

        # validate and extract duration
        duration = self.kwargs.get("duration")
        if duration is not None:
            if not isinstance(duration, (int, float)) or duration < 0:
                raise ValueError("duration must be a non-negative number")
        else:
            duration = total_num_frames / fps if fps > 0 else 0.0

        metadata = {
            "total_num_frames": total_num_frames,
            "fps": fps,
            "duration": duration,
            "video_backend": "jpeg_sequence",
            "frames_indices": frames_indices,
            "do_sample_frames": self.kwargs.get("do_sample_frames", False),
        }

        def decode_frames() -> npt.NDArray:
            load_frame = partial(
                self.image_io.load_base64,
                "image/jpeg",
            )
            return np.stack(
                [np.asarray(load_frame(frame_data)) for frame_data in frame_parts]
            )

        return decode_frames, metadata

    def load_base64(
        self, media_type: str, data: str
    ) -> tuple[DecodedFrames, dict[str, Any]]:
        if media_type.lower() == "video/jpeg":
            decode_frames, metadata = self._prepare_jpeg_sequence(data)
            return decode_frames(), metadata

        return self.load_bytes(pybase64.b64decode(data, validate=True))

    def load_base64_ref(
        self, media_type: str, data: str
    ) -> MediaRef[tuple[DecodedFrames, dict[str, Any]]]:
        if media_type.lower() == "video/jpeg":
            decode_frames, metadata = self._prepare_jpeg_sequence(data)
            spec = self.get_decode_spec().extend(
                video_backend="jpeg_sequence",
                image_decode=dict(self.image_io.get_decode_spec().settings),
            )
            return MediaRef(
                lambda: (decode_frames(), metadata),
                data.encode(),
                spec,
            )
        return super().load_base64_ref(media_type, data)

    def load_file(self, filepath: Path) -> tuple[DecodedFrames, dict[str, Any]]:
        return self.load_bytes(filepath.read_bytes())

    def encode_base64(
        self,
        media: npt.NDArray,
        *,
        video_format: str = "JPEG",
    ) -> str:
        video = media

        if video_format == "JPEG":
            encode_frame = partial(
                self.image_io.encode_base64,
                image_format=video_format,
            )

            return ",".join(encode_frame(Image.fromarray(frame)) for frame in video)

        msg = "Only JPEG format is supported for now."
        raise NotImplementedError(msg)

_prepare_jpeg_sequence(data)

Split a base64 jpeg frame sequence and validate its arguments.

Splitting and argument validation (num_frames/frames_indices/ total_num_frames/duration) run eagerly; the per-frame decode and np.stack are deferred into the returned callable.

Source code in vllm/multimodal/media/video.py

def _prepare_jpeg_sequence(
    self, data: str
) -> tuple[Callable[[], npt.NDArray], dict[str, Any]]:
    """Split a base64 jpeg frame sequence and validate its arguments.

    Splitting and argument validation (num_frames/frames_indices/
    total_num_frames/duration) run eagerly; the per-frame decode and
    np.stack are deferred into the returned callable.
    """
    if self.num_frames > 0:
        frame_parts = data.split(",", self.num_frames)[: self.num_frames]
    elif self.num_frames == 0:
        raise ValueError("num_frames must be greater than 0 or -1")
    else:
        frame_parts = data.split(",")

    total = len(frame_parts)
    fps = float(self.kwargs.get("fps", 1))

    # validate and extract frames_indices
    frames_indices = self.kwargs.get("frames_indices")
    if frames_indices is not None:
        if not (
            isinstance(frames_indices, list)
            and all(isinstance(i, int) for i in frames_indices)
        ):
            raise ValueError("frames_indices must be a list of integers")
        if len(frames_indices) != total:
            raise ValueError(
                f"frames_indices length ({len(frames_indices)}) must "
                f"match number of frames sent ({total})"
            )
    else:
        frames_indices = list(range(total))

    # validate and extract total_num_frames
    total_num_frames = self.kwargs.get("total_num_frames", total)
    if not isinstance(total_num_frames, int) or total_num_frames < 1:
        raise ValueError("total_num_frames must be a positive integer")
    if total_num_frames < total:
        raise ValueError(
            f"total_num_frames ({total_num_frames}) must be >= "
            f"number of frames sent ({total})"
        )

    # validate and extract duration
    duration = self.kwargs.get("duration")
    if duration is not None:
        if not isinstance(duration, (int, float)) or duration < 0:
            raise ValueError("duration must be a non-negative number")
    else:
        duration = total_num_frames / fps if fps > 0 else 0.0

    metadata = {
        "total_num_frames": total_num_frames,
        "fps": fps,
        "duration": duration,
        "video_backend": "jpeg_sequence",
        "frames_indices": frames_indices,
        "do_sample_frames": self.kwargs.get("do_sample_frames", False),
    }

    def decode_frames() -> npt.NDArray:
        load_frame = partial(
            self.image_io.load_base64,
            "image/jpeg",
        )
        return np.stack(
            [np.asarray(load_frame(frame_data)) for frame_data in frame_parts]
        )

    return decode_frames, metadata