python-plugin.en.md

Write a Python Plugin

From starting a project to publishing it, in eight steps. The example is the greeter the template makes: it provides the greeter service and uses the translate service, which another plugin provides, to translate its greeting. It needs Python 3.12 or later; the commands below use uv, and pip with venv works too.

1. Start a project

uvx rutis-host new greeter --lang python
cd greeter
uv sync
uv run python -m unittest discover -s tests

The project:

File	What it is
src/greeter/__init__.py	The plugin
tests/test_plugin.py	Its tests, which need no host
rutis.dev.json	The rows of a local run (step 7)
pyproject.toml	Depends on rutis; dev dependency rutis-host; entry point rutis.plugins
.github/workflows/publish.yml	Tests and publishes to PyPI on a v* tag (step 8)

2. Provide a service

The generated src/greeter/__init__.py:

"""greeter: a rutis plugin."""

# Guide: https://github.com/arcships/rutis/blob/main/docs/guide/python-plugin.en.md

from rutis import define_plugin


class Greeter:
    def __init__(self, greeting: str) -> None:
        self.greeting = greeting

    async def hello(self, name: str) -> str:
        return f"{self.greeting}, {name}!"


def apply(ctx, config):
    # config has the schema's default when the row leaves greeting out.
    ctx.provide("greeter", Greeter(config["greeting"]))


plugin = define_plugin(
    apply,
    # The services it uses (step 3 of the guide): inject=["translate"], then
    # ctx.use("translate") in apply.
    inject=[],
    # The services it provides, so a host knows them before the plugin runs:
    # the class's public methods.
    provides={"greeter": Greeter},
    # The JSON Schema of its config (step 4).
    config={"type": "object", "properties": {"greeting": {"type": "string", "default": "Hello"}}},
)

• define_plugin(apply, …) defines the plugin. When the plugin starts, rutis calls apply(ctx, config); apply may be an async def.
• ctx.provide("greeter", …) provides the service named greeter: other plugins, plugins in other languages and the host use it by that name. The service's methods are the object's methods whose names do not start with _.
• provides says which services the plugin provides and their methods, so a host knows them without running the plugin.

3. Use a service

greeter translates its greeting with the translate service. Replace src/greeter/__init__.py with:

from rutis import define_plugin


class Greeter:
    def __init__(self, translate, greeting: str) -> None:
        self.translate = translate
        self.greeting = greeting

    async def hello(self, name: str) -> str:
        return f"{await self.translate.text(self.greeting)}, {name}!"


def apply(ctx, config):
    ctx.provide("greeter", Greeter(ctx.use("translate"), config["greeting"]))


plugin = define_plugin(
    apply,
    inject=["translate"],
    provides={"greeter": Greeter},
    config={"type": "object", "properties": {"greeting": {"type": "string", "default": "Hello"}}},
)

inject lists the services the plugin uses, and ctx.use("translate") gets one. The plugin starts once all of them are there; when one goes, the plugin stops, and it starts again when the service is back. greeter need not know which plugin provides translate or in which language; call its methods with await.

A service the plugin can do without goes in optional = ["llm"] (or define_plugin(apply, optional=["llm"])) instead of inject: rutis does not wait for it, and ctx.get("llm") gives it, or None when it was not there as the plugin started (one that comes later is not leased for the row, and is sure to be seen only after the row restarts). See plugin API: optional services; it needs rutis-host 0.20 or later.

The template's test now fails: it gives no translate. Step 6 replaces it.

4. Configuration

config is the JSON Schema of the plugin's config. Before apply, rutis fills in the schema's defaults and validates the config: when the row leaves greeting out, config["greeting"] is "Hello"; when the config does not match, the row fails with config: … and apply is not called.

Where the config comes from: in a test, load(plugin, config=…) passes it (step 6); when running, it is the plugin's row in the host (rutis.dev.json in step 7, the host's rutis.json).

5. Emit and listen to events

After each greeting, hello emits the greeter/greeted event; the plugin listens to it to count the greetings, and count returns the number. Replace src/greeter/__init__.py with:

from rutis import define_plugin


class Greeter:
    def __init__(self, ctx, greeting: str) -> None:
        self.ctx = ctx
        self.translate = ctx.use("translate")
        self.greeting = greeting
        self.greeted = 0

    async def hello(self, name: str) -> str:
        word = await self.translate.text(self.greeting)
        self.ctx.emit("greeter/greeted", name)
        return f"{word}, {name}!"

    async def count(self) -> int:
        return self.greeted


def apply(ctx, config):
    greeter = Greeter(ctx, config["greeting"])

    def greeted(name):
        greeter.greeted += 1

    ctx.on("greeter/greeted", greeted)
    ctx.provide("greeter", greeter)


