Writing a plugin

Every AI-powered and mail-transport component in Oaken is a plugin seam. The five base classes in oaken.plugins.base are the extension points:

A few things worth knowing before you pick one:

BaseLLMBackend is the swap point for the AI harness. Implement it to replace Ollama with LM Studio, vLLM, or any other model server. The one required attribute is locality: set it to "local" (this machine) or "remote". A secure workspace refuses any backend not marked "local", so forgetting it is treated as remote. Register the class under oaken.llm_backends and set ai.llm_backend in config.toml to activate it. No enricher logic changes.

BaseEmbedder feeds search. It is global, not per-workspace. Switching it triggers a re-index: the vector partition is keyed to model_id, so vectors from the old model are never mixed with the new ones. Name model_id specifically enough to be unique (lmstudio:nomic-embed-text-v1.5, not just nomic-embed-text).

Enrichers are additive; backends and embedders are selected. Every installed enricher runs on every message. Only one LLM backend and one embedder are active at a time (chosen by workspace and by the [ai] config respectively). A BaseLLMBackend you install but do not activate in a workspace does nothing.

The base classes are optional. The contract is structural: any object with the right members satisfies it. In oaken.plugins.contracts each kind has a parallel Protocol class — that is what the host validates against. The Base* classes just supply safe no-op defaults and _default-marked placeholders so the host can tell "not overridden" from "explicitly implemented". Subclassing a base class is the short path; implementing the protocol directly is always valid.

Providers are the most wanted: each one lets Oaken Mail work with another mail service. Enrichers are the ones people will most often write for themselves. Both are meant to be short. The exact signatures are on the Plugin contracts page, which is generated from the code.

Subclass BaseProvider and override pull. That is a working, read-only provider.

from oaken.plugins.base import BaseProvider
from oaken.plugins.contracts import PullBatch, RawMessage


class MyProvider(BaseProvider):
    version = "0.1"

    async def pull(self, cursor):
        # cursor is whatever you returned last time (None the first time). Keep it opaque.
        for page in my_service.list_changes(since=cursor):
            yield PullBatch(
                messages=[RawMessage(m.id, m.rfc822_bytes) for m in page.new],
                deleted_provider_ids=page.deleted_ids,
                next_cursor=page.token,
            )

Add methods to add abilities. The host works out your capabilities from what you wrote, so there is no second list to keep in step:

ctx (also on self.ctx once started) carries your settings, a logger, the clock, and config, the table [plugins.<kind>.<name>] from the user's config.toml. Credentials belong in the OS keyring, never in config.

(The kind is part of the name: [plugins.triager.ai], [plugins.enricher.summary], [plugins.embedder.ollama] and [plugins.llm_backend.ollama] are four different tables. The older flat [plugins.<name>] still works as a fallback and logs a warning; oaken plugins set writes the qualified form.)

A plugin the host cannot load (a broken import, a failed check) is switched off, and the reason shows beside it; fix it and use its pill or oaken plugins enable NAME to try again. A plugin that needs settings it has not been given is also off until it has them: declare them with requires_config = ("base_url",) on the class, and the host leaves it off, saying "needs configuration: set base_url in [plugins.llm_backend.ollama-remote]", then starts it by itself on the next health pass or reload once the keys are set. (ollama-remote is one: it exists to reach another machine, so it needs its address.) A plugin that cannot start because it was never set up (no table at all) is treated the same way. Turning a plugin off yourself is never undone by a retry.

Raise oaken.core.errors.AuthError (or a subclass) when credentials are missing or refused. Core pauses that account's schedule instead of retrying, and the other accounts carry on. Core starts one instance of your class per account, so ctx.account_id tells you which one you are and is the natural name for your keyring entry.

You never touch the databases. You return data and core stores it.

You may return the contract's dataclasses, but you do not have to. A dict, a list, a tuple or a plain string is coerced into the right type, and mistakes come back as a readable message:

