Step-by-step guide to building FastCMS plugins — API routes, database models, event hooks, Admin UI pages, and persistent settings.

Plugin Development Guide

This guide walks you through building a real FastCMS plugin from scratch. The example builds a Task Manager plugin with its own API routes, database table, event hook, and admin page.

Prerequisites

FastCMS running locally
Python 3.11+
Familiarity with FastAPI and SQLAlchemy

1. Create the Plugin Directory

Every plugin is a Python package inside plugins/:

mkdir -p plugins/task_manager
touch plugins/task_manager/__init__.py

2. Write `init.py` — The Entry Point

__init__.py is the only required file. It must define PLUGIN_META and a register(ctx) function.

# plugins/task_manager/__init__.py

PLUGIN_META = {
    "id": "task-manager",
    "name": "Task Manager",
    "version": "1.0.0",
    "author": "Your Name",
    "description": "Manage tasks linked to any collection.",
}


def register(ctx) -> None:
    """Called once at server startup. Register everything here."""
    from .routes import router
    from .hooks import on_record_deleted

    # Add API routes under /api/v1/plugins/task-manager/...
    ctx.include_router(router, prefix="/task-manager")

    # React to record deletions in any collection
    ctx.on_record_delete(None, on_record_deleted)

    # Add a sidebar page in the Admin UI
    ctx.add_admin_page(
        nav_id="task-manager",
        label="Tasks",
        icon="fa-check-square",
        url="/admin/plugins/task-manager",
    )

PLUGIN_META fields

Field	Required	Description
`id`	✓	URL-safe unique identifier (used in API paths)
`name`	✓	Human-readable display name
`version`	✓	Semantic version string
`author`	—	Plugin author name
`description`	—	Short description shown in the Admin UI

3. Add API Routes

# plugins/task_manager/routes.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.dependencies import require_auth, get_db
from .models import Task

router = APIRouter()


@router.get("/tasks")
async def list_tasks(
    db: AsyncSession = Depends(get_db),
    user=Depends(require_auth),
):
    result = await db.execute(
        select(Task).where(Task.user_id == user["id"])
    )
    return result.scalars().all()


@router.post("/tasks", status_code=201)
async def create_task(
    body: dict,
    db: AsyncSession = Depends(get_db),
    user=Depends(require_auth),
):
    task = Task(user_id=user["id"], title=body["title"])
    db.add(task)
    await db.commit()
    await db.refresh(task)
    return task

Routes are mounted at /api/v1/plugins/{prefix}/.... With prefix="/task-manager" the endpoints become:

GET  /api/v1/plugins/task-manager/tasks
POST /api/v1/plugins/task-manager/tasks

Authentication helpers

Dependency	Description
`require_auth`	Requires a valid user JWT or API key
`require_admin`	Requires admin role
`get_db`	Returns an async `AsyncSession`

4. Define Database Models

Inherit from Base and Alembic will track the table automatically.

# plugins/task_manager/models.py
import uuid
from datetime import datetime
from sqlalchemy import String, Boolean, ForeignKey, DateTime
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base


class Task(Base):
    __tablename__ = "plugin_task_manager_tasks"

    id: Mapped[str] = mapped_column(
        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
    )
    user_id: Mapped[str] = mapped_column(String(36), ForeignKey("users.id"), nullable=False)
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    completed: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

Convention: Prefix table names with plugin_{plugin_name}_ to avoid conflicts with core tables.

Generate a migration

alembic revision --autogenerate -m "add task_manager tables"
alembic upgrade head

