Cached inbox — typed get_or_fetch with TTL
Use @ext.cache_model + ctx.cache.get_or_fetch to cache expensive list fetches per-user
ctx.cache is the right tool when a list fetch is expensive (network round-trip, pagination, token cost) but the data stays valid for tens of seconds. Register a Pydantic model at module scope with @ext.cache_model, then call ctx.cache.get_or_fetch in your handler — the SDK returns the cached value on a hit and calls your fetcher on a miss, writing the result automatically. This pattern is used directly in mail-client/cache_models.py and the handlers that read inbox pages.
from __future__ import annotations
import logging
from datetime import datetime, timezone
from pydantic import BaseModel
from imperal_sdk import Extension, ChatExtension
from imperal_sdk.chat import ActionResult
log = logging.getLogger(__name__)
# ── Extension bootstrap ───────────────────────────────────────────────────────
ext = Extension(
"mail-client",
version="1.0.0",
display_name="Mail Client",
description="Mail Client extension — read and send email with AI assistance.",
icon="icon.svg",
actions_explicit=True,
)
chat = ChatExtension(
ext=ext,
tool_name="tool_mail_chat",
description="Mail Client — read, search, and send email.",
)
# ── Cache model registration ─────────────────────────────────────────────────
#
# Must happen at module scope, before any handler calls ctx.cache.get/set.
# TTL is NOT set here — it is set per ctx.cache.set / get_or_fetch call.
# SDK invariant I-CACHE-TTL-CAP-300S: TTL must be in [5, 300] seconds.
# Production ref: mail-client/cache_models.py
@ext.cache_model("inbox_page")
class InboxPage(BaseModel):
account_id: str
folder: str
messages: list[dict]
total: int = 0
has_more: bool = False
fetched_at: str = ""
@ext.cache_model("unread_summary")
class UnreadSummary(BaseModel):
unread: int = 0
accounts: list[str] = []
fetched_at: str = ""
# ── Chat function: list inbox with cache ─────────────────────────────────────
@chat.function(
"list_inbox",
description="List inbox messages for the user. Returns the first page from cache or fetches fresh.",
action_type="read",
)
async def list_inbox(
ctx,
account_id: str = "",
folder: str = "INBOX",
) -> ActionResult:
"""Return cached inbox page, or fetch and cache on miss."""
uid = ctx.user.imperal_id
if not uid or not account_id:
return ActionResult.error("No account specified.")
# Key syntax: [A-Za-z0-9_\-:], max 128 chars (I-CACHE-KEY-SAFETY).
# Use : as namespace separator — standard across all production extensions.
cache_key = f"inbox:{account_id}:{folder}:0"
async def _fetch_inbox() -> InboxPage:
"""Fetcher called on cache miss. Must return an InboxPage instance."""
resp = await ctx.http.get(
"http://mail-service/v1/inbox",
params={"user_id": uid, "account_id": account_id, "folder": folder, "page": 0},
)
body: dict = resp.json() if resp.ok else {}
return InboxPage(
account_id=account_id,
folder=folder,
messages=body.get("messages", []),
total=body.get("total", 0),
has_more=body.get("has_more", False),
fetched_at=datetime.now(timezone.utc).isoformat(),
)
# get_or_fetch: returns cached InboxPage on hit; calls _fetch_inbox on miss,
# writes the result with ttl_seconds=60, and returns it.
page = await ctx.cache.get_or_fetch(
cache_key, InboxPage,
fetcher=_fetch_inbox,
ttl_seconds=60,
)
return ActionResult.success(
data={
"messages": page.messages[:20],
"total": page.total,
"has_more": page.has_more,
},
summary=f"{len(page.messages[:20])} message(s) in {folder}.",
)
# ── Chat function: get unread summary with cache ──────────────────────────────
@chat.function(
"get_unread_summary",
description="Return total unread count and list of accounts with unread messages.",
action_type="read",
)
async def get_unread_summary(ctx) -> ActionResult:
"""Return cached unread summary, or fetch and cache on miss."""
uid = ctx.user.imperal_id
if not uid:
return ActionResult.error("No user context.")
cache_key = f"unread:{uid}"
async def _fetch_unread() -> UnreadSummary:
resp = await ctx.http.get(
"http://mail-service/v1/unread",
params={"user_id": uid},
)
body: dict = resp.json() if resp.ok else {}
return UnreadSummary(
unread=body.get("unread", 0),
accounts=body.get("accounts", []),
fetched_at=datetime.now(timezone.utc).isoformat(),
)
summary = await ctx.cache.get_or_fetch(
cache_key, UnreadSummary,
fetcher=_fetch_unread,
ttl_seconds=60,
)
return ActionResult.success(
data={"unread": summary.unread, "accounts": summary.accounts},
summary=f"{summary.unread} unread message(s).",
)Walk-through
Why register at module scope.
The SDK's CacheClient resolves model names by object identity (registered_cls is cls). The lookup happens at every ctx.cache.set / ctx.cache.get call. If @ext.cache_model has not run before the first cache call, the client raises ValueError (I-CACHE-MODEL-REGISTRATION-REQUIRED). Module-scope decoration guarantees the registration runs during import, before any handler is invoked.
@ext.cache_model("name").
Accepts one positional argument — the registration key for this model within the extension namespace. The name is stable across deploys: renaming it silently invalidates all existing cache entries. Use snake_case that describes the cached shape ("inbox_page", "unread_summary", "catalog").
ctx.cache.get_or_fetch(key, model, fetcher, ttl_seconds).
The canonical pattern for read handlers. Returns the cached model instance if present; otherwise calls fetcher() (which must return an instance of model), writes the result with ttl_seconds, and returns it. The fetcher is an async callable with no arguments — capture ctx, uid, and other closure variables via the enclosing scope.
TTL range.
TTL must be in [5, 300] seconds (I-CACHE-TTL-CAP-300S). Passing a value outside this range raises ValueError before any network call. For inbox data that changes on user action but is read frequently, 60s is a reasonable default. For slower-moving data (folder lists, schema catalogs) use up to 300s.
Key safety.
Allowed characters are [A-Za-z0-9_\-:], maximum 128 characters (I-CACHE-KEY-SAFETY). Use : as a namespace separator — this is the convention in every production extension. Never include raw email addresses or arbitrary user input directly in the key; use account_id or imperal_id which satisfy the charset constraint.
Cache is per-user.
CacheClient is constructed with the authenticated user's imperal_id baked in. You cannot share a cache entry across users. If you include uid in the key, be aware the client already namespaces by user_id at the wire level — the key you see in logs will be prefixed with the user namespace.
Invalidation on write.
After a write action (e.g. archiving a message), call ctx.cache.delete(cache_key) to invalidate the stale entry. The next read will call the fetcher and warm the cache again. Pair the invalidation with effects=["__panel__sidebar"] in the @chat.function declaration to trigger panel refresh at the same time.
64 KB envelope cap
The full JSON envelope — model name, version, data, and timestamp — must be at most 64 KB (I-CACHE-VALUE-SIZE-CAP-64KB). For large data sets (hundreds of messages, large schema catalogs), split into multiple focused models. The mail-client production extension uses five separate models (InboxPage, InboxMessages, InboxManifest, UnreadSummary, AccountList) rather than one large blob.
Cross-links
@ext.cache_model reference
Full signature, key-safety rules, TTL range, value size cap, registration flow, and all common pitfalls.
Cache vs store concept
When to use ctx.cache versus ctx.store — TTL, persistence, size limits, and cross-user sharing.
Recipe — skeleton [data surface](/en/reference/glossary/)
The skeleton counterpart: ambient LLM context refreshed on a timer, not on request.