Webhooks

A webhook is an HTTP endpoint your extension exposes to the outside world. External systems — payment processors, source control platforms, notification services, or any custom caller — send HTTP requests to it when something happens on their side. Your handler receives the request and acts on it.

Webhooks are event-driven: a caller decides when to send a request. This is the defining difference from all other extension trigger types.

Webhooks vs other extension trigger types

Extension code runs in four fundamentally different modes. Knowing when each fires prevents reaching for the wrong primitive.

Webhook vs @chat.function

A @chat.function is invoked when the LLM decides a user's message requires an action. The user is present, authenticated, and the context is fully populated. The LLM controls when it fires.

A webhook fires when an external system decides to send a request — there is no user, no chat session, and no LLM routing involved. Use webhooks when the trigger lives outside the platform entirely.

Webhook vs @ext.schedule

Both run outside a user chat session. The difference is the trigger:

• Webhook: event-driven — runs because an external system sent a request.
• Scheduled task: timer-driven — runs because time has passed.

If you need to poll an external API for changes, prefer @ext.schedule. If the external system can push notifications to you, prefer @ext.webhook — it eliminates polling entirely.

Webhook vs @ext.on_event

Both react to events, but the event origin is different:

• Webhook: the event source is external — a third-party HTTP POST.
• @ext.on_event: the event source is internal — another extension emitting to the platform event bus.

For inter-extension communication, use @ext.on_event. For third-party integrations, use @ext.webhook.

Routing model

The platform maps incoming requests to webhook handlers using the URL path. When you register:

from imperal_sdk import Extension

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

@ext.webhook("stripe", method="POST", secret_header="Stripe-Signature")
async def handle_stripe(ctx, headers: dict, body: str, query_params: dict) -> dict:
    return {"status": "ok"}

The platform exposes the endpoint at:

POST https://panel.imperal.io/v1/ext/{app_id}/webhook/stripe
GET  https://panel.imperal.io/v1/ext/{app_id}/webhook/stripe   (also accepted)

Where {app_id} is your extension's manifest/folder app_id — not the Python value you typed into Extension("X", ...). If those drift, the deployed URL uses the manifest value; the SDK helper ctx.webhook_url(path) always returns the correct one.

The path suffix after webhook/ is exactly the path argument you pass to @ext.webhook.

What the platform does for you:

• panel.imperal.io/v1/ext/* is routed internally to your registered webhook handler — you don't configure or address the gateway directly.
• The platform accepts both GET (OAuth callbacks, verification challenges) and POST (server-to-server hooks) on the same path.
• Each request is dispatched to your @ext.webhook(path=...) handler with ctx already populated.

You can register multiple webhook handlers in one extension, each with a different path:

from imperal_sdk import Extension

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

@ext.webhook("stripe", method="POST", secret_header="Stripe-Signature")
async def handle_stripe(ctx, headers: dict, body: str, query_params: dict) -> dict:
    return {"status": "ok"}

@ext.webhook("github", method="POST", secret_header="X-Hub-Signature-256")
async def handle_github(ctx, headers: dict, body: str, query_params: dict) -> dict:
    return {"status": "ok"}

Each path maps to exactly one handler. If two handlers use the same path, the second registration overwrites the first.

GET handlers — OAuth callbacks + verification challenges

method="GET" covers two production-supported use cases since 2026-05-13:

1. OAuth 2.0 authorization code redirects. Spotify, GitHub OAuth, Google sign-in — after the user approves the consent screen, the provider 302-redirects the user-agent to your registered redirect URI with ?code=...&state=... in the query string. Your handler exchanges the code for an access/refresh token pair and stores them via ctx.secrets.set(...).

@ext.webhook("/callback", method="GET")
async def oauth_callback(ctx, headers, body, query_params):
    code = query_params.get("code", "")
    state = query_params.get("state", "")
    # ...resolve state → imperal_id, exchange code for tokens, store refresh_token...
    return {"status": "ok"}

The redirect URI you register in the provider's developer console (Spotify Dashboard → Redirect URIs, etc.) is the value of ctx.webhook_url("/callback") — the canonical public URL. The Imperal Panel Secrets tab shows it automatically with a Copy button.

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

2. One-time URL verification. Slack, Facebook, Twilio handshakes echo a challenge token in query_params before activating the subscription.

from imperal_sdk import Extension

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

@ext.webhook("events", method="GET")
async def verify_subscription(ctx, headers: dict, body: str, query_params: dict) -> dict:
    challenge = query_params.get("hub.challenge", "")
    if challenge:
        return {"status_code": 200, "body": challenge}
    return {"status_code": 400, "error": "no challenge"}

Request envelope

Every webhook handler receives:

The body arrives as a raw string by design — HMAC verification must be computed over the exact bytes the sender signed. If you parsed JSON first, you could not verify the signature. Always verify first, then parse.

import hashlib
import hmac
import json
from imperal_sdk import Extension

ext = Extension(
    "example",
    version="1.0.0",
    display_name="Example",
    description="Example extension for documentation purposes.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.webhook("events", method="POST", secret_header="X-My-Sig")
async def handle(ctx, headers: dict, body: str, query_params: dict) -> dict:
    # 1. Verify FIRST — over the raw body string
    secret = await ctx.config.require("webhook_secret")
    received = headers.get("x-my-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"}

    # 2. Parse AFTER verification
    payload = json.loads(body)
    return {"status": "ok"}

The webhook context

Webhook handlers run in a minimal context. ctx.user.imperal_id is "__webhook__" — a system identity, not a real user.

Available in webhook context:

Not available in webhook context:

ctx.tenant, ctx.ai, ctx.billing, ctx.notify, ctx.storage, ctx.skeleton, ctx.progress

Accessing user data

Webhooks do not arrive with a user identity — the external caller does not know which platform user the event relates to. You must resolve the user yourself, using a mapping you stored when the user connected the external service.

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

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


@ext.webhook("payment", method="POST", secret_header="X-Pay-Sig")
async def handle_payment(ctx, headers: dict, body: str, query_params: dict) -> dict:
    secret = await ctx.config.require("pay_webhook_secret")
    received = headers.get("x-pay-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)
    external_customer_id = payload.get("customer_id", "")

    # Resolve external ID → internal imperal_id using stored mapping
    mapping_doc = await ctx.store.get("customer_mappings", external_customer_id)
    if not mapping_doc:
        await ctx.log(f"payment: no mapping for customer {external_customer_id}", level="warning")
        return {"status": "ok"}  # Acknowledge anyway — don't leak internal state

    imperal_id = mapping_doc.data.get("imperal_id", "")
    user_ctx = ctx.as_user(imperal_id)

    # Now operate in the user's partition
    await user_ctx.store.create("payment_events", payload)
    await user_ctx.notify("A payment was processed on your account.")
    return {"status": "ok"}

Security model

Signature verification is mandatory

The webhook endpoint is publicly reachable. Any caller who knows the URL can send a request. Without signature verification, an attacker can:

• Trigger arbitrary actions in your extension with fabricated payloads.
• Flood your endpoint with fake events, consuming resources.
• Escalate privileges by crafting payloads that impersonate legitimate events.

Every production webhook handler must verify the request signature before processing the payload.

The SDK does not verify for you (audit S4)

Constant-time comparison is required

Use hmac.compare_digest rather than == for all signature comparisons. The == operator short-circuits — it returns early when strings diverge. An attacker can exploit the resulting timing difference to guess the signature byte by byte. hmac.compare_digest takes the same time regardless of where the strings differ.

import hmac

received = "abc123"
expected = "abc123"

# WRONG — timing-vulnerable
if received == expected:
    pass

# CORRECT — constant-time
if hmac.compare_digest(received, expected):
    pass

Store secrets in admin config, not source code

Never hardcode the shared secret or HMAC key in your extension code. Use ctx.config.require("webhook_secret") to retrieve it from the admin-configured extension settings. This keeps the secret out of version control and allows rotation without a redeploy.

Replay protection

If the external provider includes a timestamp in the signed payload (Stripe does with t=<timestamp> in its header), check that the timestamp is within an acceptable window — typically 5 minutes — before accepting the request. This prevents an attacker who captured a valid request from replaying it later.

import time


def check_replay(parsed_timestamp: str) -> dict | None:
    """Return an error dict if the timestamp is outside the tolerance window."""
    _TOLERANCE_SECONDS = 300  # 5 minutes
    event_timestamp = int(parsed_timestamp)
    if abs(int(time.time()) - event_timestamp) > _TOLERANCE_SECONDS:
        return {"status_code": 400, "error": "timestamp out of tolerance"}
    return None

Idempotency

External webhook systems retry failed deliveries. A 500 response, a network timeout, or a slow handler that does not respond within the provider's timeout window will all trigger a retry. If your handler is not idempotent, retries produce duplicate records.

The idempotency key pattern

Most providers include a stable event ID in the payload. Use it as the document key in your store. 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(
    "event-store",
    version="1.0.0",
    display_name="Event Store",
    description="Event Store — durably records external events for audit.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.webhook("inbound", method="POST", secret_header="X-Event-Sig")
async def handle_inbound(ctx, headers: dict, body: str, query_params: dict) -> dict:
    secret = await ctx.config.require("event_webhook_secret")
    received = headers.get("x-event-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("id", "")
    if not event_id:
        return {"status_code": 400, "error": "missing event id"}

    # Idempotency: if already stored, acknowledge without re-processing
    existing = await ctx.store.get("events", event_id)
    if existing:
        await ctx.log(f"inbound: duplicate event {event_id} — skipped")
        return {"status": "ok"}

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

Respond quickly, process asynchronously

Some providers enforce a strict response timeout (Stripe: 30 seconds; GitHub: 10 seconds). If your processing logic is slow — calling external APIs, writing many records, fanning out across users — respond with {"status": "accepted"} immediately and enqueue the work separately.

The simplest approach: write the raw payload to the store, respond immediately, and let a scheduled task pick up and process queued events.

Failure modes

Provider does not receive a response

If your handler raises an unhandled exception, the platform returns a 500 to the caller. The provider may retry. Wrap your handler body in a try/except and return a 200 even for non-critical processing errors — an unprocessable event should not cause repeated retries.

Malformed payload

Never assume the body is valid JSON. Call json.loads(body) inside a try/except and return 400 if parsing fails. Log the failure for debugging.

import json
import logging
from imperal_sdk import Extension

log = logging.getLogger(__name__)
ext = Extension(
    "parser-example",
    display_name="Parser Example",
    description="Parser Example — demonstrates safe JSON parsing in webhooks.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.webhook("data", method="POST")
async def safe_parse(ctx, headers: dict, body: str, query_params: dict) -> dict:
    try:
        payload = json.loads(body)
    except (json.JSONDecodeError, ValueError) as exc:
        await ctx.log(f"data webhook: invalid JSON — {exc}", level="warning")
        return {"status_code": 400, "error": "invalid JSON"}
    event_type = payload.get("type", "unknown")
    await ctx.log(f"data webhook: received type={event_type}")
    return {"status": "ok"}

Missing or empty secret_header value

Always check that the signature header is present before attempting verification. An empty or missing header should return 401 immediately.

Federal invariants

Every write action performed inside a webhook handler is subject to the same audit requirements as any other extension write:

• Use ctx.store.create / ctx.store.update / ctx.store.delete for all store mutations — direct database access bypasses the audit log.
• If the webhook triggers a consequential action (crediting a user, deleting records), log it with ctx.log(msg, level="info") so it appears in the extension dashboard.
• Do not perform write actions for unauthenticated webhooks — verify the signature first, every time.

When to use webhooks

Use @ext.webhook when:

• An external system sends push notifications when something happens (payment completed, CI job finished, message received).
• You need real-time reaction to external events — polling via @ext.schedule would introduce latency.
• The trigger originates outside the platform (a third-party SaaS, a custom internal service, a hardware event source).

Do not use @ext.webhook when:

• The event originates inside the platform — use @ext.on_event instead.
• You need to poll an external system for changes — use @ext.schedule.
• A user in a chat session initiates the action — use @chat.function.
• You need to push data to a user proactively on a timer — use @ext.schedule with fan-out and ctx.notify.

Cross-references