FabledSteward/docs/plugins/writing-a-plugin.md

Writing a Plugin

This guide walks through building a complete plugin from scratch. The Traefik plugin (plugins/traefik/) is the reference implementation — read it alongside this guide.

---

Step 1: Create the Directory

plugins/
└── myplugin/
    └── __init__.py    ← start here

The directory name is the plugin's identity. It must match the name field in plugin.yaml and is used as the URL prefix (/plugins/myplugin/) and the Python import name.

---

Step 2: Write plugin.yaml

name: myplugin
version: "1.0.0"
description: "A short description of what this plugin monitors"
author: "Your Name or GitHub username"
license: "MIT"          # any SPDX identifier, e.g. MIT, Apache-2.0, GPL-3.0

# Optional: prevents loading on older app versions
min_app_version: "0.1.0"

# Optional: where the plugin's source lives
repository_url: "https://github.com/yourname/yourrepo"
homepage: "https://github.com/yourname/yourrepo/tree/main/myplugin"
tags:
  - monitoring
  - http

# Default config values — users override these via the Settings UI
# or by writing to the app_settings DB table under "plugin.myplugin"
config:
  target_url: "http://localhost:9090/metrics"
  scrape_interval_seconds: 60

---

Step 3: Define Models (if needed)

If your plugin stores data, define SQLAlchemy models using the shared Base from steward.models.base.

# plugins/myplugin/models.py
from __future__ import annotations
import uuid
from datetime import datetime
from sqlalchemy import String, Float, DateTime
from sqlalchemy.orm import Mapped, mapped_column
from steward.models.base import Base


class MyPluginMetric(Base):
    __tablename__ = "myplugin_metrics"

    id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    resource_name: Mapped[str] = mapped_column(String, nullable=False, index=True)
    scraped_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    my_value: Mapped[float] = mapped_column(Float, nullable=False)

---

Step 4: Write Migrations

Create plugins/myplugin/migrations/ with the standard Alembic layout. Copy env.py and script.py.mako from plugins/traefik/migrations/ as a starting point — the env.py is boilerplate and rarely needs changes.

Generate the initial migration:

# From the project root
alembic --config alembic.ini revision \
    --autogenerate \
    --head=steward@head \
    --branch-label=myplugin \
    -m "myplugin initial"

Edit the generated file to set depends_on:

# In the generated revision file:
depends_on = ("0004_core_head_id",)   # the core migration head ID
down_revision = None
branch_labels = ("myplugin",)

# Prefix the revision ID with the plugin name:
revision = "myplugin_001_initial"

---

Step 5: Write Scheduled Task Logic

Keep task logic in a separate file so __init__.py stays clean.

# plugins/myplugin/scheduler.py
from __future__ import annotations
import logging
from steward.core.scheduler import ScheduledTask
from steward.core.alerts import record_metric

logger = logging.getLogger(__name__)


def make_task(app) -> ScheduledTask:
    interval = int(app.config["PLUGINS"]["myplugin"]["scrape_interval_seconds"])

    async def scrape():
        await _do_scrape(app)

    return ScheduledTask(
        name="myplugin_scrape",
        coro_factory=scrape,
        interval_seconds=interval,
        run_on_startup=True,
    )


async def _do_scrape(app) -> None:
    from .models import MyPluginMetric
    from datetime import datetime, timezone

    url = app.config["PLUGINS"]["myplugin"]["target_url"]

    try:
        value = await _fetch_value(url)
    except Exception:
        logger.exception("myplugin scrape failed (url=%s)", url)
        return

    now = datetime.now(timezone.utc)

    async with app.db_sessionmaker() as session:
        async with session.begin():
            # Write to plugin's own history table
            session.add(MyPluginMetric(
                resource_name="my-resource",
                scraped_at=now,
                my_value=value,
            ))

            # Emit to plugin_metrics so alert rules can fire
            await record_metric(
                session=session,
                source_module="myplugin",
                resource_name="my-resource",
                metric_name="my_value",
                value=value,
            )


async def _fetch_value(url: str) -> float:
    import httpx
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        resp.raise_for_status()
        return float(resp.text.strip())

