@ext.cache_model reference

@ext.cache_model registers a Pydantic BaseModel subclass as a named cache shape within an extension. Once registered, the model becomes the type contract for ctx.cache.set, ctx.cache.get, and ctx.cache.get_or_fetch — the SDK refuses to persist or retrieve any value whose type is not registered, enforcing typed cache access at the SDK boundary before any network call.

Use ctx.cache for data that is expensive to recompute, short-lived (5–300 seconds), scoped to one user, and well-structured. Typical patterns: inbox page snapshots, database schema catalogs, folder lists, unread counts with metadata. For data that must outlive the TTL cap or be shared across users, use ctx.store instead.

Where it lives

@ext.cache_model is a method on the Extension instance. It decorates a class, not a function. Register all cache models in the same file that holds your ext = Extension(...) (as in notes/app.py), or in a dedicated cache_models.py module (as in mail-client/cache_models.py).

The registration must happen at module import time — before any handler calls ctx.cache.get or ctx.cache.set on that model. The web-kernel's context wiring passes the Extension instance into CacheClient; the client reverse-looks up the registered name at call time using class identity (registered_cls is cls).

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    version="1.0.0",
    display_name="My App",
    description="My App — AI-powered tool for managing your resources.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.cache_model("resource_list")
class ResourceListCache(BaseModel):
    items: list[dict]
    fetched_at: str = ""

Signature

def cache_model(
    self,
    name: str,
) -> Callable[[type], type]:
    ...

A single required positional parameter. Returns a decorator that accepts a class and returns it unchanged (after registering it).

Kwargs reference

`name`

Required. Positional. The registration key for this model within the extension namespace.

Prop

Type

Rules:

Use snake_case that describes the cached shape — e.g. "inbox_page", "catalog", "folder_stats", "resource_list".
The name becomes part of the Redis key path: imperal:extcache:{app_id}:{user_id}:{name}:{key_hash}. It must be stable across deploys — renaming it silently invalidates all existing cache entries for that model.
Duplicate name values within the same extension raise ValueError at decoration time.

Constraints

Target must be a Pydantic `BaseModel` subclass

The decorator raises TypeError immediately if the decorated class is not a pydantic.BaseModel subclass:

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App — AI-powered tool for managing your resources.",
    icon="icon.svg",
    actions_explicit=True,
)


# WRONG — plain dict subclass raises TypeError
# @ext.cache_model("bad_model")
# class BadModel(dict):
#     pass


# CORRECT — Pydantic BaseModel subclass
@ext.cache_model("resource_list")
class ResourceListCache(BaseModel):
    items: list[dict]
    fetched_at: str = ""

Class identity, not `isinstance`

The SDK reverse-lookup uses registered_cls is cls — object identity, not isinstance. This means you must register the exact class you pass to ctx.cache.set / ctx.cache.get. Do not register a parent class and pass a subclass (or vice versa):

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App — AI-powered tool for managing your resources.",
    icon="icon.svg",
    actions_explicit=True,
)


class InboxSummary(BaseModel):
    unread: int
    latest_subject: str


# WRONG — registers _InboxSummaryCache, not InboxSummary.
# ctx.cache.set(key, InboxSummary(...)) will fail — identity mismatch.
# @ext.cache_model("inbox_summary")
# class _InboxSummaryCache(InboxSummary):
#     pass


# CORRECT — register the class you actually instantiate at call sites
@ext.cache_model("inbox_summary")
class InboxSummaryCache(BaseModel):
    unread: int
    latest_subject: str

Subclass wrapper anti-pattern

The sql-db extension changelog documents a real regression caused by registering an empty subclass _DbSchemaSnapshotCache instead of DbSchemaSnapshot directly. Every ctx.cache.set(..., model=DbSchemaSnapshot) call silently fell through to a warning-and-noop path because the identity check failed. Register the exact class you use at call sites.

TTL semantics and cap

TTL is not set on the model registration — it is set per ctx.cache.set call. There is no per-model default TTL.

Enforced constraints (I-CACHE-TTL-CAP-300S):

Minimum TTL: 5 seconds
Maximum TTL: 300 seconds

Passing a value outside this range raises ValueError before any network call. You cannot cache something for more than 5 minutes via ctx.cache. For longer-lived data, use ctx.store.

Practical TTL values from production:

30s — rapidly-changing state like task counters that must stay fresh after chat writes
60s — per-minute panel data like folder lists that change on user action but are read frequently
120s — moderate: tags lists, account summaries that change infrequently
300s — slow-moving: database schema catalogs, table detail pages, static configuration snapshots

Key-safety rules (I-CACHE-KEY-SAFETY)

