@ext.skeleton reference

Live data probe decorator — section naming, return contract, TTL, alert mode, and ctx availability

@ext.skeleton registers a live data probe that the web-kernel queries on a per-user schedule. The snapshot the probe returns is stored in the web-kernel and injected into the intent classifier context on every chat turn. The LLM does not call skeleton tools explicitly — it sees their output as ambient awareness.

Use skeletons for "what does the user have right now" data: unread counts, pending tasks, schema overviews, monitor statuses. Keep them small — the entire snapshot appears before the user's message on every turn.

Not a UI primitive

@ext.skeleton is strictly a data probe consumed by the LLM. It has no rendering, no panel slot, no React component. If you are building anything visible to the user — sidebars, dashboards, editors, settings forms — use @ext.panel instead. Inside a panel, fetch state via ctx.cache (short-lived) or ctx.store (persistent). ctx.skeleton.get() is restricted to @ext.skeleton handlers by federal invariant I-SKELETON-LLM-ONLY and raises SkeletonAccessForbidden from any other context.

Where it lives

@ext.skeleton is a method on the Extension instance (not on ChatExtension). Register it from the same file that holds your ext = Extension(...), or import ext from app.py into a dedicated skeleton.py module — the pattern used in every production extension.

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.skeleton("my_section", alert=True, ttl=300)
async def skeleton_refresh_my_section(ctx) -> dict:
    count = await ctx.store.count("items")
    return {"response": {"total": count}}

Signature

def skeleton(
    self,
    section_name: str,
    *,
    alert: bool = False,
    ttl: int = 300,
    description: str = "",
) -> Callable:
    ...

All kwargs after section_name are keyword-only.

Kwargs reference

`section_name`

Required. Positional. The name for this skeleton section. The decorator registers your handler under the tool name skeleton_refresh_{section_name}.

Prop

Type

Rules:

Use snake_case — e.g. "tasks", "db_schema", "mail_inbox_summary", "web_tools".
Must be flat — no nesting or path separators. The web-kernel builds the full key imperal:skeleton:{app_id}:{user_id}:{section_name} internally.
Each call to @ext.skeleton must use a unique section_name within an extension.

Forbidden characters

The characters *, ?, [, ], :, /, and space are rejected at decoration time with ValueError. They would break the Redis key path and the web-kernel's purge helper. The error surfaces immediately at module import time — not at runtime.

`alert`

Prop

Type

When alert=True:

After every skeleton refresh, the web-kernel diffs the previous snapshot against the new one.
If the dicts differ, the web-kernel calls skeleton_alert_{section_name}(ctx, old=<prev>, new=<next>).
The alert tool returns {"response": "<message>"} — an empty string means no alert. A non-empty string surfaces as an ambient notification to the LLM.

The alert tool is registered separately using @ext.tool:

from imperal_sdk import Extension

ext = Extension(
    "my-tasks",
    display_name="Tasks",
    description="Tasks extension — manage your tasks with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.skeleton("tasks", alert=True, ttl=30)
async def skeleton_refresh_tasks(ctx) -> dict:
    overdue = await ctx.store.count("tasks", where={"overdue": True})
    today   = await ctx.store.count("tasks", where={"due_today": True})
    return {"response": {"overdue_count": overdue, "today_count": today}}


@ext.tool(
    "skeleton_alert_tasks",
    description="Alert on new overdue tasks or today's task changes.",
)
async def skeleton_alert_tasks(
    ctx,
    old: dict | None = None,
    new: dict | None = None,
) -> dict:
    if not new:
        return {"response": ""}
    overdue = new.get("overdue_count", 0)
    old_overdue = (old or {}).get("overdue_count", 0)
    if overdue > 0 and overdue > old_overdue:
        delta = overdue - old_overdue
        return {"response": f"{delta} new overdue task(s) — {overdue} total"}
    return {"response": ""}

If alert=True but no matching skeleton_alert_{section_name} tool is registered, the web-kernel silently skips change alerts. The refresh probe still runs — only the diff/alert step is skipped.

`ttl`

Prop

Type

The ttl value is stored as metadata on the registered tool and exposed in the manifest. The web-kernel reads it when bootstrapping a user's skeleton workflow for the first time. Once the workflow is running, the active TTL is the one in the Registry row — changing ttl in code only affects new installations unless the Registry row is updated by an operator.

Practical values from production:

ttl=30 — high-frequency: task counters that must stay fresh after chat writes (tasks extension)
ttl=60 — per-minute: inbox unread counts (mail-client extension)
ttl=120 — moderate: schema that changes rarely but needs to be current for SQL generation (sql-db extension)
ttl=300 — default: slow-moving state like note statistics or monitor statuses

`description`

Prop

Type

Always supply a description. It appears in the Developer Portal tool list and in imperal validate output. A good description names what data is refreshed and why the LLM needs it:

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,
)


