Handle user API keys

Pattern for accepting per-user API keys / OAuth tokens / webhook signing secrets via @ext.secret and ctx.secrets

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
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")

@chat.function(
    description="Search the user's Spotify library for tracks matching a query.",
    action_type="read",
)
async def search_my_library(ctx, params: SearchParams):
    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},
        )
    return ActionResult.success(
        {"tracks": r.json().get("tracks", {}).get("items", [])},
        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 HTTP round-trip to auth-gateway. That's intentional — it satisfies I-SECRETS-HANDLER-SCOPE-MEMORY.

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={"app_id": ctx.app_id, "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 refresh tokens — write_mode='extension'

If your extension takes the user through an OAuth flow and gets back a refresh token from the provider's callback, declare the secret with write_mode="extension" so YOUR code (not Panel UI) writes the value.

OAuth callback URL — use ctx.webhook_url(), don't hardcode (v4.2.7+):

@chat.function(
    description="Start Spotify OAuth — returns the URL the user must visit.",
    action_type="read",
)
async def start_spotify_auth(ctx):
    # Build the redirect URI from web-kernel-authoritative app_id — never hardcode
    # a string like "https://panel.imperal.io/v1/ext/spotify/webhook/callback".
    # The helper auto-tracks the correct folder/manifest name.
    redirect_uri = ctx.webhook_url("/callback")
    auth_url = (
        "https://accounts.spotify.com/authorize"
        f"?client_id={await ctx.secrets.get('spotify_client_id')}"
        f"&redirect_uri={redirect_uri}"
        f"&response_type=code"
        f"&scope=user-read-private"
    )
    return ActionResult.success({"auth_url": auth_url})

@ext.webhook("/callback", method="GET")
async def oauth_callback(ctx, headers, body, query_params):
    code = query_params.get("code")
    # ...exchange code for refresh_token...
    await ctx.secrets.set("spotify_refresh_token", refresh_token)
    return {"status": "ok"}

The end-user gets the canonical callback URL automatically rendered in Panel → Settings → Secrets with a Copy button — they paste it into the provider's developer console (Spotify Dashboard → app → Settings → Redirect URIs) before they paste their client_id/secret.

ext.secret(
    name="spotify_refresh_token",
    description="OAuth refresh token — written by extension after you authorize.",
    write_mode="extension",
    max_bytes=4096,
    rotation_hint_days=180,
)(lambda: None)

@chat.function(
    description="Complete Spotify OAuth authorization.",
    action_type="write",
)
async def authorize_spotify(ctx, params):
    # ... your OAuth dance ...
    tokens = await _exchange_code_for_tokens(params.code)
    await ctx.secrets.set("spotify_refresh_token", tokens["refresh_token"])
    await ctx.secrets.set("spotify_access_token", tokens["access_token"])
    return ActionResult.success({"authorized": True}, summary="Spotify authorized")

Panel UI renders write_mode="extension" secrets as read-only with the helper text "the extension will write this after you authorize" — the user can't paste a value there even if they wanted to.

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, request):
    signing_secret = await ctx.secrets.get("stripe_webhook_signing_secret")
    if signing_secret is None:
        return {"error": "webhook not configured"}, 503

    # Verify the signature with hmac.compare_digest (timing-safe)
    import hmac, hashlib
    sig_header = request.headers.get("Stripe-Signature", "")
    expected = hmac.new(signing_secret.encode(), request.body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig_header, f"v1={expected}"):
        return {"error": "invalid signature"}, 403

    # ...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 — that
    # would be a federal violation (I-SECRETS-NEVER-LOGGED).

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 — module-level cache violates I-SECRETS-HANDLER-SCOPE-MEMORY
_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.