Sparkth Plugin Guide¶

Quick guide for creating Sparkth plugins with API routes and MCP tools.

Plugin Config Definition¶

Define a config class for the plugin that must inherit from sparkth.lib.plugins:PluginConfig

# sparkth/plugins/myappplugin/config.py
from pydantic import Field
from sparkth.lib.plugins import PluginConfig


class MyAppPluginConfig(PluginConfig):
    config_field: str = Field(..., description="...")

    # define additional fields as required

LLM-Aware Plugins¶

If your plugin lets users pick an AI model to power some feature (e.g. answer synthesis, content generation), add llm_config_id and llm_model_override to the config class. These two fields work as a pair: llm_config_id points to one of the user's saved LLM configurations (provider + API key), and llm_model_override lets them swap in a different model from the same provider without creating a new configuration.

# sparkth/plugins/myappplugin/config.py
from pydantic import Field
from sparkth.lib.plugins import PluginConfig


class MyAppPluginConfig(PluginConfig):
    llm_config_id: int | None = Field(
        default=None,
        description="ID of an LLMConfig row for AI features. None disables AI.",
    )
    llm_model_override: str | None = Field(
        default=None,
        description="Override the model from the selected LLMConfig. None uses the config's model.",
    )

On its own, these are plain optional integers and strings — Pydantic only checks the types. Cross-field validation (does llm_config_id actually belong to this user? is llm_model_override a valid model for the linked provider?) requires a config adapter, described in the Plugin Config Adapters section below.

Valid llm_model_override values must be a legal model for the linked LLMConfig's provider. The single source of truth is PROVIDER_MODELS in sparkth/llm/providers.py.

LMS Plugins: Tools Without Credentials¶

An LMS tool reads the caller's saved settings itself with load_user_plugin_config, described in Reading the caller's stored settings, and raises PluginNotConfiguredError when there are none.

An LMS plugin ships a zero-argument <prefix>check_settings tool (e.g. mylms_check_settings) that calls the LMS with the stored settings. Its docstring is the tool description the model reads, so it carries the LMS's rules: call the tool to learn whether the LMS is set up, and never ask the user for credentials or accept them in the chat. The chat system prompt holds no LMS rules, so the model reads them only while that LMS's tools are bound. When there are no settings, the PluginNotConfiguredError message tells the model to send the user to the Settings page.

An LMS plugin should also ship a zero-argument <prefix>about tool (e.g. mylms_about) returning {"about": <text>} that explains the LMS's course structure, its vocabulary, and what "publish" means there, with the text kept in the plugin's assets/about.txt. LMSs use the same words for different things (Canvas "publish" makes content visible; Moodle has no publish step), so the chat LLM reads this before acting. See canvas_about, openedx_about and moodle_about.

Secret fields¶

Mark a credential with json_schema_extra=secret(), or secret(ConfigWidget.TEXTAREA) for a multi-line one. Sparkth stores every user's plugin config encrypted, and the user-plugins API never returns a secret field: it lists the stored ones in secrets_set instead. The settings form shows a stored secret as dots with a Replace action, and a save that leaves it out keeps the stored value.

Register the plugin configuration class¶

Register a Pydantic configuration class when your plugin has user-configurable settings that the system should validate and normalize, and always for LMS plugins, whose tools load it with load_user_plugin_config. Plugins with no user-facing configuration can skip this entirely.

When you do register one, contribute your config class to the CONFIG_SCHEMAS hook from your plugin's __init__, right after calling super().__init__(...). The system resolves config classes by the plugin's declared name (the value you pass to super().__init__(), see Plugin Names below), so no name string is needed at the call site.

# sparkth/plugins/myappplugin/plugin.py

from sparkth.plugins.myappplugin.config import MyAppPluginConfig
from sparkth.lib.config.hooks import CONFIG_SCHEMAS
from sparkth.lib.plugins import SparkthPlugin


class MyAppPlugin(SparkthPlugin):
    def __init__(self) -> None:
        super().__init__("my-app")
        CONFIG_SCHEMAS.add_item(self, MyAppPluginConfig)

