Provider Plugins

Plugins can register custom providers for Sapphire's core inference systems: TTS, STT, Embedding, and LLM. When a provider plugin is enabled, it appears in the system's settings dropdown alongside core providers.

Supported Systems

Quick Start

1. Create the provider class

Subclass the appropriate base class:

# user/plugins/my-tts/provider.py
from core.tts.providers.base import BaseTTSProvider

class MyTTSProvider(BaseTTSProvider):
    audio_content_type = 'audio/ogg'
    SPEED_MIN = 0.5
    SPEED_MAX = 2.0

    def generate(self, text, voice, speed=1.0, **kwargs):
        # Generate audio bytes from text
        # Return bytes or None on failure
        ...

    def is_available(self):
        # Return True if provider is ready
        return True

2. Declare in manifest

{
  "name": "my-tts",
  "version": "1.0.0",
  "description": "My custom TTS provider",
  "capabilities": {
    "providers": {
      "tts": {
        "key": "my_tts",
        "display_name": "My TTS",
        "entry": "provider.py",
        "class_name": "MyTTSProvider",
        "requires_api_key": true,
        "api_key_env": "MY_TTS_API_KEY"
      }
    },
    "settings": [
      {
        "key": "api_key",
        "type": "password",
        "label": "API Key",
        "default": ""
      }
    ]
  }
}

3. Enable the plugin

Settings > Plugins > toggle on. The provider appears in the TTS dropdown. Settings declared in the manifest render inline on the TTS settings page when your provider is selected.

Manifest Reference

capabilities.providers

Each key is a system name (tts, stt, embedding, llm):

"providers": {
  "tts": {
    "key": "my_provider",
    "display_name": "My Provider",
    "entry": "provider.py",
    "class_name": "MyProviderClass",
    "requires_api_key": true,
    "api_key_env": "MY_PROVIDER_API_KEY"
  }
}

Extra fields are passed through to the registry as metadata and available via registry.get_entry(key).

Privacy gate: in a private chat (the 👁 eyeball), Sapphire refuses to transcribe through a non-local STT provider or speak through a non-local TTS provider — the user's voice / response text would leave the machine. An honest is_local: true on a genuinely local provider is what keeps it usable in private chats; a missing flag means "cloud" by design.

Multi-system plugins

A single plugin can register providers for multiple systems:

"providers": {
  "tts": {
    "key": "my_router",
    "display_name": "My Router",
    "entry": "tts_provider.py",
    "class_name": "MyRouterTTSProvider"
  },
  "stt": {
    "key": "my_router",
    "display_name": "My Router",
    "entry": "stt_provider.py",
    "class_name": "MyRouterSTTProvider"
  }
}

Base Classes

TTS — BaseTTSProvider

from core.tts.providers.base import BaseTTSProvider

class MyProvider(BaseTTSProvider):
    audio_content_type = 'audio/ogg'  # or 'audio/wav', 'audio/mp3'
    SPEED_MIN = 0.5
    SPEED_MAX = 2.0

    def generate(self, text: str, voice: str, speed: float, **kwargs) -> bytes | None:
        """Generate audio bytes. Return None on failure."""
        ...

    def is_available(self) -> bool:
        """Check if provider is ready (API key set, server reachable, etc.)."""
        ...

Optional methods:

• list_voices() — return list of {"voice_id": str, "name": str} dicts for the voice picker

Streaming — supports_streaming + generate_stream(). If your backend can produce audio incrementally, declare the flag and override the method:

class MyProvider(BaseTTSProvider):
    supports_streaming = True

    def generate_stream(self, text: str, voice: str, speed: float, **kwargs):
        """Yield self-contained, independently decodable audio blobs
        (e.g. one complete OGG per chunk), in play order."""
        for piece in self._backend_stream(text, voice, speed):
            yield piece

The base class ships a fallback generate_stream() that just calls generate() once and yields the whole blob — so a provider without the flag still works when the user enables streaming TTS, it silently degrades to one-shot synthesis (no audio until the full clip is ready). Set supports_streaming = True only when you actually override the method: the TTS client reads the flag to decide whether to expect chunked semantics.

Pitch — supports_pitch. Declare supports_pitch = True when your backend applies pitch at synthesis time; the client then passes pitch= as a kwarg to generate() / generate_stream() and expects the shift already applied. Without the flag, a user pitch setting falls back to the legacy client-side decode→resample→re-encode path (slower, lossier). It's strictly one or the other — an honest flag is what prevents a double shift.

STT — BaseSTTProvider

from core.stt.providers.base import BaseSTTProvider

