@ext.webhook reference

External HTTP receiver — path routing, manual HMAC verification, action authorization

@ext.webhook registers an HTTP endpoint your extension exposes for external callers — Stripe, GitHub, Slack, custom services, or any system that can send an HTTP request. The platform routes incoming requests to your handler. Your handler is responsible for verifying the request signature. There is no built-in helper.

Use webhooks for event-driven work triggered by external systems: payment events, CI build notifications, inbound message hooks, third-party CRM triggers. For time-driven work use @ext.schedule; for LLM-triggered actions use @chat.function.

Where it lives

@ext.webhook is a method on the Extension instance, not on ChatExtension. Register it from the same file that holds your ext = Extension(...), or import ext from app.py into a dedicated handlers_webhook.py module.

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.webhook("events", method="POST", secret_header="X-My-Signature-256")
async def handle_events(ctx, headers: dict, body: str, query_params: dict) -> dict:
    """Receive external events."""
    return {"status": "ok"}

Signature

def webhook(
    self,
    path: str,
    method: str = "POST",
    secret_header: str = "",
) -> Callable:
    ...

All parameters are positional. method and secret_header have defaults.

Kwargs reference

`path`

Prop

Type

Use a short, descriptive slug: "events", "payment", "github", "callback", "inbound".
Leading slash is normalised — both "events" and "/events" register the same path.
Each @ext.webhook in an extension must use a different path.
The full public URL the external caller hits is: https://panel.imperal.io/v1/ext/{app_id}/webhook/{path} — where {app_id} is the manifest/folder app_id (not whatever you typed into Extension("X", ...) in Python — if those drift, the deployed URL uses the manifest value).
Don't hardcode the URL in your config files. Use ctx.webhook_url(path) at runtime — it auto-tracks the canonical app_id. Hardcoded URLs are the #1 cause of OAuth-callback drift bugs.
The end-user sees the canonical URL automatically in Imperal Panel → ext → Settings → Secrets (info card with Copy button per declared webhook) and you the developer see it in Dev Portal → App Details → Webhooks tab.

`method`

Prop

Type

POST — most webhook providers (Stripe, GitHub events, SendGrid, custom services).

GET — two cases, both production-supported since 2026-05-13:

OAuth callbacks. OAuth 2.0 authorization code flow redirects the user-agent to your registered redirect URI with ?code=...&state=... in the query string. That's a GET. You exchange the code for tokens inside the handler and persist via ctx.secrets.set(...). See the OAuth callback example below.
One-time URL verification. Slack, Facebook, Twilio handshakes echo a challenge token in query_params before activating the subscription.

To accept both methods on the same path, register two handlers with different method values, or branch inside one handler based on query_params.

`secret_header`

Prop

Type

secret_header only declares — it does not verify

Setting secret_header tells the platform which header carries the signature and records it in the manifest so operators know which header to configure. It does not perform any verification. Your handler must read the header from the headers dict and verify the signature using hmac.compare_digest. See the HMAC verification section below for the canonical pattern.

Handler signature

Every webhook handler receives four arguments:

async def my_handler(ctx, headers: dict, body: str, query_params: dict) -> dict:
    ...

Argument	Type	Description
`ctx`	`Context`	Minimal webhook context — `ctx.user.imperal_id` is `"__webhook__"`. See below for available attributes.
`headers`	`dict[str, str]`	Incoming request headers with hop-by-hop headers (e.g. `Connection`, `Transfer-Encoding`) stripped. Keys are lowercase.
`body`	`str`	Raw request body as a UTF-8 string. JSON payloads arrive as a JSON string — call `json.loads(body)` yourself.
`query_params`	`dict[str, str]`	URL query parameters. For `GET` verification challenges the challenge token is here.

Return value

Return a dict. The platform serializes it as a JSON response to the external caller with status 200. Return {"status": "ok"} as the minimal acknowledgement. If you need to return a specific HTTP status, include "status_code" in the dict — the platform will use it as the HTTP response code.

Available ctx attributes in webhook handlers

Webhook handlers run in a minimal context. The ctx.user actor is the platform's webhook system identity, not a real user.

