Media

The tenxgraph.storage.media module handles images, audio, video, and documents in messages. It offloads large base64 blobs to a media store as graph://media/{key} references, validates and processes images with Pillow, caches provider file uploads, and checks uploads for safety. See Send media to a model for a walkthrough.

Import paths

Almost everything is exported from tenxgraph.storage.media. upload_to_google_file_api is not re-exported there and is imported from tenxgraph.storage.media.provider_media.

from tenxgraph.storage.media import (
    # Offloading and storage
    MediaOffloadPolicy, ensure_media_offloaded, BaseMediaStore,
    InMemoryMediaStore, LocalFileMediaStore, CloudMediaStore,
    # Image processing (requires Pillow)
    MediaProcessor,
    # Provider helpers
    ProviderMediaCache, should_use_google_file_api, GOOGLE_INLINE_THRESHOLD,
    create_openai_file_attachment, create_openai_file_search_tool,
    # Resolving references
    MediaRefResolver,
    # Configuration
    MultimodalConfig, ImageHandling, DocumentHandling,
    # Security
    validate_magic_bytes, sanitize_filename, enforce_file_size,
)
from tenxgraph.storage.media.provider_media import upload_to_google_file_api

MediaOffloadPolicy

A string enum that controls when inline base64 data is offloaded to a BaseMediaStore.

from tenxgraph.storage.media import MediaOffloadPolicy

policy = MediaOffloadPolicy.THRESHOLD  # Default

ensure_media_offloaded

async def ensure_media_offloaded(message, store, policy=MediaOffloadPolicy.THRESHOLD, max_inline_bytes=50_000) -> Message

message = await ensure_media_offloaded(
    message=my_message,
    store=media_store,
    policy=MediaOffloadPolicy.THRESHOLD,
    max_inline_bytes=50_000,
)

Inspects ImageBlock, AudioBlock, VideoBlock, and DocumentBlock entries in a message. For any block with media.kind == "data" and media.data_base64, this function decodes the base64, uploads it to the store, and replaces the block’s media with a graph://media/{key} reference that keeps filename, width, height, duration_ms and sets size_bytes. The decoded size is estimated from the base64 length. The message is mutated in place and also returned. With NEVER it returns immediately.

Parameters:

Returns: The (possibly mutated) Message.

Example:

import asyncio
import base64

from tenxgraph.core.state import ImageBlock, MediaRef, Message, TextBlock
from tenxgraph.storage.media import LocalFileMediaStore, MediaOffloadPolicy, ensure_media_offloaded

store = LocalFileMediaStore(base_dir="./media")

# Create a message with inline image data
with open("photo.jpg", "rb") as f:
    b64_data = base64.b64encode(f.read()).decode()

msg = Message(
    role="user",
    content=[
        TextBlock(text="Describe this image"),
        ImageBlock(media=MediaRef(kind="data", data_base64=b64_data, mime_type="image/jpeg")),
    ],
)

# Offload the image if it is large
async def main() -> None:
    offloaded = await ensure_media_offloaded(msg, store, policy=MediaOffloadPolicy.THRESHOLD)
    # For a large image, offloaded.content[1].media is now a graph://media/{key} reference
    print(offloaded.content[1].media.kind, offloaded.content[1].media.url)


asyncio.run(main())

BaseMediaStore

Abstract interface for media storage backends. Subclasses must implement four async methods. get_metadata, get_direct_url and to_media_ref have default implementations.

from tenxgraph.storage.media import BaseMediaStore, LocalFileMediaStore, InMemoryMediaStore, CloudMediaStore

Implementations

All three are re-exported from tenxgraph.storage.media.

Wiring into a graph

from tenxgraph.storage.media import LocalFileMediaStore

media_store = LocalFileMediaStore(base_dir="./media_uploads")

app = graph.compile(
    checkpointer=my_checkpointer,
    media_store=media_store,
)

compile(media_store=...) registers the store in the graph’s dependency container as BaseMediaStore, so nodes and the API server can retrieve it. Offloading itself is done by calling ensure_media_offloaded() (shown above) on the messages you want to slim down. Once a message holds a graph://media/{key} reference, the checkpointer stores that small reference instead of the binary blob.