plugin = define_plugin(
    apply,
    inject=["translate"],
    provides={"greeter": Greeter},
    config={"type": "object", "properties": {"greeting": {"type": "string", "default": "Hello"}}},
)

ctx.emit(name, *args) emits an event and returns at once, without waiting for the listeners. ctx.on(name, listener) listens to one, with nothing to declare: it hears the event alike whether this plugin, another plugin, a plugin in another language or the host emits it. The plugin's listeners go when it stops. The other ways to emit: Plugin API: events.

6. Test

load from rutis.testing starts a host for the test (rutis-host test, installed by the project's dev dependency rutis-host) and loads the plugin on the real Python runtime, in the test's process: with a config and a fake translate, it gives you its service to call and its events to listen to, and unloads the plugin when the async with ends. Replace tests/test_plugin.py with:

import asyncio
import unittest

from rutis.testing import load

from greeter import plugin


class FakeTranslate:
    async def text(self, text: str) -> str:
        return {"Hello": "Bonjour", "Hi": "Salut"}.get(text, text)


class Plugin(unittest.TestCase):
    def test_greets_with_the_translated_greeting(self):
        async def go():
            async with load(plugin, config={"greeting": "Hi"}, services={"translate": FakeTranslate()}) as t:
                greeted = []
                t.on("greeter/greeted", greeted.append)
                self.assertEqual(await t.service("greeter").hello("Ada"), "Salut, Ada!")
                # The test can emit too; parallel waits for every listener, by then the emit before it has arrived
                await t.parallel("greeter/greeted", "Bo")
                self.assertEqual(greeted, ["Ada", "Bo"])
                self.assertEqual(await t.service("greeter").count(), 2)

        asyncio.run(go())


if __name__ == "__main__":
    unittest.main()

uv run python -m unittest discover -s tests

The checks are the host's own: its loader fills in the schema's defaults and validates the config, and a config that does not match makes load raise PluginTestError with the row's state line (greeter: config: greeting should be a string, got 3); events are dispatched by the host. unittest has no time limit, so the test tool gives the calls of the plugin's service one: a call that does not return within 30 seconds is cancelled and fails the test with PluginTestError: greeter.hello() did not return within 30s (code timeout). A plugin that is not up makes t.service() throw the row's state line, exactly as dev says it (greeter: waiting for translate (nothing provides it yet), code not-running), and await t.waiting() gives that line too. The rest of the test tool: Plugin API: the Python test tool. A function the plugin hands the test is here in this process, its own: await it for an async def, get its value for a plain def; a function from another process is a RemoteFunction, await f(...) (the same as f.call_async(...)), and a bare f(...) is only for a sync method (How values cross). When a host gives Python types for its services, see Types from the host.

7. Run it locally and try it

rutis-host dev runs the plugin in a local host. The rows of rutis.dev.json say which plugins run:

{
  "format": 1,
  "rows": [
    { "id": "greeter", "config": { "greeting": "Hello" } },
    { "id": "translate", "name": "py:fake_translate" }
  ]
}

• The row whose id is the project's plugin name (greeter) is the plugin itself, and config is its config. The project's plugin name is the entry point's name under [project.entry-points."rutis.plugins"] in pyproject.toml (the template writes greeter = "greeter": the left-hand one), or [project]'s name when there is no entry point; not the module's name, nor a distribution name you changed. It is the name in the running greeter that dev prints on start. With another id (say the entry point became tool = "my_tool" and the row still says my-tool), the row is not the plugin itself but another plugin's row without a name: my-tool: unresolved: invalid entry: "my-tool" has no name.
• The other row is a fake translate: py:fake_translate is the module fake_translate in the project's src/, below. It runs only locally and is not in the published package (pyproject.toml packages only src/greeter).

from rutis import define_plugin


class FakeTranslate:
    async def text(self, text: str) -> str:
        return {"Hello": "Bonjour", "Hi": "Salut"}.get(text, text)


def apply(ctx, config):
    ctx.provide("translate", FakeTranslate())


plugin = define_plugin(apply, provides={"translate": FakeTranslate})

uv run rutis-host dev

rutis-host dev: running greeter; changes reload it (Ctrl-C ends)
greeter: starting
translate: running
rutis-host dev: services: greeter (greeter), rutis/loader (host), rutis/sessions (host), translate (translate)
greeter: running

Each row prints a line when its state changes, and the rows' lines may come in another order each time; rutis-host dev: services: gives the services there are now and the row that provides each; it is printed once every row is running, waiting or failed, and again whenever the services change, before or after the rows' state lines.

In another terminal, in the project's directory, see which services there are and which row provides each, then call one (status and call are not on Windows):

$ uv run rutis-host status
rows:
  greeter: running
  translate: running
services:
  greeter: greeter
  rutis/loader: host
  rutis/sessions: host
  translate: translate
runtimes:
  py: running (pid 55011)
calls:
  none
$ uv run rutis-host call greeter.hello '"Ada"'
"Bonjour, Ada!"
$ uv run rutis-host call greeter.count
1

Each argument of call is a JSON value, strings in quotes; the result is printed as JSON, and an error goes to standard error with exit status 1 (see Calling a service).

A change to the code reloads it: change the ! that hello returns to !! and save, and dev prints

greeter: reloaded (src/greeter/__init__.py changed)
rutis-host dev: services: greeter (greeter), rutis/loader (host), rutis/sessions (host), translate (translate)

call greeter.hello '"Ada"' now gives "Bonjour, Ada!!". Only the row whose file changed reloads; a change to rutis.dev.json takes effect while it runs too. When another module of the plugin's package changes (src/greeter/helpers.py, with from .helpers import … in __init__.py), the package is imported again with it, and dev prints greeter: reloaded (src/greeter/helpers.py changed); modules outside the plugin's package (another top-level module, a package installed in the environment) are not imported again: after changing one, restart dev. So keep helper code in the plugin's package. See Developing a plugin with dev.

For a fake service in TypeScript, put it and its package.json in dev/, and have the Node runtime use dev/ as its project in rutis.dev.json:

{
  "format": 1,
  "runtimes": { "node": { "project": "dev" } },
  "rows": [
    { "id": "greeter", "config": { "greeting": "Hello" } },
    { "id": "translate", "name": "./dev/fake-translate.ts" }
  ]
}

In dev/, npm install @arcships/rutis @arcships/rutis-runtime (the version of rutis-host); a row's name, as every path in the file, is relative to the directory of rutis.dev.json. A package.json at the project's root too leaves it a Python project for dev, unless the entry file that package.json names (src/index.ts, main, …) is there.

8. Publish

uv build
uv publish

Or push a tag such as v0.1.0: the template's workflow tests and publishes with PyPI trusted publishing (configure a trusted publisher for the repository on PyPI first).

• Package name: a short name like greeter is most likely taken on PyPI, so change name in pyproject.toml before publishing, to your-name-greeter for example. The entry point's name stays, and so does the host's row name py:greeter:

  [project.entry-points."rutis.plugins"]
  greeter = "greeter"

• Dependency range: give rutis a lower bound only (rutis>=0.19; rutis-host new writes its own version). The plugin and the host's Python runtime use the same rutis package, so an upper bound would downgrade the host or keep it from upgrading (see Plugin API: versions).

How a host installs your plugin: rutis-host: how a host installs plugins.

Debugging

--debug has the Python runtime read this terminal: a breakpoint() in the plugin stops in pdb; in the terminal, w shows the stack, p <variable> a variable, c continues.

uv run rutis-host dev --debug

• While it is stopped at a breakpoint, the whole Python runtime is stopped: calls to its other rows wait too.
• Without --debug the runtime has no terminal: breakpoint() raises a RuntimeError, which fails the call it is in (the row, in apply), and the row runs on: rutis-host call prints rutis-host: call to greeter.hello failed: RuntimeError: breakpoint() needs a terminal, and this Python runtime has none: run rutis-host dev --debug. With PYTHONBREAKPOINT set (debugpy, remote-pdb, …), the hook it names runs instead.
• The runtime shares the terminal with dev, so Ctrl-C reaches it too: it exits at once (after a breakpoint as well: pdb leaves Ctrl-C alone), and the plugin's cleanups may not run.
• What the plugin prints, and what it writes to standard error, appears in dev's output.
• With a graphical debugger (VS Code and others): install debugpy in the project's environment, call debugpy.listen(5678) in apply, and attach the debugger to 127.0.0.1:5678.
• Tests (step 6) run the plugin in the test process, where breakpoint() works without --debug.

Types from the host

A Rust host can give the Python types of its services (the module rutis-host types --python writes, see Embed in a Rust application: declarations), usually in its plugin template. Annotate what ctx.use returns with them, and your editor and pyright check the calls before anything runs; without the annotation it runs the same:

from deskmate_types import Notes  # the host's module (written by rutis-host types --python deskmate_types.py)


def apply(ctx, config):
    notes: Notes = ctx.use("notes")
    ...

A call that does not match, as pyright reports it:

error: Argument of type "Literal[42]" cannot be assigned to parameter "value" of type "str" in function "put"

Advanced: plugin API

Plugins as values, the other ways to emit events, scopes, objects with methods, how values pass, the rest of the test tool: see Plugin API. To run on another machine, see the remote section of rutis-host.en.md.