Attribute	Available?	Notes
`ctx.user`	Yes	`imperal_id == "__webhook__"` — not a real user
`ctx.user.imperal_id`	Yes	Always `"__webhook__"`
`ctx.tenant`	No — `None`	No tenant in webhook context
`ctx.store`	Yes	System namespace — not user-scoped
`ctx.http`	Yes	Outbound HTTP for callbacks and acknowledgements
`ctx.config`	Yes	Admin-configured extension settings (e.g. stored shared secret). Not for credentials — see `ctx.secrets`.
`ctx.secrets`	Yes (v4.2.2+)	Encrypted user credentials. In webhook context the writes target `__webhook__` user — use `ctx.as_user(uid).secrets` to write a real user's secret after you resolve which user the event maps to.
`ctx.extensions`	Yes	IPC calls to other extensions
`ctx.log(msg)`	Yes	Structured log in extension dashboard
`ctx.webhook_url(path)`	Yes (v4.2.7+)	Returns the canonical public URL for any `@ext.webhook` path declared by this extension. See below.
`ctx.ai`	No	Not available in webhook context
`ctx.billing`	No	Not available in webhook context
`ctx.notify`	No	No real user — notify requires a user context
`ctx.cache`	Yes (v4.2.7+)	Available since ContextFactory inject — same caveat about webhook namespace.
`ctx.storage`	No	Not available in webhook context
`ctx.skeleton`	No	Raises `SkeletonAccessForbidden` — guarded to skeleton handlers only
`ctx.progress`	No	Only meaningful in user-facing tool calls
`ctx.as_user(uid)`	Yes	Obtain a user-scoped context for writing user data

Writing user data from webhooks

To write into a specific user's store partition after receiving a webhook event, call ctx.as_user(user_id) where user_id is an imperal_id you previously stored (e.g. when the user connected the external service). Then use user_ctx.store, user_ctx.notify, etc. to act on their data.

Internal storage and manifest

The decorator stores the handler in two places:

_webhooks[path] = WebhookDef(path=path, func=func, method=method, secret_header=secret_header)
# Also registered as a synthetic ToolDef so the platform can dispatch it:
_tools["__webhook__{path}"] = ToolDef(name="__webhook__{path}", func=func, ...)

The manifest includes a webhooks array:

{
  "webhooks": [
    {
      "path": "events",
      "method": "POST",
      "secret_header": "X-Hub-Signature-256"
    }
  ]
}

The synthetic __webhook__{path} tool is never exposed to the LLM and does not appear in the manifest's tools array. It is an internal dispatch mechanism.

Critical: secret verification is manual (audit S4)

No built-in HMAC helper — you must verify manually

The SDK provides no ctx.verify_webhook_secret(...) method or similar helper. The secret_header kwarg only declares which header carries the signature for manifest and operator documentation purposes. Your handler must read the header, retrieve the shared secret, and verify using Python's stdlib hmac.compare_digest. Omitting verification is a critical security gap — any caller on the public internet can trigger your handler.

Why `hmac.compare_digest` — not `==`

Comparing two strings with == takes less time when the strings diverge early (short-circuit evaluation). An attacker can exploit this timing difference to guess the signature one byte at a time. hmac.compare_digest uses a constant-time algorithm that always inspects every byte, making timing attacks infeasible.

import hashlib
import hmac

received_sig = "abc123"
expected_sig = "abc123"

# WRONG — vulnerable to timing attack
if received_sig == expected_sig:
    pass

# CORRECT — constant-time comparison
if hmac.compare_digest(received_sig, expected_sig):
    pass

Canonical HMAC verification pattern

import hashlib
import hmac
import json
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.webhook("events", method="POST", secret_header="X-My-Signature-256")
async def handle_events(ctx, headers: dict, body: str, query_params: dict) -> dict:
    """Verify HMAC-SHA256 and process the event."""
    # 1. Retrieve the shared secret from admin config (never hardcode)
    secret = await ctx.config.require("webhook_secret")

    # 2. Read the signature header the provider sent
    received_sig = headers.get("x-my-signature-256", "")
    if not received_sig:
        await ctx.log("webhook: missing signature header", level="warning")
        return {"status_code": 401, "error": "missing signature"}

    # 3. Compute the expected signature over the raw body
    expected_sig = hmac.new(
        secret.encode(), body.encode(), hashlib.sha256
    ).hexdigest()

    # 4. Constant-time comparison — rejects timing attacks
    if not hmac.compare_digest(received_sig, expected_sig):
        await ctx.log("webhook: signature mismatch", level="warning")
        return {"status_code": 401, "error": "invalid signature"}

    # 5. Signature verified — safe to process
    payload = json.loads(body)
    event_type = payload.get("type", "")
    await ctx.log(f"webhook: received {event_type}")
    return {"status": "ok"}

