@ext.on_event + @ext.emits reference

Event bus — publish typed events, subscribe with handlers running in minimal context

@ext.on_event and @ext.emits are the two decorators that connect your extension to the internal platform event bus. One declares what your extension emits; the other subscribes to events from any publisher.

Both decorators are methods on the Extension instance, not on ChatExtension.

Where they live

Register event subscriptions and emission declarations from the same file that holds your ext = Extension(...), or import ext from app.py into a dedicated handlers_events.py module — the pattern used in sql-db.

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 resources.",
    icon="icon.svg",
    actions_explicit=True,
)

@ext.emits("my-app.item_created")
async def _declare_item_created() -> None:  # pragma: no cover
    pass

@ext.on_event("other-app.item_created")
async def on_item_created(ctx, event: dict) -> None:
    # Handle the event with minimal context — see warning below
    pass

`@ext.emits`

Signature

def emits(
    self,
    event_type: str,
    *,
    schema_ref: str | None = None,
) -> Callable:
    ...

@ext.emits is a declaration decorator — it records what event types your extension publishes so the manifest can advertise them under events.emits[]. It does not make your extension emit anything automatically. Actual emission happens at runtime via ctx.extensions.emit(event_type, data) inside a @chat.function or other handler.

Kwargs reference

Prop

Type

Cross-namespace enforcement

The event_type MUST be prefixed with {app_id}.. This is enforced with a ValueError at decoration time — at module import, before deployment. You cannot declare that your extension emits events on behalf of another extension.

from imperal_sdk import Extension

# app_id is "billing"
ext = Extension(
    "billing",
    version="1.0.0",
    display_name="Billing",
    description="Billing — manage subscriptions and top-ups.",
    icon="icon.svg",
    actions_explicit=True,
)

# Correct — prefix matches app_id
@ext.emits("billing.topup_completed")
async def _declare_billing_events() -> None:  # pragma: no cover
    pass

The following raise ValueError immediately at import time:

@ext.emits("payments.topup_completed") — prefix "payments" does not match app_id "billing"
@ext.emits("billing_topup_completed") — no dot separator; not a dotted event type

Handler body convention

The function decorated with @ext.emits is treated as a stub. Its body is never called at runtime. The convention is an empty async function with # pragma: no cover to exclude it from coverage:

from imperal_sdk import Extension

ext = Extension(
    "notes",
    version="1.0.0",
    display_name="Notes",
    description="Notes — create, organize, and search notes with AI.",
    icon="icon.svg",
    actions_explicit=True,
)

@ext.emits("notes.created")
@ext.emits("notes.updated")
@ext.emits("notes.deleted")
@ext.emits("notes.folder_created")
@ext.emits("notes.folder_deleted")
async def _declare_events() -> None:  # pragma: no cover
    pass

This is the exact pattern used in the notes extension (notes/app.py), where twelve event types are declared by stacking the decorator.

Manifest output

Each @ext.emits call appends one entry to events.emits[] in imperal.json:

{
  "events": {
    "emits": [
      { "type": "notes.created" },
      { "type": "notes.updated", "schema_ref": "https://example.com/schema.json" }
    ]
  }
}

`@ext.on_event`

Signature

def on_event(
    self,
    event_type: str,
) -> Callable:
    ...

@ext.on_event subscribes your handler to a platform event type. When any extension emits an event matching event_type, the web-kernel dispatches your handler.

Kwargs reference

Prop

Type

Handler signature

Every event handler receives two positional arguments:

async def my_handler(ctx, event: dict) -> None:
    ...

Argument	Type	Description
`ctx`	`Context \| None`	Minimal context — see critical warning below
`event`	`dict`	The raw event payload dict emitted by the publisher

The return value is ignored. Handlers are fire-and-forget from the publisher's perspective.

Handler ctx is minimal — ctx.http, ctx.store, ctx.cache will crash

The web-kernel dispatches @ext.on_event handlers with a minimal or None context. Accessing ctx.http, ctx.store, ctx.cache, ctx.ai, or any other client from inside an event handler will raise AttributeError at runtime.

The underlying cause: the web-kernel passes ctx=None to event handlers in the current dispatch path. Any attribute access on None raises AttributeError: 'NoneType' object has no attribute '...'. The web-kernel swallows the error and logs it — your handler silently fails without blocking the publisher.

This is a known platform constraint documented in sql-db/events.py. The comment there reads: "Any attempt to do ctx.cache.set from inside an @ext.on_event handler crashes with AttributeError: 'NoneType' object has no attribute 'cache'".

What is safe inside an event handler:

Reading fields from the event dict parameter
Calling pure Python functions that do not require ctx
Logging via the standard logging module

