Complete API surface

from imperal_sdk import Extension

ext = Extension(
    "my-app",                      # app_id — positional
    version="1.0.0",
    display_name="My App",
    description="A longer description — at least 40 characters (V14).",
    icon="icon.svg",
    actions_explicit=True,
    capabilities=["my-app:read", "my-app:write"],
)

from imperal_sdk import ChatExtension
from pydantic import BaseModel, Field

chat = ChatExtension(ext, tool_name="my-app", description="My App assistant")

class SendEmailParams(BaseModel):
    to: str = Field(description="Recipient email address")
    subject: str = Field(description="Email subject line")
    body: str = Field(description="Email body text")

@chat.function(
    description="Send an email message via the user's connected mail account.",
    action_type="write",
    chain_callable=True,
    effects=["email.send"],
    event="email.sent",
    id_projection=None,
)
async def send_email(ctx, params: SendEmailParams):
    # params is auto-validated Pydantic model
    return ActionResult.success({"sent": True}, summary="Email sent")

@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,
    }}

@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.schedule("nightly_archive", cron="0 3 * * *")
async def nightly_archive(ctx):
    # Fan-out pattern: iterate all installed users
    async for user_id in ctx.store.list_users():
        user_ctx = ctx.as_user(user_id)
        # ... do per-user work

@ext.webhook("/incoming", method="POST", secret_header="X-Webhook-Secret")
async def receive_webhook(ctx, headers: dict, body: str, query_params: dict):
    # HMAC verification must be done manually — no SDK helper
    sig = headers.get("X-Webhook-Secret", "")
    # ... verify and process
    return {"ok": True}

# Subscribe
@ext.on_event("email.received")
async def handle_email(ctx, event):
    # ctx.http.* is NOT available here — minimal context only
    # ctx.store and ctx.notify are available
    await ctx.notify(f"New email from {event.data.get('from')}")

# Declare emission (must prefix with own app_id)
@ext.emits("my-app.item.created")
async def _declare_item_created(): ...

from pydantic import BaseModel

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

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

# Then in a handler:
cached = await ctx.cache.get("inbox", InboxSummary)
if cached is None:
    fresh = await fetch_inbox_summary(ctx)
    await ctx.cache.set("inbox", fresh, ttl_seconds=60)

@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="openai_api_key",
    description="Your OpenAI API key (sk-proj-...).",
    required=True,
    write_mode="user",         # 'user' | 'extension' | 'both'
    max_bytes=200,
    rotation_hint_days=90,
)(lambda: None)

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

@ext.on_upgrade("2.0.0")
async def on_upgrade_2_0_0(ctx):
    # Runs when user upgrades from any version to 2.0.0
    await ctx.store.update("settings", "main", {"migrated": True})

@ext.on_uninstall
async def on_uninstall(ctx):
    pass  # cleanup

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

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

@ext.health_check
async def health_check(ctx):
    # Called every 60s by web-kernel
    return HealthStatus(healthy=True)

async def my_handler(ctx, params):
    ctx.user.imperal_id   # str — canonical user ID (W1 canonical)
    ctx.user.email        # str | None
    ctx.user.tenant_id    # str
    ctx.user.role         # str
    ctx.user.scopes       # list[str]
    ctx.user.attributes   # dict — arbitrary admin-set attributes
    ctx.tenant            # TenantContext | None
    ctx.store             # StoreProtocol | None — per-user document store
    ctx.db                # DBProtocol | None — raw SQL access
    ctx.ai                # AIProtocol | None — LLM completion
    ctx.skeleton          # _SkeletonAccessGuard — only in @ext.skeleton context
    ctx.billing           # BillingProtocol | None
    ctx.notify            # NotifyProtocol | None — push/in-app notifications
    ctx.storage           # StorageProtocol | None — file storage
    ctx.http              # HTTPProtocol | None — outbound HTTP
    ctx.tools             # ToolsProtocol | None — cross-tool discovery/call
    ctx.config            # ConfigProtocol | None — extension config
    ctx.extensions        # ExtensionsProtocol | None — IPC + event emit
    ctx.cache             # CacheClient (property) — short-lived Pydantic cache
    ctx.time              # TimeContext — timezone, utc_offset, now_utc, now_local
    ctx.agency_id         # str | None
    ctx.agency_theme      # dict | None

doc    = await ctx.store.create("collection", {"key": "value"})
doc    = await ctx.store.get("collection", doc_id)
page   = await ctx.store.query("collection", where={"status": "active"}, limit=100)
doc    = await ctx.store.update("collection", doc_id, {"key": "new_value"})
ok     = await ctx.store.delete("collection", doc_id)
count  = await ctx.store.count("collection", where={"status": "active"})
# System context only:
async for user_id in ctx.store.list_users(): ...
page   = await ctx.store.query_all("collection", where={...})