@ext.skeleton(
    "db_schema",
    alert=True,
    ttl=300,
    description="Active database schema cache — tables, columns, row counts.",
)
async def skeleton_refresh_db_schema(ctx) -> dict:
    return {"response": {"table_count": 0}}

Return contract

Every @ext.skeleton handler must return a dict with a single "response" key whose value is a dict of scalar fields.

Rules:

The outer {"response": ...} wrapper is required. The web-kernel's skeleton_save_section activity unwraps it before storing.
The inner dict should contain flat scalar values — integers, booleans, short strings.
Nested structures (short lists of dicts) are acceptable for recent-item arrays, but keep the total payload under ~1KB. Large payloads consume classifier token budget on every chat turn.
The function is idempotent — the web-kernel may call it on a configurable tick regardless of whether data changed.

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.skeleton("status", ttl=300)
async def skeleton_refresh_status(ctx) -> dict:
    total   = await ctx.store.count("items")
    pending = await ctx.store.count("items", where={"status": "pending"})
    active  = await ctx.store.count("items", where={"status": "active"})
    return {"response": {
        "total":   total,
        "pending": pending,
        "active":  active,
    }}

Keep the payload small

Every skeleton section is included in the intent classifier context on every chat turn. A 10KB skeleton payload adds roughly 2,500 tokens before the user's message even arrives. Prefer counts and flags over full lists. Expose detail on demand via @chat.function.

Returning `ActionResult` (advanced)

mail-client/skeleton.py returns ActionResult.success(data=..., summary=...) instead of a plain dict. This is accepted by the web-kernel — the data field is used as the section payload. Use this pattern when you also want a summary string surfaced in observability logs. For most extensions, a plain dict is simpler and equally valid.

Registered tool name

The decorator registers a tool named skeleton_refresh_{section_name} in the extension manifest. You should never register a tool with this name manually — use @ext.skeleton instead. Validator rule MANIFEST-SKELETON-1 flags manual @ext.tool("skeleton_refresh_*") registrations as errors.

The registered tool runs with system-level scopes (["*"]) — no capability scope declaration is required.

ctx in skeleton handlers

Skeleton handlers receive the full Context object. Every ctx attribute available to regular tools is available to skeleton handlers, including:

ctx.user — UserContext with imperal_id, id, email, role, attributes
ctx.tenant — TenantContext | None
ctx.store — StoreClient for persistent per-user data
ctx.http — HTTPProtocol for outbound HTTP calls
ctx.cache — CacheClient for short-lived Pydantic snapshots (5–300s TTL)
ctx.ai — AIProtocol for LLM completions
ctx.notify — send in-app notifications (useful for alerting new items)
ctx.config — read admin-configured extension settings
ctx.billing — check limits before expensive operations
ctx.time — web-kernel-injected time context (timezone, now_local, is_business_hours)

`ctx.skeleton.get()` — reading previous snapshot

Inside a @ext.skeleton handler, ctx.skeleton.get(section) returns the previously stored snapshot for that section. This is useful for diff-based alerting without a separate alert tool.

ctx.skeleton is guarded

ctx.skeleton.get() raises SkeletonAccessForbidden if called outside a @ext.skeleton-decorated handler. Validator rule V24 (AST scan) flags any ctx.skeleton access in @chat.function or @ext.panel handlers at validate time. Use ctx.cache or ctx.store in those contexts — they are the correct data sources for panels and chat functions.

Refresh lifecycle

On installation (first user):
  └─ Web-kernel bootstraps a per-user skeleton workflow with the registered TTL.

Every TTL seconds (per user):
  ├─ Web-kernel calls skeleton_refresh_{section_name}(ctx) on a worker.
  │   └─ Handler fetches fresh data, returns {"response": {...}}.
  ├─ Web-kernel stores snapshot via skeleton_save_section activity.
  └─ If alert=True and snapshot changed:
       └─ Web-kernel calls skeleton_alert_{section_name}(ctx, old=..., new=...).
            └─ Non-empty response string = ambient alert injected into classifier.

On every chat turn:
  └─ Web-kernel reads all stored skeleton sections for the user.
       └─ Injected into classifier context — LLM sees them as ambient facts.

On uninstall:
  └─ Skeleton state is purged (I-SKEL-LIVE-INVALIDATE: unreachable < 2s).
       └─ Chat history cannot leak the extension's skeleton data.

Federal guarantees

Auto-derive

The web-kernel discovers skeleton sections from the skeleton_refresh_{X} naming convention — no manual section listing in the Registry.

Continue-as-new safe

Skeleton workflows are designed to run indefinitely. The web-kernel auto-rotates the workflow history to avoid hitting size limits (I-SKELETON-CAN-ROTATE).

