Writing plugins¶

Plugins are .py files that add REPL commands. Drop a file into a plugin/ folder and it loads automatically. No compilation, no registration, no restart.

Security model¶

Plugin auto-load runs arbitrary Python in-process — same trust shape as conftest.py, direnv, Makefile, pip install setup.py, or any other tool that reads code from a directory. Treat termapy_cfg/<name>/plugin/ and termapy_cfg/plugin/ the same way you treat those: don't point termapy at config folders you don't trust.

If you need a stronger guarantee — for example when running --mcp with an LLM driver, or in CI, or against a freshly cloned project repo — set the env var:

# bash / zsh
export TERMAPY_TRUSTED_PLUGINS_ONLY=1

# PowerShell
$env:TERMAPY_TRUSTED_PLUGINS_ONLY = "1"

Any boolean token works in place of 1 (on, true, yes, ...); a value outside that vocabulary refuses to start, naming the variable.

With that flag set, termapy skips both filesystem plugin discovery passes (global termapy_cfg/plugin/ and per-cfg termapy_cfg/<name>/plugin/) in every frontend — the TUI, the CLI and the MCP server alike. Only built-in commands — the ones shipped inside the wheel, in site-packages — will load. The trust boundary collapses to "your Python environment," which is the boundary every other Python tool already uses.

This flag lives in the environment, not in the cfg file. A hostile cfg cannot disable its own gate.

A second related gate is TERMAPY_OS_CMD_ENABLED: when truthy, /os is allowed to spawn shell commands. Was a cfg key (os_cmd_enabled) through v0.65; same rationale forced the move to env-var-only in v0.66.

Quick start: copy and modify¶

The fastest way to write a plugin is to copy an existing one:

1. Copy probe.py from the demo plugins folder
2. Rename it to your_plugin.py
3. Change the command name, help text, and handler logic
4. Drop it into termapy_cfg/plugin/ (all configs) or termapy_cfg/<config>/plugin/ (one config)

How plugins work¶

Whenever a config becomes current — at startup and on every /cfg.load, in the TUI, the CLI and the MCP server alike — termapy scans the plugin/ folders for .py files: the global folder first, then the config's own, later overriding earlier by name. A switch drops the previous config's plugins, transforms, hooks and converters before the new ones load. A folder plugin can never take the name of one of the app's own commands; it is skipped with a warning instead. Each file is imported and checked for a COMMAND object at module level. If found, that command is registered in the REPL, and users can invoke it by typing its name with the command prefix (e.g. /hello).

The COMMAND object tells termapy:

• name: what the user types to invoke it (/name)
• args: the argument syntax shown in /help ({optional} or <required>)
• help: one-line description shown in /help
• handler: the Python function to call when the command runs

The handler function¶

The handler is where your plugin logic lives. It is called whenever a user types your command in the REPL input, or when a .run script contains your command. The handler receives two arguments:

• ctx (PluginContext): your interface to the terminal, serial port, config, and filesystem. This is the only API your plugin needs.
• args (str): everything the user typed after the command name. For /hello world, args is "world". For /hello, args is "".

The handler can do anything: print output, send commands to the serial device, read responses, write files, or chain other REPL commands.

Plugin file structure¶

A minimal plugin:

from termapy.plugins import CmdResult, Command, PluginContext

def _handler(ctx: PluginContext, args: str) -> CmdResult:
    """Called when the user types /hello."""
    name = args.strip() or "world"
    ctx.io.result(f"Hello, {name}!")
    return CmdResult.ok(value=f"Hello, {name}!")

# ── COMMAND (must be at end of file) ──────────────────────────────────────────
COMMAND = Command(
    name="hello",
    args="{name}",          # {braces} = optional, <angle> = required
    help="Say hello.",
    handler=_handler,
)

The COMMAND object must be defined after all the functions it references. Termapy looks for this specific name.

Name the handler _handler, and a subcommand's handler _handler_<sub> (_handler_list for /tool.list, _handler_crc_calc for /proto.crc.calc). Every built-in, demo and example plugin follows that shape, and termapy's own test suite enforces it on them, so a new command is a copy of any existing one.

A command is not the only thing a plugin file can export. Any of these makes the file a plugin, and one file may export several:

A file exporting none of them is skipped with a warning.

The PluginContext shape¶

The ctx object is a thin shell over five handles -- four capability domains plus the privileged internal escape hatch -- with a few plain fields (cfg, config_path, prefix) and universal methods (dispatch, ns, arg, flag, wait_for_match) beside them:

def _handler(ctx: PluginContext, args: str) -> CmdResult:
    if not ctx.cfg.get("encoding"):           # plain config (read-only)
        ...
    ctx.io.result("Hello", "green")           # final answer (quiet+)
    ctx.io.output("listing line")             # bulk data (normal+)
    ctx.io.status("loading...")               # progress chatter (verbose only)

    with ctx.serial.io():                     # claim serial for sync read
        ctx.serial.write(b"AT\r")
        resp = ctx.serial.read_raw()

    folder = ensure_folder(ctx.fs.cap_dir)    # termapy.folders: data folders appear on first write
    ctx.fs.open_file(folder / "out.csv")      # open in system viewer

    ctx.ui.confirm("Sure?")                   # TUI-only dialog (gated)

    ctx.dispatch("/var.set X 5")              # re-route a command
    ns = ctx.ns("my_plugin")                 # session-scoped storage (a dict)
    ns["counter"] = ns.get("counter", 0) + 1
    ctx.io.output(f"Usage: {ctx.prefix}foo")  # active REPL prefix (from cfg)

    return CmdResult.ok(value=...)

The five handles:

• ctx.io -- write to user (terminal, log, fallback notifications)
• ctx.serial -- connect/disconnect, read/write the serial port, observe bytes
• ctx.fs -- per-config directories and file opening
• ctx.ui -- TUI-only operations (dialogs, screenshots); raises in CLI
• ctx.internal -- privileged escape hatch for built-ins; external plugins should avoid it

Capability gating¶

Some handle methods are gated on CapabilitySet flags. Calling a gated method without declaring the capability raises MissingCapability, which the dispatcher converts to CmdResult.fail. The fix is to declare what your command needs:

COMMAND = Command(
    name="ask", help="Prompt user.",
    needs=CapabilitySet(confirm_dialog=True),  # refuse to dispatch in CLI
    handler=_handler,
)

The dispatcher refuses to invoke a handler whose needs aren't satisfied, so most capability mismatches fail loudly before the handler runs.

Gated methods today:

• ctx.fs.open_file -- requires gui_apps (no-op in headless/SSH)
• ctx.ui.confirm -- requires confirm_dialog (implies block_until)
• ctx.ui.notify -- requires ui_notify (use ctx.io.notify for the always-works fallback)
• ctx.ui.clear_screen -- requires tui_mode (use ctx.io.clear_screen)
• ctx.ui.screenshot -- requires screen_capture (no fallback)
• ctx.ui.exit_app -- requires tui_mode
• ctx.wait_for_match -- requires block_until

Returning scriptable values¶

Handlers return CmdResult to indicate success or failure:

• CmdResult.ok(value="...") -- success, value is available to scripts
• CmdResult.ok(value="") -- success, no scriptable data (pure side-effect)
• CmdResult.fail(msg="...") -- failure with error message

value= is required on CmdResult.ok(). Pass the data your command produces (a path, a count, a parsed response, a toggle state). For pure side-effect commands (/cls, /exit) pass value="" explicitly so the "no scriptable data" intent is visible at the call site.

Paths auto-resolve to absolute strings. If value is a pathlib.Path, the constructor calls str(path.resolve()) for you, so handlers can write CmdResult.ok(value=path) directly -- no manual str() / .resolve() boilerplate.

Scripts run in quiet mode (or via $(VAR) <- /cmd) read the value field; the type checker catches missing-value= calls so a "I forgot" gap can't ship silently.

data= is the optional structured twin of value. Agent consumers (the MCP server, --json) read it as real JSON -- a dict, list, or number, never a prose string. Most commands don't need it; listings and info commands benefit most.

A list returns CmdResult.table. Build records of raw values and declare the columns; the records become data, and termapy prints the table (title with count, header, dash row, rows) unless the caller wants data (--json, /term.request, MCP). The handler renders nothing:

from termapy.plugins import CmdResult, Column

_COLUMNS = (
    Column("NAME", lambda record: record["name"]),
    Column("TEMP", lambda record: f"{record['temp_c']:.1f}", right=True),
)

def _handler(ctx: PluginContext, args: str) -> CmdResult:
    records = [{"name": name, "temp_c": read_temp(name)} for name in sensors()]
    names = "\n".join(record["name"] for record in records)
    return CmdResult.table("Sensors", "sensor", _COLUMNS, records, value=names)

When each row is one value under a unique name, pass data= the mapping (data={record["name"]: record["temp_c"] for record in records}): a consumer then looks a name up instead of searching, and the table still shows the rows.

A status page returns CmdResult.record. The same idea for one record: the facts with real JSON types become data, and the (label, text) rows are what a person reads, aligned with cyan labels. termapy prints the rows unless the caller wants data, so a JSON answer carries each fact once. A row's optional third element colors its text; leave a row out to omit it, put None between rows for a blank line, and a TextLine("heading", "green") among them for a plain line:

def _handler(ctx: PluginContext, args: str) -> CmdResult:
    temp_c = read_temp()
    data = {"temp_c": temp_c, "alarm": temp_c > 80}
    rows = [
        ("temp", f"{temp_c:.1f} C"),
        ("alarm", "yes" if data["alarm"] else "no", "red" if data["alarm"] else ""),
    ]
    return CmdResult.record(data, rows, value=f"{temp_c:.1f}")

Handlers that ignore wants_data keep working: their prose reaches structured consumers via the response's output_lines.

def _handler(ctx: PluginContext, args: str):
    temp = read_temperature()
    ctx.io.result(f"Temperature: {temp}C")
    return CmdResult.ok(value=str(temp))

Examples that should set value=:

• Query commands (/port.baud_rate returns "115200")
• State toggles (/echo on returns "on")
• Computed values (/proto.crc.calc returns the CRC)
• Ping timings, version strings, variable values

Examples that should not:

• Pure side-effect commands (/cls, /edit, /cap.stop)
• Commands that print a page of prose (/help)

Usage errors: raise UsageError, never hand-write "Usage:" strings¶

When the arguments don't match your synopsis, raise UsageError instead of returning a hand-written usage message:

from termapy.plugins import CmdResult, Command, PluginContext, UsageError

def _handler(ctx: PluginContext, args: str):
    if not args.strip():
        raise UsageError()                     # bare: just the usage line
    try:
        n = int(args)
    except ValueError:
        raise UsageError(f"Invalid count: {args.strip()!r}") from None
    ...

The dispatcher catches it and renders the canonical line from your registered declaration -- the command's dotted name, its args= string (or the synthesized params synopsis), and the active REPL prefix:

Invalid count: 'xyz'
Usage: /mycmd {N}

One owner renders every usage line, so the error message, /help, and a re-configured prefix can never disagree -- and your args= declaration is the single source of truth for the synopsis.