Plugin Config Adapters (LLM-Aware Plugins)¶

If your plugin config includes llm_config_id and llm_model_override, register a config adapter so the system validates them at write time and resolves them for the frontend at read time. Without an adapter the fields are stored as plain values with no cross-field checks.

What the adapter does¶

LLMConfigAdapter (import from sparkth.lib.llm) hooks into the config pipeline via three methods — preprocess_config (validates on save), postprocess_config (resolves read-only fields for the frontend), and sync_cache (a no-op override point for cache invalidation). See the LLMConfigAdapter reference for the exact signatures and behaviour.

Adding an adapter for your plugin¶

For the default behaviour — LLM config ownership check plus model override validation — register LLMConfigAdapter itself in your plugin's __init__, alongside the config schema. There is no adapter class to write:

# sparkth/plugins/myappplugin/plugin.py
from sparkth.lib.config.hooks import CONFIG_ADAPTERS, CONFIG_SCHEMAS
from sparkth.lib.llm import LLMConfigAdapter
from sparkth.lib.plugins import SparkthPlugin

from sparkth.plugins.myappplugin.config import MyAppPluginConfig


class MyAppPlugin(SparkthPlugin):
    def __init__(self) -> None:
        super().__init__("my-app")
        CONFIG_SCHEMAS.add_item(self, MyAppPluginConfig)
        CONFIG_ADAPTERS.add_item(self, LLMConfigAdapter())

That's it. preprocess_config and postprocess_config are now wired in automatically for every POST and PUT to your plugin's config endpoint.

Subclass only when you need to change that behaviour — see Overriding adapter methods below. A subclass that only says pass adds a file and an import without changing anything, and the base class is what the config pipeline calls either way.

Overriding adapter methods¶

Override a method when the default behaviour isn't enough. Always call super() first in preprocess_config and postprocess_config so the base LLM validation still runs.

Custom preprocessing — validate an extra field before saving:

from typing import Any
from sqlmodel.ext.asyncio.session import AsyncSession
from sparkth.lib.llm import LLMConfigAdapter


class MyAppPluginConfigAdapter(LLMConfigAdapter):
    async def preprocess_config(
        self,
        *,
        session: AsyncSession,
        user_id: int,
        incoming_config: dict[str, Any],
    ) -> dict[str, Any]:
        result = await super().preprocess_config(
            session=session,
            user_id=user_id,
            incoming_config=incoming_config,
        )
        if result.get("webhook_url") and not result["webhook_url"].startswith("https://"):
            raise ValueError("webhook_url must use HTTPS.")
        return result

Custom postprocessing — attach extra derived fields to the response:

class MyAppPluginConfigAdapter(LLMConfigAdapter):
    async def postprocess_config(
        self,
        *,
        session: AsyncSession,
        user_id: int,
        stored_config: dict[str, Any],
    ) -> dict[str, Any]:
        result = await super().postprocess_config(
            session=session,
            user_id=user_id,
            stored_config=stored_config,
        )
        result["some_derived_field"] = compute_something(result)
        return result

Cache invalidation — clear a Redis key when config is updated:

from sparkth.lib.cache import get_cache_service
from sparkth.lib.settings import get_settings


class MyAppPluginConfigAdapter(LLMConfigAdapter):
    async def sync_cache(
        self,
        *,
        session: AsyncSession,
        user_id: int,
        stored_config: dict[str, Any],
    ) -> None:
        settings = get_settings()
        cache = get_cache_service(settings.REDIS_URL, settings.REDIS_KEY_TTL)
        await cache.delete(cache.make_key("myappplugin", str(user_id)))

Plugin Names¶

Every plugin declares its own name explicitly by passing it positionally to super().__init__(). Nothing is derived from the class name, and renaming the class never changes the plugin's identity:

class MyAppPlugin(SparkthPlugin):
    def __init__(self) -> None:
        super().__init__("my-app")

