# loopy.multimodal

## `loopy.multimodal.MultiModalMessage` (class)

A message with text and media content.

```python
@dataclass
class MultiModalMessage:
    """A message with text and media content."""

    text: str
    media: list[MediaContent] = field(default_factory=list)

    @property
    def has_media(self) -> bool:
        return len(self.media) > 0

    @property
    def images(self) -> list[MediaContent]:
        return [m for m in self.media if m.type == MediaType.IMAGE]

    @property
    def audio(self) -> list[MediaContent]:
        return [m for m in self.media if m.type == MediaType.AUDIO]

    def to_openai(self) -> list[dict[str, Any]]:
        """Convert to OpenAI multi-modal format."""
        content: list[dict[str, Any]] = []

        # Add images first
        for media in self.media:
            if media.type == MediaType.IMAGE:
                content.append(media.to_openai())

        # Add text
        if self.text:
            content.append({"type": "text", "text": self.text})

        return content

    def to_anthropic(self) -> list[dict[str, Any]]:
        """Convert to Anthropic multi-modal format."""
        content: list[dict[str, Any]] = []

        for media in self.media:
            content.append(media.to_anthropic())

        if self.text:
            content.append({"type": "text", "text": self.text})

        return content
```

## `loopy.multimodal.MultiModalBuilder` (class)

Build multi-modal messages easily.

Example:
    msg = (MultiModalBuilder()
        .text("What's in this image?")
        .image("photo.jpg")
        .image("https://example.com/chart.png")
        .build())

```python
class MultiModalBuilder:
    """
    Build multi-modal messages easily.

    Example:
        msg = (MultiModalBuilder()
            .text("What's in this image?")
            .image("photo.jpg")
            .image("https://example.com/chart.png")
            .build())
    """

    def __init__(self):
        self._text = ""
        self._media: list[MediaContent] = []

    def text(self, content: str) -> MultiModalBuilder:
        """Add text content."""
        self._text = content
        return self

    def image(self, source: str) -> MultiModalBuilder:
        """Add image from file or URL."""
        if source.startswith(("http://", "https://")):
            self._media.append(MediaContent.from_url(source, MediaType.IMAGE))
        else:
            self._media.append(MediaContent.from_file(source))
        return self

    def audio(self, source: str) -> MultiModalBuilder:
        """Add audio from file or URL."""
        if source.startswith(("http://", "https://")):
            self._media.append(MediaContent.from_url(source, MediaType.AUDIO))
        else:
            self._media.append(MediaContent.from_file(source))
        return self

    def file(self, path: str) -> MultiModalBuilder:
        """Add any file as media."""
        self._media.append(MediaContent.from_file(path))
        return self

    def build(self) -> MultiModalMessage:
        """Build the multi-modal message."""
        return MultiModalMessage(text=self._text, media=self._media)
```

## `loopy.multimodal.MediaContent` (class)

A piece of media content.

```python
@dataclass
class MediaContent:
    """A piece of media content."""

    type: MediaType
    data: str  # base64 or URL
    mime_type: str = ""
    metadata: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_file(cls, path: str) -> MediaContent:
        """Load media from file."""
        file_path = Path(path)
        if not file_path.exists():
            raise FileNotFoundError(f"Media file not found: {path}")

        suffix = file_path.suffix.lower()
        mime_map = {
            ".png": "image/png",
            ".jpg": "image/jpeg",
            ".jpeg": "image/jpeg",
            ".webp": "image/webp",
            ".gif": "image/gif",
            ".mp3": "audio/mpeg",
            ".wav": "audio/wav",
            ".mp4": "video/mp4",
            ".pdf": "application/pdf",
        }

        mime_type = mime_map.get(suffix, "application/octet-stream")
        media_type = (
            MediaType.IMAGE
            if mime_type.startswith("image/")
            else MediaType.AUDIO
            if mime_type.startswith("audio/")
            else MediaType.VIDEO
            if mime_type.startswith("video/")
            else MediaType.DOCUMENT
        )

        data = base64.b64encode(file_path.read_bytes()).decode()
        return cls(
            type=media_type,
            data=data,
            mime_type=mime_type,
            metadata={"filename": file_path.name, "size": file_path.stat().st_size},
        )

    @classmethod
    def from_url(
        cls,
        url: str,
        media_type: MediaType = MediaType.IMAGE,
        *,
        allow_private: bool = True,
    ) -> MediaContent:
        """Create media from URL (no download).

        The URL is passed through to the provider, which fetches it
        server-side — so keep *allow_private* True only for operator-supplied
        URLs. Set False to reject internal/loopback destinations when the URL
        can come from model output or untrusted content.

        Raises:
            ValueError: If the URL scheme is not http(s) or it has no host.
        """
        validate_outbound_url(url, allow_private=allow_private)
        return cls(
            type=media_type,
            data=url,
            metadata={"url": url},
        )

    def to_openai(self) -> dict[str, Any]:
        """Convert to OpenAI vision API format.

        Returns a dict suitable for use in the ``content`` array
        of an OpenAI chat completion request.
        """
        if self.data.startswith(("http://", "https://")):
            return {"type": "image_url", "image_url": {"url": self.data}}
        return {
            "type": "image_url",
            "image_url": {"url": f"data:{self.mime_type};base64,{self.data}"},
        }

    def to_anthropic(self) -> dict[str, Any]:
        """Convert to Anthropic Messages API format.

        Returns a dict suitable for use in the ``content`` array
        of an Anthropic message request.
        """
        if self.data.startswith(("http://", "https://")):
            return {"type": "image", "source": {"type": "url", "url": self.data}}
        return {
            "type": "image",
            "source": {"type": "base64", "media_type": self.mime_type, "data": self.data},
        }
```