What crashes:

ctx.cache.*
ctx.store.*
ctx.http.*
ctx.ai.*
ctx.notify.*
Any attribute access on ctx at all

Workaround: If you need context-aware side effects in response to an event, do the work inline inside the @chat.function that caused the event, where the full ctx is available. The sql-db extension uses this pattern: cache mutations happen inline in the write handler (fn_execute_sql), not inside @ext.on_event.

Handler isolation

A failure inside @ext.on_event does not propagate to the publisher. Events are dispatched fire-and-forget. If your handler raises, the exception is caught and logged by the web-kernel — the original emitting extension sees no error and the user's chat session is unaffected.

This isolation works in both directions: a slow handler does not delay the publisher, and a crashing handler does not surface to the user.

Manifest output

Each @ext.on_event appends one entry to events.subscribes[] in imperal.json:

{
  "events": {
    "subscribes": [
      { "type": "email.received", "handler": "on_email_received" }
    ]
  }
}

`ctx.extensions.emit`

To actually publish an event at runtime, call ctx.extensions.emit from inside any handler that has a live context — typically a @chat.function.

Signature

async def emit(event_type: str, data: dict) -> None:
    ...

Prop

Type

ctx.extensions.emit publishes to a Redis stream. It is fire-and-forget — the coroutine returns as soon as the event is enqueued, without waiting for any subscriber handler to complete.

The call does not raise if there are no subscribers for the event type.

Event type naming rules

All event types use scope.action format:

Must contain at least one dot. "notes_created" is invalid; "notes.created" is valid.
When emitting from an extension, the prefix must be the extension's app_id. The SDK enforces this at decoration time with a ValueError.
When subscribing, any dotted type is valid. You may subscribe to events emitted by any extension.
Use lowercase, snake_case. "notes.folder_created" not "notes.FolderCreated".
Use a noun-verb or noun.past_participle pattern. "notes.created", "sql.ddl_executed", "table.touched", "email.received".

Cross-extension event flow

A typical publish-subscribe pair across two extensions:

notes (publisher)                   my-app (subscriber)
─────────────────                   ────────────────────
@ext.emits("notes.created")         @ext.on_event("notes.created")
                                    async def on_note_created(ctx, event):
                                        # ctx is minimal — read event only

In @chat.function:
  await ctx.extensions.emit(
      "notes.created",
      {"note_id": "...", "title": "..."},
  )
           │
           └─► web-kernel Redis pub/sub ─► on_note_created(None, event_dict)

The publisher (notes) calls ctx.extensions.emit at runtime. The web-kernel routes the payload to all registered @ext.on_event handlers for "notes.created" across all installed extensions.

The publisher does not need to know which extensions are subscribed. The subscriber does not need the publisher to be installed — if no event arrives, the handler simply never fires.

Examples

Example 1 — Emit from a write `@chat.function`

This is the canonical pattern: declare the emission with @ext.emits, then call ctx.extensions.emit inside the handler.

import logging
from pydantic import BaseModel
from imperal_sdk import Extension
from imperal_sdk.chat import ChatExtension, ActionResult

log = logging.getLogger(__name__)

ext = Extension(
    "inventory",
    version="1.0.0",
    display_name="Inventory",
    description="Inventory — track and manage items across locations.",
    icon="icon.svg",
    actions_explicit=True,
)
chat = ChatExtension(ext, tool_name="inventory", description="Inventory assistant")


@ext.emits("inventory.item_created")
async def _declare_events() -> None:  # pragma: no cover
    pass


class CreateItemParams(BaseModel):
    name: str
    location: str
    quantity: int


@chat.function(  # type: ignore[attr-defined]
    "create_item",
    action_type="write",
    chain_callable=True,
    effects=["create:item"],
    description="Create a new inventory item at the given location.",
)
async def create_item(ctx, params: CreateItemParams) -> ActionResult:
    item = await ctx.store.create("items", {
        "name": params.name,
        "location": params.location,
        "quantity": params.quantity,
    })
    await ctx.extensions.emit("inventory.item_created", {
        "item_id": item.id,
        "name": params.name,
        "location": params.location,
    })
    return ActionResult.success(
        data={"item_id": item.id},
        summary=f"Created item '{params.name}' at {params.location}.",
    )

This pattern is safe: the handler only reads from the event dict and uses the standard logging module. No ctx attributes are accessed.

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)