gmail: pull() returned something invalid (messages.0.raw_bytes: Input should be a valid bytes)
embedder 'mine': embed() returned 3 vectors for 4 texts

An error fails that one job and is logged. It never stops core. What core passes on to the rest of the system is always the strict, frozen type.

from oaken.plugins.base import BaseEmbedder


class MyEmbedder(BaseEmbedder):
    model_id = "my-model"
    dimensions = 384

    async def embed(self, texts):
        return [my_model.encode(t) for t in texts]

An LLM backend is the same shape: set model_id and context_window, override generate(request). The request carries a system instruction and user parts. Mail content is always an UntrustedText, never part of the system text.

from pydantic import BaseModel
from oaken.plugins.base import enricher


class Result(BaseModel):
    words: int


@enricher("word-count", output=Result)
async def word_count(msg):
    return {"words": len(msg.body.unwrap().split())}

The decorator builds the class. Core validates the result against Result and publishes its JSON schema. Change the version whenever the output could change: it is part of the cache key, so a new version re-runs the enricher and an unchanged one never does. An enricher that needs a model sets needs_llm=True.

What an enricher is given

An CleanMessage has the message's cleaned text (body), subject, a few display headers, every header in raw_headers (lower-case names, all values), the attachments' metadata (attachments: name, type, size, whether core flagged it; never the bytes), the date, whether it is_reply, and, for a reply, the stored parent_summary of the message before it. Everything derived from the mail is UntrustedText.

Reaching a model

An enricher never gets a backend. It is handed ctx.llm in start() and asks through it:

class Answer(SafeOutput):
    title: str


TITLE = PromptBuilder("my.title", "1", "Give the email a three-word title.", Answer)

answer = await self.ctx.llm.generate_cached(
    TITLE,
    msg.subject,
    msg.body,
    content_hash=msg.message_id,
    account_id=msg.account_id,
    message_id=msg.message_id,
)

Core picks the workspace from the account (a secure one only reaches a local model), answers a repeated question from its cache, validates the reply against Answer and records the tokens. The instructions are a literal in your source; the mail goes in as UntrustedText parts, and a lint test fails a module that builds a request any other way. The first-party enrichers (oaken_enrichers: summary, actions, inspector) and the first-party triagers (oaken_triagers: spamcheck, rules, ai) are the worked examples. Turn one off with [plugins.enricher.<name>] enabled = false; summary and actions skip bulk mail unless you set skip_bulk = false.

The cache holds validated model answers. Core hands it the facts of a question and the cache builds the key, so a smarter key or a shared store is a plugin, not a core change. Name it in config.toml ([ai] cache = "my-cache"); with nothing named, core keeps its own SQLite cache in raw.db, which a plugin never sees.

from oaken.plugins.base import BaseCache
from oaken.plugins.contracts import CacheEntry, CacheStats


class MyCache(BaseCache):
    def key(self, *, workspace, task, content_hash, model, prompt_version):
        return (workspace, task, content_hash, model, prompt_version)

    async def get(self, key): ...  # a CacheEntry or None
    async def put(self, key, entry): ...
    async def stats(self, workspace=None): ...  # a CacheStats
    async def clear(self, workspace=None, task=None): ...  # how many were removed

model is the model that will answer, never a backend default: two models must not share a key. A cache that raises is treated as a miss on get and a lost write on put; the answer still comes back. cache_stats, clear_cache and the workspace figures read through the same cache.

You write the functionality; core controls the plugin. A plugin does not offer enable, disable, reload or settings commands of its own, and should not (each would be a dead verb on every plugin in the CLI). A person turns a plugin on and off, restarts it and edits its settings through core:

The General page (/general, the tab for the plugin's kind) has the same switch and a Reload button, and the same operations are available to any client (plugin_enable, plugin_disable, plugin_reload, get_plugin_config, check_plugin_config, set_plugin_config).

Every plugin's [plugins.<kind>.<name>] table takes keys that belong to the host, not to you, so your ctx.config never contains them:

• enabled = false keeps the plugin off. A plugin that should be opt-in (it costs money, or reaches a network) sets enabled_default = False on its class, and then runs only when its table says enabled = true (or someone runs oaken plugins enable NAME for that session). The table wins over the default, and over plugins enable.
• capabilities = ["sync.pull", "auth"] is an allow-list for a provider. It only narrows: a name your plugin does not implement does nothing, a name that does not exist is an error at start, and a list that leaves only auth is refused. auth left off hides the plugin's sign-in command. This is how a person makes Gmail receive-only at the plugin level.
• when = ["sieve:vip", "triage:invoice"] and unless = ["user:muted"] choose which messages an enricher or an action sees, by label. A label is scope:label (the scope is sieve, triage or user) or a bare label, which matches in any scope. when is any-of (absent: every message); unless drops a message carrying any of its labels. An action that every selector passes over for a route does not park the message.

A message's labels are kept by scope: sieve (set by the reader's scripts and replaced when they change), triage (added by triagers, and only ever added) and user. A triager or enricher reads them in CleanMessage.labels, and the one folder the message is filed in in CleanMessage.filepath (the user's choice, else Sieve's). A triager adds labels by putting a labels list in its verdict; it cannot file, flag, discard or remove anything. What a provider does with filepath (a folder move, a Gmail label) is the provider's business.

list_plugins reports enabled, enabled_default and allowed_capabilities for each plugin. To try settings before saving them, oaken plugins check (the operation check_plugin_config) validates them against your config_model and then starts a throwaway instance on them and asks its health(). It never touches the running plugin, so keep start() free of side effects beyond connecting. A secret is never accepted in these settings; it belongs in the keyring.

An attachment scanner is a specialized action: BaseScanner, on the scan route. Core quarantines every attachment on the way in (mode 0000; ADR 0020) and releases one only after a scanner calls it clean, so with no scanner installed nothing is ever released. You write one method:

from pathlib import Path
from oaken.plugins.base import BaseScanner
from oaken.plugins.contracts import Verdict

class MyScanner(BaseScanner):
    version = "1"

    async def scan(self, path: Path) -> Verdict:
        # path is readable only now (0400), relocked after. Stream it; do not read it whole.
        return Verdict.clean()          # or Verdict.infected("Sig-Name") / Verdict.error("why")

The base does the rest through AttachmentHandler: it lists the message's attachments, opens each one still owed a verdict (pending or a previous error), calls your scan, and records the result. It is idempotent (a clean or infected file is left alone) and safe on retry. Two rules carry the weight:

• Never clean on doubt. Only an explicit OK from your engine is clean; a failure, a timeout or an exception becomes error, and the file stays held. clean is the one verdict that releases the file.
• scan is an auto_route. It only reads and records (it releases nothing), so it is safe to run unasked; the reader can still set approval = "ask".

When a scanner is first installed, files already held are pending; rescan them with oaken reprocess scan. The first-party clamav plugin (oaken_scanners) ships on this base and talks to a running clamd; see AttachmentHandler (oaken.plugins.attachments) for the shared loops.

A plugin is found through an entry point in its package's pyproject.toml. First-party plugins use exactly the same mechanism. The entry-point name is the plugin's name.

[project.entry-points."oaken.providers"]
myservice = "my_package.provider:MyProvider"

[project.entry-points."oaken.enrichers"]
word-count = "my_package.enrichers:word_count"

Install the package into the same environment as Oaken Mail and it is found. Config chooses which plugin fills a slot ([sync] provider = "myservice", [ai] embedder = "ollama"); enrichers are all used.