MediaProcessor

Validates and optionally resizes images before they enter the pipeline. Requires Pillow (pip install "10xgraph[images]") for resizing, orientation, thumbnails and format conversion; validation works without it.

from tenxgraph.storage.media import MediaProcessor, MultimodalConfig
from tenxgraph.core.state import ImageBlock

config = MultimodalConfig(
    max_image_size_mb=10.0,
    max_image_dimension=2048,
    supported_image_types={"image/jpeg", "image/png", "image/webp", "image/gif"},
)
processor = MediaProcessor(config=config)

Constructor:

Methods:

validate_image(block: ImageBlock) -> None

Validate an ImageBlock against the allowed MIME types (a missing MIME type is treated as image/png) and the size limit, using size_bytes when set and the approximate decoded size for inline data. Raises ValueError if validation fails.

try:
    processor.validate_image(block)
except ValueError as e:
    print(f"Invalid image: {e}")

resize_image(block: ImageBlock) -> ImageBlock

Resize an inline base64 image if either dimension exceeds max_image_dimension. Returns the original block unchanged if Pillow is unavailable, the image is a URL/file_id reference, or dimensions are within limits. Requires Pillow.

resized_block = processor.resize_image(block)

process(block: ImageBlock) -> ImageBlock

Validate and resize in one call. Convenience method equivalent to validate_image() then resize_image().

block = processor.process(block)

fix_orientation(block: ImageBlock) -> ImageBlock

Apply EXIF orientation to an inline image (common with phone camera uploads). Reads the EXIF Orientation tag and transposes the pixel data, then strips the tag. Only JPEG and TIFF images are checked. Returns the original block if Pillow is unavailable, the image is not inline, or there is no EXIF rotation. Requires Pillow.

block = processor.fix_orientation(block)

full_process(block: ImageBlock) -> ImageBlock

Full pipeline: validate, fix EXIF orientation, then resize. Use this when incoming images may have rotation metadata.

block = processor.full_process(block)

generate_thumbnail(block: ImageBlock, max_dim: int = 256) -> ImageBlock

Create a JPEG thumbnail (quality 80) of an inline image. Returns the original block if Pillow is unavailable, the image is a URL/file_id reference, or both sides are already within max_dim.

thumb = processor.generate_thumbnail(block, max_dim=128)

optimize_image(block: ImageBlock, target_format: str = "JPEG", quality: int = 85) -> ImageBlock

Convert and compress an inline image to a target format. Useful for normalising uploads to a single format. target_format is a Pillow format name such as "JPEG", "PNG" or "WEBP"; quality (1-100) affects lossy formats. Returns the original block if Pillow is unavailable.

optimized = processor.optimize_image(block, target_format="WEBP", quality=80)

ProviderMediaCache

Content-addressed cache mapping file hashes to provider file references. Prevents re-uploading identical binary content to Google, OpenAI, or other providers. Reads are safe from multiple threads; writes use a plain in-memory dict and are not synchronized.

from tenxgraph.storage.media import ProviderMediaCache

cache = ProviderMediaCache(max_entries=1000)
key = ProviderMediaCache.content_key(b"example bytes")
cache.put("google", key, "files/abc123")  # any reference object
print(cache.get("google", key))  # files/abc123

Constructor:

Methods:

content_key(data: bytes) -> str (static)

Compute a content-addressed key using SHA-256 hex digest.

key = ProviderMediaCache.content_key(file_bytes)

get(provider: str, key: str) -> Any | None

Look up a cached provider file reference. Returns None if not found.

ref = cache.get("google", key)

put(provider: str, key: str, reference: Any) -> None

Store a provider file reference. If the provider bucket holds max_entries, the oldest entry is evicted first.

cache.put("google", key, upload_result)

clear(provider: str | None = None) -> None

Clear cached references for a specific provider (or all if provider=None).

cache.clear("google")  # Clear Google cache
cache.clear()  # Clear all

