Writing a plugin¶

A plugin is a module with a name, a version, and a configuration schema — plus the one property a module cannot have: it can be installed by somebody else, validated before it runs, and attributed when it does.

The trust boundary comes first¶

Plugins are trusted, in-process Python code. Adding one is equivalent to importing a library: it runs with the full privileges of the process. There is no sandbox, no permission model, and no protection from a plugin that misbehaves — only the decision to add it (ADR 0007). There is also no entry-point discovery, so nothing can become a plugin by being installed: app.add_plugin(...) is the only way in, and startup stays something a reader can follow from the application's own code.

That is the whole reason the metadata on this page exists. name, api_version, and config_type are not documentation — they are the three things the runtime can check before a line of the plugin runs.

The shape¶

examples/learn_extend/plugin_audit.py

"""An in-tree plugin: a capability, two routes, its configuration, and a plugin that requires it.

A distribution usually ships several plugins in one package, which is why both classes live here.
The first one is self-contained; the second one declares ``requires = ("audit",)`` and therefore
cannot be registered before it — the registry refuses the wrong order instead of discovering it at
request time.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Annotated

from starlette.requests import Request

from xyberos import App, Inject, Plugin
from xyberos.http import read_json_object


@dataclass(frozen=True)
class AuditConfig:
    """Typed configuration, read from ``[plugins.audit]`` or the environment."""

    prefix: str = "audit"
    limit: int = 100


class AuditLog:
    """The capability this plugin contributes to the application."""

    def __init__(self, prefix: str, limit: int) -> None:
        self.prefix = prefix
        self.limit = limit
        self.events: list[str] = []

    def record(self, event: str) -> None:
        """Record one event, keeping only the most recent ``limit`` of them."""
        if len(self.events) >= self.limit:
            self.events.pop(0)
        self.events.append(event)


class AuditPlugin(Plugin[AuditConfig]):
    """Contributes the log, a route that reads it, and a route that writes to it."""

    name = "audit"
    api_version = "1"
    config_type = AuditConfig

    def __init__(self) -> None:
        self.started = False
        self.stopped = False

    def register(self, app: App, config: AuditConfig) -> None:
        app.provide(AuditLog, instance=AuditLog(config.prefix, config.limit))
        app.get("/audit")(self._read)
        app.post("/audit")(self._record)

    async def _read(self, log: Annotated[AuditLog, Inject()]) -> dict[str, object]:
        return {"prefix": log.prefix, "events": list(log.events)}

    async def _record(
        self, request: Request, log: Annotated[AuditLog, Inject()]
    ) -> dict[str, object]:
        payload = await read_json_object(request)
        log.record(str(payload.get("event", "")))
        return {"events": list(log.events)}

    async def start(self) -> None:
        """Runs after app-scoped providers are constructed, before they are closed."""
        self.started = True

    async def stop(self) -> None:
        """Runs before app-owned resources are closed, and after every provider was constructed."""
        self.stopped = True


class ReportsPlugin(Plugin[None]):
    """A dependent plugin: it needs the log, so the registry makes the order a requirement."""

    name = "reports"
    api_version = "1"
    requires = ("audit",)
    config_type = None

    def register(self, app: App, config: None) -> None:
        app.get("/reports")(self._summary)

    async def _summary(self, log: Annotated[AuditLog, Inject()]) -> dict[str, object]:
        return {"count": len(log.events)}

Two plugins live in one module because that is how a distribution usually ships: AuditPlugin contributes a capability and two routes, and ReportsPlugin declares requires = ("audit",) because it reads the capability the first one registers. Everything a plugin contributes goes through the same app.provide and app.get calls the application itself uses — there is no second API to learn.

Registering them¶

examples/learn_extend/plugins01_registration.py

"""Walkthrough — registering plugins: what they contribute, what they read, and when they run.

Run it with:

    python -m learn_extend.plugins01_registration