Alembic auto-discovers plugin models because migrations/env.py imports plugins/*/models.py at startup.

5. React to Events

# plugins/task_manager/hooks.py
from app.core.logging import get_logger

logger = get_logger(__name__)


async def on_record_deleted(event) -> None:
    """Log whenever any record is deleted."""
    logger.info(
        f"[task-manager] Record deleted: "
        f"{event.collection_name}/{event.record_id}"
    )

ctx.on_record_create("posts", handler)    # Only posts collection
ctx.on_record_update(None, handler)       # All collections
ctx.on_record_delete("posts", handler)    # Only posts collection
ctx.on_any_event(handler)                 # All event types, all collections

Event object

Every handler receives an event object with these attributes:

Attribute	Type	Description
`event.type`	`EventType`	`record.created`, `record.updated`, or `record.deleted`
`event.collection_name`	`str`	Collection where the event fired
`event.record_id`	`str \| None`	ID of the affected record
`event.data`	`dict`	Full record data payload
`event.timestamp`	`str`	ISO-8601 timestamp

6. Plugin Settings

Settings are stored per-plugin in the plugin_settings database table and accessible at startup without an async DB call.

def register(ctx) -> None:
    # Read a setting (returns default if not set)
    webhook_url = ctx.get_setting("webhook_url", default="")
    max_tasks = ctx.get_setting("max_tasks_per_user", default=100)

    if webhook_url:
        # Use the setting ...
        pass

Update settings via the admin REST API:

PATCH /api/v1/admin/plugins/task-manager/settings
Authorization: Bearer ADMIN_TOKEN
Content-Type: application/json

{"webhook_url": "https://example.com/hook", "max_tasks_per_user": 50}

Read current settings:

GET /api/v1/admin/plugins/task-manager/settings
Authorization: Bearer ADMIN_TOKEN

7. Admin UI Page

Serve an admin page with your own Jinja2 template. Create a route inside register():

def register(ctx) -> None:
    from fastapi import APIRouter, Request, Depends
    from fastapi.responses import HTMLResponse
    from fastapi.templating import Jinja2Templates
    from pathlib import Path
    from app.core.dependencies import require_admin_ui

    _templates = Jinja2Templates(
        directory=str(Path(__file__).parent / "templates")
    )
    _admin = APIRouter()

    @_admin.get("/task-manager", response_class=HTMLResponse)
    async def admin_page(request: Request, user=Depends(require_admin_ui)):
        return _templates.TemplateResponse(
            "page.html",
            {"request": request, "user": user, "active": "task-manager"},
        )

    ctx.include_router(_admin, prefix="/task-manager-admin", tags=["Task Manager Admin"])

    ctx.add_admin_page(
        nav_id="task-manager",
        label="Tasks",
        icon="fa-check-square",
        url="/admin/plugins/task-manager",
    )

Create plugins/task_manager/templates/page.html extending base.html:

{% extends "base.html" %}
{% block title %}Tasks{% endblock %}

{% block content %}
<div class="p-6">
  <h1 class="text-2xl font-bold text-gray-900">Task Manager</h1>
  <p class="text-sm text-gray-500 mt-1">Manage tasks across your collections.</p>
</div>
{% endblock %}

8. Complete Plugin Structure

A full-featured plugin looks like:

plugins/
└── task_manager/
    ├── __init__.py      # REQUIRED: PLUGIN_META + register(ctx)
    ├── routes.py        # FastAPI APIRouter
    ├── models.py        # SQLAlchemy models (inherit from Base)
    ├── hooks.py         # Async event handlers
    └── templates/
        └── page.html    # Jinja2 admin page (extends base.html)

9. Testing Your Plugin

Use pytest with the client and db fixtures from tests/conftest.py:

# plugins/task_manager/tests/test_routes.py
import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_list_tasks_requires_auth(client: AsyncClient):
    resp = await client.get("/api/v1/plugins/task-manager/tasks")
    assert resp.status_code == 401


@pytest.mark.asyncio
async def test_create_task(client: AsyncClient, auth_headers: dict):
    resp = await client.post(
        "/api/v1/plugins/task-manager/tasks",
        json={"title": "Write docs"},
        headers=auth_headers,
    )
    assert resp.status_code == 201
    assert resp.json()["title"] == "Write docs"

Best Practices

Practice	Reason
Prefix table names with `plugin_{name}_`	Avoid collisions with core tables
Use `require_auth` / `require_admin` on every route	Never expose unprotected endpoints
Use SQLAlchemy ORM — never raw SQL	Prevents SQL injection
Keep `register()` fast and synchronous	It runs in the startup lifespan
Use `ctx.get_setting()` for any user-configurable value	Settings are editable without code changes
Handle exceptions in hook handlers	A crashing hook logs but doesn't affect the request

Reference: PluginContext API

ctx is a PluginContext instance passed to register(). These are all the methods available:

# Routes
ctx.include_router(router: APIRouter, *, prefix: str = "", tags=None) -> None

# Event hooks
ctx.on_record_create(collection: str | None, handler: Callable) -> None
ctx.on_record_update(collection: str | None, handler: Callable) -> None
ctx.on_record_delete(collection: str | None, handler: Callable) -> None
ctx.on_any_event(handler: Callable) -> None

# Admin UI
ctx.add_admin_page(*, nav_id: str, label: str, icon: str, url: str) -> None

# Settings (read-only during registration)
ctx.get_setting(key: str, default: Any = None) -> Any

← Plugin Overview

Plugin Development Guide

On this page