## `loopy.multimodal.MediaType` (class)

Supported media types.

```python
class MediaType(str, Enum):
    """Supported media types."""

    IMAGE = "image"
    AUDIO = "audio"
    VIDEO = "video"
    DOCUMENT = "document"
```

## `loopy.multimodal.ImageFormat` (class)

Image formats.

```python
class ImageFormat(str, Enum):
    """Image formats."""

    PNG = "png"
    JPEG = "jpeg"
    WEBP = "webp"
    GIF = "gif"
```

## `loopy.multimodal.RealtimeSession` (class)

v0.7.10 - Async iterator over realtime events from a transport.

Lightweight adapter that consumes raw WebSocket frames from any
:class:`RealtimeTransport` and yields normalized
:class:`RealtimeEvent` instances. Useful for voice-first agent
loops, transcription-only bots, and OpenAI Realtime clients.

Example `` `` ``
    from loopy.multimodal import RealtimeSession, RealtimeTransport

    class MyOpenAITransport:
        async def send(self, payload): ...
        async def recv(self): ...
        async def close(self): ...

    async with RealtimeSession(MyOpenAITransport()) as session:
        await session.send({"type": "session.update", "session": {...}})
        async for event in session:
            if event.type == RealtimeEventType.TRANSCRIPT_DELTA:
                print(event.transcript, end="", flush=True)
