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]| Hook | Fires when | Typical use |
|---|---|---|
@ext.on_install | User installs the extension for the first time | Provision default store documents, log install event |
@ext.on_uninstall | User or admin removes the extension | Tear down user data, log uninstall |
@ext.on_enable | Admin re-enables a previously disabled extension | Warm up caches, re-register scheduled workers |
@ext.on_disable | Admin disables the extension for a user or tenant | Suspend background jobs, flush volatile state |
@ext.on_upgrade | Extension version bumps for an existing install | Schema migrations, data backfills |
@ext.health_check | Web-kernel ping every 60 seconds | Probe 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:
| Attribute | Available | Notes |
|---|---|---|
ctx.user | Yes | Canonical user installing the extension |
ctx.user.imperal_id | Yes | Primary user identifier |
ctx.tenant | Yes | Tenant context |
ctx.store | Yes | Use for initial data provisioning |
ctx.http | Yes | External HTTP calls |
ctx.config | Yes | Read admin-configured defaults |
ctx.notify | Yes | Welcome notification |
ctx.cache | Yes | Warm up initial cache entries |
ctx.ai | Yes | Available but discouraged — avoid LLM calls during install |
ctx.skeleton | No | Guarded — 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.
passProduction 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:
| Attribute | Available | Notes |
|---|---|---|
ctx.user | No / system | ctx.user.imperal_id == "__system__" — do not query user-specific data |
ctx.http | Yes | Probe external backends |
ctx.store | No | Per-user store requires a user context |
ctx.cache | No | Per-user cache requires a user context |
ctx.config | Yes | Read admin configuration |
ctx.notify | No | Notifications are per-user |
ctx.skeleton | No | Guarded — 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 value | Web-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:
| Hook | V22 requirement | Common failure |
|---|---|---|
on_install | No mandatory kwargs | Adding unexpected **kwargs is fine; web-kernel calls with (ctx,) |
on_uninstall | No mandatory kwargs | — |
on_enable | No mandatory kwargs | — |
on_disable | No mandatory kwargs | — |
on_upgrade | Must accept from_version kwarg | async def upgrade(ctx): ... — missing from_version |
health_check | No 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:
passRun 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 excepthealth_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
passQuick reference
| Decorator | Args | Fires | Context | Return |
|---|---|---|---|---|
@ext.on_install | none | User installs | User context | None |
@ext.on_uninstall | none | User uninstalls | User context | None |
@ext.on_enable | none | Admin re-enables | User context | None |
@ext.on_disable | none | Admin disables | User context | None |
@ext.on_upgrade(version) | version: str | Version bump | User context + from_version kwarg | None |
@ext.health_check | none | Every 60s | System context (no per-user data) | dict with "status" |
Cross-references
@ext.tool reference
Generic tool registration — internal helpers, skeleton alert handlers, and cross-extension callable targets.
@ext.skeleton reference
Live data probe decorator — section naming, return contract, TTL, alert mode.
@ext.schedule reference
Cron-driven background tasks — system context, fan-out with list_users + as_user.
Validators reference
All V1–V24 validators including V9 (missing health_check) and V22 (lifecycle signatures).