Watchdog respawn

A parent session workflow watches each skeleton handle and respawns it on terminal child state within seconds (I-SKELETON-WATCHDOG).

Live invalidation

Uninstall purges the user's skeleton state immediately — data is unreachable within 2 seconds, never leaked to chat (I-SKEL-LIVE-INVALIDATE).

Production examples

Counter skeleton — tasks

From tasks/skeleton.py. Uses alert=True with a short TTL to keep LLM-visible counters fresh after chat writes.

from imperal_sdk import Extension

ext = Extension(
    "tasks",
    display_name="Tasks",
    description="Tasks extension — manage and track your tasks with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.skeleton(
    "tasks",
    alert=True,
    ttl=30,
    description="Background: today/overdue/upcoming counts + recent tasks + active projects.",
)
async def skeleton_refresh_tasks(ctx) -> dict:
    """Refresh task counters and recent activity. Idempotent — safe per tick."""
    if not ctx.user.imperal_id:
        return {"response": {"note": "no user on context"}}

    overdue_page  = await ctx.store.query("tasks", where={"overdue": True})
    today_page    = await ctx.store.query("tasks", where={"due_today": True})
    recent_page   = await ctx.store.query("tasks", order_by="-updated_at", limit=5)

    return {"response": {
        "connected":      True,
        "overdue_count":  overdue_page.total,
        "today_count":    today_page.total,
        "recent_tasks": [
            {"task_id": d.id, "title": d.data.get("title", "")[:80]}
            for d in recent_page.data
        ],
    }}


@ext.tool(
    "skeleton_alert_tasks",
    description="Alerts: overdue tasks, due-today, sudden spike in backlog.",
)
async def skeleton_alert_tasks(
    ctx,
    old: dict | None = None,
    new: dict | None = None,
) -> dict:
    if not new:
        return {"response": ""}
    overdue     = new.get("overdue_count", 0)
    old_overdue = (old or {}).get("overdue_count", 0)
    if overdue > 0 and overdue > old_overdue:
        delta = overdue - old_overdue
        label = f"{delta} new overdue task(s) (now {overdue} total)"
        return {"response": label}
    return {"response": ""}

Schema skeleton — sql-db

From sql-db/skeleton.py. Uses alert=True to detect table additions and removals. Also mirrors the snapshot to ctx.cache so @chat.function write-time validators can read the current schema without accessing ctx.skeleton.

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,
)


@ext.skeleton(
    "db_schema",
    alert=True,
    ttl=300,
    description="Active database schema cache — tables, columns, row counts.",
)
async def skeleton_refresh_db_schema(ctx) -> dict:
    """Refresh schema for the user's active connection. Idempotent — safe per tick."""
    try:
        resp = await ctx.http.get(
            "http://db-service/v1/schema",
            params={"user_id": ctx.user.imperal_id},
        )
        if not resp.ok:
            return {"response": {
                "database": "", "connection": "", "table_count": 0, "tables": [],
            }}
        raw = resp.json()
        tables = [
            {"name": t["name"], "rows": t.get("rows", 0), "columns": t.get("columns", [])}
            for t in raw.get("tables", [])
        ]
        return {"response": {
            "database":    raw.get("database", ""),
            "connection":  raw.get("connection", ""),
            "table_count": len(tables),
            "tables":      tables,
        }}
    except Exception as exc:
        return {"response": {
            "database": "", "connection": "", "table_count": 0,
            "tables": [], "error": str(exc)[:120],
        }}


@ext.tool(
    "skeleton_alert_db_schema",
    description="Alert on schema changes (tables added or removed).",
)
async def skeleton_alert_db_schema(
    ctx,
    old: dict | None = None,
    new: dict | None = None,
) -> dict:
    if not old or not new:
        return {"response": ""}
    old_tables = {t["name"] for t in old.get("tables", [])}
    new_tables  = {t["name"] for t in new.get("tables", [])}
    added   = new_tables - old_tables
    removed = old_tables - new_tables
    if not added and not removed:
        return {"response": ""}
    parts = []
    if added:
        parts.append(f"New tables: {', '.join(sorted(added))}")
    if removed:
        parts.append(f"Removed tables: {', '.join(sorted(removed))}")
    return {"response": f"Schema changed — {'; '.join(parts)}"}

Status-gauge skeleton — web-tools

From web-tools/skeleton.py. Reads monitors from ctx.store, aggregates by status, and surfaces counts the classifier can reason about instantly.

import asyncio

from imperal_sdk import Extension