```

The ``websockets`` library itself is NOT a dependency - users wire
in their preferred WebSocket client via the ``transport=`` argument.
This keeps the loopy-agent core dependency surface at just
``httpx`` + ``pydantic`` while leaving the door open for voice /
realtime use cases.

```python
class RealtimeSession:
    """v0.7.10 - Async iterator over realtime events from a transport.

    Lightweight adapter that consumes raw WebSocket frames from any
    :class:`RealtimeTransport` and yields normalized
    :class:`RealtimeEvent` instances. Useful for voice-first agent
    loops, transcription-only bots, and OpenAI Realtime clients.

    Example `` `` ``
        from loopy.multimodal import RealtimeSession, RealtimeTransport

        class MyOpenAITransport:
            async def send(self, payload): ...
            async def recv(self): ...
            async def close(self): ...

        async with RealtimeSession(MyOpenAITransport()) as session:
            await session.send({"type": "session.update", "session": {...}})
            async for event in session:
                if event.type == RealtimeEventType.TRANSCRIPT_DELTA:
                    print(event.transcript, end="", flush=True)
    ```

    The ``websockets`` library itself is NOT a dependency - users wire
    in their preferred WebSocket client via the ``transport=`` argument.
    This keeps the loopy-agent core dependency surface at just
    ``httpx`` + ``pydantic`` while leaving the door open for voice /
    realtime use cases.
    """

    __slots__ = ("_transport", "_closed", "_events", "_pump_task")

    def __init__(self, transport: RealtimeTransport) -> None:
        self._transport = transport
        self._closed = False
        self._events: asyncio.Queue[RealtimeEvent] = asyncio.Queue()
        self._pump_task: asyncio.Task[None] | None = None

    async def __aenter__(self) -> RealtimeSession:
        # Start the background pump that drains the transport into the queue.
        self._pump_task = asyncio.create_task(self._pump())
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        await self.close()

    async def send(self, payload: dict[str, Any]) -> None:
        """Send a payload upstream through the transport."""
        if self._closed:
            raise RuntimeError("RealtimeSession is closed")
        await self._transport.send(payload)

    async def close(self) -> None:
        """Close the transport and mark the session done."""
        if self._closed:
            return
        self._closed = True
        # Cancel pump first so it cannot put more events after we close.
        if self._pump_task is not None and not self._pump_task.done():
            self._pump_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):  # noqa: BLE001
                await self._pump_task
        try:
            await self._transport.close()
        finally:
            # Always emit a CLOSED event so consumers see the terminal.
            with contextlib.suppress(RuntimeError):
                # Queue may already be closed if pump was cancelled.
                await self._events.put(RealtimeEvent(type=RealtimeEventType.CLOSED, data={}))

    def __aiter__(self) -> RealtimeSession:
        return self

    async def __anext__(self) -> RealtimeEvent:
        if self._closed and self._events.empty():
            raise StopAsyncIteration
        try:
            return await asyncio.wait_for(self._events.get(), timeout=0.05)
        except asyncio.TimeoutError:
            # Background pump will refill the queue. Loop again.
            return await self.__anext__()

    async def _pump(self) -> None:
        """Internal: drain the transport until closed, normalising events.

        The pump does NOT call :meth:`close` - that would re-enter
        ``close`` which is awaiting this task. It simply stops putting
        new events and exits; ``close`` is responsible for cancelling
        this task and emitting the final ``CLOSED`` event.
        """
        try:
            while not self._closed:
                try:
                    payload = await self._transport.recv()
                except Exception as e:  # noqa: BLE001 - transport errors become ERROR events
                    logger.warning("RealtimeSession transport error: %s", e)
                    break
                if payload is None:
                    break
                try:
                    await self._events.put(_build_event(payload))
                except Exception:  # noqa: BLE001
                    # Queue may be closed during shutdown.
                    break
        except asyncio.CancelledError:
            pass
```

## `loopy.multimodal.RealtimeEvent` (class)

v0.7.10 - Normalized realtime event surfaced to the agent loop.

```python
@dataclass
class RealtimeEvent:
    """v0.7.10 - Normalized realtime event surfaced to the agent loop."""

    type: RealtimeEventType
    data: dict[str, Any] = field(default_factory=dict)
    timestamp: float = field(default_factory=lambda: __import__("time").time())

    @property
    def transcript(self) -> str:
        """Concatenated transcript text (from ``transcript.delta`` events)."""
        return str(self.data.get("transcript", ""))

    @property
    def audio_bytes(self) -> bytes:
        """Raw audio payload (from ``audio.delta`` events)."""
        return self.data.get("audio", b"")
```

## `loopy.multimodal.RealtimeEventType` (class)

v0.7.10 - Subset of OpenAI Realtime event types we support natively.

```python
class RealtimeEventType(str, Enum):
    """v0.7.10 - Subset of OpenAI Realtime event types we support natively."""

    SESSION_CREATED = "session.created"
    TRANSCRIPT_DELTA = "transcript.delta"
    TRANSCRIPT_DONE = "transcript.done"
    AUDIO_DELTA = "audio.delta"
    TOOL_CALL = "tool.call"
    ERROR = "error"
    CLOSED = "closed"
```

## `loopy.multimodal.RealtimeTransport` (class)

v0.7.10 - Pluggable WebSocket transport for ``RealtimeSession``.

Any object exposing ``async send(payload)``, ``async recv()``, and
``async close()`` can drive a ``RealtimeSession``. Loopy ships no
concrete WebSocket implementation - users wire in their preferred
client (``websockets``, ``openai-agents`` realtime client, etc.).

```python
class RealtimeTransport(Protocol):
    """v0.7.10 - Pluggable WebSocket transport for ``RealtimeSession``.

    Any object exposing ``async send(payload)``, ``async recv()``, and
    ``async close()`` can drive a ``RealtimeSession``. Loopy ships no
    concrete WebSocket implementation - users wire in their preferred
    client (``websockets``, ``openai-agents`` realtime client, etc.).
    """

    async def send(self, payload: dict[str, Any]) -> None: ...

    async def recv(self) -> dict[str, Any] | None: ...

    async def close(self) -> None: ...
```
