Imperal Docs
SDK Reference

Lifecycle hooks reference

@ext.on_install / on_uninstall / on_enable / on_disable / on_upgrade / health_check — web-kernel-driven extension lifecycle events

Lifecycle hooks let an extension react to installation, removal, enable/disable transitions, version upgrades, and periodic health pings driven by the web-kernel. Unlike @chat.function or @ext.tool, lifecycle hooks are never called by the LLM — they are called by the web-kernel runtime on specific platform events.

All six hooks are methods on the Extension instance. Register them from app.py, or import ext from app.py into a dedicated lifecycle.py module.


The extension lifecycle state machine

An extension for a given user passes through a well-defined set of states:

[not installed]

      ▼  install
  [installed / enabled]
      │                  │
      ▼  disable         ▼  uninstall
  [disabled]         [not installed]

      ▼  enable
  [installed / enabled]

      ▼  upgrade (version bump)
  [installed / enabled — new version]
HookFires whenTypical use
@ext.on_installUser installs the extension for the first timeProvision default store documents, log install event
@ext.on_uninstallUser or admin removes the extensionTear down user data, log uninstall
@ext.on_enableAdmin re-enables a previously disabled extensionWarm up caches, re-register scheduled workers
@ext.on_disableAdmin disables the extension for a user or tenantSuspend background jobs, flush volatile state
@ext.on_upgradeExtension version bumps for an existing installSchema migrations, data backfills
@ext.health_checkWeb-kernel ping every 60 secondsProbe backend connectivity, report status

One hook per event per Extension instance

Each of on_install, on_uninstall, on_enable, on_disable, and health_check may be registered at most once per Extension instance. Re-registering overwrites the previous handler silently. on_upgrade is keyed by version string and can have multiple handlers — one per version bump.


Common patterns

Idempotency in install and enable hooks

on_install and on_enable may be called more than once in edge cases (e.g. retry after a transient error). Write handlers defensively:

from imperal_sdk import Extension
from imperal_sdk.chat import ActionResult

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.on_install
async def on_install(ctx) -> None:
    # Use set() (upsert) rather than create() to stay idempotent.
    await ctx.store.set(
        f"config/{ctx.user.imperal_id}",
        {"initialised": True, "theme": "default"},
    )

No side effects in health checks

@ext.health_check runs every 60 seconds. Do not mutate store data, send notifications, or call destructive APIs from inside it:

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.health_check
async def health(ctx) -> dict:
    # READ-ONLY operations only.
    try:
        resp = await ctx.http.get("https://my-backend/health")
        return {
            "status": "ok" if resp.status_code == 200 else "degraded",
            "version": ext.version,
        }
    except Exception:
        return {"status": "degraded", "version": ext.version}

Schema migrations in on_upgrade

Use from_version to gate migrations to the right version window:

from imperal_sdk import Extension

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


@ext.on_upgrade("2.0.0")
async def upgrade_to_v2(ctx, from_version: str | None = None) -> None:
    # Only do work when upgrading from v1.x.
    if from_version and from_version.startswith("1."):
        # Migrate legacy documents to the new schema.
        docs = await ctx.store.query_all("items")
        for doc in docs:
            if "legacy_field" in doc.data:
                await ctx.store.update(
                    "items",
                    doc.id,
                    {**doc.data, "new_field": doc.data.pop("legacy_field")},
                )

@ext.on_install

Signature

def on_install(self, func: Callable) -> Callable:
    ...

Applied directly — no arguments. The decorator stores the handler in ext._lifecycle["on_install"].

When it fires

Once, when a user installs the extension for the first time. If the user reinstalls after a previous uninstall, the hook fires again.

Available ctx attributes

on_install receives a full user context — the installing user's identity is present:

AttributeAvailableNotes
ctx.userYesCanonical user installing the extension
ctx.user.imperal_idYesPrimary user identifier
ctx.tenantYesTenant context
ctx.storeYesUse for initial data provisioning
ctx.httpYesExternal HTTP calls
ctx.configYesRead admin-configured defaults
ctx.notifyYesWelcome notification
ctx.cacheYesWarm up initial cache entries
ctx.aiYesAvailable but discouraged — avoid LLM calls during install
ctx.skeletonNoGuarded — raises SkeletonAccessForbidden

Return contract

