Web-kernel context (ctx)

Every handler — chat function, skeleton, panel, scheduled job — receives ctx as its first argument. This page covers every attribute, every method, every guarantee.

Why ctx exists

Two reasons:

🔐

Federal trust boundary

ctx is built by the [web-kernel](/en/reference/glossary/) after auth + tenant + scope checks pass. By the time your handler sees it, the identity is verified — fail-closed.

🧰

Capability surface

HTTP, storage, billing, notifications — everything you need is on ctx. You never import third-party clients directly.

The shape

async def my_handler(ctx, params):
    ctx.user.imperal_id    # str — canonical user id
    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              # per-user document store
    ctx.db                 # raw SQL access
    ctx.ai                 # LLM completion client
    ctx.billing            # billing: limits, subscription, usage
    ctx.notify             # push/in-app notifications
    ctx.storage            # file upload/download/list
    ctx.http               # outbound HTTP client
    ctx.tools              # cross-tool discovery and call
    ctx.config             # extension config reader (non-sensitive settings)
    ctx.secrets            # per-user encrypted credentials (EXT-SECRETS-V1, v4.2.2+)
    ctx.extensions         # IPC + event emit
    ctx.cache              # short-lived Pydantic-typed cache (property)
    ctx.time               # timezone, utc_offset, now_utc, now_local, hour_local, is_business_hours
    ctx.agency_id          # str | None
    ctx.agency_theme       # dict | None
    ctx.webhook_url(path)  # build canonical webhook URL (v4.2.7+)

ctx.lang does not exist

There is no ctx.lang attribute in the SDK. There is no ctx.api, no ctx.llm, no ctx.history, no ctx.prior, no ctx.cancel_event, no ctx.now, no ctx.translate.

If your extension needs the user's language, read it from ctx.user.attributes. Language detection is handled at the web-kernel classifier layer, not exposed per-call to extension code.

Identity — `ctx.user`

Prop

Type

Always key state by ctx.user.imperal_id

Never key by display_name (changes), email (can change), or any other field. imperal_id is opaque, stable, and federal-canonical. Multi-tenant safety depends on this.

Tenant — `ctx.tenant_id`

For multi-org tenancy, ctx.user.tenant_id separates org boundaries.

# Federal-clean storage key
storage_key = f"tenant:{ctx.user.tenant_id}:user:{ctx.user.imperal_id}:notes:{note_id}"

When your extension serves only direct users (no agency tier), tenant_id == imperal_id. The pattern still works.

Document store — `ctx.store`

Per-user document store with CRUD + query operations.

# Create
doc = await ctx.store.create("tasks", {"title": "Buy milk", "status": "pending"})

# Read
doc = await ctx.store.get("tasks", doc_id)

# Query with filtering
page = await ctx.store.query("tasks", where={"status": "pending"}, limit=50)
for item in page.data:
    item.id           # str
    item.data         # dict — your payload
    item.created_at   # datetime

# Update
updated = await ctx.store.update("tasks", doc_id, {"status": "done"})

# Delete
ok = await ctx.store.delete("tasks", doc_id)

# Count
n = await ctx.store.count("tasks", where={"status": "pending"})

System context only (scheduled tasks):

async for user_id in ctx.store.list_users():
    user_ctx = ctx.as_user(user_id)
    # per-user work

LLM — `ctx.ai`

LLM completion client. Do not import anthropic or openai directly.

result = await ctx.ai.complete(
    prompt="Summarize this text: ...",
    model="",   # empty = use admin-configured default
)
result.text         # completion text
result.tokens_used  # int

Outbound HTTP client. Strongly preferred over raw requests or httpx — ctx.http adds auth headers, audit logs, retries, and circuit breaking. Not enforced by a federal validator; convention only.

# GET
resp = await ctx.http.get("https://api.example.com/data")

# POST with JSON
resp = await ctx.http.post("https://api.example.com/items", json={"key": "value"})

# PUT / PATCH / DELETE
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")

# Response
resp.status_code   # int
resp.text          # str
resp.json()        # dict (raises if not JSON)
resp.headers       # dict

ctx.http, not ctx.api

The outbound HTTP client is at ctx.http. There is no ctx.api attribute. Earlier documentation incorrectly called this ctx.api.

Notifications — `ctx.notify`

Send push or in-app notifications to the user.

# Preferred style
await ctx.notify("Your export is ready.")

# Explicit send with channel
await ctx.notify.send("Your export is ready.", channel="in_app")

Short-lived cache — `ctx.cache`

Pydantic-typed per-user cache. TTL 5-300 seconds.

Model must be registered via @ext.cache_model on the Extension instance:

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

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

# In handler:
cached = await ctx.cache.get("inbox", InboxSummary)
if cached is None:
    fresh = InboxSummary(unread=5, latest_subject="Meeting notes")
    await ctx.cache.set("inbox", fresh, ttl_seconds=120)