• Load. It imports the plugin and checks it: the api_version, that every capability you declare has a real method behind it, and that the members of its kind are present. It reports every problem it finds at once.
• Start. start(ctx) is called, with a timeout.
• Watch. health() is polled on an interval. A plugin that fails to load, fails to start or fails a health check is marked unhealthy, logged, announced as a PluginHealthChanged event, and left out of lookups. A plugin whose start() failed is retried at each health check, so a service that was down comes back by itself.
• Offer operations. If the plugin has an operations() method, each PluginOperation it returns becomes <plugin>.<name> in the operation registry, under the same confirm gate. They start off over MCP and cannot mark mail read.
• Stop. stop() is called once at shutdown.

Core never stops because a plugin did. start, stop and health are all optional to define.

Beyond the five kinds, a plugin of any kind can add capabilities with four optional methods. Each is collected from the healthy plugins at start-up, validated like the rest of the contract, and given a plugin-scoped context with no database handle — you return data, core stores it. See ADR 0012.

Add an operation

Define operations() returning PluginOperation values. Each becomes <plugin>.<name> in the operation registry, under the same confirm gate and audit as core's own; a plugin's operations start off over MCP and can never mark mail read.

Add a job kind

Define job_handlers() to run your own durable background work. Core registers each as plugin.<name>.<kind> on the job runner — the same queue, retries, leases and per-kind concurrency as core's own kinds.

from pydantic import BaseModel
from oaken.plugins.base import BaseEmbedder
from oaken.plugins.contracts import PluginJobHandler


class Recount(BaseModel):
    text: str


class Count(BaseModel):
    words: int


async def recount(ctx, payload: Recount) -> Count:  # ctx: PluginJobContext (config, clock, log)
    return Count(words=len(payload.text.split()))


class MyPlugin(BaseEmbedder):
    model_id, dimensions = "mine", 384

    async def embed(self, texts): ...

    def job_handlers(self):
        return [PluginJobHandler("recount", Recount, recount, result_model=Count)]

The handler gets its payload (validated against payload_model) and a PluginJobContext; a returned result is validated against result_model and stored on the job.

Add a pipeline stage

Define pipeline_stages() to derive something after the enrichers for each message. Core stores the result under plugin.<name>.<stage> with the same content-hash idempotency as an enricher, so a rerun or a reprocess is a cache hit.

from oaken.plugins.contracts import PluginStage, PluginStageContext, PluginStageInput


async def tag(ctx: PluginStageContext, msg: PluginStageInput) -> dict:
    # msg.enrichments holds this message's enricher results (summary, actions, ...);
    # msg.triage is its triage verdict, or None when it is unrated
    priority = (msg.triage or {}).get("priority", 3)
    return {"tag": "urgent" if priority == 0 else "normal"}


class MyPlugin(BaseEnricher):
    version = "1"

    async def enrich(self, msg): ...

    def pipeline_stages(self):
        return [PluginStage("tag", tag, version="1")]

Bump a stage's version when its output could change; the old result is recomputed, an unchanged version is a cache hit. A stage failure is logged and never fails the message's enrichment.

Add a scheduled task

Define scheduled_tasks() to run work on a schedule. Core seeds each into the scheduler as plugin.<name>.<task> only if absent, so a schedule you later edit or remove is left alone (the same rule as the sync schedules). The task fires a job_kind — usually your own from job_handlers() — with a fixed payload.

from oaken.plugins.contracts import PluginScheduledTask


class MyPlugin(BaseEmbedder):
    model_id, dimensions = "mine", 384

    async def embed(self, texts): ...

    def job_handlers(self):
        return [PluginJobHandler("recount", Recount, recount, result_model=Count)]

    def scheduled_tasks(self):
        return [
            PluginScheduledTask("nightly", "plugin.mine.recount", Recount(text=""), every=86400.0)
        ]  # once a day; or cron="0 3 * * *"

Add a provider, embedder, model backend, enricher or vector store

These are the five kinds above — subclass the base class, set a few attributes, override the method your kind needs, and register an entry point. The earlier sections are the worked examples.

Every hook has a conformance suite (next section): JobHandlerContract, PipelineStageContract and ScheduledTaskContract, plus the five kind suites.

