Plugin authoring¶

The classes and hooks a plugin author uses. See the plugin development guide for the how-to; this page is the generated reference.

Plugin framework¶

sparkth.lib.plugins ¶

Public API for the plugin framework.

All plugins and external modules import the plugin-authoring surface from here. Nothing outside sparkth/lib/plugins should import from sparkth.core.plugins, sparkth.core.plugins.base, sparkth.core.plugins.config_base, sparkth.core.plugins.middleware or sparkth.core.plugins.service directly.

PluginConfig ¶

Bases: BaseModel

Base class for all plugin configs

lms_tool_prefix `classmethod` ¶

lms_tool_prefix() -> str | None

Return the tool-name prefix for this LMS plugin (e.g. "openedx_"), or None if this plugin is not an LMS.

Used to detect whether any active tools belong to this LMS so that the credential injection can short-circuit the database call when no LMS tools are present. Override in each LMS config class.

to_lms_credentials_hint ¶

to_lms_credentials_hint() -> str | None

Return a human-readable, newline-formatted block of credentials for the LLM system prompt, or None if credentials are incomplete or this plugin is not an LMS.

Override in each LMS config class. The returned string will be included verbatim in the system message that instructs the LLM to use these credentials automatically when calling LMS tools.

PluginService ¶

Business logic related to Plugin persistence and state.

apply_postprocess `async` `staticmethod` ¶

apply_postprocess(
    plugin_name: str,
    session: AsyncSession,
    user_id: int,
    stored_config: dict[str, Any],
) -> dict[str, Any]

Run the plugin's postprocess adapter if one is registered.

apply_preprocess `async` `staticmethod` ¶

apply_preprocess(
    plugin_name: str,
    session: AsyncSession,
    user_id: int,
    incoming_config: dict[str, Any],
) -> dict[str, Any]

Run the plugin's preprocess adapter if one is registered.

get_all `async` ¶

get_all(
    session: AsyncSession,
    include_disabled: bool = True,
    include_deleted: bool = False,
) -> Sequence[Plugin]

Get all plugins.

Parameters:

Name	Type	Description	Default
`session`	`AsyncSession`	Database session	required
`include_disabled`	`bool`	Whether to include disabled plugins	`True`
`include_deleted`	`bool`	Whether to include soft-deleted plugins	`False`

Returns:

Type	Description
`Sequence[Plugin]`	list of Plugin objects

get_or_create_all `async` ¶

get_or_create_all() -> None

Ensure every loaded plugin has a row in the database.

Called once at startup. Fetches existing plugins in a single query and upserts all loaded plugins in one transaction. Only config_schema is refreshed on existing rows — enabled is left untouched so a plugin a user disabled stays disabled across restarts.

initial_config `staticmethod` ¶

initial_config(schema: dict[str, Any]) -> dict[str, Any]

Populate config dict with all keys from schema set to None.

validate_user_config `staticmethod` ¶

validate_user_config(
    plugin: Plugin, user_config: dict[str, Any]
) -> dict[str, Any]

Validate and normalize user configuration against plugin's Pydantic config model.

Uses the plugin.config_schema directly instead of dynamically loading config.py.

Raises:

Type	Description
`ConfigValidationError`	if config_schema is not a subclass of PluginConfig or if validation fails

SparkthPlugin ¶

SparkthPlugin(name: str)

Base class for Sparkth plugins.

All plugins should inherit from this class. The loader constructs every plugin as plugin_class(plugin_name), so __init__ must accept the derived plugin_name as its first positional argument and pass it through to super().__init__(). Register routes, tools, and the config schema from within __init__.

Example:

from sparkth.lib.mcp.hooks import MCP_TOOLS, Tool

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

    async def my_tool(self, payload: MyPayload) -> dict:
        """Describe what the tool does (becomes the MCP tool description)."""
        ...

Initialize the plugin with its derived name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Unique identifier for the plugin (e.g., "canvas")	required

repr ¶

__repr__() -> str

Return string representation of the plugin.

UserPluginResponse ¶

Bases: BaseModel

Response model for user plugin information.

get_plugin_loader ¶

get_plugin_loader() -> PluginLoader

Get the singleton PluginLoader instance.

This function is safe and efficient to call multiple times.

Returns:

Name	Type	Description
`PluginLoader`	`PluginLoader`	The global plugin loader instance

Route registration¶

sparkth.lib.routes ¶

register_router ¶

register_router(
    plugin: SparkthPlugin, router: APIRouter
) -> None

Register a router associated to a plugin.

The plugin routes will automatically be prefixed with "/api/v1/" and the plugin name will be associated to the route tags.

MCP tools¶

sparkth.lib.mcp.hooks ¶

Tool `dataclass` ¶

Tool(
    handler: Callable[..., Any], category: str | None = None
)

An MCP tool a plugin contributes to the :data:MCP_TOOLS hook.

The plugin registers it from its __init__ with MCP_TOOLS.add_item(self, Tool(self.my_method, category="...")). The tool's name and description are derived from the bound handler (its name and docstring); the input schema is auto-generated from the handler signature.

generate_input_schema ¶

generate_input_schema(
    func: Callable[..., Any],
) -> dict[str, Any]

Auto-generate a JSON Schema from a function signature using type hints.

resolve_schema_refs ¶

resolve_schema_refs(
    schema: Any, defs: dict[str, Any]
) -> Any

Recursively resolve all $ref references inline within a JSON schema.

type_to_json_schema ¶

type_to_json_schema(py_type: type[Any]) -> dict[str, Any]

Convert a Python type to a JSON Schema type definition.

Config hooks¶

sparkth.lib.config ¶

get_plugin_adapter ¶

get_plugin_adapter(
    plugin_name: str,
) -> LLMConfigAdapter | None

Return the config adapter a plugin contributed, looked up by plugin name.

get_plugin_config_schema ¶

get_plugin_config_schema(
    plugin_name: str,
) -> type[PluginConfig] | None

Return the config class a plugin contributed, looked up by plugin name.

iter_plugin_adapters ¶

iter_plugin_adapters() -> Iterator[
    tuple[str, LLMConfigAdapter]
]

Yield (plugin_name, adapter) for every plugin that contributed a config adapter.

Like CONFIG_SCHEMAS, CONFIG_ADAPTERS is populated when plugins are instantiated at the process entrypoint; this iterator assumes that has already happened.

iter_plugin_config_schemas ¶

iter_plugin_config_schemas() -> Iterator[
    tuple[str, type[PluginConfig]]
]

Yield (plugin_name, config_class) for every plugin that contributed a config.

Plugins are instantiated once per process at the entrypoint (the FastAPI lifespan, the standalone MCP server, or the migration runner), which is what populates CONFIG_SCHEMAS; this iterator assumes that has already happened.