imperal-sdk 1.5.14

SDK for building Imperal Cloud extensions

Imperal SDK

Build AI agents. Ship to marketplace. Get paid.

The SDK for the first AI Cloud OS.

Getting Started | Features | Docs (source) | OpenAPI | Discord | Marketplace

---

What is Imperal?

Imperal is an AI Cloud Operating System — a complete platform where developers build AI-powered extensions (agents, tools, workflows), users install and use them, and everyone benefits from a shared ecosystem.

Think Shopify for AI agents. You build it. Users install it. The platform handles auth, billing, LLM routing, real-time sync, and everything else.

The SDK is how you build for the platform. One pip install. Five minutes to your first working AI agent.

pip install imperal-sdk

---

5-Minute Quickstart

imperal init my-agent --template chat
cd my-agent

That generates a complete AI agent:

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

ext = Extension("my-agent", version="1.0.0")
chat = ChatExtension(ext, tool_name="my_agent", description="My first AI agent")


class GreetParams(BaseModel):
    name: str = "World"


@chat.function("greet", description="Greet someone", action_type="read")
async def fn_greet(ctx, params: GreetParams) -> ActionResult:
    """Say hello to someone."""
    return ActionResult.success(
        data={"message": f"Hello, {params.name}!"},
        summary=f"Greeted {params.name}",
    )

Validate it:

imperal validate
# Extension: my-agent v1.0.0
# Tools: 1, Functions: 1, Events: 0
# 0 errors, 0 warnings

Test it:

from imperal_sdk.testing import MockContext

async def test_greet():
    ctx = MockContext(role="user")
    result = await fn_greet(ctx, GreetParams(name="World"))
    assert result.status == "success"
    assert result.data["message"] == "Hello, World!"

---

What You Get

For Extension Developers

Feature	Description
Typed Everything	Generic ActionResult[T], Page[T], typed Protocols, Pydantic params — full IDE autocomplete
Machine-Validated Contracts	JSON Schema for imperal.json (v1.5.9), ActionResult + Event payloads (v1.5.10), OpenAPI 3.x for Auth Gateway / Registry / Sharelock Cases (v1.5.11). See docs/openapi/.
10 SDK Clients	ctx.store, ctx.ai, ctx.billing, ctx.skeleton, ctx.notify, ctx.storage, ctx.http, ctx.config, ctx.extensions, ctx.time
Error Hierarchy	ImperalError > APIError > NotFoundError, RateLimitError — catch what you need
MockContext	Drop-in test replacement with 10 mock clients. Test without a server.
CLI Tools	imperal init, imperal validate, imperal dev — scaffold, validate, develop
Lifecycle Hooks	@ext.on_install, @ext.on_upgrade("0.9.0"), @ext.on_uninstall, @ext.health_check
Events System	@ext.on_event("deal.created") — subscribe to typed platform events
Webhooks	@ext.webhook("/stripe", secret_header="Stripe-Signature") — external HTTP ingestion
Inter-Extension API	@ext.expose("get_deal") + ctx.extensions.call("crm", "get_deal") — typed cross-extension calls
UI Contributions	Panels, Widgets, Commands, Context Menus, Settings, Themes — declare UI from Python
Validator	12 rules (V1-V12) catch issues before deployment. Claude-friendly fix reports.
Pagination	Page[T] with cursor, iteration, auto-pagination — built into every list operation

For the Platform

Feature	What It Means
BYOLLM	Users bring their own LLM keys. Anthropic, OpenAI, Google, Ollama, any OpenAI-compatible API.
Multi-Model Routing	Different models for routing (fast/cheap), execution (accurate), navigation (conversational)
2-Step Confirmation	Destructive actions require user approval. Kernel-enforced, not extension-optional.
RBAC + Scopes	Fine-grained permissions per user, per extension, per function
Token Economy	Built-in billing: token wallet, usage metering, subscription plans, marketplace payouts
Automation Engine	Event-driven rules with cron scheduling. Users create "if X then Y" without code.
Self-Hosted	Run on your servers. Your data never leaves.

---

Extension Architecture

@chat.function("create_deal", action_type="write", event="deal.created")
     |
     v
[ Kernel Pipeline ]
     |
     +-- Scope Check (RBAC)
     +-- 2-Step Confirmation (if destructive)
     +-- Execute Function
     +-- Truth Gate (ActionResult.status)
     +-- Event Publishing (automations)
     +-- Action Recording (audit trail)
     |
     v