class MyProvider(BaseSTTProvider):
    def _transcribe_impl(self, audio_path: str) -> str | None:
        """Transcribe an audio file. Return raw text or None.

        Do NOT apply the Whisper hallucination filter here — the base
        class's concrete `transcribe_file()` wraps your impl and filters
        the result automatically (since 2.6.4). That way wakeword,
        browser STT, and any future continuous-listen consumer all get
        filtered output uniformly without each having to remember.
        """
        ...

    def is_available(self) -> bool:
        ...

Important: providers implement _transcribe_impl (abstract). The base class's concrete transcribe_file() is what every consumer of the STT system calls — it invokes _transcribe_impl, runs the result through is_whisper_hallucination() (core/stt/hallucination.py), and returns None for known canned-phrase hallucinations like "Thank you", "[music]", "Thanks for watching" that Whisper produces on silence/noise. Your provider should NOT also filter — that would double-filter and possibly hide useful text.

Embedding — BaseEmbeddingProvider

Embedding providers are the most contract-sensitive of the four systems — every vector your provider writes is stamped with your provider's identity, so changing identity later invalidates the data. Read this section carefully.

The registry enforces a duck-typed contract at register time (_validate_plugin_provider_class): subclassing BaseEmbeddingProvider is optional. Required members are embed(), available, and PROVIDER_ID. Subclass if you like the structure; otherwise just implement the contract.

import numpy as np

class MyProvider:  # or: class MyProvider(BaseEmbeddingProvider)
    # REQUIRED — stable identifier stamped on every stored vector. Read-path
    # filters by this; changing it invalidates all prior writes until the user
    # runs a re-embed. Don't rename casually. Include the model version so
    # upgrading the model is a clean swap, not a silent semantic shift.
    PROVIDER_ID = 'my-plugin:my-model-v1'

    # Advertised dimension — the actual dim stamped on write is derived from
    # the vector returned by embed(). This constant is used for contract
    # checks at register time.
    DIMENSION = 384

    @property
    def provider_id(self) -> str:
        return self.PROVIDER_ID

    @property
    def dimension(self) -> int:
        return self.DIMENSION

    @property
    def available(self) -> bool:
        """True once the model is loaded and ready. Called frequently — make
        this cheap (cache the load state)."""
        ...

    def embed(self, texts: list, prefix: str = 'search_document'):
        """Embed a list of strings.

        Returns a numpy array of shape (N, DIMENSION), dtype=float32, with
        every row L2-normalized (unit length). Return None on failure.

        The `prefix` arg is a task hint (e.g. 'search_document', 'search_query')
        used by some models (Nomic). If your model doesn't use prefixes, ignore
        it.
        """
        ...

Canary contract — at register-time, Sapphire runs _canary_embed() to verify your provider is actually usable. It embeds a short test string and checks:

1. embed() returns a numpy-convertible array of shape (1, D) with D > 0.
2. Dtype is float32 (call .astype(np.float32) on your output).
3. All values are finite (no NaN/Inf).
4. L2 norm of the vector is in [0.90, 1.10] (the unit-vector drift band). Outside that: hard fail, provider falls back to NullEmbedder. In the drift zone [0.90, 0.95] or [1.05, 1.10]: accepted, but a warning is logged pointing at your normalization — tighten it before it drifts further.

A provider that fails the canary is disabled at the registry level and Sapphire boots with NullEmbedder (vector search off, FTS still works). The failure reason is logged loudly.

Provenance & re-embed flow — every stored vector row carries (embedding, embedding_provider, embedding_dim). When a user switches providers:

1. GET /api/embedding/integrity reports how many rows are stamped with what provider/dim. The Settings UI warns before the swap.
2. PUT /api/settings/batch with EMBEDDING_PROVIDER + confirm_embedding_swap: true → swap fires.
3. Old vectors are still on disk but invisible to vector search (filtered out by embedding_provider = ? in SELECTs). FTS text search still works on all rows.
4. POST /api/embedding/reembed walks every orphaned row, regenerates embeddings under the new active provider, re-stamps the row with new (provider, dim).

Your plugin doesn't have to do anything special — this all works as long as your PROVIDER_ID is stable and your embed() contract is honored.

Reference implementations: the built-in embedding providers in core/embeddings/__init__.py (LocalEmbedder, RemoteEmbedder) show the full pattern: lazy load, CPU/remote inference, L2-normalized, canary-clean output. For a plugin-shipped provider, the manifest + class contract above is complete — pair a capabilities.providers.embedding entry with a class that honors the canary contract.

Gotchas:

• The registry calls your class's __init__() with no arguments. Lazy-load the model inside available or inside embed(), not in __init__ — otherwise boot stalls on every restart.
• stamp_embedding(vec, embedder) requires passing your embedder instance explicitly. Don't use the no-arg form from plugin code; the default path reads the active singleton, which may be a DIFFERENT provider by the time you get to the call (race during swap). Hold your reference.
• If your provider talks to a remote API, handle is_available correctly — returning False when unconfigured is a legal state (canary passes with "provider reports unavailable, skipping canary"). Returning True while the API is down causes the canary to fail loudly, which is also fine.

