mm-cache-client 0.0.3

Tiny aiohttp client for the mm-cache distributed cache service

mm-cache-client

mm-cache-client is a tiny async Python client for the mm-cache service.

Install

pip install mm-cache-client==0.0.3

Usage

from mm_cache_client import MMCacheClient, MMCacheConfig

async with MMCacheClient(MMCacheConfig(base_url="http://mm-cache:8080")) as cache:
    hit = await cache.resolve(
        model="moonshotai/Kimi-K2.5",
        content_type="image_url",
        url="https://example.com/cat.png",
    )
    if hit is not None:
        mm_hash = hit.mm_hash
        mm_kwargs = hit.mm_kwargs

Plugin fast path:

sha = MMCacheClient.alias_sha(
    model=model_id,
    content_type="image_url",
    url=url,
)
hit = await cache.resolve_sha(model=model_id, sha=sha)
if hit is not None:
    return hit.mm_hash, hit.mm_kwargs

Presigned URL churn:

hit = await cache.get_payload(model=model_name, mm_hash=mm_hash)
if hit is not None:
    mm_kwargs = hit.mm_kwargs

Resolve or compute on miss:

from mm_cache_client import CacheHit

async def encode() -> CacheHit:
    return CacheHit(mm_hash=mm_hash, mm_kwargs=mm_kwargs)

hit = await cache.resolve_or_call(
    model=model_name,
    content_type="image_url",
    url=url,
    on_miss=encode,
)

API

• alias_sha(model, content_type, url)
• BASETEN_MM_HASH_HEADER
• MMCacheClient.alias_sha(model=..., content_type=..., url=...)
• MMCacheClient.resolve(...)
• MMCacheClient.resolve_or_call(...)
• MMCacheClient.resolve_sha(...)
• MMCacheClient.resolve_alias(...)
• MMCacheClient.get_payload(...)
• MMCacheClient.put_resolve(...)
• MMCacheClient.put_sha(...)
• MMCacheClient.put_payload(...)

Integration patterns

image_encoder_plugin:

sha = MMCacheClient.alias_sha(model=model_id, content_type="image_url", url=url)
hit = await cache.resolve_sha(model=model_id, sha=sha)
if hit is not None:
    return hit.mm_hash, hit.mm_kwargs

Encoder backend:

hit = await cache.resolve(model=model_id, content_type="image_url", url=url)
if hit is not None:
    return hit.mm_hash, hit.mm_kwargs

hit = await cache.get_payload(model=model_id, mm_hash=mm_hash)
if hit is not None:
    return hit.mm_hash, hit.mm_kwargs

See:

• examples/plugin_integration.py
• examples/encoder_integration.py