Examples

Stripe webhook with HMAC-SHA256 verification

Stripe signs webhooks using its own multi-part Stripe-Signature header format: t=<timestamp>,v1=<sig>. The canonical verification reconstructs the signed payload from the timestamp and raw body, then compares against the v1 element.

import hashlib
import hmac
import json
import logging
import time
from imperal_sdk import Extension

log = logging.getLogger(__name__)
ext = Extension(
    "billing-sync",
    version="1.0.0",
    display_name="Billing Sync",
    description="Billing Sync — keeps your billing records in sync with Stripe.",
    icon="icon.svg",
    actions_explicit=True,
)

_STRIPE_TOLERANCE_SECONDS = 300


@ext.webhook("stripe", method="POST", secret_header="Stripe-Signature")
async def handle_stripe(ctx, headers: dict, body: str, query_params: dict) -> dict:
    """Verify Stripe webhook signature and dispatch payment events."""
    stripe_secret = await ctx.config.require("stripe_webhook_secret")

    sig_header = headers.get("stripe-signature", "")
    if not sig_header:
        return {"status_code": 400, "error": "missing Stripe-Signature"}

    # Parse: t=<ts>,v1=<sig1>[,v1=<sig2>]
    parts: dict[str, list[str]] = {}
    for chunk in sig_header.split(","):
        if "=" not in chunk:
            continue
        k, v = chunk.split("=", 1)
        parts.setdefault(k.strip(), []).append(v.strip())

    timestamp_strs = parts.get("t", [])
    signatures = parts.get("v1", [])
    if not timestamp_strs or not signatures:
        return {"status_code": 400, "error": "malformed Stripe-Signature"}

    try:
        event_timestamp = int(timestamp_strs[0])
    except ValueError:
        return {"status_code": 400, "error": "non-integer timestamp"}

    # Replay protection: reject events older than tolerance window
    now = int(time.time())
    if abs(now - event_timestamp) > _STRIPE_TOLERANCE_SECONDS:
        await ctx.log("stripe webhook: replay rejected", level="warning")
        return {"status_code": 400, "error": "timestamp out of tolerance"}

    # Reconstruct the signed payload: "<timestamp>.<raw_body>"
    signed_payload = f"{event_timestamp}.{body}"
    expected = hmac.new(
        stripe_secret.encode(), signed_payload.encode(), hashlib.sha256
    ).hexdigest()

    if not any(hmac.compare_digest(expected, sig) for sig in signatures):
        await ctx.log("stripe webhook: signature mismatch", level="warning")
        return {"status_code": 401, "error": "invalid signature"}

    event = json.loads(body)
    event_type: str = event.get("type", "")
    await ctx.log(f"stripe webhook: {event_type}")

    if event_type == "invoice.payment_succeeded":
        customer_id: str = event.get("data", {}).get("object", {}).get("customer", "")
        await ctx.log(f"stripe: payment succeeded for customer {customer_id}")
        # Retrieve the internal user_id you stored when the user connected Stripe
        doc = await ctx.store.get("stripe_customers", customer_id)
        if doc:
            user_ctx = ctx.as_user(doc.data.get("imperal_id", ""))
            await user_ctx.notify("Your payment was processed successfully.")

    return {"status": "ok"}

GitHub webhook with hex signature verification

GitHub sends an X-Hub-Signature-256 header: sha256=<hex_digest>. Strip the algorithm prefix before comparing.

