Source code for ralph.mcp.multimodal._manifest_entry

"""ManifestEntry — an entry in the session-scoped multimodal manifest."""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass, field

ByteLoader = Callable[[], bytes | None]


[docs] @dataclass class ManifestEntry: """An entry in the session-scoped multimodal manifest.""" artifact_id: str uri: str mime_type: str title: str modality: str identity_key: str = "" cache_path: str = "" source_path: str = "" source_uri: str = "" _raw_bytes: bytes | None = field(default=None, repr=False) _byte_loader: ByteLoader | None = field(default=None, repr=False, compare=False) @property def raw_bytes(self) -> bytes: """Return artifact bytes, rehydrating from the loader when needed.""" return self.load_bytes() or b""
[docs] def load_bytes(self) -> bytes | None: """Return artifact bytes from memory or a backing replay source.""" if self._raw_bytes is not None: return self._raw_bytes if self._byte_loader is None: return None return self._byte_loader()
[docs] def set_replay_source( self, *, cache_path: str = "", source_path: str = "", source_uri: str = "", byte_loader: ByteLoader | None = None, retain_raw_bytes: bool = False, ) -> None: """Attach a durable replay source and optionally release in-memory bytes.""" if cache_path: self.cache_path = cache_path if source_path: self.source_path = source_path if source_uri: self.source_uri = source_uri if byte_loader is not None: self._byte_loader = byte_loader if not retain_raw_bytes: self._raw_bytes = None
[docs] def resource_list_entry(self) -> dict[str, object]: """Return the entry shape for a resources/list response.""" return { "uri": self.uri, "name": self.title, "description": f"{self.modality} artifact: {self.title}", "mimeType": self.mime_type, }
__all__ = ["ByteLoader", "ManifestEntry"]