ActionResult.success(data={"deal_id": "d1"}, summary="Deal created")

Every function call goes through the full kernel pipeline. Security, billing, events — all automatic. You just write the business logic.

---

SDK Type System

# Generic ActionResult — works with dict or Pydantic models
from pydantic import BaseModel

class Deal(BaseModel):
    id: str
    name: str
    value: float

result = ActionResult.success(data=Deal(id="d1", name="Acme", value=50000), summary="Created")
result.to_dict()  # {"status": "success", "data": {"id": "d1", "name": "Acme", "value": 50000}, ...}

# Cursor-based pagination
page = await ctx.store.query("deals", where={"status": "open"}, limit=20)
for deal in page:  # Page[Document] supports iteration
    print(deal.data["name"])
if page.has_more:
    next_page = await ctx.store.query("deals", cursor=page.cursor)

# Error hierarchy — catch what you need
from imperal_sdk.errors import NotFoundError, RateLimitError
try:
    deal = await ctx.store.get("deals", "nonexistent")
except NotFoundError as e:
    print(f"{e.resource} '{e.id}' not found")  # "deal 'nonexistent' not found"

---

Multi-Model LLM

Extensions never import LLM libraries directly. The platform handles provider routing, failover, and per-user model preferences.

# Extensions use ctx.ai — the platform routes to the right model
result = await ctx.ai.complete("Summarize this deal", model="")  # uses user's configured model

# Or use the LLM provider directly for advanced use cases
from imperal_sdk import get_llm_provider
provider = get_llm_provider()
resp = await provider.create_message(
    messages=[{"role": "user", "content": "Hello"}],
    purpose="execution",       # routing | execution | navigate
    user_id="imp_u_xxx",       # BYOLLM: uses the user's own API key
)

Supported: Anthropic (Claude), OpenAI (GPT), Google (Gemini), Ollama, vLLM, LM Studio, any OpenAI-compatible API.

---

System Features

Extensions have access to platform-level capabilities through the SDK. No kernel SDK needed.

Scheduled Tasks (Cron)

@ext.schedule("daily_report", cron="0 9 * * *")
async def daily_report(ctx):
    """Runs every day at 9 AM UTC."""
    stats = await ctx.store.query("metrics", where={"date": "today"})
    await ctx.notify.push(title="Daily Report", body=f"{len(stats)} events today")
    return ActionResult.success(summary="Report sent")

@ext.schedule("hourly_sync", cron="0 * * * *")
async def hourly_sync(ctx):
    """Sync data every hour."""
    data = await ctx.http.get("https://api.example.com/data")
    await ctx.store.create("synced_data", data.json())

Dynamic Scheduling (User-Created Intervals)

For user-driven schedules (e.g., monitors with custom intervals), use a single cron + last_run_at check:

import time

@ext.schedule("monitor_runner", cron="0 * * * *")  # check every hour
async def monitor_runner(ctx):
    """Run monitors that are due based on user-configured intervals."""
    monitors = await ctx.store.query("monitors", where={"active": True})
    now = time.time()
    for m in monitors:
        interval_sec = m["interval_hours"] * 3600  # 1h, 6h, 12h, 24h
        if now - m.get("last_run_at", 0) >= interval_sec:
            await run_scan(ctx, m["id"])
            await ctx.store.update("monitors", m["id"], {"last_run_at": now})

This is the standard production pattern. No ctx.scheduler API needed — the cron trigger + per-record interval check handles any user-configured frequency.

When to use which:

Pattern	Use Case
@ext.schedule(cron=...)	Fixed intervals: daily reports, hourly syncs, cleanup
Cron + last_run_at	Dynamic: user-created monitors, per-item schedules
@ext.on_event(...)	Reactive: trigger on events from other extensions

Push Notifications

@chat.function("send_alert", description="Send push notification", action_type="write")
async def send_alert(ctx, params: AlertParams) -> ActionResult:
    await ctx.notify.push(
        title=params.title,
        body=params.message,
    )
    return ActionResult.success(summary=f"Alert sent: {params.title}")

Event System (Cross-Extension)

# Subscribe to events from other extensions
@ext.on_event("mail.received")
async def on_new_email(ctx, event):
    """Triggered when any email arrives."""
    subject = event.data.get("subject", "")
    if "urgent" in subject.lower():
        await ctx.notify.push(title="Urgent email!", body=subject)

