Writing plugins¶

A plugin is a container that serves the routes core calls. With the SDK, a plugin is a manifest, a settings class and one handler per kind, given to create_plugin_app. This page shows each kind, then settings, the Dockerfile and the tests.

Creating a plugin¶

spot-cli plugin new <id> creates the plugin's repository from spot-plugin-template, at the version your SPOT release pins, in the current directory. It needs uv and git. Without the wrapper, the command it runs is:

uvx --from copier==9.19.1 copier copy --vcs-ref v1.0.0 \
    --data plugin_id=analyzer-url https://codeberg.org/SPOT_Project/spot-plugin-template.git analyzer-url

copier asks five things, and a sixth, renovate_day, when spot_project is yes; --data NAME=VALUE answers one ahead and --defaults takes the default of the others:

• plugin_id: the plugin's id, which names its container spot-<id>.
• kind: analyzer, context_provider, mail_retriever or responder.
• description: what the plugin does, in one line.
• egress: none, indicators or content, as the manifest declares it.
• spot_project: whether the plugin is one of the SPOT Project's, which adds its pipeline and policy files.

Then cd analyzer-url && git init -b main && uv lock && make test. The plugin starts with one example check, to replace with its own. To bring a later version of the template into the plugin, or to adopt the template in an existing plugin, follow the template's README.

Layout¶

Every plugin made from the template has this layout, shown here for an analyzer. The other kinds replace src/services/ and the tests of the handler with their own (src/sync.py, src/respond.py or src/poll.py).