The synopsis grammar is validated when your plugin loads: "" = no args, {braces} = optional, <angle> = required, tight alternation ({on|off}), nesting allowed. Square brackets, doubled {{, spaced |, or unbalanced groups reject the plugin with a clear error instead of rendering wrong in /help.

Typed parameters: params=¶

Instead of parsing args by hand, a command can declare its parameters and let the dispatcher parse, coerce, and validate them -- the synopsis, /help PARAMETERS block, usage errors, and the MCP tool schema are all synthesized from the same declaration:

from termapy.plugins import CmdResult, Command, PluginContext
from termapy.plugins.params import ParamSpec

def _handler(ctx: PluginContext, args: str) -> CmdResult:
    count = ctx.arg("count")            # already an int, range-checked
    delay = ctx.arg("delay")            # a duration in seconds
    ...
    return CmdResult.ok(value=count)

COMMAND = Command(
    name="blink",
    help="Blink the LED.",
    handler=_handler,
    params=[
        ParamSpec("count", "int", positional=True, required=True, min=1, max=100,
                  help="how many blinks"),
        ParamSpec("delay", "duration", default=0.5, help="gap between blinks"),
    ],
)

Positionals are matched by order, keywords as name=value; rest=True takes the remainder of the line (values with spaces), variadic=True binds a repeatable positional to a list. ctx.arg(name) returns the coerced value and ctx.flag(name) a declared --flag. A $(*NAME) reference resolves to exactly one argument whatever it holds.

File arguments: files=¶

When the first argument names a file in one of the config's data folders, declare the folder and the TUI and CLI complete the name as it is typed:

COMMAND = Command(
    name="replay",
    args="<filename>",
    help="Replay a capture.",
    handler=_handler,
    files="cap",    # a folders.FOLDERS name: run, proto, cap, ...
)

An unknown folder name fails when the plugin loads.

Setting commands: bare queries, never mutates¶

A command that reads/writes an on/off flag or a multi-value option follows the stty/git config rule: a bare invocation shows the current value and changes nothing. Never flip on bare, and never treat an unrecognized argument as a flip.

from termapy.scripting import SETTING_QUERY, SETTING_TOGGLE, parse_bool_setting

def _handler(ctx, args):
    try:
        action = parse_bool_setting(args)
    except ValueError as e:                  # unrecognized -> error, not a flip
        return CmdResult.fail(msg=str(e))    # "Invalid value: X (use on/off/... or toggle)"
    flags = ctx.ns("flags")
    if action is SETTING_QUERY:               # bare: show, don't mutate
        pass
    elif action is SETTING_TOGGLE:            # the explicit `toggle` verb
        flags["myflag"] = not flags.get("myflag", False)
    else:                                     # a bool token: set
        flags["myflag"] = action
    state = "on" if flags.get("myflag") else "off"
    ctx.io.result(state)
    return CmdResult.ok(value=state)

Synopsis: {on|off|toggle} for a boolean; {val1|val2|...|cycle} for an enum (use scripting.next_in_cycle to advance). A TUI button that flips on click dispatches the explicit toggle/cycle verb, not the bare command.

Dynamic help for runtime state¶

If your command owns runtime state that a user should see right on its help page - a loaded file, an open connection, a count of cached items - set long_help to a function instead of a string. The function takes the PluginContext and returns a string. It's invoked at render time, so whatever it reads from ctx.ns(...) or ctx.cfg is live.

def _dynamic_long_help(ctx):
    profile = ctx.ns("active_profile") or {}
    commands = profile.get("commands") or {}
    if commands:
        rev = profile.get("profile_revision") or "(none)"
        state = (
            f"Active profile: {len(commands)} device command(s) "
            f"(rev {rev})."
        )
    else:
        state = "No profile loaded."
    return f"""{state}

Load a device profile with /profile.load <path>, or fetch one from
the connected device with /profile.load cmd=<command>."""

COMMAND = Command(
    name="profile",
    help="Device profile commands.",
    long_help=_dynamic_long_help,   # a function, not a string
    handler=_handler,
)

When the user runs /help profile, the DESCRIPTION section calls this function and the first line reflects the current state. No change to the rendering path, no extra registration - the long_help field just accepts either form.

Two caveats:

1. Read ctx defensively. Use ctx.ns("x").get("k", default) rather than indexing blindly. Help may be invoked at any moment, including before your plugin's state is populated.
2. Never raise. The renderer catches exceptions and substitutes (dynamic help failed: <error>) so /help never crashes, but a noisy fallback is worse than a thoughtful default like "(not loaded)".

Reusable helpers (termapy.help_dynamic)¶

Most dynamic help lines fall into a handful of shapes, so the built-ins share a small helper module. Prefer these over hand-rolling - the output is green-on-default and uniform across every command.

from termapy.help_dynamic import (
    state_line,   # "Current <label> = <value>" in green
    folder_line,  # "<N> files in <folder>/" in green
    port_status,  # "Connected: COM3 @ 115200 8N1" or "Not connected"
    cfg_status,   # "Active cfg = demo (2 configs available)"
    ns_count,     # len(ctx.ns(name)), guards a missing ns
    compose,      # join non-empty parts with a blank line between
    green,        # any text as a green state line, escaped
)

def _long_help(ctx):
    return compose(
        folder_line(ctx, "run", noun="script"),
        "Run a .run script from the run/ folder.",
    )

compose drops empty parts, so a callable that returns "" for "no state yet" collapses gracefully. For a single-value command that needs no prose, you can pass the helper directly:

"baud_rate": Command(
    help="Show or set baud rate.",
    long_help=lambda ctx: state_line("baud rate", ctx.cfg["serial"]["baud_rate"]),
    handler=_baud_handler,
),

Serial I/O pattern¶

Most plugins follow this pattern: send a command, read the response, do something with it.

def _handler(ctx: PluginContext, args: str) -> CmdResult:
    if not ctx.serial.is_connected():
        return CmdResult.fail(msg="Not connected.")

    encoding = ctx.cfg.get("encoding", "utf-8")
    eol = ctx.cfg.get("eol", "\r")

    with ctx.serial.io():                # suppress terminal, claim serial
        ctx.serial.drain()               # discard stale bytes
        ctx.serial.write(f"YOUR_COMMAND{eol}".encode(encoding))
        raw = ctx.serial.read_raw()      # read response with timeout
        text = raw.decode(encoding, errors="replace").strip()

    ctx.io.result(text)
    return CmdResult.ok(value=text)

Key points:

• ctx.serial.io() suppresses the normal terminal display during I/O
• ctx.serial.drain() clears any leftover bytes before your command
• ctx.serial.write() sends raw bytes; you add the line ending
• ctx.serial.read_raw() waits for a complete response (timeout-based framing)

Even cleaner: declare the connection requirement on Command and let the dispatcher gate the call:

COMMAND = Command(
    name="hello", help="Say hello to the device.",
    needs=CapabilitySet.SERIAL_CONNECTED,
    handler=_handler,
)

The recurring needs have named profiles -- CapabilitySet.SERIAL_CONNECTED, .INTERACTIVE, .GUI_APPS, .SCREEN_CAPTURE, .TUI_MODE, .BLOCK_UNTIL, .SERIAL_INTERACTIVE -- hover one in your editor for exactly what it gates. Spell out an unusual combination inline, as the confirm_dialog example above does.

The handler then doesn't need to check is_connected -- the dispatcher returns Not connected. automatically when the port is down.

Bytes and hex¶

Read bytes the user typed with termapy.protocol.parse_data when the payload may hold text: hex in any spelling or case (01 03, 0x01 0x03, 0103) and quoted text ("AT\r"), freely mixed, exactly what /proto.send accepts. Use parse_hex when it is a packet, hex only. In both, a 0x prefix marks one byte, so 0xDEADBEEF is refused with the fix in the message. Print and return hex in uppercase: format_hex(data) for display (01 0A FF), data.hex().upper() for a scriptable value.

from termapy.protocol import format_hex, parse_data

def _handler(ctx: PluginContext, args: str) -> CmdResult:
    try:
        data = parse_data(args)
    except ValueError as e:
        return CmdResult.fail(msg=f"Parse error: {e}")
    ctx.io.output(f"  {format_hex(data)}")
    return CmdResult.ok(value=data.hex().upper())

Useful library imports¶

Beyond termapy.plugins, three pyserial-adjacent subpackages are available for plugins that need typed device contracts, binary protocol parsing, or USB chip lookup. Each is self-contained (no Textual / no engine deps) and re-exported via its package's __init__:

# Device profile schema, loader, type registry (validate typed args,
# resolve profile-local types, walk a v2 profile dict).
from termapy.profile import (
    TypeRegistry, load_profile, validate_profile, profile_command_view,
)

# Binary protocol toolkit (format-spec parser, CRC catalog, .pro runner,
# visualizer loader).  See protocol-testing.md for the format-spec
# language and serial-tools.md for the CRC catalog.
from termapy.protocol import (
    parse_format_spec, apply_format, FrameCollector, get_crc_registry,
)

# USB lookup tables (VID -> vendor, VID:PID -> chip + max baud,
# manufacturer-string -> short alias).  Pure data + lookup helpers;
# pyserial gives you the VID/PID/manufacturer, these add value on top.
from termapy.usb import chip, vendor_for, mfg

Each subpackage's __init__.py carries a library-usage docstring with a worked example. Treat them as importable libraries: a plugin that crunches binary frames doesn't need to vendor a CRC catalog itself.

PluginContext API reference¶

Output (ctx.io)¶

All handler output flows through three semantic channels gated by the session's output level (silent/quiet/normal/verbose). Use the plain methods for unstructured text (with optional color) and the _markup variants when the text contains Rich markup tags like [bold red]Warning![/].

Text inside markup is escaped. Rich reads [word] as a style tag and drops one it doesn't know without an error, so a device reply like [ok] disappears from the line. Pass every value you put into a _markup string through escape_markup; text with no markup of its own belongs on the plain channels, which never parse it:

from termapy.plugins import escape_markup

ctx.io.output_markup(f"  [cyan]{escape_markup(name)}[/] = {escape_markup(value)}")
ctx.io.output(f"  {name} = {value}")  # no markup, nothing to escape

CmdResult.record and CmdResult.table escape their rows for you.

Errors flow through return CmdResult.fail(msg=...) -- the dispatcher paints the red error line. Do not write red error text via the output channels.

Forbidden in handler code: ctx.io._write and ctx.io._write_markup (the underscore-prefixed primitives that bypass level gating) are engine-internal. A test guard fails the suite if a built-in calls either directly.

Config¶

Serial port (ctx.serial)¶

Filesystem (ctx.fs)¶

TUI-only (ctx.ui)¶

These raise MissingCapability when called from CLI without declaring the matching capability on Command.needs.

Top-level¶

Subcommands¶

Use sub_commands for related operations (e.g. /tool.run, /tool.status):

COMMAND = Command(
    name="tool",
    help="A tool with subcommands.",
    sub_commands={
        "run":    Command(args="<file>", help="Run.", handler=_run),
        "status": Command(help="Show status.", handler=_status),
    },
)

An interior node with no handler lists its children when invoked bare, and /tool run x redirects to /tool.run x (the space form). Declare bare_sub="status" to make the bare form dispatch that child instead -- termapy's own /mem runs /mem.info this way (bare shows state). A bare_sub naming no subcommand fails at load. picker="run" (or cfg, port, proto) opens that built-in picker on a bare call in the TUI; other hosts run your bare_sub if you declare one, else your handler. /help prints the declarations as the command's bare form.

Example plugins¶

The demo config ships with four plugins of increasing complexity:

• cmd.py: minimal. Wraps a single AT command in a custom name.
• probe.py: intermediate. Send/receive cycle with formatted output, good starting template.
• temp_plot.py: advanced. Repeated sampling, response parsing, ASCII sparkline visualization.
• traffic.py: advanced. RX/TX byte tap (/traffic.count, /traffic.hexdump, /traffic.rate, /traffic.snoop); demonstrates ctx.serial.rx_observer() / ctx.serial.tx_observer() context managers for passive monitoring without disrupting the normal pipeline.

temp_plot.py is the best example for plugins that send and parse a single device response. It shows:

• Declaring needs=CapabilitySet.SERIAL_CONNECTED for connection gating
• Reading config for encoding and line ending
• Using ctx.serial.io() for a multi-read loop
• Parsing numeric values from device responses
• Handling edge cases (no data, invalid count)
• Rendering results with Rich markup

traffic.py is the best example for plugins that watch the byte stream without disrupting normal operation. It shows:

• Using ctx.serial.rx_observer(cb) and ctx.serial.tx_observer(cb) as the canonical passive-tap pattern -- observers are released on every exit path including exceptions, no try/finally needed
• Coordinating an observer with a threading.Event for "wait for X" semantics (the snoop subcommand)
• Letting normal device output continue flowing through the pipeline while traffic is also being measured / logged in the background

Run /temp_plot and /traffic.count AT+VER in demo mode to see them in action, then read the source.

Using AI coding tools¶

temp_plot.py was generated in one shot by Claude Code with full project context. If you use an AI coding assistant with access to the termapy source, describing what you want often produces a working plugin on the first try. The key is that the AI can see probe.py, the device protocol, and the PluginContext API all at once.

Without full project context, expect to iterate. The serial I/O timing and response parsing are device-specific and hard to get right from an API reference alone.

For more on how termapy itself was built with LLM tooling, see On AI assistance.