Cache keys must satisfy:

Allowed characters: [A-Za-z0-9_\-:]
Maximum length: 128 characters

Violating either constraint raises ValueError at ctx.cache.get/set/delete call time, before any network call.

Use : as a namespace separator within keys — this is the canonical convention from production extensions:

folders:{user_id} — per-user folder list
catalog:{conn_id} — per-connection database catalog
table:{conn_id}:{database}:{table_name} — per-table detail snapshot
inbox:{email}:INBOX:0 — per-account inbox page

Key scope: always per-user

ctx.cache is always scoped to the authenticated user — CacheClient is constructed with user_id baked in. You cannot share a cache entry across users. If multiple users could legitimately share data (e.g. a shared database catalog), either accept the redundancy or use a shared store layer outside ctx.cache. Never include raw user IDs in the key unless you control the construction — the client already namespaces by user_id at the wire level.

Value size cap (I-CACHE-VALUE-SIZE-CAP-64KB)

The full JSON envelope — including model name, version, data, and timestamp — must be at most 64 KB. The SDK computes the envelope byte size before the PUT request and raises ValueError if exceeded.

For large data sets (e.g. database catalogs with hundreds of tables), split the data into multiple focused cache models rather than one large blob:

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "sql-db",
    display_name="SQL Database",
    description="SQL Database extension — run queries and explore schemas with AI.",
    icon="icon.svg",
    actions_explicit=True,
)


# WRONG — one giant blob of all tables risks exceeding the 64 KB envelope cap
# @ext.cache_model("all_tables")
# class AllTablesCache(BaseModel):
#     tables: list[dict]  # can easily exceed 64 KB for large databases


# CORRECT — three focused models, each well under the limit
@ext.cache_model("catalog")
class CatalogCache(BaseModel):
    conn_id: str
    databases: list[dict]
    fetched_at: str = ""


@ext.cache_model("tables_page")
class TablesPageCache(BaseModel):
    conn_id: str
    database: str
    items: list[dict]
    total_count: int = 0
    fetched_at: str = ""


@ext.cache_model("table_detail")
class TableDetailCache(BaseModel):
    conn_id: str
    database: str
    table: str
    columns: list[dict]
    fetched_at: str = ""

Registration flow

The web-kernel discovers registered cache models by accessing ext._cache_models on the Extension instance it wired into the worker. The flow is:

Module import:
  └─ @ext.cache_model("name") decorates MyModel
       └─ ext._cache_models["name"] = MyModel

Worker startup:
  └─ Web-kernel constructs CacheClient(app_id, user_id, ..., extension=ext)
       └─ CacheClient holds a reference to ext

At ctx.cache.set(key, value, ttl_seconds=N):
  └─ CacheClient._resolve_model_name(type(value))
       └─ iterates ext._cache_models.items()
       └─ returns name where registered_cls is type(value)
       └─ raises ValueError if not found (I-CACHE-MODEL-REGISTRATION-REQUIRED)
  └─ SDK builds envelope {"model": name, "version": 1, "data": ..., "cached_at": ...}
  └─ Validates key (length, charset), TTL (range), envelope size
  └─ PUT to auth gateway /v1/internal/extcache/{app}/{user}/{name}/{key_hash}

At ctx.cache.get(key, model=MyModel):
  └─ Same reverse-lookup for model_name
  └─ GET from auth gateway
  └─ 404 → None
  └─ Envelope version or model name mismatch → None
  └─ Returns model.model_validate(data)

ctx.cache API reference

Once models are registered, use ctx.cache in any handler:

`ctx.cache.get(key, model)`

Fetch a cached value by key. Returns None on a cache miss or envelope mismatch.

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "notes",
    display_name="Notes",
    description="Notes extension — create and manage notes with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.cache_model("folders_list")
class FoldersCacheEntry(BaseModel):
    folders: list[dict]


async def example_get(ctx) -> None:
    entry = await ctx.cache.get(f"folders:{ctx.user.imperal_id}", FoldersCacheEntry)
    if entry is not None:
        folders = entry.folders

`ctx.cache.set(key, value, ttl_seconds)`

Write a cached value. The model must be registered. TTL must be 5–300 seconds.

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "notes",
    display_name="Notes",
    description="Notes extension — create and manage notes with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.cache_model("folders_list")
class FoldersCacheEntry(BaseModel):
    folders: list[dict]


async def example_set(ctx) -> None:
    folders_data = [{"id": "f1", "name": "Work"}, {"id": "f2", "name": "Personal"}]
    entry = FoldersCacheEntry(folders=folders_data)
    await ctx.cache.set(f"folders:{ctx.user.imperal_id}", entry, ttl_seconds=60)