# Publish events from your functions
@chat.function("create_deal", action_type="write", event="crm.deal_created")
async def create_deal(ctx, params: DealParams) -> ActionResult:
    deal = await ctx.store.create("deals", params.dict())
    return ActionResult.success(data=deal, summary="Deal created")
    # Platform auto-publishes crm.deal_created event — other extensions can subscribe

System Tray (v1.5.4)

Inject icons, badges, and dropdown panels into the OS top bar:

from imperal_sdk import ui

@ext.tray("unread", icon="Mail", 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.tray("alerts", icon="Bell", tooltip="Active alerts")
async def tray_alerts(ctx, **kwargs):
    alerts = await ctx.store.query("alerts", where={"active": True}, limit=5)
    return ui.Stack([
        ui.Badge(str(len(alerts)), color="red"),
        # Dropdown panel — shown when user clicks the tray icon
        ui.List(items=[
            ui.ListItem(id=a["id"], title=a["title"], subtitle=a["severity"])
            for a in alerts
        ]),
    ])

Webhooks (External HTTP)

@ext.webhook("/stripe", method="POST", secret_header="Stripe-Signature")
async def handle_stripe(ctx, headers, body, query_params):
    """Receive Stripe webhook at POST /v1/ext/{app_id}/webhook/stripe"""
    import json
    data = json.loads(body)
    if data["type"] == "payment_intent.succeeded":
        await ctx.store.create("payments", {"amount": data["data"]["object"]["amount"]})
    return {"received": True}

Inter-Extension Calls (IPC)

# Expose a method for other extensions to call
@ext.expose("get_deal", action_type="read")
async def api_get_deal(ctx, deal_id: str) -> ActionResult:
    deal = await ctx.store.get("deals", deal_id)
    return ActionResult.success(data=deal)

# Call another extension's exposed method
result = await ctx.extensions.call("crm", "get_deal", deal_id="d123")

---

System Prompt Guidelines

Important: Extensions must NOT identify as a specific assistant. The platform injects OS identity automatically.

# WRONG — deploy validation will fail (R10)
chat = ChatExtension(ext, tool_name="my_tool", description="...",
    system_prompt="You are a CRM assistant for Imperal Cloud.")

# CORRECT — describe what the module DOES, not what the AI IS
chat = ChatExtension(ext, tool_name="my_tool", description="...",
    system_prompt="CRM module — manage deals, contacts, and pipelines.")

The kernel injects the AI identity ({assistant_name}) and full platform capabilities into every LLM call. Your system_prompt should only contain module-specific rules and capabilities.

---

Testing

Every extension is testable without a server:

from imperal_sdk.testing import MockContext

async def test_full_workflow():
    ctx = MockContext(role="admin", config={"api_url": "https://example.com"})

    # MockStore — in-memory, full CRUD
    doc = await ctx.store.create("deals", {"name": "Big Deal", "value": 50000})
    assert doc.id is not None

    page = await ctx.store.query("deals", where={"name": "Big Deal"})
    assert len(page) == 1

    # MockAI — configurable responses
    ctx.ai.set_response("summarize", "This is a big deal worth $50K")
    result = await ctx.ai.complete("Summarize this deal")
    assert "50K" in result.text

    # MockHTTP — register mock endpoints
    ctx.http.mock_get("api.example.com", {"status": "ok"})
    resp = await ctx.http.get("https://api.example.com/health")
    assert resp.ok

    # MockNotify — verify notifications were sent
    await ctx.notify.send("Deal created!", channel="email")
    assert len(ctx.notify.sent) == 1

---

Validation

$ imperal validate ./my-extension

── Imperal Extension Validator v1.0 ────────────────────

Extension: crm v1.0.0
Tools: 1, Functions: 12, Events: 5

RESULTS: 1 error, 2 warnings

  ERROR  [V5] @chat.function 'create_deal' must return ActionResult
         Fix: Add return type annotation: -> ActionResult

  WARN   [V10] @chat.function 'update_deal' (action_type="write") has no event=
         Fix: Add event='crm.update_deal' to the decorator

  WARN   [V9] No @ext.health_check registered
         Fix: Add @ext.health_check decorator to a health check function

12 rules. Catches type issues, missing annotations, banned imports, missing events. Runs in CI, in CLI, and at kernel load time.

---

Lifecycle & Events

@ext.on_install
async def setup(ctx):
    """Called once when user installs your extension."""
    await ctx.store.create("settings", {"initialized": True})

@ext.on_upgrade("0.9.0")
async def migrate(ctx):
    """Called when upgrading from 0.9.x."""
    await ctx.store.query("deals")  # migrate data

@ext.health_check
async def health(ctx):
    """Called every 60s. Return health status."""
    return HealthStatus.ok({"connections": 5})

@ext.on_event("email.received")
async def on_email(ctx, event):
    """React to platform events from other extensions."""
    await ctx.notify.send(f"New email: {event.data['subject']}")

@ext.webhook("/stripe", method="POST", secret_header="Stripe-Signature")
async def handle_stripe(ctx, request):
    """Receive webhooks from external services."""
    data = request.json()
    return WebhookResponse.ok({"received": True})

---

Declarative UI (v1.5.0)

Build full Panel UI from Python — zero React, zero rebuilds. 57 components across 7 modules.

from imperal_sdk import Extension, ui

ext = Extension("inventory", version="1.0.0")

@ext.panel("sidebar", slot="left", title="Inventory", icon="Package",
           default_width=300, refresh="on_event:item.created,item.deleted")
async def panel_sidebar(ctx, **kwargs):
    """Panel handler — returns UINode tree. Auto-discovered by the platform."""
    items = await ctx.store.query("items", limit=50)
    return ui.Stack([
        ui.Header("Inventory", level=3, subtitle=f"{len(items)} items"),
        ui.Button("+ New Item", variant="primary", on_click=ui.Call("create_item")),
        ui.List(
            items=[
                ui.ListItem(
                    id=i["id"], title=i["name"],
                    subtitle=f"Qty: {i['qty']}",
                    badge=ui.Badge("Low", color="red") if i["qty"] < 5 else None,
                    expandable=True,
                    expanded_content=[
                        ui.KeyValue(items=[
                            {"key": "SKU", "value": i["sku"]},
                            {"key": "Price", "value": f"${i['price']}"},
                        ], columns=2),
                        ui.Button("Delete", variant="danger",
                                  on_click=ui.Call("delete_item", item_id=i["id"])),
                    ],
                )
                for i in items
            ],
            searchable=True,
        ),
    ])

55 components:

• Layout (8): Stack, Grid, Tabs, Page, Section, Row, Column, Accordion
• Display (9): Text, Header, Icon, Image, Code, Markdown, Divider, Empty, Html
• Interactive (7): Button, Card, Menu, Dialog, Tooltip, Link, SlideOver
• Input (11): Input, TextArea, Toggle, Select, MultiSelect, Form, Slider, DatePicker, FileUpload, RichEditor, TagInput
• Data (11): List, ListItem, DataTable, DataColumn, Stat, Stats, Badge, Avatar, KeyValue, Timeline, Tree
• Feedback (5): Alert, Progress, Chart, Loading, Error
• Actions (4): Call, Navigate, Send, Open

Zero-rebuild panel discovery: @ext.panel() auto-publishes to the config store. New extensions show panels instantly — no frontend changes.

v1.5.0 highlights:

• ui.Html(theme="light") — white-bg email rendering with auto-resize iframe
• ui.List(selectable=True, bulk_actions=[...]) — multi-select with checkbox hover, bulk action bar
• ui.List(on_end_reached=action) — infinite scroll with IntersectionObserver sentinel
• ui.Stack(sticky=True) — pin toolbars to top of scroll container
• System padding: horizontal Stacks get px-3 py-1 by default

System features: Pagination, infinite scroll, multi-select, bulk actions, drag-drop, hover actions, search, expandable cards, inline editing, collapsible sections — all kernel-enforced

---

Project Structure

my-extension/
  main.py              # Extension + ChatExtension + @chat.functions
  imperal.json         # Manifest (optional — auto-discovered from code)
  tests/
    test_main.py       # Tests using MockContext
  requirements.txt     # imperal-sdk>=1.0.0

---

Links

• Documentation: docs.imperal.io — source in docs/
• OpenAPI Specs: docs/openapi/ — Auth Gateway, Registry, Sharelock Cases (229 endpoints)
• Platform: imperal.io
• Discord: discord.gg/imperal
• PyPI: pypi.org/project/imperal-sdk
• GitHub: github.com/imperalcloud
• License: AGPL-3.0

---

Built by Imperal, Inc.

The AI Cloud OS.