upload_to_google_file_api

Upload a file to Google’s File API and return a types.Part reference built from the uploaded file URI. If cache already holds the content, returns the cached reference without uploading. Requires the google-genai extra. All arguments after mime_type are keyword-only.

import asyncio

from tenxgraph.storage.media import ProviderMediaCache
from tenxgraph.storage.media.provider_media import upload_to_google_file_api

cache = ProviderMediaCache()


async def main() -> None:
    with open("photo.jpg", "rb") as f:
        file_bytes = f.read()
    # Requires Google credentials in the environment for google.genai.Client()
    part = await upload_to_google_file_api(
        file_bytes,
        "image/jpeg",
        display_name="photo.jpg",
        cache=cache,
    )
    # part is a google.genai.types.Part referencing the uploaded file
    print(part)


asyncio.run(main())

Parameters:

Returns: A google.genai.types.Part referencing the uploaded file.

should_use_google_file_api

Check whether a file size exceeds GOOGLE_INLINE_THRESHOLD (20 MB, exported from the same module). Returns True if the file should be uploaded via the File API instead of sent inline.

from tenxgraph.storage.media import should_use_google_file_api

print(should_use_google_file_api(25 * 1024 * 1024))  # True
print(should_use_google_file_api(1024))  # False

Parameters:

Returns: True if size is greater than 20 MB (20 * 1024 * 1024 bytes), False otherwise.

OpenAI file helpers

Build request fragments for OpenAI’s Files API and file_search tool.

create_openai_file_search_tool

Create a file_search tool dict for OpenAI.

from tenxgraph.storage.media import create_openai_file_search_tool

tool = create_openai_file_search_tool(["file-abc123", "file-xyz789"])
print(tool)
# {"type": "file_search", "file_search": {"vector_store_ids": []}}

Parameters:

Returns: A tool dict for OpenAI’s tools parameter. The current implementation does not use file_ids: it always returns an empty vector_store_ids list.

create_openai_file_attachment

Create a message attachment referencing an uploaded OpenAI file.

from tenxgraph.storage.media import create_openai_file_attachment

attachment = create_openai_file_attachment("file-abc123", tools=["file_search"])
print(attachment)
# {"file_id": "file-abc123", "tools": [{"type": "file_search"}]}

Parameters:

Returns: An attachment dict for OpenAI’s attachments parameter.

MultimodalConfig

Per-agent configuration for how media is validated and delivered to the provider.

from tenxgraph.core.graph import Agent
from tenxgraph.storage.media import DocumentHandling, ImageHandling, MultimodalConfig

config = MultimodalConfig(
    image_handling=ImageHandling.BASE64,
    document_handling=DocumentHandling.EXTRACT_TEXT,
    max_image_dimension=2048,
    max_image_size_mb=10.0,
)
agent = Agent(model="gpt-4o", multimodal_config=config)

Offload behavior is not configured here. It is driven by MediaOffloadPolicy and the media_store passed to graph.compile().

MediaRefResolver

Resolves MediaRef objects into concrete content parts that providers expect. Fetches graph://media/{key} references from the media store and can generate signed direct URLs instead of re-uploading bytes on every turn.

from tenxgraph.storage.media import MediaRefResolver

resolver = MediaRefResolver(media_store=media_store)

# Optional: cache signed URLs in a shared backend (any cache object you provide)
resolver = resolver.with_cache(
    cache_backend=my_cache,
    expiration_seconds=3600,
    refresh_buffer_seconds=60,
)

part = await resolver.resolve_for_openai(media_ref, model="gpt-4o")

Constructor:

Methods:

with_cache(cache_backend: Any, expiration_seconds: int = 3600, refresh_buffer_seconds: int = 60) -> MediaRefResolver

Attach or update a shared cache backend for signed URLs. Mutates the resolver and returns self for chaining.

resolver = resolver.with_cache(my_cache, expiration_seconds=7200)

resolve_for_openai(ref: MediaRef, model: str | None = None) -> dict[str, Any]