`ctx.cache.get_or_fetch(key, model, fetcher, ttl_seconds)`

The most ergonomic pattern: returns the cached value if present, otherwise calls fetcher(), writes the result with ttl_seconds, and returns it. The fetcher must return an instance of model.

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "notes",
    display_name="Notes",
    description="Notes extension — create and manage notes with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.cache_model("folders_list")
class FoldersCacheEntry(BaseModel):
    folders: list[dict]


async def load_sidebar(ctx) -> list:
    uid = ctx.user.imperal_id

    async def _fetch_folders():
        resp = await ctx.http.get(
            "http://notes-api/v1/folders",
            params={"user_id": uid},
        )
        data = resp.json()
        return FoldersCacheEntry(folders=data.get("folders", []))

    entry = await ctx.cache.get_or_fetch(
        f"folders:{uid}", FoldersCacheEntry,
        fetcher=_fetch_folders, ttl_seconds=60,
    )
    return entry.folders

`ctx.cache.delete(key)`

Delete all cache entries for a given key across all registered model namespaces. Used for explicit cache invalidation after a write action.

Production examples

Per-user folder and tag cache — notes

From notes/app.py. Three focused models covering different sidebar data types. Short TTLs (30–120s) keep data fresh after note operations.

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "notes",
    display_name="Notes",
    description="Notes extension — create and manage notes with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.cache_model("folders_list")
class FoldersCacheEntry(BaseModel):
    folders: list[dict]


@ext.cache_model("tags_list")
class TagsCacheEntry(BaseModel):
    tags: list[str]


@ext.cache_model("folder_stats")
class FolderStatsCacheEntry(BaseModel):
    counts: dict

At call sites in panels.py, the pattern is consistently get_or_fetch with a per-user key and a short TTL:

# In a panel handler (illustrative — not a complete runnable block):
# entry = await ctx.cache.get_or_fetch(
#     f"folders:{uid}", FoldersCacheEntry,
#     fetcher=lambda: ..., ttl_seconds=60,
# )

Multi-model schema cache — sql-db

From sql-db/app.py. Four models of increasing granularity, with TTLs tuned to how often each changes. The skeleton handler mirrors schema data into db_schema_snapshot; the panel sidebar reads catalog and tables_page for O(1) cold-start rendering.

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "sql-db",
    display_name="SQL Database",
    description="SQL Database extension — run queries and explore schemas with AI.",
    icon="icon.svg",
    actions_explicit=True,
)

CATALOG_CACHE_TTL = 300
TABLES_PAGE_CACHE_TTL = 300
TABLE_DETAIL_CACHE_TTL = 600  # capped at 300 by CacheClient; set at call site


class DbSchemaTable(BaseModel):
    name: str
    rows: int = 0
    columns: list[dict] = []


@ext.cache_model("db_schema_snapshot")
class DbSchemaSnapshot(BaseModel):
    database: str = ""
    connection: str = ""
    table_count: int = 0
    tables: list[DbSchemaTable] = []
    note: str = ""


class CatalogDb(BaseModel):
    name: str
    table_count: int = 0
    schema_version: str = ""


@ext.cache_model("catalog")
class CatalogCache(BaseModel):
    conn_id: str
    databases: list[CatalogDb] = []
    fetched_at: str = ""


@ext.cache_model("tables_page")
class TablesPageCache(BaseModel):
    conn_id: str
    database: str
    items: list[dict] = []
    total_count: int = 0
    fetched_at: str = ""


@ext.cache_model("table_detail")
class TableDetailCache(BaseModel):
    conn_id: str
    database: str
    table: str
    columns: list[dict] = []
    fetched_at: str = ""

Inbox page and manifest — mail-client

From mail-client/cache_models.py. When the extension has many models, register them in a dedicated module. Import-order matters — register before the panel and handler modules are imported.

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "mail-client",
    display_name="Mail Client",
    description="Mail Client extension — read and send email with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


class InboxPage(BaseModel):
    messages: list[dict]
    page: int = 0
    total: int = 0


class UnreadSummary(BaseModel):
    unread: int
    accounts: list[str] = []


class InboxManifest(BaseModel):
    preloaded: int = 0
    total_pages: int = 0


# Programmatic registration — identical to @ext.cache_model("inbox_page")
ext.cache_model("inbox_page")(InboxPage)
ext.cache_model("unread_summary")(UnreadSummary)
ext.cache_model("inbox_manifest")(InboxManifest)

Both the decorator form @ext.cache_model("name") and the programmatic form ext.cache_model("name")(MyClass) are equivalent. Use the programmatic form when you need to separate model class definitions from the ext import to avoid circular imports.

