Decorators reference

from imperal_sdk import Extension, ChatExtension, ActionResult
from pydantic import BaseModel, Field

ext = Extension("my-app", display_name="My App", description="...", icon="icon.svg", actions_explicit=True)
chat = ChatExtension(ext, tool_name="my-app", description="My App assistant")

class CreateTaskParams(BaseModel):
    title: str = Field(description="Task title — what needs to be done")
    due_at: str | None = Field(None, description="Optional ISO datetime when the task is due")

@chat.function(
    description="Create a new task for the user.",
    action_type="write",
    chain_callable=True,
    effects=["task.create"],
    event="task.created",
    id_projection=None,
)
async def create_task(ctx, params: CreateTaskParams):
    doc = await ctx.store.create("tasks", {"title": params.title})
    return ActionResult.success({"task_id": doc.id}, summary=f"Created task: {params.title}")

@ext.skeleton("tasks_summary", alert=True, ttl=300)
async def refresh_tasks_skeleton(ctx) -> dict:
    total   = await ctx.store.count("tasks")
    pending = await ctx.store.count("tasks", where={"status": "pending"})
    return {"response": {
        "total":   total,
        "pending": pending,
    }}

from imperal_sdk import Extension, ui

ext = Extension("my-app", ...)

@ext.panel(
    "sidebar",
    slot="left",
    title="My Sidebar",
    icon="📜",
    refresh="on_event:my-ext.item.created,my-ext.item.updated",
    default_width=280,
    min_width=200,
    max_width=500,
)
async def sidebar(ctx, **params):
    items = await ctx.store.query("items")
    if not items.data:
        return ui.Empty(message="No items yet.")
    return ui.List(items=[ui.Text(i.data["label"]) for i in items.data])

@ext.panel("detail", slot="right", title="Detail", default_width=320)
async def detail(ctx, **params):
    item_id: str = params.get("item_id", "")
    if not item_id:
        return ui.Empty(message="Select an item.")
    doc = await ctx.store.get("items", item_id)
    return ui.Stack(direction="v", children=[
        ui.Text(doc.data["title"]),
    ])

@ext.schedule("nightly_archive", cron="0 3 * * *")
async def nightly_archive(ctx):
    async for user_id in ctx.store.list_users():
        user_ctx = ctx.as_user(user_id)
        await process_user(user_ctx)

import hmac, hashlib

@ext.webhook("/incoming", method="POST", secret_header="X-Hub-Signature-256")
async def receive_webhook(ctx, headers: dict, body: str, query_params: dict):
    # Verify HMAC manually — no SDK helper
    sig = headers.get("X-Hub-Signature-256", "")
    secret = ctx.config.require("webhook_secret")
    expected = "sha256=" + hmac.new(
        secret.encode(), body.encode(), hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(sig, expected):
        return {"error": "Invalid signature"}
    payload = json.loads(body)
    # process...
    return {"ok": True}

@ext.on_event("email.received")
async def handle_email_received(ctx, event):
    subject = event.data.get("subject", "")
    await ctx.notify(f"New email: {subject}")

@ext.emits("my-app.item.created", schema_ref="ItemCreatedEvent")
async def _declare_item_created(): ...

@ext.emits("my-app.item.deleted")
async def _declare_item_deleted(): ...

from pydantic import BaseModel

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

@ext.cache_model("inbox_summary")
class _InboxSummary(InboxSummary):
    pass

@ext.expose("get_summary", action_type="read")
async def get_summary(ctx, **kwargs):
    return {"summary": "..."}

@ext.tray("unread", icon="📧", tooltip="Unread messages")
async def tray_unread(ctx, **kwargs):
    count = await ctx.store.count("messages", where={"read": False})
    return ui.Stack([ui.Badge(str(count), color="red" if count > 0 else "gray")])

ext.secret(
    name="spotify_api_key",
    description="Your Spotify API key (from developer.spotify.com).",
    required=True,
    write_mode="user",
    max_bytes=200,
)(lambda: None)

@chat.function
async def search(ctx, query: str):
    api_key = await ctx.secrets.get("spotify_api_key")
    # Use api_key for the duration of this handler call only.
    ...

@ext.tool("skeleton_alert_tasks", scopes=[], description="Alert when tasks change.")
async def skeleton_alert_tasks(ctx):
    return {"changed": True}

@ext.on_install
async def on_install(ctx):
    await ctx.store.create("settings", {"theme": "dark"})

@ext.on_upgrade("2.0.0")
async def upgrade_to_2_0(ctx):
    await ctx.store.update("settings", "main", {"schema_v": 2})

@ext.on_uninstall
async def on_uninstall(ctx): ...

@ext.on_enable
async def on_enable(ctx): ...

@ext.on_disable
async def on_disable(ctx): ...

@ext.health_check
async def health_check(ctx):
    return HealthStatus(healthy=True)