ext = Extension(
    "web-tools",
    display_name="Web Tools",
    description="Web Tools extension — monitor websites and APIs with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.skeleton(
    "web_tools",
    ttl=300,
    description=(
        "Refresh web-tools monitor statuses from last scan snapshots. "
        "Provides instant context: how many monitors are critical/warning."
    ),
)
async def skeleton_refresh_web_tools(ctx) -> dict:
    """Load monitors and their last snapshot statuses for instant AI context."""
    try:
        page = await ctx.store.query(
            "wt_monitors",
            where={"owner_id": ctx.user.imperal_id},
            limit=50,
        )
        if not page.data:
            return {"response": {
                "total": 0, "critical": 0, "warning": 0, "ok": 0,
            }}

        snap_ids = [m.data.get("last_snapshot_id") for m in page.data]

        async def _get_snap(snap_id: str | None):
            if snap_id:
                return await ctx.store.get("wt_snapshots", snap_id)
            return None

        snaps = await asyncio.gather(*[_get_snap(sid) for sid in snap_ids])

        critical = warning = ok = 0
        for m, snap in zip(page.data, snaps):
            status = snap.data.get("status", "unknown") if snap else "unknown"
            if status == "critical":
                critical += 1
            elif status == "warning":
                warning += 1
            elif status == "ok":
                ok += 1

        return {"response": {
            "total":    len(page.data),
            "critical": critical,
            "warning":  warning,
            "ok":       ok,
        }}
    except Exception as exc:
        return {"response": {
            "error": str(exc)[:120], "total": 0,
            "critical": 0, "warning": 0, "ok": 0,
        }}

Common pitfalls

Wrong return shape

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


# WRONG — missing outer {"response": ...} wrapper
@ext.skeleton("tasks_wrong", ttl=300)
async def skeleton_refresh_tasks_wrong(ctx) -> dict:
    return {"total": 5, "pending": 2}  # type: ignore[return-value]


# CORRECT
@ext.skeleton("tasks", ttl=300)
async def skeleton_refresh_tasks(ctx) -> dict:
    return {"response": {"total": 5, "pending": 2}}

Registering the tool manually

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


# WRONG — flagged by MANIFEST-SKELETON-1; use @ext.skeleton instead
@ext.tool("skeleton_refresh_tasks_manual")
async def skeleton_refresh_tasks_manual(ctx) -> dict:
    return {"response": {"total": 5}}


# CORRECT
@ext.skeleton("tasks", ttl=30)
async def skeleton_refresh_tasks(ctx) -> dict:
    total = await ctx.store.count("tasks")
    return {"response": {"total": total}}

Payload too large

from imperal_sdk import Extension

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


# WRONG — full message body in skeleton; 10KB+ wastes classifier token budget
@ext.skeleton("mail_inbox_large", ttl=60)
async def skeleton_refresh_mail_inbox_large(ctx) -> dict:
    msgs = await ctx.http.get("https://mail-api/v1/messages?limit=100")
    return {"response": {"messages": msgs.json()}}  # entire list — avoid this


# CORRECT — counts and short metadata only; expose full messages via @chat.function
@ext.skeleton("mail_inbox", ttl=60)
async def skeleton_refresh_mail_inbox(ctx) -> dict:
    stats = await ctx.http.get("https://mail-api/v1/inbox/stats")
    s = stats.json()
    return {"response": {
        "unread_total": s.get("unread", 0),
        "accounts_connected": s.get("account_count", 0),
    }}

Testing

Use MockContext(tool_type="skeleton") to pass the _SkeletonAccessGuard check when calling ctx.skeleton.get() from test code. Import the handler function directly from your module to test it in isolation.

import pytest
from imperal_sdk.testing import MockContext, MockStore

# Assume skeleton_refresh_tasks is importable from your module
# from skeleton import skeleton_refresh_tasks

@pytest.mark.asyncio
async def test_skeleton_refresh_tasks_overdue():
    ctx = MockContext(user_id="test_user", tool_type="skeleton")
    ctx.store = MockStore()
    await ctx.store.create("tasks", {"overdue": True, "title": "Fix prod"})
    await ctx.store.create("tasks", {"overdue": False, "title": "Write docs"})

    # Call your handler directly
    # result = await skeleton_refresh_tasks(ctx)
    # assert result["response"]["overdue_count"] == 1

@ext.skeleton reference

Where it lives

Signature

Kwargs reference

`section_name`

`alert`

`ttl`

`description`

Return contract

Returning `ActionResult` (advanced)

Registered tool name

ctx in skeleton handlers

`ctx.skeleton.get()` — reading previous snapshot

Refresh lifecycle

Federal guarantees

Auto-derive

Continue-as-new safe

Watchdog respawn

Live invalidation

Production examples

Counter skeleton — tasks

Schema skeleton — sql-db

Status-gauge skeleton — web-tools

Common pitfalls

Wrong return shape

Registering the tool manually

Payload too large

Testing

Cross-references

Skeletons concept

Decorators reference

@chat.function reference

On this page