---

Step 6: Write Routes (if needed)

# plugins/myplugin/routes.py
from quart import Blueprint, current_app, render_template
from steward.auth.middleware import require_role
from steward.models.users import UserRole
from .models import MyPluginMetric

myplugin_bp = Blueprint("myplugin", __name__, template_folder="templates")


@myplugin_bp.get("/")
@require_role(UserRole.viewer)
async def index():
    async with current_app.db_sessionmaker() as db:
        from sqlalchemy import select
        result = await db.execute(
            select(MyPluginMetric).order_by(MyPluginMetric.scraped_at.desc()).limit(50)
        )
        rows = result.scalars().all()
    return await render_template("myplugin/index.html", rows=rows)


@myplugin_bp.get("/widget")
@require_role(UserRole.viewer)
async def widget():
    """HTMX fragment for the dashboard widget."""
    async with current_app.db_sessionmaker() as db:
        from sqlalchemy import select
        result = await db.execute(
            select(MyPluginMetric).order_by(MyPluginMetric.scraped_at.desc()).limit(1)
        )
        latest = result.scalar_one_or_none()
    return await render_template("myplugin/widget.html", latest=latest)

---

Step 7: Write Templates

Templates live in plugins/myplugin/templates/myplugin/ (the extra nesting avoids naming collisions with core templates).

