Build a complete extension

From zero to a working "Tasks Mini" extension — exercises every decorator, UI primitive, ctx surface, and federal pattern. Approximately 30 minutes.

This tutorial walks you through building Tasks Mini — a complete, production-pattern extension that lets users manage a personal task list through chat and a two-panel UI. It deliberately exercises every major surface of the SDK so that by the end you have touched every concept page and reference in the docs.

What you will build

Tasks Mini gives users three chat commands, two panels, a skeleton that surfaces pending counts, and a daily cleanup job:

Surface	What it does
`list_tasks` — read	Returns the user's task list with optional status filter
`create_task` — write	Creates a new task; refreshes the sidebar
`delete_task` — destructive	Deletes a task after user confirmation (KAV gate)
`update_task` — write	Updates a task's title or content; refreshes the editor
Left sidebar panel	Scrollable task list; clicking a task opens the editor
Center editor panel	Rich-text task detail with a save button
Skeleton — `pending_tasks`	Surfaces pending count to the LLM intent classifier
Daily cleanup schedule	Archives completed tasks older than 30 days for every user

By the end of this tutorial:

Your extension validates with imperal validate . — zero ERRORs.
A pytest suite covers every handler path.
You understand why each line is written the way it is.

Prerequisites

Python 3.11 or newer
pip install "imperal-sdk[dev]>=5.0.0" — SDK + test extras. The >=5.0.0 pin is required: SDK 5.0.0 introduced the unified chain orchestrator (kernel-side chain_executor is now the sole multi-tool dispatcher; the SDK's in-process LLM router was removed). Tutorials in this section assume the 5.x contract. 5.0.1 (current) adds the additive Federal Typed Return Contract used in Step 4 below.
Familiarity with async Python and Pydantic basics
You have read Local dev setup — this tutorial skips env-setup steps covered there

No running web-kernel is needed for any step until the final deploy section. All tests run locally with MockContext.

Step 1: Project skeleton

Every production extension uses the same flat file layout. Create the directory and files now — you will fill them in through the steps below.

icon.svg

imperal.json

app.py

models.py

handlers_tasks.py

handlers_update.py

panels.py

skeleton.py

schedule.py

mkdir tasks-mini && cd tasks-mini
mkdir tests && touch tests/__init__.py
touch app.py models.py handlers_tasks.py handlers_update.py panels.py skeleton.py schedule.py

icon.svg — drop any valid SVG with a viewBox attribute. A minimal placeholder:

<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2">
  <path d="M9 11l3 3L22 4"/><path d="M21 12v7a2 2 0 0 1-2 2H5a2 2 0 0 1-2-2V5a2 2 0 0 1 2-2h11"/>
</svg>

V21 requires: .svg extension, file ≤ 100 KB, valid SVG root with a viewBox attribute, no embedded base64 raster images. The validator rejects files that miss any of these.

imperal.json starts empty — you will generate it from code in Step 10.

Why this layout? The production reference extension (tasks, notes) separates handler logic by domain into handlers_*.py files so that each file stays well under 300 lines. app.py wires together the Extension and ChatExtension instances and imports all handler modules so their decorators register.

See Local dev setup — file structure for the full convention.

Step 2: Define typed params

All @chat.function handlers must accept a Pydantic BaseModel as their second argument (V17). Define all param models in models.py at module scope — the SDK inspects func.__globals__ to verify module-scope placement and auto-derive the JSON schema.

models.py

from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field


class ListTasksParams(BaseModel):
    status: Literal["all", "pending", "done"] = Field(
        "all",
        description="Filter tasks by status. 'all' returns everything.",
    )


class CreateTaskParams(BaseModel):
    title: str = Field(
        description="Task title — a short phrase describing what needs to be done.",
        min_length=1,
        max_length=200,
    )
    content: str = Field(
        "",
        description="Optional rich-text body with additional detail.",
    )


class UpdateTaskParams(BaseModel):
    task_id: str = Field(description="ID of the task to update.")
    title: str | None = Field(None, description="New title. Omit to keep existing.")
    content: str | None = Field(None, description="New body content. Omit to keep existing.")
    status: str | None = Field(
        None,
        description="New status. One of 'pending', 'done', 'archived'. Omit to keep existing.",
    )


class DeleteTaskParams(BaseModel):
    task_id: str = Field(description="ID of the task to permanently delete.")

Why Field(description=...)? The SDK emits each field's description into the LLM tool-use schema. The LLM reads these descriptions to decide what value to pass — quality field descriptions reduce hallucinations and improve tool selection precision.

Why Literal["all", "pending", "done"]? An enum type generates a JSON schema enum constraint. The LLM's Pydantic feedback loop (SDK v4.1.0) validates the model before passing it to your handler — invalid values are caught and retried automatically rather than crashing the handler.

See Testing typed Pydantic params and the full @chat.function reference.

Step 3: `app.py` — wire Extension and ChatExtension

app.py is the entry point. It constructs the two central objects (ext and chat) and imports the other modules so their decorators register.

app.py

from __future__ import annotations

from imperal_sdk import Extension, ActionResult  # noqa: F401 — re-exported for handler imports
from imperal_sdk.chat import ChatExtension

ext = Extension(
    "tasks-mini",
    version="1.0.0",
    display_name="Tasks Mini",
    description=(
        "Tasks Mini — a lightweight personal task manager. "
        "Create, read, update, and delete tasks through chat or the panel UI."
    ),
    icon="icon.svg",
    actions_explicit=True,
)

chat = ChatExtension(
    ext=ext,
    tool_name="tasks-mini",
    description=(
        "Tasks Mini assistant. Ask me to list your tasks, create a new one, "
        "update an existing task, or delete one."
    ),
)

<Callout type="warn" title="`tool_name=` and `system_prompt=` are deprecated in SDK 5.0.0 — removed in 5.1.0">
  As of SDK 5.0.0 the `ChatExtension.__init__` kwargs `tool_name=` and `system_prompt=` are no-op and emit a `DeprecationWarning`. The in-process LLM router that consumed them was removed — multi-tool reasoning now routes through the kernel `chain_executor` driven by the classifier (see [Unified Chain Orchestrator changelog entry](/en/changelog#500--2026-05-15--unified-chain-orchestrator)).

  - **`tool_name=`** — the kernel resolves tool routing from `app_id` + per-function names; the orchestrator-tool concept no longer exists. The kwarg remains tolerated in 5.0.x for backward compatibility; the constructor emits a deprecation warning and is scheduled for removal in 5.1.0.
  - **`system_prompt=`** — move the classifier-readable prose into `Extension(description=...)` (Tasks-Mini already does this above) and into each `@chat.function(description=...)`. Both fields are surfaced to the classifier and narrator.

  New extensions can omit both kwargs entirely once 5.1.0 ships. The example above keeps `tool_name=` to remain copy-pasteable on the current 5.0.x line.
</Callout>

# Import handler modules so their @chat.function / @ext.panel decorators register.
# Order matters only when there are circular imports — there are none here.
import handlers_tasks   # noqa: E402, F401
import handlers_update  # noqa: E402, F401
import panels           # noqa: E402, F401
import skeleton         # noqa: E402, F401
import schedule         # noqa: E402, F401


@ext.on_install
async def on_install(ctx: object) -> None:
    """Create the tasks collection index on first install."""
    # ctx.store creates documents on demand — nothing to provision here.
    # Real extensions might run a SQL migration: await ctx.db.session().execute(...)
    await ctx.log("tasks-mini installed")  # type: ignore[attr-defined]


@ext.health_check
async def health_check(ctx: object) -> dict:
    # Return a plain dict — the web-kernel accepts {"healthy": bool, "message": str}
    return {"healthy": True, "message": "tasks-mini OK"}

Key decisions:

actions_explicit=True is the default and required by V19. It tells the web-kernel that every write/destructive tool has explicit chain_callable=True in its decorator.
description on Extension must be ≥ 40 characters and must not equal the app_id (V14). The description above is 94 characters.
display_name must be ≥ 3 characters and not equal app_id (V15). "Tasks Mini" satisfies both.
@ext.on_install satisfies V12 (INFO: missing on_install hook). It is good practice even when no provisioning is needed.
@ext.health_check satisfies V9 (WARN: missing health check).

See Decorators reference — extension class and the Extension constructor reference.

Step 4: Implement `list_tasks` — read tool

Read tools never mutate state. The LLM calls them freely and without a confirmation gate.

handlers_tasks.py

from __future__ import annotations

from imperal_sdk import ActionResult
from imperal_sdk.chat import ChatExtension

from app import chat
from models import ListTasksParams, CreateTaskParams, DeleteTaskParams


@chat.function(
    "list_tasks",
    description=(
        "List the user's tasks. Returns task IDs, titles, status, and creation dates. "
        "Accepts an optional status filter: 'all' (default), 'pending', or 'done'."
    ),
    action_type="read",
)
async def list_tasks(ctx: object, params: ListTasksParams) -> ActionResult:
    from imperal_sdk import Context  # type: ignore[attr-defined]
    ctx_typed: Context = ctx  # type: ignore[assignment]

    where: dict | None = None
    if params.status != "all":
        where = {"status": params.status}

    page = await ctx_typed.store.query("tasks", where=where, order_by="created_at", limit=100)

    tasks = [
        {
            "task_id": doc.id,
            "title": doc.data.get("title", ""),
            "status": doc.data.get("status", "pending"),
            "created_at": str(doc.created_at),
        }
        for doc in page.data
    ]

    status_label = params.status if params.status != "all" else "all"
    summary = f"Found {len(tasks)} {status_label} task(s)."
    return ActionResult.success({"tasks": tasks, "count": len(tasks)}, summary=summary)

Why order_by="created_at"? Most recent tasks first — the ordering you want in every list view. The store client accepts any field name present in the documents.

Why no error handling here? ctx.store.query returns an empty Page when there are no results — it does not raise. The only failure modes (network error, auth failure) are genuine infrastructure faults best left as unhandled exceptions so the web-kernel's generic error path fires them cleanly. Add try/except only when you can provide actionable recovery context.

See Error handling idioms and the ActionResult reference.

Step 4.5: Declare the return shape — `data_model=` (Federal Typed Return Contract, v5.0.1)

The handler above ships a perfectly valid 5.0.0-era read tool, but the v5.0.1 Federal Typed Return Contract wants every action_type="read" function to also declare the shape of ActionResult.data so the platform can validate downstream chain references at classify time.

Add a Pydantic TaskRecord model to models.py and wire it through data_model=:

models.py (append)

from typing import Literal

class TaskRecord(BaseModel):
    task_id: str = Field(description="Stable task identifier.")
    title: str = Field(description="Task title.")
    content: str = Field("", description="Task body.")
    status: Literal["pending", "done", "archived"] = Field(
        "pending", description="Lifecycle status."
    )
    created_at: str = Field(description="ISO-8601 creation timestamp.")

Then update the list_tasks decorator (the rest of the handler body stays the same):

handlers_tasks.py (replace the list_tasks decorator)

from imperal_sdk.chat.action_result import ActionResult
from models import TaskRecord  # noqa: F401 — used by data_model=

@chat.function(
    "list_tasks",
    description=(
        "List the user's tasks. Returns task IDs, titles, status, and creation dates. "
        "Accepts an optional status filter: 'all' (default), 'pending', or 'done'."
    ),
    action_type="read",
    data_model=TaskRecord,   # v5.0.1 — typed return contract
)
async def list_tasks(ctx: object, params: ListTasksParams) -> ActionResult[TaskRecord]:
    ...

Why both data_model=TaskRecord and -> ActionResult[TaskRecord]? Either alone is enough. The SDK resolves _return_model in this order:

Explicit data_model= kwarg — wins outright.
Direct -> SomeBaseModel return annotation.
-> ActionResult[T] generic — T is extracted via typing.get_args.
None of the above — _return_model = None and validator V23 surfaces a WARN.

Setting both gives the next reader a redundant signal and lets mypy/pyright type-check the return statements without weakening the runtime contract.

What the platform does with _return_model:

Emits a return_schema block into imperal.json so the kernel knows the field set without re-deriving from prose.
Validates $REF:<app_id>[<n>].path references in classifier-emitted action plans against real field names — a chain step that wrote $REF:tasks-mini[0].titel (typo) is hard-rejected at plan time instead of silently resolving to None at runtime.
Renders return_fields in the classifier envelope so the LLM sees what shape it can read without scraping previous tool outputs.
Runs data.model_validate(...) at emit time. In 5.0.1 mismatch logs a structured warning only; the call still succeeds.

Field-name symmetry rule (v5.0.1 changelog). Keep field names consistent across *Params (input) and the return data_model. The drift class we are closing: a notes extension whose input model had content_text but whose record exposed content, so $REF:notes[0].content_text silently resolved to None at runtime. Pick one spelling and use it on both sides.

Validator severity. V23 (read tools missing data_model) ships as WARN in 5.0.1 and can be promoted to ERROR via IMPERAL_VALIDATOR_V23_SEVERITY=error after third-party adoption. V24 (write/destructive tools missing data_model) is WARN-only and stays advisory — declaring data_model on writes still lets the chain narrator and audit ledger describe the resulting entity shape without re-deriving from prose. Apply the same data_model=TaskRecord kwarg to create_task, update_task, and delete_task if you want to clear V24 too.

See @chat.function reference — data_model and the v5.0.1 Federal Typed Return Contract changelog entry.

Step 5: Implement `create_task` — write tool

Write tools mutate state. They require chain_callable=True (enforced by V19 when actions_explicit=True) and should declare effects and event for compliance and panel refresh.

handlers_tasks.py (continued)

@chat.function(
    "create_task",
    description=(
        "Create a new task for the user with a required title and optional body content. "
        "Returns the new task ID. Refreshes the task sidebar panel after creation."
    ),
    action_type="write",
    chain_callable=True,
    effects=["task.create"],
    event="tasks-mini.task.created",
)
async def create_task(ctx: object, params: CreateTaskParams) -> ActionResult:
    from imperal_sdk import Context  # type: ignore[attr-defined]
    ctx_typed: Context = ctx  # type: ignore[assignment]

    if not params.title.strip():
        return ActionResult.error("Task title cannot be blank.", retryable=True)

    doc = await ctx_typed.store.create(
        "tasks",
        {
            "title": params.title.strip(),
            "content": params.content,
            "status": "pending",
        },
    )

    await ctx_typed.log(f"task created: {doc.id}")

    return ActionResult.success(
        {"task_id": doc.id, "title": params.title},
        summary=f"Created task: \"{params.title}\".",
        refresh_panels=["sidebar"],
    )

refresh_panels=["sidebar"] tells the web-kernel to re-fetch only the sidebar panel after this call succeeds. Passing None (the default) refreshes all panels — a sensible default when you are not sure which panels need updating. Passing [] refreshes nothing. The sidebar panel is registered as "sidebar" in Step 6.

effects=["task.create"] is a compliance tag. V20 warns if write/destructive tools omit it. Severity has tracked the SDK major: WARN in v4 (legacy), ERROR-eligible in v5+ once Dev Portal flips the validator promotion env (default still WARN in 5.0.x — see the V20 row in the validator table below). Use verb:noun format ("task.create", "task.delete", "email.send", "pii.read").

event="tasks-mini.task.created" declares the SSE event name. Panel refresh modes set to "on_event:tasks-mini.task.created" will re-fetch automatically when this event fires.

See chat-function reference — effects and event and Refresh after write recipe.

Step 6: Implement `delete_task` — destructive tool

Destructive tools go through the KAV (Key Action Verification) confirmation gate. The user sees a confirmation card in chat before the action executes. Return clear, specific error messages — they appear verbatim in the confirmation UI and the audit ledger.

handlers_tasks.py (continued)

@chat.function(
    "delete_task",
    description=(
        "Permanently delete a task by its task ID. "
        "This action is irreversible and requires user confirmation before executing."
    ),
    action_type="destructive",
    chain_callable=True,
    effects=["task.delete"],
    event="tasks-mini.task.deleted",
    id_projection="task_id",
)
async def delete_task(ctx: object, params: DeleteTaskParams) -> ActionResult:
    from imperal_sdk import Context  # type: ignore[attr-defined]
    ctx_typed: Context = ctx  # type: ignore[assignment]

    doc = await ctx_typed.store.get("tasks", params.task_id)
    if doc is None:
        return ActionResult.error(
            f"Task {params.task_id!r} was not found. It may have already been deleted.",
            retryable=False,
        )

    title = doc.data.get("title", params.task_id)
    deleted = await ctx_typed.store.delete("tasks", params.task_id)
    if not deleted:
        return ActionResult.error(
            "Could not delete the task. Please try again.",
            retryable=True,
        )

    await ctx_typed.log(f"task deleted: {params.task_id}")

    return ActionResult.success(
        {"task_id": params.task_id, "title": title},
        summary=f"Deleted task: \"{title}\".",
        refresh_panels=["sidebar"],
    )

action_type="destructive" triggers the KAV confirmation gate automatically. The user must explicitly confirm before your handler executes. You do not need to implement confirmation logic — the web-kernel handles it.

id_projection="task_id" tells the chain executor which field in DeleteTaskParams is the resource ID. This is required when the tool name does not follow the simple verb-prefix heuristic (e.g., delete_task would be correctly inferred, but remove_completed_task would not — id_projection removes the ambiguity).

Two distinct error cases: doc is None (not found, not retryable — repeating the call produces the same result) vs. deleted is False (transient failure, retryable). The retryable flag controls whether the web-kernel offers a "Try again" button in the chat UI.

See Error handling — retryable vs non-retryable and @chat.function reference — id_projection.

Step 7: Implement `update_task` — write tool

Place update in its own file to demonstrate the multi-file pattern.

handlers_update.py

from __future__ import annotations

from imperal_sdk import ActionResult

from app import chat
from models import UpdateTaskParams


@chat.function(
    "update_task",
    description=(
        "Update the title, body content, or status of an existing task identified by task_id. "
        "Pass only the fields you want to change; omit the rest to keep existing values."
    ),
    action_type="write",
    chain_callable=True,
    effects=["task.update"],
    event="tasks-mini.task.updated",
    id_projection="task_id",
)
async def update_task(ctx: object, params: UpdateTaskParams) -> ActionResult:
    from imperal_sdk import Context  # type: ignore[attr-defined]
    ctx_typed: Context = ctx  # type: ignore[assignment]

    doc = await ctx_typed.store.get("tasks", params.task_id)
    if doc is None:
        return ActionResult.error(
            f"Task {params.task_id!r} not found.",
            retryable=False,
        )

    patch: dict = {}
    if params.title is not None:
        stripped = params.title.strip()
        if not stripped:
            return ActionResult.error("Task title cannot be blank.", retryable=True)
        patch["title"] = stripped
    if params.content is not None:
        patch["content"] = params.content
    if params.status is not None:
        allowed_statuses = {"pending", "done", "archived"}
        if params.status not in allowed_statuses:
            return ActionResult.error(
                f"Invalid status {params.status!r}. Use 'pending', 'done', or 'archived'.",
                retryable=True,
            )
        patch["status"] = params.status

    if not patch:
        return ActionResult.error(
            "Nothing to update — provide at least title, content, or status.",
            retryable=True,
        )

    updated_doc = await ctx_typed.store.update("tasks", params.task_id, patch)
    new_title = updated_doc.data.get("title", params.task_id)

    return ActionResult.success(
        {"task_id": params.task_id, "title": new_title},
        summary=f"Updated task: \"{new_title}\".",
        refresh_panels=["editor", "sidebar"],
    )

refresh_panels=["editor", "sidebar"] refreshes both panels — the editor shows new content and the sidebar shows the new title. Panel IDs must match the panel_id strings registered with @ext.panel.

Partial update pattern: Only include fields that have changed in the patch dict. ctx.store.update merges the patch into the existing document — fields not in the patch retain their current values.

The sidebar shows all tasks as a list. Clicking a task opens its editor in the center slot. On first load, auto_action claims the center slot with the first task automatically.

panels.py

from __future__ import annotations

from imperal_sdk import ui
from app import ext


@ext.panel(
    "sidebar",
    slot="left",
    title="Tasks",
    icon="☑️",
    default_width=280,
    min_width=220,
    max_width=460,
)
async def sidebar(ctx: object, active_task_id: str = "", **kwargs: object) -> object:
    from imperal_sdk import Context  # type: ignore[attr-defined]
    ctx_typed: Context = ctx  # type: ignore[assignment]

    page = await ctx_typed.store.query("tasks", where=None, order_by="created_at", limit=200)
    tasks = page.data

    if not tasks:
        return ui.Empty(
            message="No tasks yet — ask me to create one.",
            icon="☑️",
        )

    items = [
        ui.ListItem(
            id=doc.id,
            title=doc.data.get("title", doc.id),
            subtitle="done" if doc.data.get("status") == "done" else "pending",
            badge=ui.Badge(label="done", color="green") if doc.data.get("status") == "done" else None,
            selected=(doc.id == active_task_id),
            on_click=ui.Call("__panel__editor", task_id=doc.id),
            actions=[
                {
                    "icon": "Trash2",
                    "on_click": ui.Call("delete_task", task_id=doc.id),
                    "confirm": f"Delete \"{doc.data.get('title', doc.id)}\"?",
                }
            ],
        )
        for doc in tasks
    ]

    root = ui.Stack(
        children=[ui.List(items=items, searchable=True)],
        gap=2,
    )

    # Claim center slot on first load when no task is already active.
    if tasks and not active_task_id:
        root.props["auto_action"] = ui.Call("__panel__editor", task_id=tasks[0].id)

    return root

auto_action: The Imperal Panel frontend reads leftPanel.props.auto_action once per discovery cycle. When the sidebar is fetched and active_task_id is empty (first load), setting auto_action causes the frontend to immediately call the editor panel with the first task's ID — no user click required.

if tasks and not active_task_id: The guard has two parts: tasks ensures we never fire auto_action when the list is empty (there is nothing to open), and not active_task_id prevents re-firing when the user has already selected a task and the sidebar refreshes (e.g., after a create_task).

actions=[{"icon": ..., "on_click": ..., "confirm": ...}]: Inline actions appear as hover buttons on each list row. The "confirm" key renders a confirmation popover before the action fires — separate from the KAV gate (which acts on the chat/action level). Both can coexist: the popover is a UI-level guard, the KAV gate is a web-kernel-level guard.

ui.Call("__panel__editor", task_id=doc.id): Panel calls use the synthetic tool name __panel__{panel_id}. The task_id kwarg becomes a panel param — the editor handler receives it as task_id in **kwargs.

See Master-detail panel recipe, Auto-action on load recipe, and Concepts: panels.

Step 9: Add the center editor panel

The editor shows task details and lets the user save changes. Returning ui.Empty() at batch discovery (when task_id is absent) is essential — see the explanation below.

`center_overlay=True` is required for slot="center"

Federal v4.1.8+ contract: declaring slot="center" is not enough for the Imperal Panel host to render your panel. You must also pass center_overlay=True so the web-kernel writes the flag into unified_config. Without it, the SDK accepts the registration silently but the frontend has no render path for the slot. See Panels concept — How slot="center" activates and federal invariant I-PANEL-RENDERING-CONTRACT.

panels.py (continued)

@ext.panel(
    "editor",
    slot="center",
    title="Task editor",
    icon="📝",
    center_overlay=True,   # v4.1.8+ — REQUIRED for slot="center" to activate
)
async def editor(ctx: object, task_id: str = "", **kwargs: object) -> object:
    from imperal_sdk import Context  # type: ignore[attr-defined]
    ctx_typed: Context = ctx  # type: ignore[assignment]

    # At batch discovery the host calls this panel with no params (task_id == "").
    # Return ui.Empty() so the slot is reserved but visually shows a placeholder.
    # This allows auto_action on the sidebar to fire and claim the center slot.
    if not task_id:
        return ui.Empty(
            message="Select a task from the sidebar.",
            icon="🖱️",
        )

    doc = await ctx_typed.store.get("tasks", task_id)
    if doc is None:
        return ui.Error(
            message="Task not found. It may have been deleted.",
            retry=ui.Call("__panel__sidebar"),
        )

    title = doc.data.get("title", "")
    content = doc.data.get("content", "")
    status = doc.data.get("status", "pending")
    is_done = status == "done"

    return ui.Stack(
        children=[
            ui.Stack(
                children=[
                    ui.Header(text=title, level=2),
                    ui.Badge(
                        label="Done" if is_done else "Pending",
                        color="green" if is_done else "gray",
                    ),
                ],
                direction="h",
                gap=3,
                align="center",
            ),
            ui.Divider(),
            ui.RichEditor(
                content=content,
                placeholder="Add task details here...",
                on_save=ui.Call("update_task", task_id=task_id),
                param_name="content",
                toolbar=True,
            ),
            ui.Form(
                children=[
                    ui.Input(
                        placeholder="Rename task...",
                        value=title,
                        param_name="title",
                    ),
                ],
                action="update_task",
                submit_label="Save title",
                defaults={"task_id": task_id},
            ),
            ui.Stack(
                children=[
                    ui.Button(
                        label="Mark done" if not is_done else "Reopen",
                        variant="secondary",
                        on_click=ui.Call(
                            "update_task",
                            task_id=task_id,
                            status="done" if not is_done else "pending",
                        ),
                        icon="✅" if not is_done else "RotateCcw",
                    ),
                    ui.Button(
                        label="Delete task",
                        variant="destructive",
                        on_click=ui.Call("delete_task", task_id=task_id),
                        icon="Trash2",
                    ),
                ],
                direction="h",
                gap=2,
            ),
        ],
        gap=4,
    )

Why ui.Empty() and not return None? Returning None from a panel handler leaves leftPanel as null after the discovery cycle. The frontend's auto_action effect only fires when leftPanel is non-null. ui.Empty() returns a valid UINode, which sets leftPanel — allowing auto_action to proceed and open the first task automatically. return None is appropriate only when you want the slot to be truly absent from the layout.

ui.RichEditor(on_save=...) binds Ctrl+S in the editor to a ui.Call. The param_name="content" kwarg tells the editor to merge the current editor HTML content as content in the call params. This flows into update_task as params.content.

ui.Form(action="update_task", defaults={"task_id": task_id}) wires a form submit to the update_task function. The defaults dict pre-populates hidden fields — here task_id is injected so the handler always knows which task to update without the user needing to type it.

See UI primitives reference, UI actions reference, and Concepts: panels.

Step 10: Add the skeleton — pending count

Skeletons are live data probes that feed the LLM intent classifier. The web-kernel calls the skeleton refresh handler on a TTL schedule and stores the result as structured LLM context. This lets the classifier say "you have 3 pending tasks" without making a live store call.

skeleton.py

from __future__ import annotations

from app import ext


@ext.skeleton("pending_tasks", alert=True, ttl=300)
async def refresh_pending_tasks(ctx: object) -> dict:
    from imperal_sdk import Context  # type: ignore[attr-defined]
    ctx_typed: Context = ctx  # type: ignore[assignment]

    total = await ctx_typed.store.count("tasks")
    pending = await ctx_typed.store.count("tasks", where={"status": "pending"})
    done = total - pending

    return {
        "response": {
            "total_tasks": total,
            "pending_tasks": pending,
            "done_tasks": done,
        }
    }

Return contract: The return value must be {"response": {...}} where the inner dict contains flat scalar values (strings, ints, bools). The LLM reads these values directly from the skeleton context — use human-readable key names.

What the classifier actually sees

The web-kernel compresses every skeleton section before injecting it into the classifier prompt. A handler that returns rich data is silently simplified — knowing the compression rules is the difference between an effective skeleton and one whose contents never reach the LLM.

What this shows: the same pending_tasks skeleton with a long recent_titles array — before and after kernel compression.

Before kernel compression — your handler returns:

{
  "response": {
    "total_tasks": 47,
    "pending_tasks": 12,
    "done_tasks": 35,
    "recent_titles": [
      "Write the Q3 report",
      "Schedule team off-site",
      "Review pull request #4521",
      "Refactor the auth middleware (this title is intentionally very long to demonstrate truncation behaviour)",
      "Deploy v2 to staging",
      "Update the team handbook",
      "...19 more items..."
    ]
  }
}

After kernel compression — the classifier prompt receives:

- pending_tasks (cached ~12s ago): total_tasks=47, pending_tasks=12, done_tasks=35, recent_titles=list[25]

Why the array collapses to list[25]: the kernel's _render_skeleton_value (federal invariant I-SKELETON-SMALL-LIST-INLINE) only expands lists of dicts up to five items. The 25 plain-string titles fall outside that path and reduce to a shape hint. To make titles classifier-visible, return only the first five in recent_titles (the kernel will expand them inline with id and title if you make each entry a dict), or rely on the fact-ledger — when the user last called list_tasks, the verbatim result is preserved for the next five turns.

See Concepts: skeletons — what the kernel does with your skeleton for the full compression rule table.

alert=True: Enables change-alerting via the paired skeleton_alert_pending_tasks tool. When the pending count changes, the web-kernel can surface a notification. Omit this flag if your skeleton data changes too frequently (e.g., every second).

ttl=300: A hint to platform operators. The authoritative TTL lives in the web-kernel Registry row configured per extension. 300 seconds (5 minutes) is the default.

Access guard: ctx.skeleton.get() raises SkeletonAccessForbidden if called outside a @ext.skeleton handler. Regular tool handlers must use ctx.store or ctx.cache for data — never read from the skeleton inside a @chat.function.

See Concepts: skeletons and Decorator skeleton reference.

Step 11: Add the scheduled job — daily cleanup

The schedule handler runs with system context (ctx.user.imperal_id == "__system__"). To act on behalf of individual users, use ctx.store.list_users() + ctx.as_user() to fan out.

schedule.py

from __future__ import annotations

from datetime import datetime, timezone, timedelta

from app import ext


@ext.schedule("daily_cleanup", cron="0 2 * * *")
async def daily_cleanup(ctx: object) -> None:
    """Archive completed tasks older than 30 days for every user."""
    from imperal_sdk import Context  # type: ignore[attr-defined]
    ctx_typed: Context = ctx  # type: ignore[assignment]

    cutoff = datetime.now(timezone.utc) - timedelta(days=30)
    archived = 0
    errors = 0

    async for user_id in ctx_typed.store.list_users("tasks"):
        user_ctx = ctx_typed.as_user(user_id)
        try:
            page = await user_ctx.store.query(
                "tasks",
                where={"status": "done"},
                order_by="created_at",
                limit=500,
            )
            for doc in page.data:
                updated_at_raw = doc.updated_at
                if updated_at_raw is None:
                    continue
                # updated_at is a datetime or ISO string depending on store impl
                if isinstance(updated_at_raw, str):
                    updated_at = datetime.fromisoformat(updated_at_raw)
                else:
                    updated_at = updated_at_raw
                if updated_at.tzinfo is None:
                    updated_at = updated_at.replace(tzinfo=timezone.utc)

                if updated_at < cutoff:
                    await user_ctx.store.update("tasks", doc.id, {"status": "archived"})
                    archived += 1
        except Exception as exc:  # noqa: BLE001
            errors += 1
            await ctx_typed.log(
                f"daily_cleanup: error for user {user_id}: {exc}",
                level="error",
            )

    await ctx_typed.log(
        f"daily_cleanup complete: {archived} archived, {errors} user errors",
        level="info",
    )

cron="0 2 * * *" runs at 02:00 UTC every day. Use standard 5-field cron format: minute hour day-of-month month day-of-week. The web-kernel executes your handler inside an ICNLI Worker activity — the same runtime as regular tool calls, with the same timeout constraints.

ctx.store.list_users("tasks") is a system-context-only method that yields user IDs with documents in the "tasks" collection. It raises RuntimeError if called from a non-system context.

ctx.as_user(user_id) returns a derived Context scoped to that user. Store reads and writes through this context are isolated to the user's data — the same isolation as when the user calls the tool directly.

Per-user try/except: Swallow individual user errors so one bad user record does not abort the job for everyone else. Log the error for observability but do not re-raise.

See Concepts: scheduled tasks and Decorator schedule reference.

Step 12: Validate the manifest

Before writing tests, run the validator to catch any structural issues:

# From inside the tasks-mini/ directory
imperal validate .

Expected output (zero errors):

Validation complete: 0 errors, 2 warnings, 0 info
  WARN V10  create_task: event= field present but no @ext.emits declaration found
  WARN V20  create_task: effects declared — WARN in v4 legacy, ERROR-eligible in v5+ (env-promotable)

The V10 warning about missing @ext.emits is expected — @ext.emits is optional for event publishing when you rely on SSE panel refresh rather than cross-extension subscriptions. You can silence it by adding @ext.emits("tasks-mini.task.created") to app.py.

What the validator checks:

Rule	What it catches
V1	`app_id` format: `[a-z0-9][a-z0-9-]*[a-z0-9]`
V14	`description` ≥ 40 chars, ≠ `app_id`
V15	`display_name` ≥ 3 chars, ≠ `app_id`
V16	Each `@chat.function` description ≥ 20 chars
V17	Each `@chat.function` has a Pydantic `BaseModel` param
V18	Each `@chat.function` has `-> ActionResult` return annotation
V19	`actions_explicit=True` AND write/destructive tools have `chain_callable=True`
V20	`write`/`destructive` tools declare `effects=[...]` — ERROR in SDK v5+ (was WARN in v4 legacy; v5.0.x still emits as WARN, Dev Portal can promote per-extension via env)
V21	`icon.svg` exists, ≤ 100 KB, valid SVG with `viewBox`
V22	Lifecycle hook signatures match the SDK contract (`on_refresh(ctx, message=None)` etc.)
V23	Read tools SHOULD declare `data_model=...` so `$REF` paths can be validated against the schema. WARN by default in 5.0.1; promotable to ERROR via `IMPERAL_VALIDATOR_V23_SEVERITY=error`. Auto-satisfied by `-> ActionResult[T]` or `-> SomeBaseModel` return annotations.
V24	Write/destructive tools SHOULD declare `data_model=...` (WARN-only — advisory; lets the chain narrator + audit ledger describe the resulting entity without re-deriving from prose).
V24-AST	No `ctx.skeleton.*` attribute access inside `@chat.function` handler bodies (ERROR — AST walk). Handlers use `ctx.api` / `ctx.store`; skeleton is the LLM context cache, refreshed via `@ext.skeleton` handlers.
V25	Manifests MUST NOT contain `tool_<ext>_chat` orchestrator-tool entries (ERROR). The SDK 5.0.0 manifest emitter no longer produces them; this validator rejects legacy manifests carried over from v4.
V31	`Extension(system=True)` allowed only for first-party Imperal authors. SDK 5.0.1+ uses an env-driven local check (`IMPERAL_FIRSTPARTY_AUTHOR_IDS=<comma-list>`); the SDK no longer ships an embedded allowlist. The Dev Portal enforces the authoritative allowlist server-side at publish.

After validation passes, generate imperal.json:

imperal build .

This writes imperal.json from your Python Extension object. Commit this file — the web-kernel reads it at install time.

See the Validators reference for the full rule list.

Step 13: Write tests

Every handler path needs a test. The imperal_sdk.testing module provides MockContext with pluggable in-memory store, AI, billing, and notify mocks — no running web-kernel needed.

tests/test_tasks_mini.py

from __future__ import annotations

import pytest
from imperal_sdk.testing import MockContext
from imperal_sdk import ActionResult

# Import the handlers so the decorators register before the tests run.
import sys, os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

import app  # noqa: F401 — registers all decorators
from models import ListTasksParams, CreateTaskParams, UpdateTaskParams, DeleteTaskParams
from handlers_tasks import list_tasks, create_task, delete_task
from handlers_update import update_task
from skeleton import refresh_pending_tasks


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

@pytest.fixture()
def ctx() -> MockContext:
    return MockContext(user_id="user_test_01", role="user", tool_type="tool")


@pytest.fixture()
def skel_ctx() -> MockContext:
    """Skeleton handlers require tool_type='skeleton'."""
    return MockContext(user_id="user_test_01", role="user", tool_type="skeleton")


@pytest.fixture()
def sys_ctx() -> MockContext:
    """Schedule handlers run in system context."""
    return MockContext(user_id="__system__", role="system", tool_type="tool", is_system=True)


# ---------------------------------------------------------------------------
# list_tasks
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_list_tasks_empty(ctx: MockContext) -> None:
    result: ActionResult = await list_tasks(ctx, ListTasksParams(status="all"))
    assert result.status == "success"
    assert result.data["tasks"] == []
    assert result.data["count"] == 0


@pytest.mark.asyncio
async def test_list_tasks_returns_tasks(ctx: MockContext) -> None:
    await ctx.store.create("tasks", {"title": "Buy milk", "status": "pending"})
    await ctx.store.create("tasks", {"title": "Read docs", "status": "done"})

    result: ActionResult = await list_tasks(ctx, ListTasksParams(status="all"))
    assert result.status == "success"
    assert result.data["count"] == 2


@pytest.mark.asyncio
async def test_list_tasks_status_filter(ctx: MockContext) -> None:
    await ctx.store.create("tasks", {"title": "Task A", "status": "pending"})
    await ctx.store.create("tasks", {"title": "Task B", "status": "done"})

    result: ActionResult = await list_tasks(ctx, ListTasksParams(status="pending"))
    assert result.status == "success"
    assert result.data["count"] == 1
    assert result.data["tasks"][0]["status"] == "pending"


# ---------------------------------------------------------------------------
# create_task
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_create_task_success(ctx: MockContext) -> None:
    result: ActionResult = await create_task(ctx, CreateTaskParams(title="Write tests"))
    assert result.status == "success"
    assert "task_id" in result.data
    assert result.data["title"] == "Write tests"
    assert result.refresh_panels == ["sidebar"]


@pytest.mark.asyncio
async def test_create_task_blank_title(ctx: MockContext) -> None:
    result: ActionResult = await create_task(ctx, CreateTaskParams(title="   "))
    assert result.status == "error"
    assert result.retryable is True


@pytest.mark.asyncio
async def test_create_task_stores_document(ctx: MockContext) -> None:
    await create_task(ctx, CreateTaskParams(title="My task", content="Details here"))
    page = await ctx.store.query("tasks", where={"status": "pending"})
    assert any(d.data["title"] == "My task" for d in page.data)


# ---------------------------------------------------------------------------
# delete_task
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_delete_task_success(ctx: MockContext) -> None:
    doc = await ctx.store.create("tasks", {"title": "To delete", "status": "pending"})
    result: ActionResult = await delete_task(ctx, DeleteTaskParams(task_id=doc.id))
    assert result.status == "success"
    assert result.data["task_id"] == doc.id
    assert result.refresh_panels == ["sidebar"]


@pytest.mark.asyncio
async def test_delete_task_not_found(ctx: MockContext) -> None:
    result: ActionResult = await delete_task(ctx, DeleteTaskParams(task_id="does-not-exist"))
    assert result.status == "error"
    assert result.retryable is False


# ---------------------------------------------------------------------------
# update_task
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_update_task_title(ctx: MockContext) -> None:
    doc = await ctx.store.create("tasks", {"title": "Old title", "status": "pending"})
    result: ActionResult = await update_task(
        ctx, UpdateTaskParams(task_id=doc.id, title="New title")
    )
    assert result.status == "success"
    assert result.data["title"] == "New title"
    assert "editor" in (result.refresh_panels or [])


@pytest.mark.asyncio
async def test_update_task_not_found(ctx: MockContext) -> None:
    result: ActionResult = await update_task(
        ctx, UpdateTaskParams(task_id="ghost-id", title="Anything")
    )
    assert result.status == "error"
    assert result.retryable is False


@pytest.mark.asyncio
async def test_update_task_blank_title_rejected(ctx: MockContext) -> None:
    doc = await ctx.store.create("tasks", {"title": "Real title", "status": "pending"})
    result: ActionResult = await update_task(
        ctx, UpdateTaskParams(task_id=doc.id, title="  ")
    )
    assert result.status == "error"
    assert result.retryable is True


@pytest.mark.asyncio
async def test_update_task_nothing_to_update(ctx: MockContext) -> None:
    doc = await ctx.store.create("tasks", {"title": "Unchanged", "status": "pending"})
    result: ActionResult = await update_task(
        ctx, UpdateTaskParams(task_id=doc.id)  # no title or content
    )
    assert result.status == "error"


# ---------------------------------------------------------------------------
# Skeleton
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_skeleton_returns_response_key(skel_ctx: MockContext) -> None:
    result = await refresh_pending_tasks(skel_ctx)
    assert "response" in result
    assert "total_tasks" in result["response"]
    assert "pending_tasks" in result["response"]


@pytest.mark.asyncio
async def test_skeleton_counts_correctly(skel_ctx: MockContext) -> None:
    await skel_ctx.store.create("tasks", {"title": "A", "status": "pending"})
    await skel_ctx.store.create("tasks", {"title": "B", "status": "pending"})
    await skel_ctx.store.create("tasks", {"title": "C", "status": "done"})

    result = await refresh_pending_tasks(skel_ctx)
    assert result["response"]["total_tasks"] == 3
    assert result["response"]["pending_tasks"] == 2
    assert result["response"]["done_tasks"] == 1

Run the suite:

cd tasks-mini
pytest tests/ -v

Expected output:

tests/test_tasks_mini.py::test_list_tasks_empty PASSED
tests/test_tasks_mini.py::test_list_tasks_returns_tasks PASSED
tests/test_tasks_mini.py::test_list_tasks_status_filter PASSED
tests/test_tasks_mini.py::test_create_task_success PASSED
tests/test_tasks_mini.py::test_create_task_blank_title PASSED
tests/test_tasks_mini.py::test_create_task_stores_document PASSED
tests/test_tasks_mini.py::test_delete_task_success PASSED
tests/test_tasks_mini.py::test_delete_task_not_found PASSED
tests/test_tasks_mini.py::test_update_task_title PASSED
tests/test_tasks_mini.py::test_update_task_not_found PASSED
tests/test_tasks_mini.py::test_update_task_blank_title_rejected PASSED
tests/test_tasks_mini.py::test_update_task_nothing_to_update PASSED
tests/test_tasks_mini.py::test_skeleton_returns_response_key PASSED
tests/test_tasks_mini.py::test_skeleton_counts_correctly PASSED

14 passed in 0.38s

MockContext key behaviours:

Mock	What it provides
`ctx.store`	In-memory store with `create`, `get`, `query`, `update`, `delete`, `count`
`ctx.log`	Records log messages in `ctx.log.messages`
`ctx.notify`	Records notifications in `ctx.notify.sent`
`tool_type="skeleton"`	Required for handlers decorated with `@ext.skeleton` — other values raise `SkeletonAccessForbidden`
`is_system=True`	Enables `ctx.store.list_users()` and `ctx.as_user()`

See Testing extensions guide and MockContext reference.

Step 14: Run the compiled-code check

Run py_compile on every Python file before committing. This is a mandatory pre-deploy gate (rule 4 in the platform's mandatory rules):

python3 -m py_compile app.py models.py handlers_tasks.py handlers_update.py panels.py skeleton.py schedule.py

No output means all files compile cleanly. A syntax error will produce a SyntaxError message pointing to the file and line.

If you have pyright installed:

pyright --level error .

Pyright catches type annotation mismatches before they surface in production.

Step 15: Local iteration with `imperal dev`

Once tests pass and the validator is clean, you can run against a local ICNLI Worker instance:

imperal dev --extension .

This loads your extension into the local worker runtime. Chat with the assistant in the Imperal Panel (or the local dev UI if configured). Use the tasks-mini extension to:

Ask "show my tasks" → exercises list_tasks
Ask "create a task called Write the tutorial" → exercises create_task, sidebar should refresh
Click a task in the sidebar → editor opens via auto_action
Press Ctrl+S in the editor → update_task fires via RichEditor.on_save
Ask "delete task X" → KAV confirmation card appears; confirm → delete_task fires

See Local dev setup — running your extension for dev environment prerequisites.

Step 16: Deploy via the Developer Portal

When the extension is ready for production:

Run imperal build . to ensure imperal.json is current.
Run imperal validate . to verify the full federal validator set (V14-V22 + V23 + V24 + V24-AST + V25 + V31) passes locally.
Upload your packaged extension at the Imperal Developer Portal — drag-and-drop your built artefact, which triggers the review queue.
The portal re-runs every validator server-side — any ERROR must be resolved before the extension is published. V25 (tool_*_chat orchestrator entries) is enforced as a hard publish gate: any manifest carrying a legacy orchestrator-tool from the pre-5.0.0 emitter will be rejected.
After approval, the extension appears in the Marketplace and can be installed by tenants.

What you have covered

This tutorial exercised the full SDK surface. Here is a checklist of every API you used and the reference page for each:

API	Tutorial step	Reference
`Extension(app_id, display_name, description, icon, actions_explicit)`	Step 3	Decorators reference
`ChatExtension(ext, tool_name, description)`	Step 3	chat-function reference
`@chat.function(action_type="read")`	Step 4	chat-function reference
`@chat.function(action_type="write", chain_callable, effects, event)`	Steps 5, 7	chat-function reference
`@chat.function(action_type="destructive", id_projection)`	Step 6	chat-function reference
`@chat.function(data_model=...)` + `-> ActionResult[T]` (Federal Typed Return Contract)	Step 4.5	chat-function reference — data_model
Pydantic `BaseModel` params at module scope	Step 2	chat-function reference — Pydantic params
`ActionResult.success(data, summary, refresh_panels)`	Steps 4–7	UI actions reference
`ActionResult.error(msg, retryable)`	Steps 5–7	Error handling guide
`ctx.store.query`, `.create`, `.get`, `.update`, `.delete`, `.count`	Steps 4–11	ctx reference
`ctx.log(message, level)`	Steps 5, 6, 11	ctx reference
`@ext.panel(slot="left", default_width, min_width, max_width)`	Step 8	Concepts: panels
`@ext.panel(slot="center", center_overlay=True)`	Step 9	Concepts: panels
`root.props["auto_action"]` on left panel	Step 8	Auto-action on load recipe
`ui.List`, `ui.ListItem` with `actions`	Step 8	UI primitives reference
`ui.Empty`, `ui.Error`, `ui.Header`, `ui.Divider`	Steps 8, 9	UI primitives reference
`ui.Stack`, `ui.Badge`, `ui.Button`	Steps 8, 9	UI primitives reference
`ui.RichEditor`, `ui.Form`, `ui.Input`	Step 9	UI primitives reference
`ui.Call`, `ui.Navigate`	Steps 8, 9	UI actions reference
`@ext.skeleton(section_name, alert, ttl)`	Step 10	Decorator skeleton reference
Skeleton return contract `{"response": {...}}`	Step 10	Concepts: skeletons
`@ext.schedule(name, cron)`	Step 11	Decorator schedule reference
`ctx.store.list_users()` + `ctx.as_user()` fan-out	Step 11	Concepts: scheduled tasks
`@ext.on_install`, `@ext.health_check`	Step 3	Decorators reference
V14-V22 + V23 + V24 + V24-AST + V25 + V31 validator rules	Step 12	Validators reference
`MockContext`, `MockStore`	Step 13	Testing extensions guide
`pytest` + `pytest-asyncio`	Step 13	Testing extensions guide