analyzer-url/
├── .copier-answers.yml   # the template's version and the answers
├── .dockerignore
├── .gitignore
├── CHANGELOG.md
├── README.md
├── docker-compose.yml    # runs the plugin on the spot-plugins network
├── pyproject.toml        # spot-sdk-python==2.1.0b3, uvicorn; the testing extra for the tests
├── uv.lock               # written by uv lock
├── Dockerfile            # the spot.plugin.* labels of the manifest
├── Makefile              # test, test:coverage, test:plugin, quality
├── src/
│   ├── __init__.py
│   ├── config.py         # Settings(PluginSettings)
│   ├── manifest.py       # MANIFEST
│   ├── main.py           # app = create_plugin_app(...)
│   └── services/
│       ├── __init__.py
│       └── analysis.py   # the handler of an analyzer
└── tests/
    ├── emls/
    │   ├── corpus.yaml   # optional: settings, expectations
    │   ├── benign/*.eml
    │   └── phish/*.eml
    ├── __init__.py
    ├── conftest.py       # the app as core calls it, the handler on a view
    ├── test_analysis.py  # the example check
    ├── test_emls.py      # corpus.yaml against the messages
    └── test_main.py      # the manifest, the labels, the settings

tests/emls/ may also hold a corpus.yaml, which testing the image reads.

The SDK does not serve the app: the template adds uvicorn to the plugin's dependencies and the Dockerfile runs uvicorn src.main:app --host 0.0.0.0 --port 8000.

The app¶

def create_plugin_app(
    manifest: Manifest,
    settings_cls: type[PluginSettings],
    *,
    analyze=None,
    respond=None,
    sync=None,
    lifespan=None,
    health=None,
) -> FastAPI: ...

• Each kind the manifest declares needs its handler (analyze for an analyzer, respond for a responder, sync for a context provider; a mail retriever has none), and each handler its kind: create_plugin_app raises ValueError otherwise.
• The settings are read once, when the app is created: a missing required setting stops the plugin at startup. Core restarts the plugin when its settings change.
• GET /health and GET /manifest always exist. POST /internal/analyze, POST /internal/respond, and POST /internal/sync with GET /internal/state exist for the handlers given. /health reports the plugin as healthy, unless a health check says otherwise (see Reporting readiness).
• Every /internal/* route checks Authorization: Bearer <token> against the plugin's own token, in constant time, before it reads the request: a missing or wrong token is 401, and so is every call when the plugin has no token, which the app warns about at startup.
• A handler may be async def (or an object whose __call__ is async def), which runs on the event loop, or a plain def, which runs in a worker thread. Use def for blocking work (a model's inference, a blocking library) and async def with the SDK's async clients.
• When a handler raises, the call is answered 500 ({"detail": "analysis failed"}) and core records the analyzer as unknown. The log line names the job, the exception's type and where it was raised, never its message, which may quote the email. When the plugin knows why it cannot decide, it should answer unknown itself, with a summary.
• A request that does not validate is answered 422 with the fields refused, never the values sent.
• Load heavy resources (a model) once per process: at startup in a lifespan (an async context manager, given to FastAPI as is), at import time, or on first use with functools.cache. A model plugin gives load, which loads in the background and reports it in /health (see Model plugins).
• The app is a FastAPI app; routes you add to it yourself are not token-checked.

Reporting readiness¶

A plugin that loads a model in the background is not ready at once: until the model is loaded, or for good if loading failed, its analyses can only answer unknown. A health check reports that state in GET /health, so core's plugin status and the dashboard show it. create_plugin_app's load does all of this for a model plugin (see Model plugins); the check below is how a plugin reports a state of its own.

import asyncio
import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI
from spot_sdk import PluginHealth

logger = logging.getLogger(__name__)
classifier: Classifier | None = None
failed = False


async def load() -> None:
    global classifier, failed
    try:
        classifier = await asyncio.to_thread(Classifier.load, "/models/distilbert")
    except Exception as error:
        logger.error("The model did not load: %s", type(error).__name__)
        failed = True


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    loading = asyncio.create_task(load())  # the app answers /health meanwhile
    yield
    loading.cancel()


async def health() -> PluginHealth:  # reads the state load() keeps
    if classifier is not None:
        return PluginHealth(status="healthy")
    if failed:
        return PluginHealth(status="unhealthy", reason="the model did not load")
    return PluginHealth(status="starting", reason="loading the model")


def analyze(request: AnalyzeRequest, settings: Settings) -> AnalyzerVerdict:
    if classifier is None:
        return AnalyzerVerdict(
            verdict=VerdictClass.UNKNOWN,
            score=0,
            confidence=0,
            summary="The model is not loaded",
        )
    probability = classifier.classify(request.email.visible_text)
    ...


app = create_plugin_app(
    MANIFEST, Settings, analyze=analyze, lifespan=lifespan, health=health
)

• GET /health calls the check on each request, and always answers 200, so that liveness checks pass while the plugin loads: {"status": "starting", "version": "1.0.0", "reason": "loading the model"}. status is healthy, starting or unhealthy; without a health check, it is healthy.
• /health is open to anyone who reaches the plugin: the reason is a few words of the plugin's own, at most 200 characters without control characters, never an email's content, a setting or an exception's message. Core's logs and the dashboard show it.
• Prefer an async def check that reads the state the plugin keeps, as above, and returns at once. A def check runs in the thread pool the handlers run in, so under load it waits behind them, and /health answers late.
• Report the conditions you expect, such as starting or "the model did not load", as a PluginHealth: never raise them. A check that raises, or returns no valid PluginHealth, answers unhealthy with the reason the health check failed, and logs an ERROR line, by the exception's type and place only, on every call.

Writing an analyzer¶

The handler gets an AnalyzeRequest and the settings, and answers an AnalyzerVerdict.

from spot_sdk import (
    AnalyzerVerdict,
    AnalyzeRequest,
    Egress,
    Manifest,
    PluginKind,
    PluginSettings,
    Signal,
    VerdictClass,
    create_plugin_app,
)
from spot_sdk.signals import ANOMALY_SIGNALS, HEADER_REPLY_TO_MISMATCH

MANIFEST = Manifest(
    id="analyzer-rules",
    version="1.0.0",
    kind=PluginKind.ANALYZER,
    egress=Egress.NONE,
)


class Settings(PluginSettings):
    anomaly_weight: int = 5


def analyze(request: AnalyzeRequest, settings: Settings) -> AnalyzerVerdict:
    email = request.email
    signals = [
        Signal(
            name=ANOMALY_SIGNALS[anomaly.code],
            weight=settings.anomaly_weight,
            evidence=anomaly.detail or "",
            location=anomaly.location,
        )
        for anomaly in email.anomalies
    ]
    sender = email.from_.domain if email.from_ else ""
    for reply_to in email.reply_to:
        if sender and reply_to.domain != sender:
            signals.append(
                Signal(
                    name=HEADER_REPLY_TO_MISMATCH,
                    weight=40,
                    evidence=f"From {sender}, Reply-To {reply_to.domain}",
                    location="header:Reply-To",
                )
            )
    score = min(100, sum(max(signal.weight, 0) for signal in signals))
    verdict = VerdictClass.SUSPICIOUS if score >= 50 else VerdictClass.BENIGN
    return AnalyzerVerdict(verdict=verdict, score=score, confidence=70, signals=signals)


app = create_plugin_app(MANIFEST, Settings, analyze=analyze)

This example keeps everything in one module. A plugin made from the template splits it: MANIFEST in src/manifest.py, Settings in src/config.py, the handler in src/services/analysis.py and the app in src/main.py.

What the request holds:

• request.email is the parsed message (see the email view). Read visible_text rather than html, authentication rather than the raw headers, and urls rather than searching the bodies yourself: the parser has already dealt with encodings, hidden text and forged headers.
• visible_text comes without the characters that render as nothing (zero-width spaces and joiners, soft hyphens, tags), which split words a reader sees whole; invisible_characters notes that some were removed. The subject, the headers, display names, anchor text and file names keep them, as written, and invisible_characters notes the ones that hold some, with bidi_control for bidi overrides (the right-to-left-override file name): call spot_sdk.mime.visible on them before matching words, rather than keeping your own table.
• request.previous holds the outcome of the earlier stages, by stage name.
• request.retrieval_limits caps Knowledge Store queries; KnowledgeClient.for_analysis(request) applies them.
• request.deadline_ms is what is left of the analyzer's timeout.

Signal names are dotted and lower-case. Use the constants of spot_sdk.signals for the official names, and your own prefix for the others. A positive weight pushes the score up; a negative one is evidence the email is legitimate.

Fetching the message or a part¶

The view holds the metadata of attachments, not their content. An analyzer with the content:read scope fetches what it needs from the gateway:

from spot_sdk import SpotClient

SCANNED_TYPES = {"application/pdf", "text/html"}
MAX_SCANNED_BYTES = 5_000_000
MAX_FETCHES = 3


async def analyze(request: AnalyzeRequest, settings: Settings) -> AnalyzerVerdict:
    spot = SpotClient.from_settings(settings)
    wanted = [
        part
        for part in request.email.attachments
        if part.sha256 is not None  # an attached message: its leaves are in parts
        and part.content_type in SCANNED_TYPES
        and part.size_bytes <= MAX_SCANNED_BYTES
    ]
    for part in wanted[:MAX_FETCHES]:
        data, content_type = await spot.part(request.email.id, part.part_id)
        ...

Fetch only what the analyzer reads, and a few parts at most: every fetch makes the gateway read the stored message and parse it again.

An attached message (message/rfc822) is a container, like a multipart part: it has no content of its own (sha256 is None), and its leaves (1.2.1, ...) are fetched one by one. SpotClient.raw(email_id) returns the whole message as received.

Reading the Knowledge Store¶

from spot_sdk import KnowledgeClient


async def analyze(request: AnalyzeRequest, settings: Settings) -> AnalyzerVerdict:
    sender = request.email.from_.display_name if request.email.from_ else ""
    kb = KnowledgeClient.for_analysis(request)
    executives = await kb.fetch(tags="employee+executive", text=sender, top_k=3)
    ...

Asking a language model¶

LLMClient speaks the OpenAI chat completions API, as Ollama, vLLM and llama.cpp do. The email always goes in its own user message, between delimiters it cannot forge, and the system prompt says it is untrusted data. With schema, the answer is validated against a Pydantic model.

The email data is cut to max_bytes bytes of UTF-8 (8 000 by default), so a hostile message cannot overflow the model's context. A tokenizer never makes more tokens than bytes, so the email costs at most about max_bytes tokens, typically a quarter of that: set the server's context (num_ctx for Ollama) to at least max_bytes plus the system prompt plus max_tokens plus about 512 tokens (the SDK's notice, the delimiters and the chat template), and no email can overflow it. Tool results are not counted: bound what your tools return.

from typing import Annotated, Literal

import httpx2
from pydantic import BaseModel, Field
from spot_sdk import LLMClient, LLMOutputError
from spot_sdk.signals import LLM_PREFIX


class Settings(PluginSettings):
    llm_url: str = "http://ollama:11434/v1"
    llm_model: str = "llama3.1:8b"


class Assessment(BaseModel):
    verdict: Literal["malicious", "suspicious", "spam", "benign"]
    score: int = Field(ge=0, le=100)
    indicators: list[Annotated[str, Field(pattern=r"^[a-z0-9_]+$")]]
    reasoning: str


SYSTEM = "You are a phishing analyst. Assess the email."


async def analyze(request: AnalyzeRequest, settings: Settings) -> AnalyzerVerdict:
    llm = LLMClient(
        settings.llm_url,
        settings.llm_model,
        options={"temperature": 0},
        timeout=request.deadline_ms / 1000,  # core stops waiting then anyway
    )
    try:
        assessment = await llm.complete(SYSTEM, request.email, schema=Assessment)
    except (LLMOutputError, httpx2.HTTPError) as error:
        return AnalyzerVerdict(
            verdict=VerdictClass.UNKNOWN,
            score=0,
            confidence=0,
            summary=f"The model gave no usable answer ({type(error).__name__})",
        )
    return AnalyzerVerdict(
        verdict=VerdictClass(assessment.verdict),
        score=assessment.score,
        confidence=60,
        signals=[
            Signal(name=LLM_PREFIX + indicator, weight=10)
            for indicator in assessment.indicators
        ],
        summary=assessment.reasoning,
    )

With tools=[...] (OpenAI function definitions), a model may answer with tool calls instead: complete then returns LLMToolCalls for the analyzer to run. It never calls the model again on its own: the analyzer runs the calls and sends their results back with tool_rounds, one LLMToolRound per round, oldest first, until the model answers. Bound the rounds:

from spot_sdk import LLMToolCalls, LLMToolRound

TOOLS = [LOOKUP_DOMAIN]  # OpenAI function definitions


async def assess(llm: LLMClient, email: EmailView) -> Assessment | LLMToolCalls:
    rounds: list[LLMToolRound] = []
    answer = await llm.complete(SYSTEM, email, schema=Assessment, tools=TOOLS)
    while isinstance(answer, LLMToolCalls) and len(rounds) < 3:
        results = {call.id: await run_tool(call) for call in answer.calls}
        rounds.append(LLMToolRound(calls=answer.calls, results=results))
        answer = await llm.complete(
            SYSTEM, email, schema=Assessment, tools=TOOLS, tool_rounds=rounds
        )
    return answer

Each round needs one result per call, by call id: LLMToolRound refuses a result for no call of its round, and a call without a result. Tool output often quotes the email (a domain, a URL), so the client treats it like the email: each result sits between delimiter lines the result cannot forge, and the system prompt says results are untrusted data too. Never put tool output in the system prompt yourself.

Model plugins¶

A model plugin answers with a model's phishing probability, as analyzer-nlp and analyzer-ml do. The SDK holds what they share: the background load and its readiness, the verdict of a probability, the model's files from a URL, the URL as a log line may show it, and the text the model reads with the manifest its files carry.

from pathlib import Path

from spot_sdk import AnalyzeRequest, AnalyzerVerdict, PluginSettings, create_plugin_app
from spot_sdk.artifacts import fetch
from spot_sdk.model_input import ModelManifest, model_text, read_manifest
from spot_sdk.signals import ML_PHISHING_PROBABILITY


class Settings(PluginSettings):
    # A local directory, or the http(s) URL of a .tar.gz, .tgz, .tar or .zip
    models_dir: str = "/app/saved_models"


class Model:
    def __init__(self, directory: Path) -> None:
        self.manifest: ModelManifest = read_manifest(directory)  # spot-model.json
        self.classifier = Classifier.read(directory)


def load(settings: Settings) -> Model:
    return Model(fetch(settings.models_dir))


def analyze(
    request: AnalyzeRequest, settings: Settings, model: Model
) -> AnalyzerVerdict:
    email = request.email
    text = model_text(email.subject, email.visible_text, model.manifest.text)
    return AnalyzerVerdict.from_probability(
        model.classifier.phishing_probability(text),
        signal=ML_PHISHING_PROBABILITY,
        location="visible_text",
        model_version=model.manifest.label,
    )


app = create_plugin_app(MANIFEST, Settings, load=load, analyze=analyze)

• load(settings) runs once the app starts, after the plugin's own lifespan startup (its logging set up, say), in a daemon thread: the app answers meanwhile. Until it returns, /health answers starting (loading) and /internal/analyze answers unknown. When it raises, the plugin is unhealthy for good, with the exception's type as the reason (the load failed (FetchError)), and the failure is logged as a handler's, by its type and place: fixing the setting restarts the plugin. Then analyze gets what load returned as its third argument, and a health check, if given, reports the plugin's state.
• AnalyzerVerdict.from_probability(p, signal=...) makes the verdict: the score is p in percent, the class follows VerdictPolicy's default thresholds, the confidence is max(p, 1 - p), and one signal carries the probability. A probability out of 0-1, or NaN, gives unknown. AnalyzerVerdict.unknown(reason) is the answer when the plugin cannot decide (No verdict: the message has no text.).
• model_text(subject, body, version) is the text the model reads. Training and inference must call this same function: a model trained on one text and served another one drifts, and the SDK is where both get it. The version is the model's manifest.text, the normalisation it was trained with; a released one never changes, another gets a new name, and an unknown one raises ValueError.
• spot-model.json sits in the model's directory (read_manifest, write_manifest, ModelManifest). Its fields: name (lowercase letters, digits and hyphens), version (up to 64 characters of letters, digits, ., _, + and -), text (the normalisation, spot-text-1 by default) and max_tokens (1 to 512, or absent). manifest.label is "<name> <version>", what the verdict gives as model_version. A missing, unreadable or invalid file raises ValueError naming it, with the failing fields.
• fetch(value) returns a local path as it is, and for an http(s) URL the directory its archive is extracted into, downloaded once into a cache. It refuses a URL with credentials (give access with a presigned URL), a member that would land outside the directory, and a link. Its errors and log lines name the URL as redact shows it. A Hugging Face id is a local path to fetch: a plugin that takes them tells them apart first.
• redact(url) keeps a URL's scheme, host, port and path, and refuses with ValueError a value it cannot show safely (an @ anywhere but in the userinfo, a % in the host, a port that is not a number). Show what it returns, never the URL itself, in a log line, an error or a verdict's details: details={"model_url": redact(url)}, for a URL only, as a local path says how the plugin's files are laid out.

Writing a context provider¶

A context provider keeps the Knowledge Store up to date with company data. Its sync handler writes the documents and returns how many.

import json
from pathlib import Path

from spot_sdk import (
    Egress,
    KnowledgeClient,
    KnowledgeDocument,
    KnowledgeTag,
    Manifest,
    PluginKind,
    PluginSettings,
    SettingField,
    SettingType,
    create_plugin_app,
)

MANIFEST = Manifest(
    id="provider-employee-dir",
    version="2.0.0",
    kind=PluginKind.CONTEXT_PROVIDER,
    egress=Egress.NONE,
    settings={
        "EMPLOYEES_FILE": SettingField(
            type=SettingType.STRING,
            default="/data/employees.json",
            description="The employee directory, as JSON",
        ),
    },
)


class Settings(PluginSettings):
    employees_file: Path = Path("/data/employees.json")


async def sync(settings: Settings) -> int:
    employees = json.loads(settings.employees_file.read_text())
    documents = [
        KnowledgeDocument(
            id=f"employee:{employee['email']}",
            content=f"{employee['name']}, {employee['title']}",
            tags=[KnowledgeTag.EMPLOYEE],
            metadata={"email": employee["email"], "title": employee["title"]},
        )
        for employee in employees
    ]
    await KnowledgeClient.from_settings(settings).replace_source(documents)
    return len(documents)


app = create_plugin_app(MANIFEST, Settings, sync=sync)

• replace_source makes the documents the provider's whole set: the store deletes the provider's other documents, so a departed employee disappears. The source is the plugin's id (SPOT_PLUGIN_ID), and the store refuses documents of another source.
• POST /internal/sync answers 202 at once with the SyncState and runs the handler in the background. While a sync runs, another call starts nothing and answers 202 with the running sync's state.
• GET /internal/state reports in_progress, last_sync_at (when the last sync started), last_finished_at, last_sync_count (the count the last successful sync returned) and last_error, cleared when the next sync starts.
• A failed sync sets last_error to the exception's type only, and is logged by its type and where it was raised: its message may hold a directory URL with credentials.
• The operator sets how often core calls the provider (sync_schedule in spot.yaml).

Writing a mail retriever¶

A mail retriever has no handler: it runs its own loop, started by the app's lifespan, and pushes each message to the gateway with SpotClient.ingest. It still serves /health and /manifest, for core and the catalog.

A message is deleted only once SPOT has stored it, or has refused it for good (400, 413, 422). Any other error passes: a 401 or 403 (a rotated token), a 429, a 5xx, or SPOT unreachable. The message is then kept for the next pass, and the log names the status or the error type, never the message. Any other failure ends the loop: the plugin logs its type and stops, rather than answer /health with no loop running.

import asyncio
import logging
import os
import signal
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from pathlib import Path

import httpx2
from fastapi import FastAPI
from spot_sdk import (
    Egress,
    Manifest,
    PluginKind,
    PluginSettings,
    SpotAPIError,
    SpotClient,
    create_plugin_app,
)

logger = logging.getLogger(__name__)

MANIFEST = Manifest(
    id="retriever-maildir",
    version="1.0.0",
    kind=PluginKind.MAIL_RETRIEVER,
    egress=Egress.NONE,
)
REFUSED_FOR_GOOD = {400, 413, 422}


class Settings(PluginSettings):
    maildir: Path = Path("/var/mail/new")
    poll_interval_s: float = 30.0


async def poll(settings: Settings) -> None:
    spot = SpotClient.from_settings(settings)
    while True:
        for path in sorted(settings.maildir.iterdir()):
            try:
                await spot.ingest(path.read_bytes())
            except SpotAPIError as error:
                logger.warning("ingest answered HTTP %s", error.status_code)
                if error.status_code not in REFUSED_FOR_GOOD:
                    break  # keep the message for the next pass
            except httpx2.TransportError as error:
                logger.warning("SPOT is unreachable: %s", type(error).__name__)
                break
            path.unlink()
        await asyncio.sleep(settings.poll_interval_s)


def stop_on_failure(task: asyncio.Task[None]) -> None:
    error = None if task.cancelled() else task.exception()
    if error is not None:
        logger.critical("The poll loop failed: %s", type(error).__name__)
        os.kill(os.getpid(), signal.SIGTERM)  # stop the server: the container restarts


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    task = asyncio.create_task(poll(Settings()))
    task.add_done_callback(stop_on_failure)
    yield
    task.cancel()


app = create_plugin_app(MANIFEST, Settings, lifespan=lifespan)

A filter in the mail path (a milter) waits for the verdict:

result = await spot.ingest(
    raw, envelope=envelope, wait_for_verdict=True, timeout_ms=20_000
)
if result.status is IngestionStatus.TIMEOUT:
    ...  # apply the plugin's own fail-open or fail-closed policy

The call then waits timeout_ms on top of the client's own timeout.

Writing a responder¶

A responder acts on the verdicts its workflows send it. It acts only in enforce mode: in observe mode (the default) it records what it would have done.

Core calls a responder at least once per job, not exactly once. It retries a call that timed out, while the first one may still complete, since the handler runs inside the request; and when core restarts between its calls and its acknowledgement of the job's result, it calls every responder of that job again. So a responder acts once per request.job_id, either way:

• it keeps the job ids it has acted on, checked and recorded under a lock so that a repeated call waits for the first one, and in a store that outlives the plugin's restarts;
• or its action is idempotent: a ticket keyed by the job id, a message already in quarantine left there.

import asyncio
import logging

from spot_sdk import (
    Egress,
    Manifest,
    PluginKind,
    PluginSettings,
    RespondRequest,
    SpotClient,
    create_plugin_app,
)

logger = logging.getLogger(__name__)

MANIFEST = Manifest(
    id="responder-quarantine",
    version="1.0.0",
    kind=PluginKind.RESPONDER,
    egress=Egress.NONE,
)
handled: set[str] = set()  # the jobs acted on: keep them in a lasting store
acting = asyncio.Lock()


async def respond(request: RespondRequest, settings: PluginSettings) -> None:
    if request.mode != "enforce":
        logger.info("job %s: would quarantine (observe mode)", request.job_id)
        return
    async with acting:  # a repeated call waits for the first one
        if request.job_id in handled:
            return  # core called again for this job: it is done
        result = await SpotClient.from_settings(settings).result(request.job_id)
        ...  # quarantine request.email_id, using result.signals
        handled.add(request.job_id)


app = create_plugin_app(MANIFEST, PluginSettings, respond=respond)

The example shows the check, not all a real responder needs:

• the set and the lock live in memory, in one process: a real responder keeps the handled job ids in a lasting store that its replicas share;
• the one lock serialises every job's response: keep the action quick, or lock per job_id;
• a crash between acting and recording the job id acts again when core calls again: only an idempotent action, such as a ticket keyed by job_id, covers that.

request.signals are the workflow's signals, each with the analyzer that emitted it. SpotClient.result(job_id) (scope results) returns the whole OrchestrationResult, and the content routes return the message.

Settings and secrets¶

A plugin declares its settings twice: in the manifest, so the dashboard renders a form, and in its PluginSettings subclass, so the plugin reads them.

from pydantic import SecretStr
from spot_sdk import Manifest, PluginKind, PluginSettings, SettingField, SettingType

MANIFEST = Manifest(
    id="analyzer-misp",
    version="2.0.0",
    kind=PluginKind.ANALYZER,
    settings={
        "MISP_URL": SettingField(type=SettingType.URL, required=True),
        "MISP_TOKEN": SettingField(type=SettingType.SECRET, secret=True),
    },
)


class Settings(PluginSettings):
    misp_url: str
    misp_token: SecretStr

Core starts the container with:

PluginSettings reads, in order of precedence: the constructor, the files of /run/secrets, the environment, then a dotenv file when the subclass sets env_file. Names are case-insensitive, so the file MISP_TOKEN fills the field misp_token. /run/secrets is read only when it exists, so the plugin runs outside a container without warnings. A validation error names the fields, never the values read, which may be secrets. Keep secrets in SecretStr fields, so they do not show in logs or reprs.

Saving a plugin's settings in SPOT restarts the plugin, which then reads the new values.

State a setting's bounds in both places too: minimum and maximum (an integer or a number) and pattern (a string, matched whole) in the manifest, so that core refuses a value out of them when it is saved, and the same constraints in the settings model, which the plugin enforces:

# In the manifest's settings
"MAX_LENGTH": SettingField(type=SettingType.INTEGER, default=512, minimum=1, maximum=512),
"MODEL_NAME": SettingField(type=SettingType.STRING, pattern="^[a-z_]+$"),

# In the settings model
max_length: int = Field(default=512, ge=1, le=512)
model_name: str = Field(default="tree", pattern="^[a-z_]+$")

assert_bounds_agree(MANIFEST, Settings) from spot_sdk.testing keeps them in step: it fails on a bound stated in one and not the other, or otherwise. Two more cases fail it. A model's pattern without its anchors (^...$) fails, as pydantic finds it anywhere in the value, where the manifest matches it whole. A number's gt and lt fail too, as no manifest bound excludes its own value: state the smallest or the largest value allowed, with ge or le and minimum or maximum (a timeout of at least 0.1 s, not of more than 0).

The manifest and the Dockerfile¶

The manifest is served by GET /manifest, and its labels are what the catalog reads from the image. Build the image with the labels the manifest renders, so both always agree:

mapfile -t labels < <(python -c '
from src.manifest import MANIFEST
for name, value in MANIFEST.labels().items():
    print(f"--label={name}={value}")
')
docker build "${labels[@]}" -t registry.example.com/spot/analyzer-rules:1.0.0 .

A plugin made from the template builds from the template's Dockerfile, whose CMD runs src.main:app.

Choose egress with care: none when the plugin sends nothing outside SPOT, indicators when it sends indicators (URLs, hashes, addresses) to an outside service, content when it sends message content (a hosted language model). A plugin without the label counts as content, and installing it needs the operator's consent.

The hosts a plugin reaches¶

From SPOT 2.1, a plugin of any class runs without a route out. It reaches the SPOT services it calls back, and the hosts its settings of type url name, through SPOT's egress gateway, and nothing else.

• Declare each outside host the plugin calls as a setting of type url, with a default when there is one: MISP_URL, LLM_URL. The admin sees what it reaches under "Reaches".
• Give a host by name, never by address. A url setting marked secret reaches nothing, since the gateway holds no key: put the host in a plain url setting and the token in a secret.
• A plugin that must fetch hosts not known in advance names a forward proxy in a url setting, and sends its fetches through it.
• SPOT 2.1 gives the plugin no proxy variable: the gateway goes through the operator's proxy for it.

Running a plugin by hand¶

To run a plugin outside SPOT's installer (your own compose file), get its token from the CLI and mount it as a secret:

spot-cli plugin token analyzer-rules --kind analyzer > spot_plugin_token

services:
  analyzer-rules:
    image: registry.example.com/spot/analyzer-rules:1.0.0
    environment:
      SPOT_PLUGIN_ID: analyzer-rules
    secrets:
      - spot_plugin_token
    networks:
      - spot-plugins
secrets:
  spot_plugin_token:
    file: ./spot_plugin_token
networks:
  spot-plugins:
    external: true

Testing¶

spot_sdk.testing has what a plugin's tests need:

from pathlib import Path

import pytest
from fastapi import FastAPI
from spot_sdk import (
    Address,
    AnalyzeRequest,
    KnowledgeClient,
    KnowledgeDocument,
    create_plugin_app,
)
from spot_sdk.signals import HEADER_REPLY_TO_MISMATCH
from spot_sdk.testing import FakeKnowledgeClient, assert_contract, make_view

from src.config import Settings
from src.manifest import MANIFEST
from src.services.analysis import analyze

EMLS = Path(__file__).parent / "emls"
TOKEN = "spot_pt_test"


@pytest.fixture
def app(monkeypatch: pytest.MonkeyPatch) -> FastAPI:
    monkeypatch.setenv("SPOT_PLUGIN_TOKEN", TOKEN)
    return create_plugin_app(MANIFEST, Settings, analyze=analyze)


@pytest.mark.parametrize("eml", sorted(EMLS.glob("*/*.eml")), ids=lambda p: p.name)
def test_every_message_gets_a_verdict(app: FastAPI, eml: Path) -> None:
    assert_contract(app, eml, token=TOKEN)