ext = Extension(
    "audit-log",
    version="1.0.0",
    display_name="Audit Log",
    description="Audit Log — record all inventory changes for compliance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.on_event("inventory.item_created")
async def on_item_created(ctx, event: dict) -> None:
    # Safe: only reads from the event dict + logs.
    # Do NOT access ctx.store, ctx.http, ctx.cache, or any ctx attribute here.
    item_id = event.get("item_id", "unknown")
    name = event.get("name", "unknown")
    log.info("audit: item_created item_id=%s name=%s", item_id, name)

Example 3 — Work inline instead of in the subscriber

When you need ctx-aware side effects in response to an event your own extension emits, do the work in the emitting handler — where the full ctx is available — rather than trying to do it in @ext.on_event. This is the pattern established in sql-db (sql-db/handlers_execute.py):

import logging
from pydantic import BaseModel
from imperal_sdk import Extension
from imperal_sdk.chat import ChatExtension, ActionResult

log = logging.getLogger(__name__)

ext = Extension(
    "data-store",
    version="1.0.0",
    display_name="Data Store",
    description="Data Store — manage structured data with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)
chat = ChatExtension(ext, tool_name="data-store", description="Data Store assistant")

CACHE_KEY = "sidebar_v1"


@ext.emits("data-store.record_written")
async def _declare_ds_events() -> None:  # pragma: no cover
    pass


async def _invalidate_sidebar_cache(ctx, *, table: str) -> None:
    """Call inline where ctx is available — not from @ext.on_event."""
    try:
        await ctx.cache.delete(f"sidebar:{table}")
    except Exception as exc:
        log.warning("cache invalidation failed: %s", exc)


class WriteRecordParams(BaseModel):
    table: str
    data: dict


@chat.function(  # type: ignore[attr-defined]
    "write_record",
    action_type="write",
    chain_callable=True,
    effects=["create:record"],
    description="Write a new record to the given table.",
)
async def write_record(ctx, params: WriteRecordParams) -> ActionResult:
    doc = await ctx.store.create(params.table, params.data)
    # Do ctx work inline — not deferred to @ext.on_event
    await _invalidate_sidebar_cache(ctx, table=params.table)
    # Fire event for cross-extension consumers
    await ctx.extensions.emit("data-store.record_written", {
        "table": params.table,
        "record_id": doc.id,
    })
    return ActionResult.success(
        data={"record_id": doc.id},
        summary=f"Written to {params.table}.",
    )

Common pitfalls

Accessing ctx inside `@ext.on_event`

Inside @ext.on_event, ctx is None. Any attribute access crashes with AttributeError. The web-kernel swallows the error — the failure is silent and the user never sees it.

Unsafe (crashes silently at runtime):

# CRASH — ctx is None; AttributeError: 'NoneType' has no 'cache'
async def on_note_created(ctx, event: dict) -> None:
    await ctx.cache.set("last_event", event)

Safe (read event dict and log only):

import logging

async def on_note_created(ctx, event: dict) -> None:
    logging.getLogger(__name__).info("note created: %s", event.get("note_id"))

Missing `@ext.emits` declaration

ctx.extensions.emit works at runtime even without a corresponding @ext.emits declaration. However, without the declaration the manifest does not advertise the event type. Other extensions cannot discover and subscribe to it reliably. Always declare every event type your extension emits.

Naming format violations

These raise ValueError at decoration time (module import), not at runtime:

Missing dot separator: @ext.emits("notes_created") — event type must be dotted
Wrong prefix: @ext.emits("mail.created") on an extension with app_id="notes" — must be prefixed by your own app_id

Use scope.action format with lowercase snake_case: "notes.created", "sql.ddl_executed".

Cross-namespace subscription

You can subscribe to any event type — subscriptions are not restricted by namespace:

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 resources.",
    icon="icon.svg",
    actions_explicit=True,
)

# Subscribing to another extension's events is allowed
@ext.on_event("notes.created")
async def on_note(ctx, event: dict) -> None:
    pass

Emitting outside your own namespace is not:

# ValueError — "notes" prefix does not match this extension's app_id ("my-app")
@ext.emits("notes.created")

@ext.on_event + @ext.emits reference

Where they live

`@ext.emits`

Signature

Kwargs reference

Cross-namespace enforcement

Handler body convention

Manifest output

`@ext.on_event`

Signature

Kwargs reference

Handler signature

Handler isolation

Manifest output

`ctx.extensions.emit`

Signature

Event type naming rules

Cross-extension event flow

Examples

Example 1 — Emit from a write `@chat.function`

Example 3 — Work inline instead of in the subscriber

Common pitfalls

Accessing ctx inside `@ext.on_event`

Missing `@ext.emits` declaration

Naming format violations

Cross-namespace subscription

Expecting synchronous side effects

Cross-references

Events concept

@chat.function reference

@ext.schedule reference

@ext.webhook reference

On this page