LLM — BaseProvider

LLM providers are more complex. See the existing providers in core/chat/llm_providers/ for reference.

Every instance is built per turn by the ONE resolver, which stamps self.conversation (the chat or task name, or None). self.request_headers() returns the user's extra_headers config with {session} (a salted per-conversation hash from self.session_id()) and {version} filled — pass it as extra_headers= on your SDK calls, or splat self._hdr_kwargs(); self.fill(obj) does the same substitution for a body dict. Nothing configured → {}, send nothing extra.

from core.chat.llm_providers.base import BaseProvider, LLMResponse, ToolCall

class MyProvider(BaseProvider):
    def health_check(self) -> bool: ...
    def chat_completion(self, messages, tools=None, generation_params=None) -> LLMResponse: ...
    def chat_completion_stream(self, messages, tools=None, generation_params=None): ...
    def format_tool_result(self, tool_call_id, function_name, result) -> dict: ...

Settings Integration

Provider plugins that declare capabilities.settings in their manifest get their settings rendered inline on the system settings page (TTS, STT, etc.) when selected. Set "settingsUI": null in the manifest to avoid a separate plugin settings page — the system page is the single config location.

{
  "settingsUI": null,
  "capabilities": {
    "providers": { ... },
    "settings": [
      {"key": "api_key", "type": "password", "label": "API Key", "default": ""},
      {"key": "model", "type": "select", "label": "Model", "default": "fast", "options": [
        {"value": "fast", "label": "Fast"},
        {"value": "quality", "label": "Quality"}
      ]}
    ]
  }
}

Reading Plugin Settings from Provider Code

def _get_api_key(self):
    try:
        from core.plugin_loader import plugin_loader
        ps = plugin_loader.get_plugin_settings('my-tts')
        if ps and ps.get('api_key', '').strip():
            return ps['api_key'].strip()
    except Exception:
        pass
    return ''

Reacting to Settings Changes (on_settings_saved)

A provider may define an optional on_settings_saved method to react when its own plugin's settings are saved - e.g. pre-download a newly selected model so the first use isn't stalled. Wired for the active TTS / STT / embedder providers.

def on_settings_saved(self, plugin_name: str, settings: dict):
    """Called by core AFTER a plugin's settings are persisted."""
    if plugin_name != "my-tts":          # you're notified for EVERY plugin save - filter
        return
    name = (settings or {}).get("model")
    if name and not self._model_path(name).exists():
        import threading
        threading.Thread(target=self._warm_download, args=(name,), daemon=True).start()

Contract:

• Opt-in - only providers that define the method are called; everyone else is a no-op.
• Filter by plugin_name - you're notified for any plugin's settings save, not just yours.
• Return immediately. It runs on the settings-save request path. Do heavy work (downloads, model loads) in a background threading.Thread(..., daemon=True), never inline, or you stall the save response.
• Runs after the atomic save, fully isolated - an exception you raise is caught + logged and can never lose the just-saved settings.
• settings is a copy - mutating it has no effect on the persisted file or the response.

Notifications and progress (core.notify)

The notification center (System > Notifications) is where a plugin tells the person something happened: a row they can click, a toast if the level earns one, a progress bar while work runs. Rows persist (user/notifications.json), reach every open tab, and never carry chat content.

from core.notify import notify, flash, progress, tqdm_for

notify("Voice download failed: Amy", level="error", body=str(e), source="my-tts",
       go={"view": "settings", "tab": "plugins"}, key="my-tts:amy")   # key: re-fire ONE row, not a pile

flash("Cache cleared", level="success")        # a toast with no row: click feedback

with progress("Downloading voice Amy", source="my-tts", key="my-tts:amy",
              go={"view": "settings", "tab": "plugins"}, done_title="Voice Amy ready",
              failed_title="Voice download failed: Amy", lazy=True) as p:
    p.update(done_bytes, total_bytes)          # or p.set(percent) / p.text("step 2 of 3")
    # leaves done; an exception leaves failed with its words and is re-raised