# Or fetch-or-miss in one call:
value = await ctx.cache.get_or_fetch(
    "inbox", InboxSummary,
    fetcher=lambda: fetch_inbox(ctx),
    ttl_seconds=120,
)

# Delete
await ctx.cache.delete("inbox")

Key format: ^[A-Za-z0-9_\-:]+$, max 128 chars. Value envelope max 64KB.

ctx.cache raises RuntimeError in minimal mock contexts (MockContext without an Extension reference). Use MockContext with a real Extension instance, or mock ctx._cache directly in unit tests.

Billing — `ctx.billing`

limits = await ctx.billing.check_limits()       # LimitsResult
sub    = await ctx.billing.get_subscription()   # SubscriptionInfo
bal    = await ctx.billing.get_balance()         # BalanceInfo
await ctx.billing.track_usage(tokens=1200, resource="llm")

File storage — `ctx.storage`

file_info = await ctx.storage.upload("reports/2026.pdf", data, "application/pdf")
data      = await ctx.storage.download("reports/2026.pdf")
ok        = await ctx.storage.delete("reports/2026.pdf")
page      = await ctx.storage.list("reports/")

Extension config — `ctx.config`

Read-only access to admin-set config values for your extension. For non-sensitive settings only — default timezone, feature flags, display preferences. Values are stored plaintext in the user store and are visible to the Panel; they are not encrypted at rest and they are not federal-grade.

api_url   = ctx.config.get("api_url", "https://default.example.com")
flag      = ctx.config.get("show_advanced", False)
section   = ctx.config.get_section("limits") # dict
all_cfg   = ctx.config.all()                 # dict

ctx.config is NOT for credentials

API keys, OAuth tokens, webhook signing secrets — these MUST flow through ctx.secrets (next section), not ctx.config. The Spotify-style bug class (Save succeeds in the Vault path but the handler still reads from ctx.config and reports "credentials not configured") comes from leaving a legacy ctx.config.get("client_id") read site in place after migrating writes. Grep your extension for every read of a credential name and route it through ctx.secrets. See Secrets concept — ctx.config vs ctx.secrets.

Credentials — `ctx.secrets` (EXT-SECRETS-V1, SDK v4.2.2+)

ctx.secrets is the federal credentials surface. The web-kernel constructs a SecretClient bound to (ext_id, imperal_id) at handler dispatch time, right after the capability boundary is injected. The client talks to the platform's secrets endpoint with a short timeout; the platform encrypts/decrypts via the KMS (AES-256-GCM, non-exportable key) and persists ciphertext to the encrypted-secrets store.

The five methods (all async):

# Read plaintext for one declared secret.
#   - Returns str    if 200 (decrypted value).
#   - Returns None   if 404 (declared but no value set yet, or dev-mode env var not set).
#   - Raises SecretNotDeclaredError   if `name` not in manifest.secrets[].
#   - Raises SecretVaultUnavailable   on auth-gw 503 or network error.
api_key: str | None = await ctx.secrets.get("spotify_api_key")

# Write plaintext.
#   - Raises SecretWriteForbidden   if manifest write_mode='user'.
#   - Raises SecretValueTooLarge    if utf-8 bytes > manifest max_bytes.
#   - Raises SecretVaultUnavailable on auth-gw error.
await ctx.secrets.set("spotify_refresh_token", "rt-...")

# Delete the stored value (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 carries the plaintext value.
statuses = await ctx.secrets.list()

Web-kernel injection — what happens at dispatch time:

After the capability boundary is injected, the web-kernel reads your secrets[] manifest entries into SecretSpec instances.
Constructs a web-kernel-owned SecretClient bound to your (ext_id, user_id) so every read/write is scoped to the current handler invocation.
Attaches the client to ctx.secrets.

Failure mode: wiring is non-fatal. If ctx.secrets cannot be constructed (e.g. unexpected platform error), the attribute stays absent and a handler calling await ctx.secrets.get(...) raises AttributeError, which surfaces as a normal extension error in chat.

Dev mode (IMPERAL_DEV_MODE=true):

get(name) reads IMPERAL_SECRET_<UPPER_NAME> env var (or returns None).
set / delete are no-ops with WARN log.
list() reflects env-var presence in the is_set field.
Manifest contract still enforced — undeclared names still raise SecretNotDeclaredError.

Federal contract — source-inspection-friendly: SecretClient has no module-level cache, no @lru_cache, no instance attribute holding plaintext between calls. Plaintext is only ever a local variable in get()'s return path. This pins I-SECRETS-HANDLER-SCOPE-MEMORY at the type level.

`required=True` dispatch-time gate is not yet implemented in web-kernel