Run the contract suite against your plugin. It is a set of pytest tests that check the behavior the host and the rest of Oaken rely on.

import pytest
from oaken.testing.contracts import EmbedderContract


class TestMyEmbedder(EmbedderContract):
    @pytest.fixture
    def embedder(self):
        return MyEmbedder()

There is a suite for each kind: ProviderContract, EmbedderContract, LLMContract, EnricherContract, VectorStoreContract and CacheContract. A provider test also supplies mailbox_ids, the ids of the mail in its test mailbox. If you prefer one call, await verify_embedder(MyEmbedder) runs every rule and raises a single error listing everything that failed.

For the extension hooks there are JobHandlerContract, PipelineStageContract and ScheduledTaskContract (and verify_job_handlers, verify_pipeline_stages, verify_scheduled_tasks): give one your plugin instance and it checks the declarations the way the host does at load.

oaken.testing.fakes has an in-memory fake of every kind (FakeProvider with a scriptable mailbox and failure injection, FakeEmbedder, FakeLLM that records every request, FakeEnricher, InMemoryVectorStore, FakeCache). Use them to test code that uses a plugin, with no network, no model and no real mail.

oaken_ollama (an embedder and an LLM backend) is the reference for anything that talks to a model server. It reads its own [plugins.<kind>.<entry-point-name>] table through ctx.config (here base_url, model, keep_alive, timeout, max_retries, backoff, batch_size), sets model_id, dimensions and context_window in start(), and an LLM backend sets locality to "local" or "remote" so a secure workspace can refuse it. Register one class under two entry points to offer two differently configured instances. See prompt safety and workspaces.

oaken_gmail is the reference for a provider that has its own auth. It keeps its credentials in the OS keyring (never in ctx.config), renews them through a TokenManager, reports a dead credential as an unhealthy health() plus a plugin event instead of crashing, and overrides authenticate, pull, set_flags and send, so it declares all four capabilities. Its tests run the shared ProviderContract against an in-memory fake Google. See Connect Gmail.

A provider does not have to read mail. oaken_smtp declares only auth and send: it is a provider that sends through your own SMTP server, with the password in the keyring. A provider must implement pull or send, and the contract suite checks only the capabilities a plugin declares. See Send through your own SMTP server.

oaken_imap is the generic reader: any IMAP server, with no knowledge of one service. It declares auth, sync.pull and flags.write, keeps what it has delivered per folder in its cursor so an interrupted sync resumes and a renumbered folder is read again, and refuses any connection that is not encrypted. See Connect any IMAP server.

Core announces what happens on an event bus, and a plugin can announce its own. Subclass PluginEvent, add plain-data fields, and publish it from start() or anywhere you hold the context:

from dataclasses import dataclass
from oaken.plugins import PluginEvent


@dataclass(frozen=True, kw_only=True)
class QuotaLow(PluginEvent):
    remaining: int


await self.ctx.publish(QuotaLow(ts=self.ctx.clock.now(), remaining=12))

The host fills in plugin with your plugin's name, so a listener can tell who sent it, and publish accepts only PluginEvent subclasses, so a plugin can't forge a core event. Clients that stream events (an MCP client) receive yours alongside core's.

Listening

A plugin can also listen. ctx.subscribe(event_type, handler) calls handler(event) for every core or plugin event of that type (Event for all of them) and returns a function that ends the subscription:

async def start(self, ctx):
    async def on_indexed(event):
        self.pending.append(event.message_id)  # queue real work; keep the handler quick

    ctx.subscribe(MessageIndexed, on_indexed)

Events are frozen, so a handler can read but not change what happened, and one that raises is logged and skipped without affecting core or other listeners. The host ends every subscription when the plugin stops.

Plugins run in the same process as core, with full access to the machine and to whatever core can reach. They are not sandboxed in version 1. The contract keeps the code tidy and the data boundaries clear; it does not contain a hostile plugin. Only install plugins you trust, as you would any Python package.