def test_a_reply_to_on_another_domain_is_a_signal() -> None:
    view = make_view(
        from_=Address(address="ceo@example.com", domain="example.com"),
        reply_to=[Address(address="ceo@evil.example", domain="evil.example")],
    )
    request = AnalyzeRequest(job_id="test", email=view, deadline_ms=30_000)
    verdict = analyze(request, Settings())
    assert HEADER_REPLY_TO_MISMATCH in [signal.name for signal in verdict.signals]


@pytest.mark.asyncio
async def test_the_store_is_read(monkeypatch: pytest.MonkeyPatch) -> None:
    kb = FakeKnowledgeClient()
    await kb.upsert(
        KnowledgeDocument(id="employee:ceo", content="Alice Martin", tags=["employee"])
    )
    monkeypatch.setattr(KnowledgeClient, "for_analysis", kb.for_analysis)
    ...

• assert_contract also checks that the same call without the token is refused. The token must be set before the app is created: build the app in a fixture, not at import.
• assert_contract first waits, up to timeout seconds (10 by default), for /health to leave starting: a plugin with load answers unknown until its model is loaded. A plugin that is then unhealthy fails the check with its reason (the load failed (FetchError)).
• assert_provider_contract(app, knowledge, token=TOKEN) checks a context provider: the token on /internal/sync and /internal/state, the 202 and the state of a sync, and the documents it writes to knowledge, valid and with the same ids from one sync to the next. Its second sync must remove a document of the source that it no longer writes: write with replace_source. Give the provider the fake, with the provider's id as its source, before the app is created:

def test_the_provider_keeps_the_contract(monkeypatch: pytest.MonkeyPatch) -> None:
    kb = FakeKnowledgeClient(source="provider-employee-dir")
    monkeypatch.setattr(KnowledgeClient, "from_settings", lambda *_a, **_k: kb)
    monkeypatch.setenv("SPOT_PLUGIN_TOKEN", TOKEN)
    app = create_plugin_app(MANIFEST, Settings, sync=sync)
    assert assert_provider_contract(app, kb, token=TOKEN).last_sync_count > 0

These tests run the app in the test's process. To test the image itself, see testing the image below.

Testing the image¶

The helpers above test the app in the plugin's process. The image is another thing: its labels, its entrypoint, its user, its limits and its network. spot-cli plugin test runs the image you built as SPOT runs it, without a SPOT stack, and sends it the messages of tests/emls/. It arrives with SPOT 2.1.0-beta.2: from that version, the command exists in spotproject/spot-cli. The engine is spot_sdk.testing, with the testing extra: pip install "spot-sdk-python[testing]". spot-cli's image installs it, at the SDK version its SPOT release pins, so the plugin is tested with the parser and the models of that release.

Running it¶

The command needs Docker and spot-cli's image given the Docker socket. It runs in that container, joins the plugin's network from there, and reads the corpus from the directory you mount:

docker build -t analyzer-rules:dev .
docker run --rm --user "$(id -u):$(id -g)" \
  --group-add "$(stat -c %g /var/run/docker.sock)" \
  -v /var/run/docker.sock:/var/run/docker.sock \
  -v "$PWD:/work" -w /work \
  --entrypoint python spotproject/spot-cli:2.1.0-beta.2 \
  -m spot_cli plugin test analyzer-rules:dev

--user makes the files the run writes, such as the --junit file, yours and not root's. --group-add gives that user the group of the Docker socket, so that it can use it. On macOS, stat takes -f %g in place of -c %g, and with Docker Desktop the socket's group may not matter.

The options:

The image is any reference the local Docker host holds: the one you just built, or one you pulled. The command never pulls. In a pipeline, the plugin-test component of spot/ci-templates runs it, from version 1.3.0.

For each run the engine:

1. reads the image's labels and builds the manifest from them (Manifest.from_labels);
2. creates a network of its own, without a route out, as SPOT runs every plugin (spot-plugins-isolated);
3. joins that network under the names api-gateway, knowledge and llm, and serves the fake SPOT there, on port 8000;
4. starts the plugin with SPOT's options (below), a new token and its settings;
5. checks the plugin, sends it the messages, and removes the network and the container, by success, by failure and on Ctrl-C.