Convert a MediaRef to an OpenAI-style content part. If model is provided, checks the capability matrix and raises UnsupportedMediaInputError for text-only models.

Returns: A dict like {"type": "image_url", "image_url": {"url": "..."}}

resolve_for_google(ref: MediaRef, model: str | None = None) -> Any

Convert a MediaRef to a google.genai.types.Part. Same capability check as resolve_for_openai when model is given. Requires the google-genai extra.

Security helpers

Validate file content, sanitize filenames, and enforce size limits.

validate_magic_bytes

Check that raw file bytes match the claimed MIME type using magic bytes (file signatures). Known types are image/png, image/jpeg, image/gif, image/webp, image/bmp, image/tiff, application/pdf and application/zip.

from tenxgraph.storage.media import validate_magic_bytes

with open("photo.jpg", "rb") as f:
    data = f.read()

if not validate_magic_bytes(data, "image/jpeg"):
    raise ValueError("File is not actually a JPEG")

Parameters:

Returns: True if magic bytes match or MIME type is unknown (unknown types pass by default); False if data is empty or bytes contradict the claim.

sanitize_filename

Sanitize a user-provided filename for safe filesystem storage. It normalizes Unicode (NFC), keeps only the basename, removes .., characters such as <>:"/\|?* and control characters, strips surrounding spaces and dots, truncates to 255 characters keeping the extension, and returns "unnamed" for empty results.

from tenxgraph.storage.media import sanitize_filename

print(sanitize_filename("../../etc/passwd"))  # passwd
print(sanitize_filename("photo (1).jpg"))  # photo (1).jpg
print(sanitize_filename(""))  # unnamed

Parameters:

Returns: A safe filename string.

enforce_file_size

Raise ValueError if data exceeds the size limit.

from tenxgraph.storage.media import enforce_file_size

with open("large_file.zip", "rb") as f:
    data = f.read()

try:
    enforce_file_size(data, max_mb=100.0)
except ValueError as e:
    print(f"File too large: {e}")

Parameters:

Raises: ValueError if data size exceeds the limit.

Common patterns

Validate and offload an image message

Validate each image block, then offload large blobs to a store so the message stays small.

from tenxgraph.core.state import ImageBlock, Message
from tenxgraph.storage.media import (
    LocalFileMediaStore,
    MediaOffloadPolicy,
    MediaProcessor,
    MultimodalConfig,
    ensure_media_offloaded,
)

processor = MediaProcessor(config=MultimodalConfig())
store = LocalFileMediaStore(base_dir="./media")


async def prepare(msg: Message) -> Message:
    # Raises ValueError for unsupported types or oversized images
    for block in msg.content:
        if isinstance(block, ImageBlock):
            processor.validate_image(block)
    return await ensure_media_offloaded(msg, store, policy=MediaOffloadPolicy.THRESHOLD)

Cache Google uploads

upload_to_google_file_api checks the cache itself, so passing cache= is enough.

from tenxgraph.storage.media import ProviderMediaCache, should_use_google_file_api
from tenxgraph.storage.media.provider_media import upload_to_google_file_api

cache = ProviderMediaCache()


async def google_part(file_bytes: bytes, mime_type: str):
    if should_use_google_file_api(len(file_bytes)):
        # Large file: upload once, reuse the reference for identical bytes
        return await upload_to_google_file_api(file_bytes, mime_type, cache=cache)
    return None  # small file: send inline instead

Validate and sanitize uploads

Check content type, size and filename before storing an upload.

from tenxgraph.storage.media import (
    BaseMediaStore,
    enforce_file_size,
    sanitize_filename,
    validate_magic_bytes,
)


async def safe_upload(store: BaseMediaStore, filename: str, data: bytes, claimed_mime: str) -> str:
    if not validate_magic_bytes(data, claimed_mime):
        raise ValueError(f"Magic bytes do not match {claimed_mime}")
    enforce_file_size(data, max_mb=50.0)
    safe_name = sanitize_filename(filename)
    # Returns the storage key; keep the sanitized name in metadata
    return await store.store(data, claimed_mime, metadata={"filename": safe_name})

Common errors