Build a CAESAR plugin

CAESAR plugins are separate Python packages that ship Legion workers without touching CAESAR's own source tree. The operator pip installs your plugin, points a few environment variables at it, and the brain starts calling it on its next restart.

If you want your worker to live inside the CAESAR repo, read Add your own tool instead — same Worker base class, but everything is wired through CAESAR's source. This page is for the out-of-tree case: contribute a worker, ship it on PyPI, and let operators choose whether to install it.

ADR-0035 covers the design decisions; this page is the operator-facing how-to.

The contract in 60 seconds

A CAESAR plugin is one Python package that declares one caesar.plugins entry point per worker it ships. The entry point points at a factory function decorated with @plugin(...). The decorator attaches a PluginManifest describing the worker; at startup Praetor reads the manifest, validates it, populates per-plugin settings from CAESAR_PLUGINS__<WORKER_ID>__* env vars, calls the factory, and the returned worker joins the in-process Legion pool. The Policy Engine gates every plugin tool through the same allowed_tools rules that gate in-tree tools — plugins do not bypass policy.

The minimum viable plugin is ~80 lines plus tests:

caesar-plugin-mqtt/
├── pyproject.toml
├── README.md
└── caesar_plugin_mqtt/
    ├── __init__.py
    └── worker.py

1. Declare the entry point

In your plugin's pyproject.toml:

[project]
name = "caesar-plugin-mqtt"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "caesar>=2.0,<3.0",
    "asyncio-mqtt>=0.16",
]

[project.entry-points."caesar.plugins"]
mqtt_publish = "caesar_plugin_mqtt:build_worker"

The entry-point name (mqtt_publish) is the worker_id Praetor will use; it must match the worker_id declared in the manifest below. The convention (not enforced) is that the distribution name is caesar-plugin-<name> for discoverability.

2. Write the factory + worker

# caesar_plugin_mqtt/worker.py
from typing import ClassVar

import asyncio_mqtt as mqtt
from pydantic import BaseModel, SecretStr

from caesar.bus.client import Bus
from caesar.legion.protocol import TaskDispatch
from caesar.legion.worker import Worker
from caesar.llm.gateway import ToolDefinition
from caesar.plugins import PluginContext, plugin


CAPABILITY = "mqtt_publish"


class MqttSettings(BaseModel):
    broker_host: str = "localhost"
    broker_port: int = 1883
    username: str | None = None
    password: SecretStr | None = None
    topic_prefix: str = "caesar/"


MQTT_PUBLISH_TOOL = ToolDefinition(
    name=CAPABILITY,
    description=(
        "Publish a short message to an MQTT topic. Use when the user "
        "asks the brain to relay a message to a device or service "
        "subscribed via MQTT."
    ),
    input_schema={
        "type": "object",
        "properties": {
            "topic": {
                "type": "string",
                "description": "MQTT topic suffix; the configured prefix is prepended.",
            },
            "payload": {"type": "string"},
        },
        "required": ["topic", "payload"],
    },
)


class MqttPublishWorker(Worker):
    """Publishes a UTF-8 string payload to a topic under the configured prefix."""

    worker_id: ClassVar[str] = "mqtt_publish"
    capabilities: ClassVar[list[str]] = [CAPABILITY]
    version: ClassVar[str] = "0.1.0"

    def __init__(self, bus: Bus, *, settings: MqttSettings) -> None:
        super().__init__(bus)
        self._settings = settings

    async def handle(self, task: TaskDispatch) -> dict[str, object]:
        topic = str(task.payload.get("topic", "")).strip("/")
        payload = str(task.payload.get("payload", ""))
        if not topic or not payload:
            raise ValueError("mqtt_publish requires non-empty 'topic' and 'payload'")
        full_topic = f"{self._settings.topic_prefix.rstrip('/')}/{topic}"
        password = (
            self._settings.password.get_secret_value()
            if self._settings.password is not None
            else None
        )
        async with mqtt.Client(
            hostname=self._settings.broker_host,
            port=self._settings.broker_port,
            username=self._settings.username,
            password=password,
        ) as client:
            await client.publish(full_topic, payload=payload.encode("utf-8"))
        return {"published": True, "topic": full_topic, "bytes": len(payload)}


@plugin(
    worker_id="mqtt_publish",
    capabilities=[CAPABILITY],
    version="0.1.0",
    requires_caesar=">=2.0,<3.0",
    settings_model=MqttSettings,
    tool_definitions=[MQTT_PUBLISH_TOOL],
)
def build_worker(ctx: PluginContext) -> Worker:
    assert isinstance(ctx.plugin_settings, MqttSettings)
    return MqttPublishWorker(ctx.bus, settings=ctx.plugin_settings)

caesar_plugin_mqtt/__init__.py simply re-exports the factory:

from caesar_plugin_mqtt.worker import build_worker