"""

from __future__ import annotations

from learn_extend.plugin_audit import AuditPlugin, ReportsPlugin
from xyberos import App, Settings
from xyberos.http import HttpAdapter
from xyberos.http.testing import create_test_client

#: ``[plugins.audit]`` in this walkthrough, in the same shape a configuration file would have it.
SETTINGS = {"plugins": {"audit": {"prefix": "req", "limit": 2}}}


def main() -> None:
    """Configure two plugins, watch the order, and read what each one contributed."""
    app = App(settings=Settings.load(environ={}, overrides=SETTINGS))
    audit = AuditPlugin()
    reports = ReportsPlugin()
    app.add_plugin(audit)
    app.add_plugin(reports)

    print(app.diagnostics().render().strip())

    with create_test_client(HttpAdapter(app)) as client:
        for event in ("ticket-opened", "reply-drafted", "reply-sent"):
            client.post("/audit", json={"event": event})
        print("GET /audit ->", client.get("/audit").json())
        print("GET /reports ->", client.get("/reports").json())

    print("lifecycle hooks ran: started =", audit.started, "stopped =", audit.stopped)


if __name__ == "__main__":
    main()

The application supplies [plugins.audit] in its settings, so the plugin's configuration arrives from the deployment rather than from the plugin's own defaults:

providers:
  AuditLog [app] from plugin:audit
plugins:
  audit 1 [registered] requires: -
  reports 1 [registered] requires: audit
GET /audit -> {'prefix': 'req', 'events': ['reply-drafted', 'reply-sent']}
GET /reports -> {'count': 2}
lifecycle hooks ran: started = True stopped = True

Four things are visible in six lines:

from plugin:audit. The provider's source is attributed automatically, so an override you did not expect is traceable to the plugin that made it. Application registrations report from application.
prefix: 'req' and the two retained events. Settings reached the plugin as a typed object, not as a dictionary — limit = 2 came from the same section and is why the list is truncated.
requires: audit before reports. Registration order is dependency order, and the reported order is the start order. A requirement that is not registered yet is refused immediately, which is why requirement cycles are impossible rather than merely forbidden.
Both hooks ran. start runs after app-scoped providers are constructed; stop runs before the application closes its own resources.

Lifecycle, in order¶

app.add_plugin(plugin) — validate metadata, build typed configuration, run register.
app.start() — validate the provider graph, construct eager app-scoped providers, start plugins in registration order.
app.stop() — stop plugins in reverse order, then close app-owned resources in reverse order.

A failure in start stops the plugins that already started and reports a PluginLifecycleError naming the plugin and the phase, with the original exception kept as the cause. A failure in stop is collected rather than raised: every remaining hook still runs, and the failures surface as plugin:<name> in ShutdownError.resources. Shutdown is the one place where stopping late is worse than reporting loudly.

The refusals¶

examples/learn_extend/plugins02_refusals.py

"""Walkthrough — the seven refusals a plugin author meets, and what each one means.

Run it with:

    python -m learn_extend.plugins02_refusals