{# plugins/myplugin/templates/myplugin/index.html #}
{% extends "base.html" %}
{% block title %}My Plugin — Steward{% endblock %}
{% block content %}
<div class="page-title">My Plugin</div>
{% for row in rows %}
  <p>{{ row.resource_name }} — {{ row.my_value }}</p>
{% endfor %}
{% endblock %}

{# plugins/myplugin/templates/myplugin/widget.html — HTMX fragment, no extends #}
{% if latest %}
  <div class="ping-row">
    <span>{{ latest.resource_name }}</span>
    <span>{{ latest.my_value }}</span>
  </div>
{% else %}
  <p class="empty">No data yet.</p>
{% endif %}

---

Step 8: Wire Up __init__.py

# plugins/myplugin/__init__.py
from __future__ import annotations

_app = None


def setup(app) -> None:
    global _app
    _app = app
    from .models import MyPluginMetric  # noqa: registers model with Base.metadata


def get_scheduled_tasks() -> list:
    from .scheduler import make_task
    return [make_task(_app)]


def get_blueprint():
    from .routes import myplugin_bp
    return myplugin_bp

---

Step 9: Enable the Plugin

Plugin config is stored in the app_settings DB table. The easiest way to enable a plugin is via the Settings UI, or by inserting directly:

INSERT INTO app_settings (key, value_json, updated_at)
VALUES ('plugin.myplugin', '{"enabled": true, "target_url": "http://localhost:9090/metrics"}', now());

On next startup, the plugin will be loaded, its migrations applied, and its blueprint and tasks registered.

---

Using record_metric()

record_metric() is how plugins feed data into the alert pipeline. Any metric you write here can be the target of an alert rule created in the UI.

from steward.core.alerts import record_metric

# Must be inside an active transaction
async with session.begin():
    await record_metric(
        session=session,
        source_module="myplugin",       # matches alert rule source_module
        resource_name="my-server",      # matches alert rule resource_name
        metric_name="response_time_ms", # matches alert rule metric_name
        value=42.3,
    )

record_metric() writes to plugin_metrics and evaluates all matching alert rules inline. Notifications are deferred outside the transaction. It propagates SQLAlchemyError on DB failure — don't swallow it.

---

Auth in Routes

Use the @require_role decorator from steward.auth.middleware:

from steward.auth.middleware import require_role
from steward.models.users import UserRole

@myplugin_bp.get("/admin-only")
@require_role(UserRole.admin)
async def admin_page():
    ...

@myplugin_bp.get("/read-only")
@require_role(UserRole.viewer)   # viewer, operator, and admin can all access
async def read_page():
    ...

Role hierarchy: admin > operator > viewer. Requiring viewer grants access to all three roles.

---

Audit Log Entries

A change made through a plugin's pages is audited with steward.core.audit.log_audit, naming the thing it changed by entity_type and entity_id. So the audit page shows that thing's current name, linked to its page, export get_audit_entities() from __init__.py:

# plugins/myplugin/__init__.py
def get_audit_entities() -> dict:
    from .models import Target
    return {"myplugin_target": (Target, "/plugins/myplugin/targets/{}")}

The model needs id and name columns. An entity deleted since keeps the name the entry recorded in its detail (detail={"name": ...}). Add the action's wording to ACTIONS in steward/audit/service.py so the row reads as a sentence.

---

Agent Tools (MCP)

Steward serves an MCP endpoint at /mcp that AI clients reach with a personal API token created on the AI access page. A plugin adds tools to it by exporting an optional register_mcp_tools(register) from __init__.py:

# plugins/myplugin/__init__.py
def register_mcp_tools(register) -> None:
    from .mcp_tools import list_targets, rescan_target
    from steward.models.users import UserRole
    register(list_targets, role=UserRole.viewer, read_only=True)
    register(rescan_target, role=UserRole.operator, read_only=False)

# plugins/myplugin/mcp_tools.py
from steward.mcp._context import current_caller, mcp_app


async def list_targets(limit: int = 50) -> dict:
    """What this tool is for, when to use it over its siblings, what each
    argument's empty value means, and what comes back. This docstring is the
    tool's description: it is all the agent knows about it."""
    app = mcp_app()                 # Quart's current_app does not exist here
    async with app.db_sessionmaker() as db:
        ...
    return {"targets": [...], "total": n, "truncated": n > limit}

• role and read_only are required. A tool is listed and callable only for callers whose role reaches role, and only while the plugin is enabled.
• Tools run outside Quart's request pipeline: use mcp_app() and current_caller() from steward.mcp._context, never current_app or session.
• Refuse by raising ValueError("what was wrong, and what to do instead"). The agent reads that text. Any other exception reaches it only as a generic failure.
• Tool names are global across core and every plugin; a name already taken fails registration and the plugin contributes no tools (logged).
• Hot-reloading the plugin replaces its tools.
• Follow the response conventions the core tools use (steward/mcp/tools/_shape.py): times via iso() plus age_s(), lists through bounded() so a cut is never silent, series through summarize(), and lookups by name or id through pick().

Contributing to get_overview

get_overview is the agent's first call: what is wrong right now, ranked. A plugin adds its own problems to it by exporting an optional mcp_overview(db) that returns findings built with steward.mcp.tools.overview.finding:

# plugins/myplugin/__init__.py
async def mcp_overview(db) -> list[dict]:
    from steward.mcp.tools.overview import finding
    return [
        finding("warning", "target_unreachable", t.name, f"{t.name} has not answered for {t.misses} scans",
                since=t.last_ok_at, ids={"target_id": t.id}, next=f'get_target(target="{t.id}")')
        for t in await _unreachable_targets(db)
    ]

• severity is "critical" (something is down or about to be) or "warning" (degraded, or heading that way). Report problems only; a healthy plugin returns [].
• It is called only while the plugin is enabled, with its own session. If it raises, the overview still returns, and names the plugin under unavailable.

---

Shipping a Plugin

Steward never downloads plugins. There is no catalog and no install button: the app runs with no outbound network.

• First-party plugins live in this repository under plugins/ and ship inside the image, so they version with core.
• A third-party plugin is a folder you put in the external plugin root, /data/plugins by default (STEWARD_PLUGIN_DIR moves it). Restart Steward and the plugin appears in Settings → Plugins, where you enable it. A bundled plugin of the same name wins over an external one.

---

Checklist

• plugins/myplugin/ directory created
• plugin.yaml with correct name (matches directory), author, license, tags
• __init__.py exports setup() and get_scheduled_tasks()
• Models import inside setup() to register with metadata
• Migrations use depends_on pointing to core head, not down_revision
• Revision IDs prefixed with plugin name
• record_metric() called inside session.begin()
• Routes use @require_role decorator
• MCP tools (if any) registered via register_mcp_tools(register) with a role and read_only
• Templates namespaced under templates/myplugin/
• Plugin enabled in app settings