Implement multimodal inputs

Author: anbuzin

examples/samples/multimodal.py

@@ -0,0 +1,40 @@
+"""Multimodal input example: send an image URL to the model.
+
+Usage:
+    uv run examples/samples/multimodal.py
+"""
+
+import asyncio
+
+import vercel_ai_sdk as ai
+
+IMAGE_URL = "https://4kwallpapers.com/images/wallpapers/hatsune-miku-3840x2160-15479.jpg"
+
+
+async def agent(llm: ai.LanguageModel, user_query: str) -> ai.StreamResult:
+    return await ai.stream_loop(
+        llm,
+        messages=[
+            ai.Message(
+                role="user",
+                parts=[
+                    ai.TextPart(text=user_query),
+                    ai.FilePart.from_url(IMAGE_URL),
+                ],
+            )
+        ],
+        tools=[],
+    )
+
+
+async def main() -> None:
+    llm = ai.ai_gateway.GatewayModel(model="anthropic/claude-opus-4.6")
+
+    async for msg in ai.run(agent, llm, "What's in this image? Be concise."):
+        if msg.text_delta:
+            print(msg.text_delta, end="", flush=True)
+    print()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/vercel_ai_sdk/__init__.py

@@ -6,6 +6,7 @@
 
 # Re-export core types
 from .core.messages import (
+    FilePart,
     HookPart,
     Message,
     Part,
@@ -40,6 +41,7 @@
     "ToolPart",
     "ToolDelta",
     "ReasoningPart",
+    "FilePart",
     "ToolLike",
     "ToolSchema",
     "Tool",

src/vercel_ai_sdk/ai_gateway/__init__.py

@@ -106,7 +106,7 @@ async def stream_events(
         output_type: type[pydantic.BaseModel] | None = None,
     ) -> AsyncGenerator[core.llm.StreamEvent]:
         """Yield ``StreamEvent`` objects from the gateway SSE stream."""
-        body = protocol_.build_request_body(
+        body = await protocol_.build_request_body(
             messages,
             tools=tools,
             output_type=output_type,

src/vercel_ai_sdk/ai_gateway/protocol.py

@@ -24,7 +24,30 @@
 # ---------------------------------------------------------------------------
 
 
-def messages_to_v3_prompt(
+async def _file_part_to_v3(part: core.messages.FilePart) -> dict[str, Any]:
+    """Convert an internal :class:`FilePart` to a v3 ``file`` content part.
+
+    Binary data is converted to a ``data:`` URL for JSON transport (matching
+    the JS SDK gateway's ``maybeEncodeFileParts``).  HTTP(S) URLs are
+    downloaded and converted to ``data:`` URLs because the gateway wire
+    format does not accept raw HTTP URLs for file content.
+    """
+    data = part.data
+    if isinstance(data, str) and core.media.data.is_downloadable_url(data):
+        downloaded, _ = await core.media.download.download(data)
+        data = downloaded
+
+    entry: dict[str, Any] = {
+        "type": "file",
+        "mediaType": part.media_type,
+        "data": core.media.data.data_to_data_url(data, part.media_type),
+    }
+    if part.filename is not None:
+        entry["filename"] = part.filename
+    return entry
+
+
+async def messages_to_v3_prompt(
     messages: list[core.messages.Message],
 ) -> list[dict[str, Any]]:
     """Convert internal ``Message`` list to ``LanguageModelV3Prompt``.
@@ -55,11 +78,12 @@ def messages_to_v3_prompt(
                 result.append({"role": "system", "content": text})
 
             case "user":
-                content: list[dict[str, Any]] = [
-                    {"type": "text", "text": p.text}
-                    for p in msg.parts
-                    if isinstance(p, core.messages.TextPart)
-                ]
+                content: list[dict[str, Any]] = []
+                for p in msg.parts:
+                    if isinstance(p, core.messages.TextPart):
+                        content.append({"type": "text", "text": p.text})
+                    elif isinstance(p, core.messages.FilePart):
+                        content.append(await _file_part_to_v3(p))
                 result.append({"role": "user", "content": content})
 
             case "assistant":
@@ -135,15 +159,15 @@ def messages_to_v3_prompt(
 # ---------------------------------------------------------------------------
 
 
-def build_request_body(
+async def build_request_body(
     messages: list[core.messages.Message],
     tools: Sequence[core.tools.ToolLike] | None = None,
     output_type: type[Any] | None = None,
     provider_options: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Build the full ``LanguageModelV3CallOptions`` request body."""
     body: dict[str, Any] = {
-        "prompt": messages_to_v3_prompt(messages),
+        "prompt": await messages_to_v3_prompt(messages),
     }
     if tools:
         body["tools"] = [

src/vercel_ai_sdk/ai_sdk_ui/adapter.py

@@ -467,9 +467,17 @@ def to_messages(
                         )
                         resolved_any_approval = True
 
+                case ui_message.UIFilePart() as fp:
+                    internal_parts.append(
+                        core.messages.FilePart(
+                            data=fp.url,
+                            media_type=fp.media_type,
+                            filename=fp.filename,
+                        )
+                    )
+
                 case (
                     ui_message.UIStepStartPart()
-                    | ui_message.UIFilePart()
                     | ui_message.UISourceUrlPart()
                     | ui_message.UISourceDocumentPart()
                 ):

src/vercel_ai_sdk/anthropic/__init__.py

@@ -23,7 +23,73 @@ def _tools_to_anthropic(tools: Sequence[core.tools.ToolLike]) -> list[dict[str,
     ]
 
 
-def _messages_to_anthropic(
+def _file_part_to_anthropic(part: core.messages.FilePart) -> dict[str, Any]:
+    """Convert a :class:`FilePart` to an Anthropic content block.
+
+    * ``image/*`` → ``{"type": "image", "source": ...}``
+    * ``application/pdf`` → ``{"type": "document", "source": ...}``
+    * ``text/plain`` → ``{"type": "document", "source": {"type": "text", ...}}``
+    * anything else → ``ValueError``
+    """
+    mt = part.media_type
+
+    if mt.startswith("image/"):
+        media_type = "image/jpeg" if mt == "image/*" else mt
+        if isinstance(part.data, str) and core.media.data.is_url(part.data):
+            return {
+                "type": "image",
+                "source": {"type": "url", "url": part.data},
+            }
+        return {
+            "type": "image",
+            "source": {
+                "type": "base64",
+                "media_type": media_type,
+                "data": core.media.data.data_to_base64(part.data),
+            },
+        }
+
+    if mt == "application/pdf":
+        if isinstance(part.data, str) and core.media.data.is_url(part.data):
+            return {
+                "type": "document",
+                "source": {"type": "url", "url": part.data},
+            }
+        return {
+            "type": "document",
+            "source": {
+                "type": "base64",
+                "media_type": "application/pdf",
+                "data": core.media.data.data_to_base64(part.data),
+            },
+        }
+
+    if mt == "text/plain":
+        # Anthropic accepts text documents with source.type="text"
+        if isinstance(part.data, bytes):
+            text_data = part.data.decode("utf-8")
+        elif core.media.data.is_url(part.data):
+            return {
+                "type": "document",
+                "source": {"type": "url", "url": part.data},
+            }
+        else:
+            import base64 as _b64
+
+            text_data = _b64.b64decode(part.data).decode("utf-8")
+        return {
+            "type": "document",
+            "source": {
+                "type": "text",
+                "media_type": "text/plain",
+                "data": text_data,
+            },
+        }
+
+    raise ValueError(f"Unsupported media type for Anthropic: {mt}")
+
+
+async def _messages_to_anthropic(
     messages: list[core.messages.Message],
 ) -> tuple[str | None, list[dict[str, Any]]]:
     """Convert internal messages to Anthropic API format.
@@ -89,12 +155,21 @@ def _messages_to_anthropic(
                 result.append({"role": "assistant", "content": content})
             if tool_results:
                 result.append({"role": "user", "content": tool_results})
-        else:
-            # User messages
-            content_text = "".join(
-                p.text for p in msg.parts if isinstance(p, core.messages.TextPart)
-            )
-            result.append({"role": "user", "content": content_text})
+        elif msg.role == "user":
+            has_files = any(isinstance(p, core.messages.FilePart) for p in msg.parts)
+            if not has_files:
+                content_text = "".join(
+                    p.text for p in msg.parts if isinstance(p, core.messages.TextPart)
+                )
+                result.append({"role": "user", "content": content_text})
+            else:
+                user_content: list[dict[str, Any]] = []
+                for p in msg.parts:
+                    if isinstance(p, core.messages.TextPart):
+                        user_content.append({"type": "text", "text": p.text})
+                    elif isinstance(p, core.messages.FilePart):
+                        user_content.append(_file_part_to_anthropic(p))
+                result.append({"role": "user", "content": user_content})
 
     # Merge consecutive same-role messages (e.g. synthetic user(tool_result)
     # followed by a real user message).
@@ -162,7 +237,7 @@ async def stream_events(
         output_type: type[pydantic.BaseModel] | None = None,
     ) -> AsyncGenerator[core.llm.StreamEvent]:
         """Yield raw stream events from Anthropic API."""
-        system_prompt, anthropic_messages = _messages_to_anthropic(messages)
+        system_prompt, anthropic_messages = await _messages_to_anthropic(messages)
         anthropic_tools = _tools_to_anthropic(tools) if tools else None
 
         kwargs: dict[str, Any] = {

src/vercel_ai_sdk/core/__init__.py

@@ -1,3 +1,19 @@
-from . import hooks, llm, messages, runtime, telemetry, tools
+from . import (
+    hooks,
+    llm,
+    media,
+    messages,
+    runtime,
+    telemetry,
+    tools,
+)
 
-__all__ = ["messages", "tools", "runtime", "hooks", "llm", "telemetry"]
+__all__ = [
+    "hooks",
+    "llm",
+    "media",
+    "messages",
+    "runtime",
+    "telemetry",
+    "tools",
+]

src/vercel_ai_sdk/core/media/__init__.py

@@ -0,0 +1,5 @@
+"""Media handling: detection, download, and data-format helpers."""
+
+from . import data, detect_media_type, download
+
+__all__ = ["data", "detect_media_type", "download"]

src/vercel_ai_sdk/core/media/data.py

@@ -0,0 +1,100 @@
+"""Data-format helpers for multimodal content.
+
+URL detection, ``data:`` URL parsing, base-64 encoding/decoding, and
+media-type inference utilities used by :class:`~vercel_ai_sdk.core.messages.FilePart`
+and the provider converters.
+"""
+
+from __future__ import annotations
+
+import base64
+import mimetypes
+
+# -- URL helpers -----------------------------------------------------------
+
+
+def is_url(data: str) -> bool:
+    """Return True if *data* looks like a URL rather than raw base-64."""
+    return data.startswith(("http://", "https://", "data:"))
+
+
+def is_downloadable_url(data: str) -> bool:
+    """Return True if *data* is an ``http(s)://`` URL that can be fetched."""
+    return data.startswith(("http://", "https://"))
+
+
+def split_data_url(url: str) -> tuple[str | None, str | None]:
+    """Parse a ``data:`` URL into ``(media_type, base64_content)``.
+
+    Returns ``(None, None)`` if the input is not a valid ``data:`` URL.
+
+    Example::
+
+        >>> split_data_url("data:image/png;base64,iVBOR...")
+        ("image/png", "iVBOR...")
+    """
+    if not url.startswith("data:"):
+        return None, None
+    try:
+        header, b64_content = url.split(",", 1)
+        # header = "data:image/png;base64"
+        mt = header.split(";")[0].split(":", 1)[1]
+        return (mt or None), (b64_content or None)
+    except (ValueError, IndexError):
+        return None, None
+
+
+# -- encoding helpers ------------------------------------------------------
+
+
+def data_to_base64(data: str | bytes) -> str:
+    """Ensure *data* is a base-64 encoded string.
+
+    * ``bytes`` -> base-64 encoded.
+    * ``str`` that is a ``data:`` URL -> base-64 content extracted.
+    * ``str`` that is an ``http(s)://`` URL -> returned as-is (caller
+      must handle).
+    * ``str`` that is not a URL -> assumed to already be base-64.
+    """
+    if isinstance(data, bytes):
+        return base64.b64encode(data).decode("ascii")
+    if data.startswith("data:"):
+        _, b64 = split_data_url(data)
+        if b64 is not None:
+            return b64
+    return data
+
+
+def data_to_data_url(data: str | bytes, media_type: str) -> str:
+    """Convert *data* to a ``data:`` URL.  Passes through existing URLs."""
+    if isinstance(data, str) and is_url(data):
+        return data
+    b64 = data_to_base64(data)
+    return f"data:{media_type};base64,{b64}"
+
+
+# -- media-type inference --------------------------------------------------
+
+
+def infer_media_type(url: str) -> str:
+    """Infer IANA media type from a URL.
+
+    * ``data:image/png;base64,...`` -> ``"image/png"``
+    * ``https://example.com/cat.jpg`` -> ``"image/jpeg"`` (via :mod:`mimetypes`)
+    * Unknown -> raises :class:`ValueError`
+    """
+    if url.startswith("data:"):
+        # data:[<mediatype>][;base64],<data>
+        rest = url[5:]  # strip "data:"
+        sep = rest.find(",")
+        meta = rest[:sep] if sep != -1 else rest
+        mt = meta.split(";")[0]
+        if mt:
+            return mt
+    else:
+        guessed, _ = mimetypes.guess_type(url)
+        if guessed:
+            return guessed
+    raise ValueError(
+        f"Cannot infer media_type from URL: {url!r}. Provide media_type explicitly."
+    )