Handle user API keys

Store and use per-user API keys in a Webbee extension with ext.secret — credentials are encrypted at rest, never logged, and scoped to each individual user.

Practical recipe for extensions that need credentials from the user — Spotify API keys, OAuth refresh tokens, webhook signing secrets, BYOLLM provider keys. Uses @ext.secret (SDK v4.2.2+) so plaintext is Vault-encrypted at rest, never logged, and scoped per-user.

Scenario

You're writing a Spotify extension. To call api.spotify.com on a user's behalf, you need their personal API token. You want:

User pastes the token once in Panel UI
Your handler reads it whenever it runs
Admin reading the database sees only ciphertext
If the user hasn't pasted it yet, the web-kernel blocks dispatch and tells them where to go

Declare the secret

In app.py, alongside your other extension decorators:

from imperal_sdk import Extension, ChatExtension, ActionResult, sdl
from pydantic import BaseModel, Field
import httpx

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

ext.secret(
    name="spotify_api_key",
    description="Your Spotify API key — get it at developer.spotify.com → your app → Show Client Secret.",
    required=True,
    write_mode="user",
    max_bytes=200,
    rotation_hint_days=90,
)(lambda: None)

@ext.secret is a no-op decorator — it registers a SecretSpec on the Extension and returns identity. The trailing (lambda: None) is a syntactic anchor; you don't need a real target.

Read the secret in a handler

class SearchParams(BaseModel):
    query: str = Field(description="What to search for in Spotify")

# A track is an sdl.Entity: core id/title/kind, plus the AudioTrack facet
# (media.audio_codec / media.bitrate_kbps) for encoding properties.
class Track(sdl.Entity, sdl.AudioTrack):
    pass

class TrackList(sdl.EntityList[Track]):
    pass

@chat.function(
    "search_my_library",
    description="Search the user's Spotify library for tracks matching a query.",
    action_type="read",
    data_model=TrackList,
)
async def search_my_library(ctx, params: SearchParams) -> ActionResult:
    api_key = await ctx.secrets.get("spotify_api_key")
    # If required=True and the user hadn't set it, the web-kernel would have
    # blocked dispatch before we got here. Inside the handler we can trust
    # the value is present. Still, return a clean error if Vault is down:
    if api_key is None:
        return ActionResult.error(
            "Spotify key isn't set — please configure it in extension settings.",
            retryable=False,
        )

    async with httpx.AsyncClient() as c:
        r = await c.get(
            "https://api.spotify.com/v1/search",
            headers={"Authorization": f"Bearer {api_key}"},
            params={"q": params.query, "type": "track", "limit": 5},
        )
    rows = r.json().get("tracks", {}).get("items", [])
    tracks = [Track(id=t["id"], title=t["name"]) for t in rows]
    return ActionResult.success(
        TrackList(items=tracks, total=len(tracks), has_more=False),
        summary=f"Found tracks for '{params.query}'",
    )

Federal rule: never store api_key outside this handler's scope. No instance attribute, no module dict, no @lru_cache. Every call to ctx.secrets.get() is a fresh fetch of the live value. That's intentional — the platform requires secret values to stay within the handler call that reads them and never be retained in memory between calls.

What the user sees

After your extension publishes via Dev Portal:

User opens Imperal Panel → Spotify → Settings → Secrets
They see one row: ● spotify_api_key ○ Not set with description text
They click Set value, paste their token (ui.Password field — input type=password, no echo, autocomplete suppressed), click Save
Value flows: Panel UI → platform secrets endpoint → KMS encrypt → ciphertext at rest (plaintext never persisted)
User goes back to chat: "find my top tracks"
Web-kernel sees required=True Spotify secret is now set → dispatches your handler
Your handler reads plaintext, calls Spotify API, returns results

The auto-injected secrets panel (SDK v4.2.4+) appears in the right sidebar without you writing any panel code. End users land there directly via the Secrets tab in your extension's App Details page.

Custom Panel UI — use `ui.Password`, not `ui.Input`

If you build your own credential-entry surface (instead of relying on the auto-injected panel), use ui.Password (SDK v4.2.6+) so the browser masks the value while the user types and suppresses autofill:

from imperal_sdk import ui