import hashlib
import hmac
import json
import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)
ext = Extension(
    "ci-bridge",
    version="1.0.0",
    display_name="CI Bridge",
    description="CI Bridge — surface GitHub CI events inside your workspace.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.webhook("github", method="POST", secret_header="X-Hub-Signature-256")
async def handle_github(ctx, headers: dict, body: str, query_params: dict) -> dict:
    """Verify GitHub HMAC-SHA256 and handle push/PR events."""
    secret = await ctx.config.require("github_webhook_secret")

    raw_sig = headers.get("x-hub-signature-256", "")
    if not raw_sig.startswith("sha256="):
        return {"status_code": 400, "error": "missing or malformed X-Hub-Signature-256"}

    received_hex = raw_sig[len("sha256="):]
    expected_hex = hmac.new(
        secret.encode(), body.encode(), hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(received_hex, expected_hex):
        await ctx.log("github webhook: signature mismatch", level="warning")
        return {"status_code": 401, "error": "invalid signature"}

    event_name = headers.get("x-github-event", "")
    payload = json.loads(body)
    repo = payload.get("repository", {}).get("full_name", "")
    await ctx.log(f"github webhook: {event_name} on {repo}")

    if event_name == "push":
        branch = payload.get("ref", "").replace("refs/heads/", "")
        pusher = payload.get("pusher", {}).get("name", "")
        await ctx.log(f"github: push to {branch} by {pusher}")

    return {"status": "ok"}

Generic JSON webhook with shared-secret verification

For services that pass a plain shared secret in a custom header (rather than an HMAC digest), compare the header value directly using hmac.compare_digest — still required to prevent timing attacks.

import hmac
import json
import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)
ext = Extension(
    "notify-bridge",
    version="1.0.0",
    display_name="Notify Bridge",
    description="Notify Bridge — receive events from external notification systems.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.webhook("inbound", method="POST", secret_header="X-Notify-Secret")
async def handle_inbound(ctx, headers: dict, body: str, query_params: dict) -> dict:
    """Verify shared secret and store inbound event."""
    expected_secret = await ctx.config.require("notify_shared_secret")
    received_secret = headers.get("x-notify-secret", "")

    # Compare with constant-time algorithm — even for plain shared secrets
    if not received_secret or not hmac.compare_digest(
        received_secret.encode(), expected_secret.encode()
    ):
        await ctx.log("inbound webhook: auth failure", level="warning")
        return {"status_code": 401, "error": "unauthorized"}

    payload = json.loads(body)
    event_id = payload.get("id", "")

    # Idempotency: skip if already processed
    existing = await ctx.store.get("inbound_events", event_id)
    if existing:
        return {"status": "already_processed"}

    await ctx.store.create("inbound_events", {"id": event_id, "data": payload})
    await ctx.log(f"inbound: stored event {event_id}")
    return {"status": "ok"}

`ctx.webhook_url`

ctx.webhook_url(path) is the canonical way to compute your extension's public webhook URL at runtime. Available since SDK v4.2.7 (2026-05-13).

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

Why use it:

The {app_id} component is sourced from the web-kernel-authoritative folder/manifest name — not whatever you typed into Extension("X", ...) in app.py. If those drift (e.g. you wrote Extension("spotify-extension", ...) but the deployed folder is spotify), the URL is still correct.
The host comes from the IMPERAL_PUBLIC_HOST env var (default panel.imperal.io). One deployment can target a different domain without code changes.
Authors stop hardcoding URLs in *_config.py files — the #1 cause of OAuth redirect-URI drift bugs.

Path normalisation:

ctx.webhook_url("/callback")      # → .../webhook/callback
ctx.webhook_url("callback")       # → .../webhook/callback  (leading slash optional)
ctx.webhook_url("/oauth/return")  # → .../webhook/oauth/return  (multi-segment OK)

Where the URL is shown to users: Imperal Panel → ext → Settings → Secrets renders a blue info card listing every @ext.webhook URL with a Copy button. End-users paste those into the OAuth provider's developer console before they configure their client_id / client_secret via @ext.secret.

OAuth callback example

Spotify-style OAuth 2.0 authorization code flow. Two pieces: a @chat.function that builds the auth URL and returns it to the user (or auto-opens it), and a @ext.webhook("/callback", method="GET") that receives the redirect.

import json
import logging
from urllib.parse import urlencode
from imperal_sdk import Extension, ChatExtension, ActionResult

log = logging.getLogger(__name__)
ext = Extension(
    "spotify",
    version="1.0.0",
    display_name="Spotify",
    description="Spotify — search, queue, manage your library via AI.",
    icon="icon.svg",
    actions_explicit=True,
)
chat = ChatExtension(ext, tool_name="spotify",
                     description="Spotify integration")

# Declare secrets — user pastes client_id / client_secret from the
# Spotify Developer Dashboard; ext writes the refresh_token after the
# OAuth dance completes.
ext.secret(name="spotify_client_id", description="From developer.spotify.com → app → Settings",
           required=True, write_mode="user", max_bytes=256)(lambda: None)
ext.secret(name="spotify_client_secret", description="From developer.spotify.com → app → View client secret",
           required=True, write_mode="user", max_bytes=512)(lambda: None)
ext.secret(name="spotify_refresh_token", description="OAuth refresh token — written by extension after authorize.",
           write_mode="extension", max_bytes=4096)(lambda: None)


@chat.function(
    description="Start Spotify OAuth — returns the URL the user must visit to authorize.",
    action_type="read",
)
async def start_spotify_auth(ctx, params):
    client_id = await ctx.secrets.get("spotify_client_id")
    if not client_id:
        return ActionResult.error(
            "Spotify client_id isn't configured. Add it in extension settings.",
            retryable=False,
        )

    # Build redirect URI from web-kernel-authoritative app_id — never hardcode.
    # Tell the user to register this URL in their Spotify app's settings.
    redirect_uri = ctx.webhook_url("/callback")

    auth_url = "https://accounts.spotify.com/authorize?" + urlencode({
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "response_type": "code",
        "scope": "user-read-private user-library-read",
    })
    return ActionResult.success(
        {"auth_url": auth_url, "redirect_uri": redirect_uri},
        summary=f"Visit {auth_url} to connect your Spotify account.",
    )


@ext.webhook("/callback", method="GET")
async def oauth_callback(ctx, headers, body, query_params):
    """Receive Spotify's 302 redirect with ?code=...&state=...

    Exchange the auth code for tokens and store the refresh_token via
    ctx.secrets.set (write_mode='extension' allows it). The user is
    not logged in to a session at this point — resolve which Imperal
    user this OAuth dance belongs to via the `state` parameter you
    set when you built the auth URL.
    """
    code = query_params.get("code", "")
    state = query_params.get("state", "")
    if not code:
        return {"status_code": 400, "error": "missing code"}

    # Resolve which user started this OAuth flow (you stored state→user
    # mapping when start_spotify_auth ran).
    mapping = await ctx.store.get("oauth_states", state)
    if not mapping:
        return {"status_code": 400, "error": "unknown state"}
    imperal_id = mapping.data.get("imperal_id", "")

    user_ctx = ctx.as_user(imperal_id)
    client_id = await user_ctx.secrets.get("spotify_client_id")
    client_secret = await user_ctx.secrets.get("spotify_client_secret")
    redirect_uri = user_ctx.webhook_url("/callback")

    # Exchange code for tokens (Spotify token endpoint omitted for brevity)
    token_response = await user_ctx.http.post(
        "https://accounts.spotify.com/api/token",
        data={
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": redirect_uri,
            "client_id": client_id,
            "client_secret": client_secret,
        },
    )
    tokens = token_response.json()
    await user_ctx.secrets.set("spotify_refresh_token", tokens["refresh_token"])
    await user_ctx.notify("Spotify connected. You can now ask me to play your music.")
    return {"status": "ok"}

The user-visible flow:

User runs the extension and is told "Visit <auth_url> to connect Spotify."
User clicks the link, Spotify shows the consent screen.
User clicks Allow — Spotify 302-redirects to https://panel.imperal.io/v1/ext/spotify/webhook/callback?code=...&state=....
The platform routes /v1/ext/* internally and dispatches the request to your oauth_callback handler with ctx already populated.
Handler exchanges the code, stores the refresh token, notifies the user.

Before any of this works, the user must register redirect_uri (the value of ctx.webhook_url("/callback")) in their Spotify Developer Dashboard's "Redirect URIs" list — the exact URL, including trailing path. The Imperal Panel Secrets tab shows the canonical URL with a Copy button to make this paste step painless.

Common pitfalls

Omitting HMAC verification entirely

Skipping signature verification means any caller on the public internet can trigger your webhook handler — including attackers who know your endpoint path. Always verify before acting on the payload.

import json
import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)
ext = Extension(
    "my-app",
    display_name="My App",
    description="My App — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


# WRONG — no verification; attacker can send arbitrary payloads
@ext.webhook("events", method="POST", secret_header="X-My-Sig")
async def insecure_handler(ctx, headers: dict, body: str, query_params: dict) -> dict:
    payload = json.loads(body)
    await ctx.store.create("events", payload)  # executed without any auth check
    return {"status": "ok"}

Using `==` for signature comparison

import hashlib
import hmac
import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)
ext = Extension(
    "my-app",
    display_name="My App",
    description="My App — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


# WRONG — timing-vulnerable comparison
@ext.webhook("events", method="POST", secret_header="X-My-Sig")
async def timing_vulnerable(ctx, headers: dict, body: str, query_params: dict) -> dict:
    secret = await ctx.config.require("webhook_secret")
    expected = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest()
    received = headers.get("x-my-sig", "")
    if received == expected:          # WRONG — short-circuits; timing attack possible
        return {"status": "ok"}
    return {"status_code": 401, "error": "invalid signature"}


# CORRECT — constant-time comparison
@ext.webhook("events_v2", method="POST", secret_header="X-My-Sig")
async def timing_safe(ctx, headers: dict, body: str, query_params: dict) -> dict:
    secret = await ctx.config.require("webhook_secret")
    expected = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest()
    received = headers.get("x-my-sig", "")
    if hmac.compare_digest(received, expected):   # constant-time
        return {"status": "ok"}
    return {"status_code": 401, "error": "invalid signature"}

Returning sensitive data in the response body

Webhook responses are logged by the platform and potentially by the external caller. Never include PII, secrets, or internal IDs in the response body.

import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)
ext = Extension(
    "my-app",
    display_name="My App",
    description="My App — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


# WRONG — returns internal identifiers to the external caller
@ext.webhook("payment", method="POST", secret_header="X-Pay-Sig")
async def leaky_handler(ctx, headers: dict, body: str, query_params: dict) -> dict:
    return {"status": "ok", "internal_user_id": "imp_u_abc123", "secret": "..."}


# CORRECT — acknowledge only
@ext.webhook("payment_v2", method="POST", secret_header="X-Pay-Sig")
async def clean_handler(ctx, headers: dict, body: str, query_params: dict) -> dict:
    return {"status": "ok"}

Missing idempotency for replays

External webhook systems retry failed deliveries. If your handler creates a record on the first call and the provider retries, you may end up with duplicates. Use the event's provider-issued ID as a document key and check for an existing record before writing.

import hashlib
import hmac
import json
import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)
ext = Extension(
    "my-app",
    display_name="My App",
    description="My App — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


# WRONG — creates duplicate records on retry
@ext.webhook("order", method="POST")
async def non_idempotent(ctx, headers: dict, body: str, query_params: dict) -> dict:
    payload = json.loads(body)
    await ctx.store.create("orders", payload)   # called twice on retry = two docs
    return {"status": "ok"}


# CORRECT — idempotent via event ID
@ext.webhook("order_v2", method="POST", secret_header="X-Order-Sig")
async def idempotent(ctx, headers: dict, body: str, query_params: dict) -> dict:
    secret = await ctx.config.require("order_webhook_secret")
    received = headers.get("x-order-sig", "")
    expected = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest()
    if not received or not hmac.compare_digest(received, expected):
        return {"status_code": 401, "error": "invalid signature"}

    payload = json.loads(body)
    event_id = payload.get("event_id", "")
    if not event_id:
        return {"status_code": 400, "error": "missing event_id"}

    existing = await ctx.store.get("orders", event_id)
    if existing:
        return {"status": "already_processed"}

    await ctx.store.create("orders", {**payload, "id": event_id})
    return {"status": "ok"}

Registering @ext.webhook on ChatExtension

@ext.webhook must be called on the Extension instance, not on ChatExtension. ChatExtension has no webhook method and will raise AttributeError.

from imperal_sdk import Extension, ChatExtension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)
chat = ChatExtension(
    ext,
    tool_name="my_app_chat",
    description="My App AI assistant.",
)

# CORRECT — register on ext, not on chat
@ext.webhook("events", method="POST", secret_header="X-My-Sig")
async def handle_events(ctx, headers: dict, body: str, query_params: dict) -> dict:
    return {"status": "ok"}

@ext.webhook reference

Where it lives

Signature

Kwargs reference

`path`

`method`

`secret_header`

Handler signature

Return value

Available ctx attributes in webhook handlers

Internal storage and manifest

Critical: secret verification is manual (audit S4)

Why `hmac.compare_digest` — not `==`

Canonical HMAC verification pattern

Examples

Stripe webhook with HMAC-SHA256 verification

GitHub webhook with hex signature verification

Generic JSON webhook with shared-secret verification

`ctx.webhook_url`

OAuth callback example

Common pitfalls

Omitting HMAC verification entirely

Using `==` for signature comparison

Returning sensitive data in the response body

Missing idempotency for replays

Registering @ext.webhook on ChatExtension

Cross-references

Webhooks concept

Decorators reference

@ext.schedule reference

Security and audit

On this page