The @ext.secret(required=True) flag is declared in the SDK and lands in the manifest, but the web-kernel does not currently block handler dispatch on missing required secrets — a handler whose required secret is unset will be invoked, and await ctx.secrets.get(name) returns None. Handle the None case explicitly in your handler (return a clean error pointing to the Panel Secrets tab) until the web-kernel-side gate ships.

See Secrets concept for the mental model, @ext.secret reference for the decorator surface, and Handle user API keys recipe for the practical pattern.

Webhook URL builder — `ctx.webhook_url(path)` (v4.2.7+)

Build the canonical public URL for any @ext.webhook path declared by this extension.

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

Path normalisation: leading slash optional, multi-segment supported.

ctx.webhook_url("callback")       # → .../webhook/callback
ctx.webhook_url("/callback")      # → .../webhook/callback
ctx.webhook_url("/oauth/return")  # → .../webhook/oauth/return

Federal contract: the {app_id} component is the web-kernel-authoritative folder/manifest name — not the Python Extension("X", ...) value. If those drift, the URL still reflects the deployed app_id (the one auth-gw and Vault row use). Authors must never hardcode the URL in *_config.py — use this helper at runtime so OAuth redirect URIs stay correct.

Host comes from IMPERAL_PUBLIC_HOST env var (default panel.imperal.io).

See @ext.webhook reference — OAuth callback example for the canonical OAuth flow.

IPC + events — `ctx.extensions`

# Call an exposed method on another extension
result = await ctx.extensions.call("notes", "get_note", note_id="123")

# Emit an event (only events registered via @ext.emits)
await ctx.extensions.emit("my-app.item.created", {"id": "456", "title": "..."})

Time — `ctx.time`

Web-kernel-injected time context. Read-only, no network call.

ctx.time.timezone          # str, e.g. "Europe/Bucharest"
ctx.time.utc_offset        # str, e.g. "+03:00"
ctx.time.now_utc           # ISO datetime string in UTC
ctx.time.now_local         # ISO datetime string in user's timezone
ctx.time.hour_local        # int, e.g. 14
ctx.time.is_business_hours # bool

Skeleton — `ctx.skeleton`

Read-only skeleton data accessor. Only usable inside @ext.skeleton handlers.

@ext.skeleton("tasks_summary", ttl=300)
async def refresh_tasks(ctx) -> dict:
    # ctx.skeleton.get() is allowed here
    prev = await ctx.skeleton.get("tasks_summary")
    # ...
    return {"response": {...}}

Calling ctx.skeleton.get() from any other handler type raises SkeletonAccessForbidden (invariant I-SKELETON-LLM-ONLY). Use ctx.store or ctx.cache for runtime data in regular tools and panels.

`ctx.as_user(user_id)` — system context only

@ext.schedule("nightly", cron="0 3 * * *")
async def nightly(ctx):
    async for user_id in ctx.store.list_users():
        user_ctx = ctx.as_user(user_id)
        data = await user_ctx.store.query("tasks")
        # per-user logic

ctx.as_user() requires ctx.user.imperal_id == "__system__". Raises RuntimeError otherwise. The returned context has the same tenant, agency_id, config, http, ai, and storage but scoped store/skeleton/notify/billing for the target user.

`ctx.progress(percent, message="")`

For long-running operations, report progress. May raise TaskCancelled if the user cancelled.

async def process_large_dataset(ctx, params):
    for i, batch in enumerate(batches):
        try:
            await ctx.progress(i / len(batches) * 100, f"Processing batch {i+1}")
        except TaskCancelled:
            return ActionResult.error("Cancelled by user.", retryable=False)
        await process(batch)

`ctx.log(message, level="info")`

await ctx.log("Processing", level="info")
await ctx.log("Slow response", level="warning")
await ctx.log("Failed", level="error")

Logs are tagged with user_id, tenant_id, ext_name and shipped to SigNoz. PII auto-masked when EXPOSE_PII_TO_LLM=False.

V22 forbids print()

Don't use print() in handlers. The validator catches it at build time. Use ctx.log instead.

What ctx is NOT

❌

Not a database

ctx doesn't store anything. Persist your own state to ctx.store or ctx.db.

❌

Not a session

ctx is per-handler-call. Don't stash references between calls — they expire.

❌

Not a chat history

ctx has no history attribute. The classifier routes by tool description, not by message content.

❌

Not Pydantic

ctx is a typed dataclass — not a Pydantic model. You don't .model_dump() it.

Federal invariants worth knowing

🆔

W1 — canonical user_id

ctx.user.imperal_id is the ONLY canonical user identifier. All audit, all storage keys, everywhere.

🔒

I-SKELETON-LLM-ONLY

ctx.skeleton.get() is gated to @ext.skeleton tools only. Raises SkeletonAccessForbidden elsewhere.

👤

I-AS-USER-1

ctx.as_user() only usable from system context (scheduled tasks).

📦

I-CACHE-MODEL-ON-EXTENSION-INSTANCE

Cache models registered on the Extension instance — not globally.