The name must be a kebab-case slug (lowercase letters, digits, single hyphens: PLUGIN_NAME_PATTERN in sparkth/core/plugins/constants.py); super().__init__() raises ValueError otherwise. It appears in URLs (/api/v1/<name>, /dashboard/<name>), keys the plugin's database row, and is what every hook lookup (CONFIG_SCHEMAS, CONFIG_ADAPTERS, the frontend metadata hooks) resolves by, so treat it as permanent once released.

The loader (sparkth/core/plugins/loader.py) constructs every plugin as plugin_class() with no arguments and indexes it under the declared name. A plugin that declares a name already taken by another plugin is skipped with an error in the logs.

Basic Plugin Structure¶

A plugin contributes its capabilities from its __init__: routes via register_router, MCP tools to MCP_TOOLS, a config schema to CONFIG_SCHEMAS, frontend metadata to DISPLAY_INFO / SIDEBAR_ENTRIES / FRONTEND_APPS and analytics event schemas via register_event_schema / register_event_schemas. It declares permissions (Permission.create), scope kinds (PermissionScope.create / ObjectlessPermissionScope.create) and exception→HTTP mappings (register_exception_handler) at module level in plugin.py instead: a second registration of the same name or exception type raises ValueError, and the plugin class is constructed more than once (by the loader and by the plugin's own tests). Bind each permission and scope kind to a constant.

# sparkth/plugins/myappplugin/plugin.py
from fastapi import APIRouter

from sparkth.plugins.myappplugin.config import MyAppPluginConfig
from sparkth.lib.analytics import AnalyticsEventSchema, register_event_schema
from sparkth.lib.config.hooks import CONFIG_SCHEMAS
from sparkth.lib.mcp.hooks import MCP_TOOLS, Tool
from sparkth.lib.permissions import Permission
from sparkth.lib.routes import register_router
from sparkth.lib.plugins import SparkthPlugin

# Create router outside the class
router = APIRouter()

MYAPP_PROCESS = Permission.create("myapp.process")


@router.get("/")
async def get_data():
    return {"message": "Hello from my plugin"}


# An analytics event this plugin emits. `event_type` MUST be namespaced under the
# plugin name; `version` lets the payload evolve without breaking older producers.
class MyAppDataProcessed(AnalyticsEventSchema):
    event_type = "myappplugin.data_processed"
    version = 1

    input_length: int


# Plugin class
class MyAppPlugin(SparkthPlugin):
    def __init__(self) -> None:
        super().__init__("my-app")  # the plugin declares its own name
        CONFIG_SCHEMAS.add_item(self, MyAppPluginConfig)
        register_router(self, router)
        MCP_TOOLS.add_item(self, Tool(process_data, category="utilities"))
        register_event_schema(self, MyAppDataProcessed)


async def process_data(input: str) -> str:
    """Process some input and return the result."""
    return f"Processed: {input}"

Scope kinds register the same way, through PermissionScope.create("course", parent=...) (or ObjectlessPermissionScope.create(...) for a singleton scope) — the scope classes come from sparkth.lib.permissions.scopes. See the permissions guide for how scope hierarchy and assignments work.

Analytics event schemas register through register_event_schema(self, MyEvent): define an AnalyticsEventSchema subclass (from sparkth.lib.analytics) declaring its own event_type and version, then register it from __init__. Once a plugin keeps its schemas in an analytics module of its own, register the module instead — register_event_schemas(self, analytics) picks up every AnalyticsEventSchema subclass defined in it, so adding an event is one edit in one file rather than a class plus a line in plugin.py that is easy to forget. It selects by base class, not by name: seam classes sharing the module, and schemas the module imports from elsewhere, are left alone. Each schema still goes through register_event_schema, so the guards below apply either way. The event_type must be namespaced under the plugin's name ("myappplugin.data_processed"), or registration raises EventNamespaceError at import; a second class claiming the same (event_type, version) raises DuplicateEventTypeError. Emit a registered event server-side with emit_event (also from sparkth.lib.analytics) — it opens its own analytics session so you need no session plumbing. It catches nothing: a failed analytics write propagates to the caller rather than being silently hidden. Emit from a background task so that failure surfaces in the logs without affecting the request being measured. Reach for the lower-level ingest_event when the caller already holds an analytics session. What to record, what a payload may carry, and which seam can safely emit are in the analytics events guide.

To read analytics back, call read_rollup (from sparkth.lib.analytics) with the continuous aggregate's name, the event types it counts, its count column and its grouping columns as dims — each mapped to the payload key it comes from, either one key for every event type or a per-event-type mapping. No SQL: on PostgreSQL it reads the aggregate, on SQLite it computes the same rows from raw_events, so dims must match the aggregate's definition. To also total a payload value per group, pass sums (output column → payload key); on PostgreSQL the aggregate needs a column of that output name summing the key. Take the range as Annotated[RangeQuery, Query()], gate the route with Depends(ANALYTICS_READ.require_in_global_scope()) (sparkth.lib.permissions) and read on get_analytics_session (sparkth.lib.db). The aggregate's DDL still goes in the analytics migration lineage (sparkth/migrations/analytics/). The chat plugin's analytics/reads.py is a worked example. To put a headline figure on the analytics Overview, build a SummaryFigure at module level (a plugin-prefixed key, a gettext_noop label, an async count(session, from_, to) and the view it links to) and register it from __init__ with register_summary_figure(self, figure). The Overview shows it to every viewer with analytics.read while the plugin is installed, whatever the plugin switches say. Mark the plugin's analytics read routes @platform_wide (see Who can reach plugin routes?) so they answer the same way.

Where do plugin routes get mounted?¶

register_router mounts the router at /api/v1/<plugin-name> automatically, using the plugin instance's declared name. The router above is reachable at http://localhost:7727/api/v1/my-app/.

Who can reach plugin routes?¶

Every route you register is gated by PluginAccessMiddleware. A request from a caller who has turned your plugin off in their settings — or a request to a plugin disabled system-wide — is answered with 403 before it reaches your handler, so a handler never has to check plugin access itself.

A route that serves platform-wide data, such as an analytics read, opts out with @platform_wide from sparkth.lib.routes, placed below the route decorator. The gate lets it through whatever either switch says; its own dependencies, such as a permission, still apply.

The gate identifies the caller from the request's bearer token. A request carrying no readable token passes straight through it: a per-user preference is meaningless without a user, so endpoints that are called by someone other than a logged-in user — an OAuth callback, an inbound webhook — keep working. Those endpoints are reachable regardless of anyone's plugin settings, so authenticate them the way you would any other public endpoint (a signed state parameter, a request signature) rather than relying on the gate.

That applies to the system-wide switch too: disabling a plugin turns its endpoints off for every identified caller, but its unauthenticated endpoints stay reachable, because the gate has no caller to check them against. Treat "disabled" as a per-caller answer, not as a kill switch for the plugin's HTTP surface.

Who can reach plugin MCP tools?¶

Your plugin's tools are gated on /ai/mcp by PluginToolAccessMiddleware (sparkth/mcp/access.py), the MCP counterpart to the route gate above. Disabling a plugin system-wide stops its tools being called and drops them from the advertised tool listing; tools registered directly on the server belong to no plugin and are never gated.

The gate resolves the owning plugin from a name register_plugin_tools stamps onto each tool at registration, and reads Plugin.enabled on every call — so disabling a plugin takes effect immediately, with no restart, and re-enabling one brings its tools back the same way.

One limit is worth knowing before you rely on it. The gate is system-wide only: /ai/mcp carries no authenticated caller, so there is no user whose per-user preference could be checked. A user who turns your plugin off in their settings can still call its tools over MCP; only an administrator disabling the plugin outright stops them.

MCP is not the only tool door. Chat applies both switches for the requesting user on every turn: a plugin switched off system-wide or by that user has none of its tools bound, no capability section or scope offered. Only /ai/mcp stays system-wide.

Because /ai/mcp has no per-user switch, a tool that must not run for a particular caller over MCP has to enforce that inside the tool.

Frontend Metadata¶

The backend is the single source of truth for what the frontend shows about a plugin. Declare it through three per-concern hooks from sparkth.lib.frontend.hooks, registered in __init__ like every other hook. Human-facing names are passed positionally:

from sparkth.lib.frontend.hooks import (
    DISPLAY_INFO,
    FRONTEND_APPS,
    SIDEBAR_ENTRIES,
    DisplayInfo,
    FrontendApp,
    SidebarEntry,
)


class MyAppPlugin(SparkthPlugin):
    def __init__(self) -> None:
        super().__init__("my-app")
        DISPLAY_INFO.add_item(self, DisplayInfo("My App", "What the plugin does, in one line", icon="sparkles"))
        SIDEBAR_ENTRIES.add_item(self, SidebarEntry("My App", icon="sparkles", order=5))
        FRONTEND_APPS.add_item(self, FrontendApp())

What each hook carries, which of them are optional, and how the declarations reach the frontend live in the frontend metadata hooks reference, generated from the module docstrings.

After adding or removing a FRONTEND_APPS declaration, regenerate the frontend's static route list with make frontend.build.plugins (it rewrites frontend/lib/plugins/generated.ts from the declarations); tests on both tiers fail while the committed list is stale.

Rendering plugin exceptions as HTTP responses¶

A plugin's domain exceptions stay HTTP-agnostic (plain Exception subclasses). To control how one renders as an HTTP response, map it to a status at module level in plugin.py — the mapping is wired onto the app at startup:

from sparkth.lib.exceptions.handlers import register_exception_handler


class MyAppError(Exception):
    """Raised when the plugin cannot process a request."""


# at module level in plugin.py:
register_exception_handler(MyAppError, 409)

Now any route that raises MyAppError returns 409 with {"detail": str(exc)} — no per-route try/except. A mapping on a base exception also catches its subclasses. An optional third argument adds response headers, such as the CORS headers a cross-origin caller needs to read the error.

Translations (Optional)¶

User-facing strings are marked with _() / lazy_gettext() from sparkth.lib.i18n. A plugin owns its catalogs: core extraction ignores sparkth/plugins/, so a plugin that marks strings ships the .po files under its own locale/ directory and registers that directory on the LOCALE_DIRS hook from its __init__, like every other hook:

from sparkth.lib.i18n import LOCALE_DIRS

LOCALE_DIRS.add_item(Path(__file__).parent / "locale")

For an in-tree plugin, create the locale/ directory and the i18n.* Make targets pick it up automatically (make i18n.extract writes its messages.pot, make i18n.init -- <lang> creates its per-language catalogs, and so on); the containment is enforced by tests/core/i18n/test_catalog_containment.py. See the translations guide.

Register in core/config.py¶

PLUGINS = [
    "sparkth.plugins.canvas.plugin:CanvasPlugin",
    "sparkth.plugins.openedx.plugin:OpenEdxPlugin",
    "sparkth.plugins.moodle.plugin:MoodlePlugin",
    # add your plugin here
]

Format: "path.to.module:ClassName"

Adding Database Models (Optional)¶

Define your models as table=True SQLModel classes and import them at the top of your plugin module. Importing the module registers the tables in SQLModel.metadata, which is all Alembic autogenerate needs — there is no separate registration step.

# Importing the model registers its table in SQLModel.metadata for Alembic.
from sparkth.plugins.my_app.models import MyModel  # noqa: F401


class MyAppPlugin(SparkthPlugin):
    def __init__(self) -> None:
        super().__init__("my-app")
        CONFIG_SCHEMAS.add_item(self, MyAppPluginConfig)
        register_router(self, router)

Then create migration:

alembic revision --autogenerate -m "my_plugin_add_model"
alembic upgrade head

MCP Tools¶

A tool is a plugin method registered with the MCP_TOOLS hook via a Tool (sparkth/lib/mcp/hooks.py). The tool's name is the method name, its description is the method's docstring, and its input schema is auto-generated from the signature. Register each tool in __init__; pass an optional category to group it.

class MyAppPlugin(SparkthPlugin):
    def __init__(self) -> None:
        super().__init__("my-app")
        MCP_TOOLS.add_item(self, Tool(my_tool, category="my-category"))


async def my_tool(param1: str, param2: int = 0) -> dict:
    """One-line summary the LLM sees as the tool description.

    Args:
        param1: First parameter
        param2: Optional parameter
    """
    return {"result": f"{param1}-{param2}"}

When a plugin has many tools, register them with a loop:

tools: list[tuple[Callable[..., Any], str]] = [
    (my_tool, "my-category"),
    (other_tool, "my-category"),
]
for handler, category in tools:
    MCP_TOOLS.add_item(self, Tool(handler, category=category))

Declaring your own audit events¶

A plugin that changes security-relevant state outside a tool (such as connecting a third-party account) declares its own audit events in a module of its own, under a category named after the plugin (googledrive.connected, slack.disconnected), registers them on AUDIT_EVENTS from sparkth.lib.audit.hooks. The classes register when the code that records them imports the module, so no side-effect import is needed. The audit events guide (docs/guides/audit-events.md) covers what a snapshot may carry, which transaction to record in, and how to test it.

Tool executions are audited¶

Tool wraps the handler with sparkth.lib.audit.audited_tool at construction: every execution, on every surface (the MCP server, chat, RAG), records a tool.invoked event before the handler runs and a tool.completed or tool.failed event after. Handlers must be async; the wrapper rejects a sync handler with a TypeError at construction. The write is fail-closed: if the invocation record cannot be committed, the tool call is refused with sparkth.lib.audit.exceptions.AuditCaptureError. Tool arguments are redacted before persistence (any auth payload and secret-named keys like token or password are replaced wholesale), and exception messages recorded on failure are scrubbed of secret-keyed values and length bounded, but do not put secrets or free-text PII in argument names or values that redaction cannot recognize. Handlers need no audit code of their own. What else to record in the audit trail, and how, is in the audit events guide.

Who is calling a tool¶

A handler that acts for a user calls current_user_id() from sparkth.lib.auth. It never takes a user id argument, because the model writes tool arguments and could name anyone.

The id is bound for every REST request that resolves get_current_user, including chat turns and the tools they run. On /ai/mcp nothing is bound, so the call raises NoAuthenticatedUser (sparkth.lib.exceptions.auth) and the MCP client sees a tool error.

Reading the caller's stored settings¶

A tool that needs the caller's stored plugin settings calls load_user_plugin_config from sparkth.lib.plugins with its plugin name and config class:

config = await load_user_plugin_config("canvas", CanvasConfig)

It never takes those settings as arguments, because the model writes tool arguments and could send anything. The settings come from the row the bound user saved on the Settings page.

It raises PluginNotConfiguredError (sparkth.lib.plugins) when the user has saved no settings (no row, or an empty config), and PluginConfigInvalidError when the saved settings no longer validate. A tool lets either propagate, and the model sees its message, which tells the user to add or to update their settings on the Settings page. For invalid settings, only the names of the failing fields are logged, never their values.

It raises NoAuthenticatedUser where no user is bound, before any database work.

Content contributors¶

A content contributor lets a plugin that owns content (e.g. a course-generation plugin) hand off one piece of it to a plugin that publishes to an LMS, without either plugin importing the other. The owning plugin registers a ContentContributor — a name, a human-facing description, and a map of builders — with register_content_contributor(self, contributor) from sparkth.lib.content.hooks. A publishing plugin resolves a contributor by name and awaits the builder registered under its own plugin name to get back the ContentBlock to create. A builder is awaited with the destination course id and the chosen option's id. The option id is None only for a contributor that offers no options: a publishing plugin refuses to build for a contributor that offers options until one is chosen, so nothing is placed by default.

The hook is keyed by the registering plugin, one contributor per plugin. A publishing tool offers and builds a contributor only while that plugin is on for the caller; a switched-off contributor is neither listed nor built. Read the switched-on contributors with enabled_content_contributors() from sparkth.lib.content.hooks, or one by name with enabled_content_contributor(name). It raises UnknownContentContributorError for a name nobody registered and ContentContributorUnavailableError when the plugin is switched off, both from sparkth.lib.content.exceptions. With no authenticated caller, only the system-wide switch applies.

builders is keyed by the publishing plugin's registered name, so its keys are the LMSes a contributor targets. open-edx is the one shipped publisher. Register one builder per LMS you can produce a block for; a publishing plugin skips any contributor that has no builder for it, so a misspelled key reads as "does not target that LMS".

A ContentBlock carries a title, a kind and attributes. Only title means the same thing everywhere: kind and attributes are interpreted by the publishing plugin, so sparkth/lib never needs to know about any particular LMS. Open edX reads kind as the XBlock category and attributes as a dict of XBlock metadata.

list_options is an optional async function that returns ContentOption(id, label) items, the choices an author picks between. A publishing tool lists them and passes the chosen id back to the builder as option_id. With list_options left None, a publishing tool refuses any option id. Options that depend on the caller read the caller with sparkth.lib.auth.current_user_id(). The builder must re-check the chosen id and never trust that it came from the list. list_options and the builders must be module-level functions, because equality is what lets each construction of a plugin register its contributor again.

# sparkth/plugins/myappplugin/plugin.py
from sparkth.lib.content.exceptions import ContentBuildError
from sparkth.lib.content.hooks import ContentBlock, ContentContributor, ContentOption, register_content_contributor

TEMPLATES = {"intro": "Introduction", "recap": "Recap"}


async def list_my_options() -> list[ContentOption]:
    return [ContentOption(key, label) for key, label in TEMPLATES.items()]


async def build_my_openedx_block(course_id: str, option_id: str | None) -> ContentBlock:
    template = option_id or "intro"
    if template not in TEMPLATES:
        raise ContentBuildError(f"Unknown template: {template}")
    return ContentBlock(TEMPLATES[template], "my-category", {"course": course_id, "template": template})


class MyAppPlugin(SparkthPlugin):
    def __init__(self) -> None:
        super().__init__("my-app")
        register_content_contributor(
            self,
            ContentContributor(
                "my-app", "What this contributes", {"open-edx": build_my_openedx_block}, list_my_options
            ),
        )

Register through register_content_contributor, not LMS_CONTENT_CONTRIBUTORS.add_item. A plugin is constructed more than once in one process — the loader builds it, and its own tests build their own instances — so each construction registers the same contributor again. The helper accepts an equal contributor from a plugin of the same name, and raises DuplicateContentContributorError when a contributor claims a registered name with a different value or from a different plugin, which is the collision worth failing on.

If a builder cannot produce its block, it reports that by raising sparkth.lib.content.exceptions.ContentBuildError — not by returning a placeholder block or letting some other exception escape. A publishing tool catches it and reports the failure through its own error contract instead of letting it propagate unguarded.

callback is an optional async function a contributor passes after list_options. A publishing tool awaits it with the chosen option's id, for example so the contributor can record that its content landed in a course. The tool awaits it once the LMS has created the block and before it writes the block's attributes, so a block the LMS refuses is never reported. A block whose attribute write then fails still counts as placed. If callback cannot record the placement, it raises sparkth.lib.content.exceptions.ContentPlacementError. The publishing tool then leaves the block's attributes unwritten and reports the block's locator, so the block points at nothing and the caller can delete it. Like the builders, callback must be a module-level function.

Chat capabilities¶

A capability is one kind of work the chat assistant does, such as designing a course. A plugin contributes one without importing chat, by registering a ChatCapability with register_chat_capability from sparkth.lib.chat.hooks. A plugin registers one capability, from its __init__, keyed by the plugin instance.

A ChatCapability has three fields:

title is the heading the capability appears under, both in the system prompt and in the scope gate's list, so no two capabilities share one.
scope is the text the scope gate judges a message against.
prompt is plain text inserted into the system prompt as written, so a brace in it needs no escaping.

For example:

# sparkth/plugins/myappplugin/plugin.py
from sparkth.lib.chat.hooks import ChatCapability, register_chat_capability
from sparkth.lib.plugins import SparkthPlugin

TITLE = "My App builder"
SCOPE = "Requests to build or change things in My App."
PROMPT = "Ask which My App item the author wants, then build it with the My App tools."

MY_CAPABILITY = ChatCapability(TITLE, SCOPE, PROMPT)


class MyAppPlugin(SparkthPlugin):
    def __init__(self) -> None:
        super().__init__("my-app")
        register_chat_capability(self, MY_CAPABILITY)

Register through register_chat_capability, not CHAT_CAPABILITIES.add_item. Constructing the plugin again is harmless: the loader and a plugin's own tests each build an instance, and equal copies fold into one. Only a different capability claiming a registered title raises DuplicateChatCapabilityError, from sparkth.lib.chat.exceptions. The loader logs the failed construction and skips that plugin.

How chat uses capabilities¶

On every request chat reads the capabilities of the plugins enabled for the requesting user, in plugin-name order. Chat contributes its own, course design. A plugin switched off system-wide or by the user contributes no section, no scope and no tools.

Scope gate. Each message is judged against every enabled scope. A message that fits none is refused with a fixed sentence before the model runs. A refused first message writes nothing; a refused later message is stored in the conversation, with the refusal as the assistant's reply. If the classification fails, the turn fails with an error and the model is not called. A failed first message writes nothing; a failed later message stores the error in the conversation.
System prompt. The prompt is chat's skeleton with each enabled capability's section under its title. The skeleton owns the date, the assistant's identity, the output language, the refusal, injection and offering-choices rules, so a section must not repeat them. A section holds the capability's workflow, plus any language or guardrail rules specific to it.
Tools. Every turn gets the tools the request asks for, minus those of a switched-off plugin.

Plugin-specific knowledge belongs in a <prefix>about tool, not in a capability's prompt; see the LMS about tool above for the convention.

Reacting to a deleted document¶

A plugin that stores rows referencing a core Document has to clear them when the document goes away. The foreign keys declare ondelete="CASCADE", but soft_delete_document is an UPDATE, not a DELETE, so that cascade never fires — a row pointing at a soft-deleted document survives indefinitely unless the plugin removes it.

Register a handler on the DOCUMENT_DELETED hook from sparkth.lib.documents to be told. Handlers run inside the deleting transaction, after the document is flagged and before the flush, so your writes commit or roll back with the deletion. Do not commit in a handler, and note that raising from one fails the whole delete — which is deliberate, since a half-applied cleanup is worse than a delete the caller can retry.

# sparkth/plugins/myappplugin/plugin.py
from sparkth.lib.documents import DOCUMENT_DELETED, Document


async def drop_my_rows(session: AsyncSession, document: Document) -> None:
    rows = (await session.exec(select(MyRow).where(MyRow.document_id == document.id))).all()
    for row in rows:
        await session.delete(row)
    await session.flush()


class MyAppPlugin(SparkthPlugin):
    def __init__(self) -> None:
        super().__init__("my-app")
        DOCUMENT_DELETED.add_item(self, drop_my_rows)

The document is passed rather than just its id, so a handler needing the owner (to attribute an analytics event, say) has it without a second query. The chat plugin registers detach_deleted_document this way to take deleted documents out of conversations.

Complete Example¶

from fastapi import APIRouter

from sparkth.lib.mcp.hooks import MCP_TOOLS, Tool
from sparkth.lib.routes import register_router
from sparkth.lib.plugins import SparkthPlugin

# Router
router = APIRouter()


@router.get("/{city}")
async def get_weather_route(city: str):
    return {"city": city, "temp": 20}


# Plugin (declares its name explicitly)
class WeatherPlugin(SparkthPlugin):
    def __init__(self) -> None:
        super().__init__("weather")
        register_router(self, router)
        MCP_TOOLS.add_item(self, Tool(self.get_weather, category="weather"))

    async def get_weather(self, city: str) -> dict:
        """Get the current weather for a city."""
        return {"city": city, "temperature": 20, "unit": "celsius"}

Testing¶

# Start the FastAPI server locally (http://0.0.0.0:7727)
make backend.up.dev

# Test API (routes mount at /api/v1/<plugin-name>)
curl http://localhost:7727/api/v1/my-app/

For real-world examples, see the sparkth/plugins/canvas/ directory.