Return None or omit a return statement. The web-kernel ignores the return value of lifecycle hooks other than @ext.health_check.

Production example — notes extension

From notes/app.py:

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)

ext = Extension(
    "notes",
    version="3.4.1",
    display_name="Notes",
    description="Notes extension — create, edit, and organize your notes with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


def _user_id(ctx) -> str:
    return ctx.user.imperal_id if ctx and hasattr(ctx, "user") and ctx.user else ""


@ext.on_install
async def on_install(ctx) -> None:
    log.info("notes installed for user %s", _user_id(ctx) or "system")

Common pitfalls

Do not assume ctx.user is non-None. In rare cases (admin bulk provisioning), the hook may receive a system context where ctx.user is None. Guard with hasattr:

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


# WRONG — crashes if ctx.user is None
@ext.on_install
async def on_install_wrong(ctx) -> None:
    uid = ctx.user.imperal_id  # AttributeError if ctx.user is None


# CORRECT — guard before accessing user attributes
@ext.on_install
async def on_install(ctx) -> None:
    uid = ctx.user.imperal_id if ctx and ctx.user else "system"
    await ctx.store.set(f"config/{uid}", {"initialised": True})

@ext.on_uninstall

Signature

def on_uninstall(self, func: Callable) -> Callable:
    ...

Applied directly — no arguments. Stored in ext._lifecycle["on_uninstall"].

When it fires

Once, when the extension is removed for a user. This is the last lifecycle event the extension receives for that user.

Available ctx attributes

Same availability as on_install — full user context is present.

Return contract

Return None. The return value is ignored.

Canonical example

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)

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.on_uninstall
async def on_uninstall(ctx) -> None:
    uid = ctx.user.imperal_id if ctx and ctx.user else "system"
    # Remove user-specific data from the store.
    docs = await ctx.store.query("items", where={"owner": uid})
    for doc in docs.data:
        await ctx.store.delete("items", doc.id)
    log.info("my-app uninstalled for user %s%d items removed", uid, len(docs.data))

Common pitfalls

Do not throw from on_uninstall. If the hook raises, the web-kernel may retry it or leave the extension in an inconsistent state. Catch exceptions internally and log them:

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)

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.on_uninstall
async def on_uninstall(ctx) -> None:
    try:
        await ctx.http.post(
            "https://my-backend/deregister",
            json={"user_id": ctx.user.imperal_id if ctx.user else ""},
        )
    except Exception as exc:
        # Log and continue — never let cleanup failures block uninstall.
        log.warning("my-app: deregister call failed during uninstall: %s", exc)

@ext.on_enable

Signature

def on_enable(self, func: Callable) -> Callable:
    ...

Applied directly — no arguments. Stored in ext._lifecycle["on_enable"].

When it fires

When an admin re-enables an extension that was previously disabled. Does not fire on initial install (which uses on_install).

Available ctx attributes

Same as on_install — full user context is present.

Return contract

Return None.

Canonical example

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)

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.on_enable
async def on_enable(ctx) -> None:
    uid = ctx.user.imperal_id if ctx and ctx.user else "system"
    log.info("my-app re-enabled for user %s", uid)
    # Invalidate caches that may have gone stale while disabled.
    await ctx.cache.delete(f"inbox_summary:{uid}")

@ext.on_disable

Signature

def on_disable(self, func: Callable) -> Callable:
    ...

Applied directly — no arguments. Stored in ext._lifecycle["on_disable"].

When it fires

When an admin disables the extension for a user or tenant. The extension continues to exist (and on_uninstall is not called) — it is merely suspended.

Available ctx attributes

Same as on_install — full user context is present.

Return contract

Return None.

Canonical example

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)

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.on_disable
async def on_disable(ctx) -> None:
    uid = ctx.user.imperal_id if ctx and ctx.user else "system"
    log.info("my-app disabled for user %s", uid)
    # Flush volatile caches — they will be stale when re-enabled.
    await ctx.cache.delete(f"inbox_summary:{uid}")

@ext.on_upgrade

Signature

def on_upgrade(self, version: str) -> Callable:
    ...

Takes one required argument: version — the target version string (e.g. "2.0.0"). Returns a decorator. Stored under key on_upgrade:{version} in ext._lifecycle.

