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

```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`.

```python
# 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:

```bash
# 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`:

```python
# 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.

```python
# 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)

```python
# 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).

```html
{# 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 %}
```

```html
{# 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

```python
# 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:

```sql
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.

```python
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`:

```python
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`:

```python
# 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`:

```python
# 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)
```

```python
# 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`:

```python
# 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