V20 and action_type effects

Cache writes are typically paired with action_type="read" in @chat.function handlers — they are warm-up side effects of reading data, not mutations of user state. No KAV confirmation is required. If a @chat.function with action_type="write" populates cache as a side effect of a write operation, that is fine — the write gate governs the user-facing action, not the cache warming.

Cache invalidation (ctx.cache.delete) inside a @chat.function write handler should be declared in effects so panels that display the now-stale data are refreshed:

from pydantic import BaseModel

from imperal_sdk import Extension, ChatExtension
from imperal_sdk.chat import ActionResult

ext = Extension(
    "notes",
    display_name="Notes",
    description="Notes extension — create and manage notes with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)
chat = ChatExtension(
    ext=ext,
    tool_name="tool_notes_chat",
    description="Notes — create and manage notes with AI assistance.",
)


@ext.cache_model("folders_list")
class FoldersCacheEntry(BaseModel):
    folders: list[dict]


@chat.function(
    name="create_folder",
    description="Create a new note folder. Immediately visible in the sidebar.",
    action_type="write",
    event="notes.folder_created",
    effects=["__panel__sidebar"],
)
async def create_folder(ctx, name: str = "") -> ActionResult:
    resp = await ctx.http.post(
        "http://notes-api/v1/folders",
        json={"user_id": ctx.user.imperal_id, "name": name},
    )
    if not resp.json().get("ok"):
        return ActionResult.error("Could not create folder.")
    # Invalidate the stale folders cache so the next sidebar render fetches fresh data
    await ctx.cache.delete(f"folders:{ctx.user.imperal_id}")
    return ActionResult.success(
        data=resp.json().get("folder", {}),
        summary=f"Created folder '{name}'.",
    )

Common pitfalls

TTL out of range

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App — AI-powered tool for managing your resources.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.cache_model("resource_list")
class ResourceListCache(BaseModel):
    items: list[dict]


async def bad_ttl(ctx) -> None:
    entry = ResourceListCache(items=[])
    # WRONG — raises ValueError: ttl_seconds must be in [5, 300] (got 600)
    # await ctx.cache.set("my-key", entry, ttl_seconds=600)

    # CORRECT — cap at 300
    await ctx.cache.set("my-key", entry, ttl_seconds=300)

Key with forbidden characters

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App — AI-powered tool for managing your resources.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.cache_model("resource_list")
class ResourceListCache(BaseModel):
    items: list[dict]


async def bad_key(ctx) -> None:
    entry = ResourceListCache(items=[])
    uid = ctx.user.imperal_id

    # WRONG — space and @ are not in [A-Za-z0-9_\-:]
    # await ctx.cache.set(f"resources for {ctx.user.email}", entry, ttl_seconds=60)

    # CORRECT — use imperal_id (no special chars) as namespace segment
    await ctx.cache.set(f"resources:{uid}", entry, ttl_seconds=60)

Mutating a cached value without re-writing

ctx.cache.get returns a deserialized Pydantic model instance. Mutating fields on it does not update the cache — the change lives only in the returned object. Call ctx.cache.set with the updated instance to persist the change.

from pydantic import BaseModel

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App — AI-powered tool for managing your resources.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.cache_model("resource_list")
class ResourceListCache(BaseModel):
    items: list[dict]


async def update_cache(ctx) -> None:
    uid = ctx.user.imperal_id
    entry = await ctx.cache.get(f"resources:{uid}", ResourceListCache)
    if entry is None:
        return

    # WRONG — this mutation is local only; cache is unchanged
    # entry.items.append({"id": "new"})

    # CORRECT — build updated instance and re-write
    updated = ResourceListCache(items=entry.items + [{"id": "new"}])
    await ctx.cache.set(f"resources:{uid}", updated, ttl_seconds=60)

Testing

Use MockContext from the SDK testing module. The mock's ctx.cache accepts any registered model from the extension instance passed to MockContext.

import pytest
from imperal_sdk.testing import MockContext

# Assume FoldersCacheEntry is importable from your module
# from app import FoldersCacheEntry, ext

@pytest.mark.asyncio
async def test_folder_cache_roundtrip():
    # ctx = MockContext(user_id="test_user", extension=ext)
    # entry = FoldersCacheEntry(folders=[{"id": "f1", "name": "Work"}])
    # await ctx.cache.set("folders:test_user", entry, ttl_seconds=60)
    # result = await ctx.cache.get("folders:test_user", FoldersCacheEntry)
    # assert result is not None
    # assert result.folders[0]["name"] == "Work"
    pass