The plugin container gets what the plugin-manager gives every plugin: all capabilities dropped, no-new-privileges, at most 512 processes, the memory and CPU of its spot.plugin.resources label (1024 MB and one CPU without it), the token as /run/secrets/spot_plugin_token, its secret settings as files in /run/secrets, its plain settings and SPOT's variables in the environment. They come from spot_sdk.hosting, which core uses too.

The corpus¶

The corpus is the directory tests/emls/, with two sub-directories. The directory a message is in is its ground truth:

tests/emls/
├── corpus.yaml        # optional
├── phish/*.eml
└── benign/*.eml

The run is refused when:

• another directory is there, or a file that is not .eml is in phish/ or benign/;
• a message is empty, or larger than 25 MiB, which core refuses at intake;
• two messages are the same bytes, since a copy narrows the intervals;
• the corpus holds no message.

corpus.yaml is optional and holds five keys. A key it does not know refuses the file.

An expectation has verdict, one class or a list of them, signals, the signal names the answer must hold, and no_signals, those it must not hold:

settings:
  min_score: 40
trusted_authserv_ids:
  - mx.example.net
knowledge: employees.jsonl
expect:
  phish/invoice.eml:
    verdict: [malicious, suspicious]
    signals: [header.reply_to_mismatch]
  benign/newsletter.eml:
    verdict: benign
    no_signals: [header.reply_to_mismatch]

To test a plugin that asks a language model, script the model. Point the plugin's model URL setting at the fake SPOT's llm name and give the answer:

settings:
  LLM_URL: http://llm:8000/v1
llm:
  answer: '{"verdict": "malicious", "score": 92, "confidence": 85, "indicators": []}'

Every request gets that answer. The run then prints no precision or recall, since they would measure the script. To measure a real model, --set the URL to one and leave llm: out.

A secret setting never comes from corpus.yaml or --set: the file is committed and the command line is logged. Give it as a file with --secret-file KEY=FILE, and it reaches the plugin as /run/secrets/KEY. A plain setting is refused as a secret file, and a setting the manifest does not declare is refused either way.

What the checks are¶

The run reports a check by name, in this order. Each passes or fails, with the reason when it fails.

A plugin that is not an analyzer is checked by the checks only.

An analyzer is called once per message, one at a time, as the orchestrator calls it: POST /internal/analyze with the token, a correlation ID, Accept-Encoding: identity and a deadline_ms, which is --timeout, 30 s by default. The whole call, the answer's read included, ends there. The answer is read up to 1 MiB and refused when it is encoded. The message is parsed with the SDK's parser, as core parses it, and the plugin fetches it from the fake SPOT.

A message fails the run when:

• the call times out, answers an HTTP error, answers more than 1 MiB, answers encoded, or answers no AnalyzerVerdict. Core records such a call as unknown; the test reports it as failed. The metrics count it in failed and total, and only the confusion matrix leaves it out;
• the answer does not meet its expect entry.

An unknown verdict does not fail the run: it is counted, and left out of the confusion matrix.

A failure is never benign. After the messages, each message whose analysis called SPOT (the gateway, the Knowledge Store or the model) is sent again with the fake SPOT failing: every route answers 503. The answer must be neither benign nor spam (the metrics count spam as legitimate); unknown, a phishing class or a failed call passes. The check is named <path> with SPOT failing. A plugin that never calls SPOT is not sent anything twice.

The fake SPOT¶

The plugin reaches it where it reaches SPOT, with the same clients:

The gateway's and the Knowledge Store's routes need the plugin's token; the model's needs none, as Ollama's does not. Any other route is a 404. With SPOT failing, every route answers 503.

The report¶

The run prints the checks and the messages as tables. --json prints the report instead, and --junit FILE writes a test case per check and per message, which GitLab shows in the merge request. The report holds the plugin's id, version, kind and egress class, the image reference and id, the SDK version, the sha256 of the corpus, each check, each message (its file, truth, status, verdict, score, signal names, duration, the reason it failed, unmet and spot_calls) and the metrics. It never holds a message, a signal's evidence, a summary or a secret. The start check's reason may hold a plain setting's value: the log lines of a plugin that stopped while it started, in which the token and the secret settings are masked.

The metrics are those of spot-cli benchmark report, counted by the same function: precision, recall, F1, F2, false positive rate, accuracy and the mean score, with the counts they come from. Malicious and suspicious predict phishing; spam and benign predict legitimate. Each rate has a 95 % Wilson interval. With a small corpus the intervals are wide, and they assume the messages are independent: near-copies make them too narrow. Metrics never fail a run. There are none when a model is scripted (llm:), since the numbers would measure the script, and none when no message was analysed.

What fails a run¶

A contract fault, an unmet expectation, or a benign (or spam) answer while SPOT fails. Metrics never fail a run.

Exit codes¶

Ctrl-C ends the run and removes what it created. Everything the run creates carries the label spot.test.run=<run>, and the engine touches nothing else. When Docker refuses a removal, the run says which label to look for.

What differs from SPOT¶

• There is no egress gateway and no proxy. A plugin has no route out in the test, whatever its class, so its url settings reach only the fake SPOT's names (api-gateway, knowledge, llm): point a model's URL at llm, and test any other outside service with a stub in the plugin's unit tests.
• The containers have no restart policy and no log rotation.
• Calls are sequential, so that each call back to SPOT belongs to one message.
• The Knowledge Store and the model are fakes, as above. Measuring a plugin with the real ones is a run on a SPOT stack.