resp = await ctx.http.get("https://api.example.com/data")
resp = await ctx.http.post("https://api.example.com/items", json={...})
resp = await ctx.http.put("https://api.example.com/items/1", json={...})
resp = await ctx.http.patch("https://api.example.com/items/1", json={...})
resp = await ctx.http.delete("https://api.example.com/items/1")
# resp.status_code, resp.json(), resp.text, resp.headers

# Per-call timeout override (v4.2.12+). Default 30s; federal cap 180s.
resp = await ctx.http.post("https://api.example.com/v1/...",
                           json={...}, timeout=120)

# Spawn a coroutine in the background; result auto-delivered to chat
# when the coroutine completes.
task_id: str = await ctx.background_task(
    coro=_my_long_op(),
    long_running=False,    # False → 180s cap, True → 1800s cap
    name="AI refinement",  # human-readable for UI/audit
)
return ActionResult.success(
    summary="Got it — running in background. I'll send the result here.",
    data={"task_id": task_id},
)

# Inject a bot message into the user's chat at any time.
# Mirrors the web-kernel's auto-promote chat-injection path, but you call it.
await ctx.deliver_chat_message(
    "Spotify connected! 🎵 You can now ask what's playing or search your library.",
    msg_type="response",       # "response" | "system" | "tool_result"
    refresh_panels=["library"], # optional — panel_ids to re-render after
)

value = await ctx.cache.get("key", MyModel)             # returns MyModel | None
await ctx.cache.set("key", value, ttl_seconds=60)       # TTL: 5-300s
await ctx.cache.delete("key")
value = await ctx.cache.get_or_fetch("key", MyModel, fetcher, ttl_seconds=60)

# Only works when ctx.user.imperal_id == "__system__"
user_ctx = ctx.as_user("imp_u_tE-J9c_NxX")

await ctx.progress(50.0, "Halfway done...")
# May raise TaskCancelled if user cancelled

# Read plaintext for one declared secret. Returns None if declared but unset.
api_key: str | None = await ctx.secrets.get("spotify_api_key")

# Write plaintext. Allowed only if manifest write_mode is 'extension' or 'both'.
await ctx.secrets.set("spotify_refresh_token", "rt-...")

# Delete (same write_mode rules apply).
was_set: bool = await ctx.secrets.delete("spotify_refresh_token")

# Cheap metadata read — does NOT decrypt, does NOT write an audit row.
is_set: bool = await ctx.secrets.is_set("spotify_api_key")

# List all declared secrets for this extension. Returns SecretStatus dataclasses
# with name, description, is_set, last_accessed_at — NEVER the value.
statuses = await ctx.secrets.list()

import pytest
from imperal_sdk.testing import MockSecretStore

@pytest.fixture
def secrets():
    return MockSecretStore({"spotify_api_key": "sk-test"})

async def test_search(ctx_factory, secrets):
    ctx = ctx_factory(secrets=secrets)
    ...

url = ctx.webhook_url("/callback")
# → "https://panel.imperal.io/v1/ext/spotify/webhook/callback"

from imperal_sdk import ActionResult

# Success
return ActionResult.success({"note_id": "123"}, summary="Note created")
return ActionResult.success(data, ui=ui.Stack([...]))
return ActionResult.success(data, refresh_panels=["sidebar", "editor"])

# Error
return ActionResult.error("Something went wrong", retryable=True)
return ActionResult.error("Permanent failure", retryable=False)

{
  "manifest_schema_version": 3,
  "app_id": "my-app",
  "version": "1.0.0",
  "name": "My App",
  "description": "A longer description — at least 40 characters.",
  "icon": "icon.svg",
  "actions_explicit": true,
  "capabilities": ["my-app:read", "my-app:write"],
  "tools": [...],
  "signals": [...],
  "schedules": [...],
  "required_scopes": [...],
  "webhooks": [...],
  "events": {"subscribes": [...], "emits": [...]},
  "exposed": [...],
  "lifecycle": {...},
  "lifecycle_hooks": {...},
  "tray": [...]
}

imperal validate ./extension          # run V14-V22+V24+V31
imperal build ./extension # generate imperal.json
python3 -m py_compile app.py          # syntax check before deploy

from imperal_sdk import generate_manifest, save_manifest, validate_extension
manifest = generate_manifest(ext)
save_manifest(ext, ".")
report = validate_extension(ext)