Unlike the other four simple hooks, on_upgrade is a parameterised decorator. You may register multiple handlers, one per version bump:

from imperal_sdk import Extension

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


@ext.on_upgrade("2.0.0")
async def upgrade_v2(ctx, from_version: str | None = None) -> None:
    ...


@ext.on_upgrade("2.1.0")
async def upgrade_v2_1(ctx, from_version: str | None = None) -> None:
    ...

The manifest section emits the list of registered versions:

{
  "lifecycle": {
    "on_upgrade": ["2.0.0", "2.1.0"]
  }
}

When it fires

When the extension version stored for a user is lower than the deployed version. The web-kernel calls the appropriate version handler(s) in ascending order during the upgrade flow.

Available ctx attributes

Same as on_install — full user context is present. from_version is injected as a keyword argument by the web-kernel.

Return contract

Return None.

V22 — required from_version parameter

Validator V22 enforces that every on_upgrade handler accepts a from_version keyword argument (or **kwargs). This is an ERROR-level rule. The handler must be able to receive the previous version string from the web-kernel:

from imperal_sdk import Extension

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


# WRONG — V22 ERROR: on_upgrade handler must accept from_version kwarg
@ext.on_upgrade("2.0.0")
async def upgrade_wrong(ctx) -> None:
    pass


# CORRECT — accepts from_version as keyword argument
@ext.on_upgrade("2.0.0")
async def upgrade_to_v2(ctx, from_version: str | None = None) -> None:
    if from_version and from_version.startswith("1."):
        # Migrate v1.x data to v2 schema.
        pass

Production example — full migration with version guard

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)

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


@ext.on_upgrade("2.0.0")
async def upgrade_to_v2(ctx, from_version: str | None = None) -> None:
    uid = ctx.user.imperal_id if ctx and ctx.user else "system"
    log.info("my-app upgrade to 2.0.0 for user %s (from %s)", uid, from_version)

    # Only migrate if the user was on v1.x.
    if not from_version or not from_version.startswith("1."):
        return

    # Rename 'label' field to 'title' across all item documents.
    docs = await ctx.store.query("items", where={"owner": uid})
    for doc in docs.data:
        if "label" in doc.data:
            updated = {**doc.data, "title": doc.data.pop("label")}
            await ctx.store.update("items", doc.id, updated)

    log.info("my-app v2 migration complete — %d items updated", len(docs.data))

Common pitfalls

on_upgrade handlers must be idempotent. The web-kernel may call them more than once if a previous attempt failed mid-way. Write handlers that detect already-migrated documents and skip them:

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)

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


@ext.on_upgrade("2.0.0")
async def upgrade_to_v2_idempotent(ctx, from_version: str | None = None) -> None:
    docs = await ctx.store.query("items")
    for doc in docs.data:
        if "label" in doc.data and "title" not in doc.data:
            # Only migrate docs that still have the old field.
            updated = {**doc.data, "title": doc.data["label"]}
            del updated["label"]
            await ctx.store.update("items", doc.id, updated)

@ext.health_check

Signature

def health_check(self, func: Callable) -> Callable:
    ...

Applied directly — no arguments. Stored in ext._health_check (a HealthCheckDef, not in _lifecycle).

Unlike the other hooks, only one health_check is allowed per extension. Registering a second one overwrites the first.

When it fires

Every 60 seconds, driven by the web-kernel's periodic health ping. The interval is fixed at 60 seconds and is declared in the manifest:

{
  "lifecycle": {
    "health_check": {"interval_sec": 60}
  }
}

Available ctx attributes

@ext.health_check receives a minimal system-level context, not a per-user context. There is no specific user whose data you can read:

AttributeAvailableNotes
ctx.userNo / systemctx.user.imperal_id == "__system__" — do not query user-specific data
ctx.httpYesProbe external backends
ctx.storeNoPer-user store requires a user context
ctx.cacheNoPer-user cache requires a user context
ctx.configYesRead admin configuration
ctx.notifyNoNotifications are per-user
ctx.skeletonNoGuarded — raises SkeletonAccessForbidden

Health check is not per-user

@ext.health_check is called once globally, not once per installed user. Do not attempt to read ctx.store or send per-user notifications from inside it. If you need per-user health data, surface it via @ext.skeleton instead.

Return contract

Return a dict with at least one field: "status". The web-kernel reads status to determine whether the extension is healthy:

status valueWeb-kernel interpretation
"ok"Extension is healthy — no action
"degraded"Extension is partially functional — web-kernel logs the state
"unreachable"Backend is unreachable — web-kernel may surface this in the Developer Portal

Include "version" in every health response for observability. Add backend-specific fields (connection counts, latency, OAuth status) as needed.

Validator V9

Validator V9 emits a WARN if no @ext.health_check is registered. The extension can still publish to the Developer Portal, but the warning appears in imperal validate output. Add a health check to every production extension:

V9 WARN: no @ext.health_check registered — the web-kernel cannot probe backend connectivity.

Production examples

notes extension — probe the notes API, report reachability:

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)

ext = Extension(
    "notes",
    version="3.4.1",
    display_name="Notes",
    description="Notes extension — create, edit, and organize your notes with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)

NOTES_API_URL = "https://notes-api.example.com"


def _auth() -> dict:
    return {"Authorization": "Bearer secret"}


def _url(path: str) -> str:
    return f"{NOTES_API_URL}{path}"


@ext.health_check
async def health(ctx) -> dict:
    try:
        r = await ctx.http.get(_url("/health"), headers=_auth())
        if not r.ok:
            return {"status": "degraded", "version": ext.version, "api": "unreachable"}
        return {"status": "ok", "version": ext.version, "api": "reachable"}
    except Exception as exc:
        log.warning("notes health check failed: %s", exc)
        return {"status": "degraded", "version": ext.version, "api": "unreachable"}

tasks extension — probe a bridge service, report bridge status:

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)

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

BRIDGE_URL = "https://tasks-bridge.example.com"


def _bridge_url(path: str = "") -> str:
    return f"{BRIDGE_URL}{path}"


def _auth_headers() -> dict:
    return {"Authorization": "Bearer secret"}


@ext.health_check
async def health(ctx) -> dict:
    try:
        r = await ctx.http.get(_bridge_url("/health"), headers=_auth_headers())
        if not r.ok:
            return {"status": "degraded", "version": ext.version, "bridge": "unreachable"}
        body = r.body if isinstance(r.body, dict) else {}
        return {"status": "ok", "version": ext.version, "bridge": body.get("status")}
    except Exception:
        return {"status": "degraded", "version": ext.version, "bridge": "unreachable"}

google-ads extension — probe microservice and report connected accounts:

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)