__all__ = ["build_worker"]

3. The manifest fields

The @plugin(...) decorator validates these at import time:

Field	Required	Purpose
worker_id	Yes	Unique id; must match the entry-point name.
capabilities	Yes	Strings the policy engine + registry use to route work.
version	Yes	Semver of the plugin worker itself.
requires_caesar	Yes	PEP 440 specifier (e.g. ">=2.0,<3.0"). Loader rejects mismatches.
settings_model	Optional	Pydantic model describing the plugin's .env-driven configuration.
tool_definitions	Optional	ToolDefinitions the brain will advertise to the LLM.

Every ToolDefinition.name must appear in capabilities — the loader enforces this so dispatch (capability lookup) and policy (allowed_tools match) share one identifier with no provenance bookkeeping.

4. PluginContext

The factory receives a PluginContext with:

@dataclass(frozen=True)
class PluginContext:
    bus: Bus                        # Legion NATS client
    audit: AuditLogger              # write rows to audit_log
    gateway: LLMGateway             # in case the plugin needs the LLM
    settings: CaesarSettings        # full settings tree (read-only)
    plugin_settings: BaseModel | None
    worker_id: str

plugin_settings is the populated, validated instance of the plugin's settings_model, or None if the plugin didn't declare one. The plugin should never read os.environ directly — config goes through the model so operators have one consistent mental model (CAESAR_PLUGINS__<WORKER_ID>__<FIELD>).

5. Settings injection

Each top-level field on the plugin's settings_model maps to an env var with the prefix CAESAR_PLUGINS__<WORKER_ID>__:

# Maps to MqttSettings(broker_host="mqtt.lan", broker_port=8883, ...)
export CAESAR_PLUGINS__MQTT_PUBLISH__BROKER_HOST=mqtt.lan
export CAESAR_PLUGINS__MQTT_PUBLISH__BROKER_PORT=8883
export CAESAR_PLUGINS__MQTT_PUBLISH__USERNAME=brain
export CAESAR_PLUGINS__MQTT_PUBLISH__PASSWORD=hunter2
export CAESAR_PLUGINS__MQTT_PUBLISH__TOPIC_PREFIX=caesar/

Nested fields use __ as the separator (the same convention every other CAESAR setting uses). Pydantic does the parsing and validation; a missing-required field, a malformed integer, or a failed validator is logged + audited as plugin.load.rejected with reason=settings_error and the plugin is skipped — Praetor keeps booting.

6. Policy: nothing changes

Plugin capabilities flow through the same allowed_tools path as in-tree tools. To enable the MQTT plugin, the operator adds:

# policy.yaml
allowed_tools:
  - mqtt_publish

…and that's it. The Policy Engine has no plugin-vs-in-tree branch and no allowed_plugins knob — there is one source of truth for what CAESAR is allowed to do.

When the LLM emits a tool_use named mqtt_publish, the brain graph constructs a GenericToolCall(tool="mqtt_publish", input=...) and calls Policy.evaluate(...). A deny produces an is_error ToolResult for the LLM and a tool.denied audit row; an allow dispatches through the registry to the plugin worker and produces a tool.called row. The shapes match exactly what the in-tree calculator, web_search, etc. produce.

Per-tool input constraints (e.g. domain_allowlist for web_search) work for plugins too — they're matched in caesar/policy/allowlist.py::_input_matches. If your plugin introduces a new constraint kind, ship that branch upstream in a PR against the policy module.

7. Operator install + controls

Once your plugin is on PyPI, operators install it like any other Python dependency:

pip install caesar-plugin-mqtt

Two opt-out knobs sit on the loader:

# Disable plugin discovery entirely (default: true).
CAESAR_PLUGINS__ENABLED=false

# Only load plugins from these distributions (default: any installed).
CAESAR_PLUGINS__ALLOWLIST='["caesar-plugin-mqtt","caesar-plugin-tts"]'

The allowlist is a defence-in-depth posture for operators who want explicit control over which pip installs expose a worker. It matches against the distribution name, not the entry-point name, so pip install caesar-plugin-mqtt is the unit of trust.

8. Audit events the loader emits

Every plugin load attempt writes one row:

• plugin.load.succeeded — plugin loaded and started.
• plugin.load.rejected — plugin failed validation; the payload.reason carries one of these stable codes:

Reason code	Meaning
not_in_allowlist	Distribution not in CAESAR_PLUGINS__ALLOWLIST.
import_error	The entry point's target couldn't be imported.
missing_manifest	Factory lacks the @plugin(...) decorator.
entry_point_name_mismatch	Entry-point name != manifest worker_id.
invalid_requires_caesar	Manifest requires_caesar isn't a valid PEP 440 specifier.
version_mismatch	Running CAESAR version doesn't satisfy requires_caesar.
worker_id_collision	Another in-tree worker or plugin already owns this worker_id.
capability_collision	Another worker already claims one of the plugin's capabilities.
settings_error	settings_model validation failed against the env vars.
factory_error	The factory raised an unexpected exception.
identity_mismatch	Returned worker's worker_id / capabilities != manifest.