• Levels: info is the log (never toasts, never counts), success toasts, warning counts yellow in the chat sidebar, error and critical count red, critical sticks until clicked. The person's toast threshold applies to rows; a flash always shows.
• go is where a click lands: {"view", "tab"?, "key"?} (tab is your plugin's name when it has a settings tab, else "plugins"; key scrolls to a data-key row) or {"hash": "#store/plugins/x"}.
• lazy=True makes no row until the first tick, so work that finds everything cached stays silent. tqdm_for(p) is a tqdm class bound to the handle, for libraries that take tqdm_class= (huggingface_hub) or read a module-level tqdm (openwakeword).
• A row never names a private chat. In a private context pass private=True: the row becomes a flash. Never put message content in a row.
• PLUGIN_NOTICE still works and is routed here: info/success flash, warning/error become rows. New code calls core.notify directly.

Lifecycle

1. Plugin loads — plugin_loader.scan() reads capabilities.providers from manifest
2. Provider registered — class is loaded via exec(), registered with the system's registry
3. User selects — dropdown in settings page, system calls switch_*_provider(key)
4. Provider created — registry creates an instance of your class
5. Plugin unloads — provider unregistered from registry, system falls back to null or previous provider

Boot ordering

If a provider plugin is the configured provider (e.g., TTS_PROVIDER=elevenlabs), the system may boot before the plugin loads. Sapphire handles this automatically — after all plugins load, it re-checks and activates any plugin provider that was configured but unavailable at boot.

Examples

• core/plugins/elevenlabs/ — TTS provider with API key, model selection, voice picker
• core/plugins/piper/ — local TTS provider with on_settings_saved + a core.notify progress row for voice downloads
• core/embeddings/__init__.py — the built-in embedding providers (LocalEmbedder, RemoteEmbedder, etc.); study these for the swap + re-embed + provenance path.

Reference for AI

PROVIDER PLUGINS (TTS / STT / embedding / LLM):

• Manifest: capabilities.providers.{tts|stt|embedding|llm} = {key (unique id), display_name, entry (default provider.py), class_name, requires_api_key?, api_key_env?, is_local?}. is_local: true ONLY for fully on-device inference; absent = treated as cloud => private chats refuse the STT/TTS provider (fail-closed privacy gate). Extra fields pass through as registry metadata.
• TTS: subclass core.tts.providers.base.BaseTTSProvider. Required: generate(text, voice, speed, **kwargs) -> bytes|None, is_available() -> bool. Class attrs: audio_content_type (default 'audio/ogg'), SPEED_MIN/SPEED_MAX (base defaults 0.5/2.5), supports_streaming (default False), supports_pitch (default False). Optional: list_voices() -> [{"voice_id", "name"}]. generate_stream(text, voice, speed, **kwargs) -> Iterator[bytes] must yield self-contained decodable blobs in play order; base default wraps generate() one-shot — set supports_streaming=True only with a real override, else streaming TTS silently degrades to one-shot. supports_pitch=True => client passes pitch= kwarg and provider shifts at synthesis; False => client-side resample fallback; never both.
• STT: subclass core.stt.providers.base.BaseSTTProvider; implement _transcribe_impl(audio_path) -> str|None (NOT transcribe_file — the base wraps it and applies the Whisper hallucination filter exactly once) + is_available().
• Embedding: duck-typed contract, subclassing optional. Required members: PROVIDER_ID (stable string stamped on every stored vector — changing it orphans data until re-embed; include model version), DIMENSION, available (cheap property), embed(texts, prefix='search_document') -> np.float32 array (N, D), rows L2-normalized, or None. Register-time canary: shape (1, D), float32, finite, L2 norm within [0.90, 1.10] (warning band outside [0.95, 1.05]); failure disables the provider and boots NullEmbedder. __init__() is called with NO args — lazy-load models. Re-embed flow: GET /api/embedding/integrity, PUT /api/settings/batch with EMBEDDING_PROVIDER + confirm_embedding_swap, POST /api/embedding/reembed. Use stamp_embedding(vec, embedder) with an explicit instance.
• LLM: subclass core.chat.llm_providers.base.BaseProvider: health_check() -> bool, chat_completion(messages, tools=None, generation_params=None) -> LLMResponse, chat_completion_stream(...), format_tool_result(tool_call_id, function_name, result) -> dict.
• Settings: capabilities.settings renders inline on the system settings page when the provider is selected; "settingsUI": null suppresses the separate plugin settings page. Read via plugin_loader.get_plugin_settings(name).
• Optional on_settings_saved(plugin_name, settings) on the active TTS/STT/embedder provider — fires after ANY plugin's settings save (filter by plugin_name), runs on the save path (do heavy work in a daemon thread), fully isolated, settings is a copy.
• Notifications: from core.notify import notify, flash, progress, tqdm_for. notify(title, level=info|success|warning|error|critical, body='', source='', go={view, tab?, key?}|{hash}, key=dedupe) -> row id (persists, every tab, toast by level); flash(msg, level) = toast only; with progress(title, source, go, key, done_title, failed_title, lazy) as p: p.update(done, total) | p.set(percent) | p.text(s); tqdm_for(p) for tqdm_class=. Never chat content in a row; private=True turns a row into a flash. Legacy publish(Events.PLUGIN_NOTICE, {plugin, message, severity}) still works (routed: warning/error -> rows, else flash).
• Boot ordering: a configured provider from a not-yet-loaded plugin is re-checked and activated after all plugins load — no plugin code needed.