@ext.panel("setup", slot="center", title="Connect Spotify", icon="🔌")
async def setup_panel(ctx, **kwargs):
    return ui.Stack(children=[
        ui.Text("Paste your Spotify API key below."),
        ui.Form(
            action="save_app_secret",
            submit_label="Save",
            defaults={"name": "spotify_api_key"},
            children=[
                ui.Password(  # ← NOT ui.Input — federal EXT-SECRETS-V1 surface
                    placeholder="sk-proj-...",
                    param_name="value",
                ),
            ],
        ),
    ])

Federal note: type="password" is a defence against shoulder-surfing, not a security control. The plaintext still travels in the POST body to auth-gateway, which is the only correctness boundary. See ui.Password.

OAuth account-connect — use the unified flow, don't hand-roll it

If your extension connects to a provider through OAuth (Google, Microsoft, Yahoo, Spotify, …), do not build the authorize URL, callback route, and token-exchange by hand. The platform ships a unified OAuth-connect flow that does all of it for you — you declare the provider, return one authorize URL, and the connected account + tokens land in the user's secrets automatically.

Earlier recipes showed a hand-rolled @ext.webhook("/callback") + manual _exchange_code_for_tokens pattern. That is deprecated. Use ext.oauth(...) + ctx.oauth_authorize_url(...) instead — it is safer (HMAC-signed state, canonical https redirect URI), far less code, and identical across every provider.

1. Declare the provider. Your OAuth client credentials live as app-scope secrets — set once by you in the Developer Portal, shared by all users (never pasted per-user):

app.py

ext.oauth(
    "google",                       # provider key from the platform registry
    collection="mail_accounts",     # where connected accounts are persisted
    scopes=["https://www.googleapis.com/auth/gmail.readonly"],
)

# Client credentials — app-scope, written by YOU via the Dev Portal Secrets tab.
# The platform reads these during the callback token-exchange.
ext.secret(
    name="google_client_id",
    description="OAuth client ID from Google Cloud Console.",
    scope="app",
)(lambda: None)
ext.secret(
    name="google_client_secret",
    description="OAuth client secret from Google Cloud Console.",
    scope="app",
)(lambda: None)

2. Start the flow — return the signed authorize URL from a chat function:

handlers_chat.py

@chat.function(
    "connect_gmail",
    description="Connect a Gmail account — returns the URL the user must visit.",
    action_type="read",
)
async def connect_gmail(ctx):
    url = await ctx.oauth_authorize_url("google")
    return ActionResult.success({"authorize_url": url})

That is the whole extension side. The platform hosts the callback at https://panel.imperal.io/v1/ext/<app>/oauth/<provider>/callback, validates the signed state, exchanges the code with the provider, fetches the account profile, and persists the connected account + tokens into your declared collection. The user sees a self-closing result window — no code in your extension renders it.

Register that callback URL in the provider's console (Google Cloud Console → Credentials → Authorized redirect URIs, Azure App Registration, Yahoo App, …) before going live.

Full reference: decorator ext.oauth reference.

Webhook signing secrets

Stripe / GitHub / Discord webhooks have a per-installation signing secret. User pastes it once from the provider dashboard, then your webhook handler verifies every incoming request:

ext.secret(
    name="stripe_webhook_signing_secret",
    description="Stripe webhook signing secret (whsec_...) — Dashboard → Developers → Webhooks → Signing secret.",
    required=False,
    write_mode="user",
    max_bytes=200,
    rotation_hint_days=90,
)(lambda: None)

@ext.webhook("/stripe-webhook", method="POST", secret_header="Stripe-Signature")
async def stripe_webhook(ctx, headers, body, query_params):
    signing_secret = await ctx.secrets.get("stripe_webhook_signing_secret")
    if signing_secret is None:
        return {"status_code": 503, "error": "webhook not configured"}

    # Verify the signature with hmac.compare_digest (timing-safe).
    # Header names arrive lowercased; body is the raw request string.
    import hmac, hashlib
    sig_header = headers.get("stripe-signature", "")
    expected = hmac.new(signing_secret.encode(), body.encode(), hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig_header, f"v1={expected}"):
        return {"status_code": 403, "error": "invalid signature"}

    # ...process the event...

Testing locally

Two options:

Env-var dev mode — fastest:

export IMPERAL_DEV_MODE=true
export IMPERAL_SECRET_SPOTIFY_API_KEY="dev-stub-token"
imperal dev

When IMPERAL_DEV_MODE=true, ctx.secrets.get("spotify_api_key") reads IMPERAL_SECRET_SPOTIFY_API_KEY from your shell instead of hitting auth-gateway. Manifest contract still enforced — undeclared names still raise.

pytest fixture — for unit tests:

import pytest
from imperal_sdk.testing import MockSecretStore

@pytest.fixture
def secrets():
    return MockSecretStore(
        {"spotify_api_key": "sk-test-12345"},
        declared={"spotify_api_key", "spotify_refresh_token"},
    )

async def test_search(ctx_factory, secrets):
    ctx = ctx_factory(secrets=secrets)
    result = await search_my_library(ctx, SearchParams(query="test"))
    assert result.ok
    # secret value is sk-test-12345; test what you call with it,
    # but don't assert on the value itself appearing in results — the
    # platform forbids secret values from ever surfacing in logs or output.

Migrating an extension with legacy `ctx.config` read sites

The single most common bug when adding @ext.secret to an existing extension is leaving a legacy ctx.config read site in place. The Save UI confirms success, the Vault row exists, but the handler still pulls from the empty ctx.config slot and reports "credentials not configured". Users blame the platform; the credential was never the problem.

# ❌ Stale legacy code that survived the migration
async def search_my_library(ctx, params):
    api_key = ctx.config.get("spotify.client_id")  # ← reads the old, plaintext store
    if not api_key:
        return ActionResult.error("Spotify credentials not configured")
    ...

When you migrate:

Add the @ext.secret(...) declaration + the new ctx.secrets.get() read site.
Grep the entire extension for the old name — client_id, api_key, token, whatever you used in ctx.config — and replace every read.
Delete the old setup panel (or refactor it to use ui.Password) — the auto-injected secrets panel covers entry.
Bump your extension version + redeploy via Dev Portal.

The V32 publish-time validator will flag legacy read sites once enforcement turns on. Until then, audit by hand — and remember it's the read site that betrays you, not the write.

What NOT to do

# ❌ NO — a module-level cache retains the secret beyond the handler call,
#         which the platform forbids (secret values must never be held in
#         memory between calls)
_API_KEY_CACHE = {}

async def search_my_library(ctx, params):
    if ctx.user.imperal_id not in _API_KEY_CACHE:
        _API_KEY_CACHE[ctx.user.imperal_id] = await ctx.secrets.get("spotify_api_key")
    api_key = _API_KEY_CACHE[ctx.user.imperal_id]
    ...

# ❌ NO — never log the value
async def search_my_library(ctx, params):
    api_key = await ctx.secrets.get("spotify_api_key")
    log.info(f"calling Spotify with key {api_key[:10]}...")  # FEDERAL VIOLATION
    ...

# ❌ NO — never return the value in a response
async def search_my_library(ctx, params):
    api_key = await ctx.secrets.get("spotify_api_key")
    return ActionResult.success({"debug_key_prefix": api_key[:5]})  # FEDERAL VIOLATION

# ❌ NO — never write user-mode secret from extension code
ext.secret(name="spotify_api_key", write_mode="user", ...)(lambda: None)

async def some_handler(ctx, params):
    await ctx.secrets.set("spotify_api_key", "abc")  # raises SecretWriteForbidden

V32 validator — what will catch bypasses at publish (coming soon)

Once V32 publish-flow enforcement turns on, the Dev Portal will AST-scan your code for credential-like field access:

# ❌ V32 ERROR — credential read without matching @ext.secret declaration
import os
sk = os.environ.get("STRIPE_KEY")

# ❌ V32 ERROR — Redis hash read of credential-like field
token = await redis.hget("user_settings", "api_token")

# ✅ V32 PASS — declared + accessed via ctx.secrets
ext.secret(name="stripe_api_key", description="...", write_mode="user")(lambda: None)
sk = await ctx.secrets.get("stripe_api_key")

# ✅ V32 WARN (not ERROR) — bypass marker captured for audit-trail
# imperal-allow-plaintext-credential: legacy bootstrap migrating in v1.5
sk = os.environ.get("STRIPE_KEY")

Use the bypass marker sparingly — it's audit-trail visible to Marketplace review.