Operators can grep for these in the dashboard audit log or with SQL:

SELECT
  json_extract(payload, '$.entry_point') AS plugin,
  json_extract(payload, '$.reason')      AS reason,
  json_extract(payload, '$.detail')      AS detail
FROM audit_log
WHERE event_type = 'plugin.load.rejected'
ORDER BY id DESC;

9. Versioning + requires_caesar

Pin requires_caesar to a major range — ">=2.0,<3.0" is the pattern. When CAESAR ships a major version bump, the plugin loader will refuse to load any plugin whose constraint doesn't admit the new version, log + audit version_mismatch, and Praetor keeps booting. Plugin authors keep their plugin compatible by either updating the constraint after testing, or by publishing a new major version of the plugin.

The loader allows pre-releases by default (so a plugin pinned to ">=2.0" will load on 2.0.0a1).

10. Testing your plugin

The same three-layer pattern that Add your own tool recommends works here. For the loader-integration layer, the relevant fixtures are in tests/_plugin_fixtures.py and tests/test_plugins_loader.py in the CAESAR repo — open them and copy the pattern that fits.

Two plugin-specific test recipes:

# Unit-test that the @plugin(...) decorator attached a manifest.
from caesar.plugins import get_manifest
from caesar_plugin_mqtt import build_worker

def test_manifest():
    manifest = get_manifest(build_worker)
    assert manifest is not None
    assert manifest.worker_id == "mqtt_publish"
    assert "mqtt_publish" in manifest.capabilities

# Integration-test the loader picks up the plugin and the LLM sees the tool.
from importlib.metadata import EntryPoint
from caesar.plugins.loader import load_plugins_with_outcomes

async def test_loads_via_loader(bus, audit, gateway, settings):
    entries = [
        EntryPoint(
            name="mqtt_publish",
            value="caesar_plugin_mqtt:build_worker",
            group="caesar.plugins",
        ),
    ]
    outcomes = await load_plugins_with_outcomes(
        bus=bus,
        audit=audit,
        gateway=gateway,
        settings=settings,
        entry_points_iter=entries,
    )
    assert outcomes[0].reason is None
    assert outcomes[0].worker is not None

For a CI build that doesn't have nats-server on PATH, construct the Bus without connecting — the loader never calls bus methods itself, only the factory does (and only after worker.start()).

11. Security posture

Plugins run same-process with Praetor. A plugin can read every file Praetor can, hit every network endpoint Praetor can, and write to every collaborator (bus, audit, gateway) that Praetor passes through PluginContext. The trust posture is identical to adding a Python dependency to your CAESAR install: pip install caesar-plugin-foo is the trust act, not the load event.

Concrete consequences:

• Read pyproject.tomls before installing. Plugin authors are expected to be honest about their dependency surface.
• Use Sigstore's gh attestation verify for plugins published with PyPI trusted-publisher attestations. CAESAR itself publishes attestations (SR-011 follow-up); plugin authors are encouraged to do the same.
• The Policy Engine is your last line. Even if a plugin is hostile, it cannot invoke a tool that isn't in allowed_tools. Keep your allow-list tight.
• Operators paranoid about which pip install surfaces a worker can pin CAESAR_PLUGINS__ALLOWLIST to a known set.

Subprocess- or container-isolated plugins were rejected for v2.0 (see ADR-0035 §Alternatives) — the operational cost is significant for a single-operator homelab and pip install itself isn't sandboxed. That's a v3.x conversation if the threat model ever justifies it.

When not to write a plugin

• Your worker is broadly useful. Submit it as a built-in Legion worker via a PR against caesar/legion/* — Add your own tool covers that path. The in-tree set is the default toolbox every CAESAR install has.
• You need cross-process isolation. Plugins are same-process. If your worker must run as a different user, with a different filesystem view, or in a different container, point a separate worker process at Praetor's NATS bus and authenticate it with an NKEY (ADR-0027). You don't need the plugin loader for that.
• You want to override an in-tree tool. The loader refuses worker_id or capability collisions on purpose. If you need to change web_search's behaviour, fork the in-tree worker or contribute upstream.

References

• ADR-0035 — the design this page implements.
• ADR-0017 — the CAESAR_<...>__<...>__<...> env-var convention plugin settings reuse.
• ADR-0028 — the generalised GenericToolCall policy plugin capabilities flow through.
• Add your own tool — the in-tree alternative to writing a plugin.
• Run a worker on another box — when you want a cross-process worker without the plugin loader.
• SECURITY-MODEL.md — the trust posture plugins inherit from.