ext = Extension(
    "google-ads",
    version="1.0.0",
    display_name="Google Ads",
    description="Google Ads extension — manage campaigns and budgets with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)

GADS_API_URL = "https://gads-service.example.com"
GADS_JWT = "service-token"


@ext.health_check
async def health(ctx) -> dict:
    """Verify microservice connectivity and report connected accounts."""
    try:
        r = await ctx.http.get(
            f"{GADS_API_URL}/health",
            headers={"Authorization": f"Bearer {GADS_JWT}"},
        )
        svc_status = "ok" if r.status_code == 200 else "degraded"
    except Exception:
        svc_status = "unreachable"

    return {
        "status": "ok" if svc_status == "ok" else "degraded",
        "version": ext.version,
        "microservice": svc_status,
    }

Common pitfalls

Never mutate data inside a health check. The handler runs every 60 seconds. Any write operation — ctx.store.create, ctx.notify, ctx.http.post to a stateful endpoint — will fire 1440 times per day per running worker instance:

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


# WRONG — writes data every 60 seconds
@ext.health_check
async def health_wrong(ctx) -> dict:
    await ctx.store.set("last_health_ping", {"ts": "now"})  # fires 1440×/day
    return {"status": "ok", "version": ext.version}


# CORRECT — read-only probe
@ext.health_check
async def health(ctx) -> dict:
    try:
        r = await ctx.http.get("https://my-backend/health")
        return {"status": "ok" if r.ok else "degraded", "version": ext.version}
    except Exception:
        return {"status": "degraded", "version": ext.version}

Never raise from a health check. An unhandled exception in @ext.health_check causes the web-kernel to mark the extension as unhealthy and may trigger alerting. Always catch at the outermost level and return status: "degraded":

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


# CORRECT — catches all exceptions, always returns a dict
@ext.health_check
async def health(ctx) -> dict:
    try:
        r = await ctx.http.get("https://my-backend/ping")
        return {"status": "ok" if r.ok else "degraded", "version": ext.version}
    except Exception as exc:
        import logging
        logging.getLogger(__name__).warning("health check failed: %s", exc)
        return {"status": "degraded", "version": ext.version}

V22 lifecycle signature validator

Validator V22 is an ERROR-level check that enforces correct handler signatures for lifecycle hooks:

HookV22 requirementCommon failure
on_installNo mandatory kwargsAdding unexpected **kwargs is fine; web-kernel calls with (ctx,)
on_uninstallNo mandatory kwargs
on_enableNo mandatory kwargs
on_disableNo mandatory kwargs
on_upgradeMust accept from_version kwargasync def upgrade(ctx): ... — missing from_version
health_checkNo mandatory kwargs

The fix for V22 errors on on_upgrade handlers is to add from_version: str | None = None as a keyword argument:

from imperal_sdk import Extension

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


# WRONG — V22 ERROR: on_upgrade handler missing from_version kwarg
@ext.on_upgrade("2.0.0")
async def upgrade_broken(ctx) -> None:
    pass


# CORRECT — V22 passes
@ext.on_upgrade("2.0.0")
async def upgrade_fixed(ctx, from_version: str | None = None) -> None:
    pass

Run imperal validate ./my-extension to check all validators including V22 before publishing.


Internal storage

The SDK stores lifecycle hooks in two places on the Extension instance:

  • ext._lifecycle: dict[str, LifecycleHook] — all hooks except health_check. Keys: "on_install", "on_uninstall", "on_enable", "on_disable", "on_upgrade:2.0.0" (version-suffixed).
  • ext._health_check: HealthCheckDef | None — the health check handler, stored separately.

The _build_lifecycle_section() method aggregates both into the lifecycle dict emitted in imperal.json:

{
  "lifecycle": {
    "on_install": true,
    "on_uninstall": true,
    "on_enable": true,
    "on_disable": true,
    "on_upgrade": ["2.0.0", "2.1.0"],
    "health_check": {"interval_sec": 60}
  }
}

Simple boolean hooks (on_install etc.) emit true when registered. on_upgrade emits a sorted list of registered version strings. health_check emits {"interval_sec": 60}.


Testing lifecycle hooks

Lifecycle handlers are plain async functions that receive a Context. Use MockContext from imperal_sdk.testing to test them without a running web-kernel:

import pytest
from imperal_sdk.testing import MockContext, MockStore


@pytest.mark.asyncio
async def test_on_install_provisions_config():
    ctx = MockContext(user_id="test_user_123", role="user")
    ctx.store = MockStore()

    # Import the handler from your module:
    # from app import on_install
    # await on_install(ctx)

    # doc = await ctx.store.get("config", "test_user_123")
    # assert doc is not None
    # assert doc.data.get("initialised") is True
    pass


@pytest.mark.asyncio
async def test_health_check_returns_ok_on_200():
    ctx = MockContext(user_id="__system__", role="system")
    # Mock an HTTP 200 response:
    # ctx.http.mock_get("https://my-backend/health", {"status": "ok"}, 200)

    # from app import health
    # result = await health(ctx)
    # assert result["status"] == "ok"
    pass


@pytest.mark.asyncio
async def test_on_upgrade_migrates_label_to_title():
    ctx = MockContext(user_id="test_user_123", role="user")
    ctx.store = MockStore()

    # Seed a v1 document.
    # await ctx.store.create("items", {"label": "Old title", "owner": "test_user_123"})

    # from app import upgrade_to_v2
    # await upgrade_to_v2(ctx, from_version="1.5.0")

    # docs = await ctx.store.query("items")
    # assert docs.data[0].data.get("title") == "Old title"
    # assert "label" not in docs.data[0].data
    pass

Quick reference

DecoratorArgsFiresContextReturn
@ext.on_installnoneUser installsUser contextNone
@ext.on_uninstallnoneUser uninstallsUser contextNone
@ext.on_enablenoneAdmin re-enablesUser contextNone
@ext.on_disablenoneAdmin disablesUser contextNone
@ext.on_upgrade(version)version: strVersion bumpUser context + from_version kwargNone
@ext.health_checknoneEvery 60sSystem context (no per-user data)dict with "status"

Cross-references

On this page