"""

from __future__ import annotations

from dataclasses import dataclass

from learn_extend.plugin_audit import AuditPlugin, ReportsPlugin
from xyberos import App, Plugin, Settings
from xyberos.errors import (
    PluginCompatibilityError,
    PluginConfigurationError,
    PluginDependencyError,
    PluginValidationError,
)


class FuturePlugin(Plugin[None]):
    """A plugin written against a later plugin API version."""

    name = "future"
    api_version = "2"
    config_type = None

    def register(self, app: App, config: None) -> None: ...


class PunctuatedPlugin(Plugin[None]):
    """A plugin whose name is not an identifier, so ``[plugins.<name>]`` cannot address it."""

    name = "audit.log"
    api_version = "1"
    config_type = None

    def register(self, app: App, config: None) -> None: ...


@dataclass(frozen=True)
class WebhookConfig:
    """A configuration schema with one setting that has no default."""

    url: str


class WebhookPlugin(Plugin[WebhookConfig]):
    """A plugin that cannot work without the required setting."""

    name = "webhook"
    api_version = "1"
    config_type = WebhookConfig

    def register(self, app: App, config: WebhookConfig) -> None: ...


class PlainPlugin(Plugin[None]):
    """A plugin that declares no ``config_type``: passing it a mapping is a mistake, not a no-op."""

    name = "plain"
    api_version = "1"
    config_type = None

    def register(self, app: App, config: None) -> None: ...


class ExplodingPlugin(Plugin[None]):
    """A plugin whose ``register`` raises: the application is left exactly as it was."""

    name = "exploding"
    api_version = "1"
    config_type = None

    def register(self, app: App, config: None) -> None:
        raise RuntimeError("the plugin could not reach its database")


def main() -> None:
    """Try each mistake on a fresh application, and print what the runtime says about it."""
    app = App()
    try:
        app.add_plugin(ReportsPlugin())
    except PluginDependencyError as error:
        print("requirement registered second:", error)

    app = App()
    try:
        app.add_plugin(FuturePlugin())
    except PluginCompatibilityError as error:
        print("plugin from a later API:", error)

    app = App()
    try:
        app.add_plugin(PunctuatedPlugin())
    except PluginValidationError as error:
        print("name that is not an identifier:", error)

    app = App()
    app.add_plugin(AuditPlugin())
    try:
        app.add_plugin(AuditPlugin())
    except PluginValidationError as error:
        print("name registered twice:", error)

    settings = Settings.load(environ={}, overrides={"plugins": {"audit": {"prefx": "typo"}}})
    app = App(settings=settings)
    try:
        app.add_plugin(AuditPlugin())
    except PluginConfigurationError as error:
        print("misspelled setting in [plugins.audit]:", error)

    app = App()
    try:
        app.add_plugin(WebhookPlugin(), config={})
    except PluginConfigurationError as error:
        print("required setting with no value:", error)

    app = App()
    try:
        app.add_plugin(PlainPlugin(), config={"url": "x"})
    except PluginConfigurationError as error:
        print("configuration without a schema:", error)

    app = App()
    try:
        app.add_plugin(ExplodingPlugin())
    except PluginValidationError as error:
        print("register hook that raised:", error)

    print("the app that met the last refusal still has", len(app.diagnostics().plugins), "plugins")


if __name__ == "__main__":
    main()

requirement registered second: plugin 'reports' requires audit, which must be registered first
plugin from a later API: plugin 'future' targets plugin API '2', but this runtime supports '1'
name that is not an identifier: plugin name 'audit.log' must be an identifier so it can be used as a configuration key
name registered twice: a plugin named 'audit' is already registered
misspelled setting in [plugins.audit]: plugin 'audit' received unknown configuration key(s): prefx
required setting with no value: plugin 'webhook' is missing the required setting 'plugins.webhook.url'
configuration without a schema: plugin 'plain' received configuration but declares no config_type: url
register hook that raised: plugin 'exploding' failed to register: the plugin could not reach its database
the app that met the last refusal still has 0 plugins

Each of those is a mistake that would otherwise be discovered at request time, or never:

Refusal	When it is raised	What it protects
`requires` not yet registered	`add_plugin`	deterministic startup order
`api_version` major mismatch	`add_plugin`, before `register` runs	a plugin written for a contract you do not have
`name` not an identifier	`add_plugin`	`[plugins.<name>]` as a configuration key
name already registered	`add_plugin`	two plugins competing for one name
unknown configuration key	`add_plugin`	a typo that would otherwise be ignored
required setting with no value	`add_plugin`	a plugin that cannot work without it
configuration without a schema	`add_plugin`	a settings section nobody reads
an exception inside `register`	`add_plugin`	a half-registered application

The last line is the point of the whole exercise: after every one of those refusals the application has no plugins at all. Validation happens before register, and a register that raises is reported as PluginValidationError naming the plugin and the original error — the framework does not unwind a partially registered plugin and pretend it succeeded.

Configuration¶

Plugin settings follow the same precedence as core settings, and the plugin is the first thing that narrows them:

the explicit mapping passed to app.add_plugin(plugin, config={...})
the environment — XYBEROS_PLUGINS__AUDIT__PREFIX
[plugins.audit] in the configuration file
the dataclass defaults

Fields may be str, int, float, bool, or their | None forms; numeric strings and the usual boolean spellings are accepted, because that is what an environment variable provides. Anything else is a PluginValidationError at registration, not a surprise at runtime. Error messages name the plugin and the setting location — plugins.audit.prefx — and never the configured value, since a settings file is allowed to contain secrets (plugin configuration).

Publishing one¶

A plugin package is an ordinary distribution with one dependency and one export:

mycompany-xyberos-webhooks/
├── pyproject.toml            # depends on the xyberos release you target
├── src/mycompany_webhooks/
│   ├── __init__.py           # exports the Plugin subclass
│   └── plugin.py             # name, api_version, config_type, register
└── tests/
    └── test_plugin.py        # imports xyberos, never an internal module

This repository ships the pattern rather than describing it. tests/fixtures/metrics_plugin is a real distribution: CI installs it with pip install --no-deps -e tests/fixtures/metrics_plugin, and tests/integration/test_plugin_packages.py imports it as an installed package and asserts that every xyberos module it imports is part of the public surface. That last assertion is what turns an internal refactor into your test failing instead of your users breaking, and it is worth copying verbatim:

Keep api_version honest. "1" means "I use only what the version-1 plugin contract promises". A runtime with a different major refuses to load you, before register runs.
Import public names only. xyberos, xyberos.plugins, and the documented xyberos.* modules.
Validate your own configuration. Unknown keys are already refused; semantic checks — a URL that must be absolute, a limit that must be positive — belong in your own code, at registration.
Test it as an installed package, not as a directory inside your application. The two differ, and the difference is exactly what a user will hit.

Testing with plugins¶

app.with_overrides({...}) clones the provider and route registrations but not the plugin registrations, so plugin lifecycle hooks do not run a second time in a test clone. A test that needs a plugin's behaviour should register the plugin on its own application; a test that needs a plugin's provider replaced should override the contract instead, which is the same mechanism every other test uses.

Deliberate limitations¶

No sandbox, permission model, or isolation of untrusted code.
No entry-point discovery; plugins are added explicitly.
A plugin cannot resolve application services from start, because the application is not READY yet. Register an eager provider whose factory does the work, or do it in register.
Plugins cannot be added, removed, or replaced after app.start().

Where to read next¶

Plugin guide — the exhaustive reference: compatibility, the catalogue, and the full metadata table.
Writing a module — the unit to write first, if you are not publishing anything.
The provider registry — what a plugin's app.provide calls actually do.