Manage Extensions¶

Extensions are installable modules — protocol decoders, hardware drivers, analyzers — that the agent loads on demand. The SDK lets you discover what's installed, inspect each one, and start, stop, or restart them.

Building your own extension?

This page covers managing installed extensions. To develop a new one, see How to Develop Extensions.

The snippets below assume an agent connected via connect():

from zelos_sdk.agent import connect

with connect() as agent:
    ...

Examples use <ext-id> as a placeholder — substitute the publisher-qualified ID of an extension you have installed (agent.extensions.list() will show what's there).

List installed extensions¶

for ext in agent.extensions.list():
    print(f"{ext.id:45} {ext.version:8} {ext.state}")

Each entry carries enough metadata to render a marketplace-style list:

Field	Description
`id`	Publisher-qualified ID like `"publisher.extension-name"`.
`version`	SemVer of the installed copy.
`state`	`ExtensionState` — `"installed"` or `"running"`.
`pid`	OS process ID when running, otherwise `None`.
`last_exit`	`ExitInfo` (`.code` / `.signal`) if the last run exited unexpectedly; `None` after a clean stop or never run.
`name`, `description`	Display strings.
`author`, `repository`, `homepage`	Marketplace links (any may be `None`).
`keywords`, `categories`	For grouping and filtering.
`host_type`	Where it runs — typically `"agent"`.
`dev_mode`	`True` for locally-installed development builds.
`zelos_version`	Compatibility range it was built for.
`app_contribution_kind`, `entry`, `icon_path`	Display and launch metadata.

Filter by state for display:

running = [e for e in agent.extensions.list() if e.state.value == "running"]
print(f"{len(running)} extension(s) currently running")

If last_exit is set on an extension that isn't running, the previous run terminated unexpectedly:

for ext in agent.extensions.list():
    if ext.state.value != "running" and ext.last_exit is not None:
        print(f"{ext.id} exited: code={ext.last_exit.code} signal={ext.last_exit.signal}")

Inspect an extension¶

agent.extensions.info(id) returns a richer ExtensionInfo. Compared to ExtensionEntry, it adds install_path and readme_path.

info = agent.extensions.info("<ext-id>")

print(info.name, info.version)
print("installed at:", info.install_path)
print("state:", info.state)
print("categories:", info.categories)

info(id) resolves to the currently-installed version automatically. Pass version="1.2.3" to pin a specific version. The same auto-resolution applies to readme(), config_schema(), and last_config().

agent.extensions.readme(id) returns the extension's README markdown — or "" if it doesn't ship one:

readme = agent.extensions.readme("<ext-id>")
print(readme or "_No README provided._")

Inspect config¶

Two helpers expose configuration state:

config_schema(id) — JSON Schema dict describing config inputs, or None if none declared.
last_config(id) — last config saved, or None if never configured. An empty dict means "configured with no fields".

schema = agent.extensions.config_schema("<ext-id>")
last = agent.extensions.last_config("<ext-id>")

if schema is None:
    print("Extension has no config schema.")
else:
    print("required fields:", schema.get("required", []))
    print("last saved config:", last)

To render a config form, walk schema["properties"] for the field list and pre-fill from last_config (or each property's default).

Start, stop, restart¶

ext_id = "<ext-id>"

# Start with a config (config is remembered as the new last_config)
result = agent.extensions.start(ext_id, {"setting": "value"})
print(f"started {result.id}@{result.version} (pid={result.pid})")

# Stop (idempotent)
agent.extensions.stop(ext_id)

# Restart with a different config
agent.extensions.restart(ext_id, {"setting": "another"})

start() and restart() return an ExtensionStart with .id, .version, and .pid.

config shapes (all three replace the saved last_config):

config={...} — apply, validate against the schema, remember as the new last config.
config={} — start with an explicit empty config; runs schema validation (will fail if the schema declares required fields).
config=None (default) — start with no config. Skips schema validation and replaces the saved config with an empty one.

None and {} are not identical: None skips validation, {} runs it. Neither preserves the previous saved config. To re-apply the previous config, read it back yourself:

last = agent.extensions.last_config(ext_id) or {}
agent.extensions.start(ext_id, last)

After a lifecycle call, agent.signals() updates automatically to reflect the change.

Signals may lag the start call

start() returns when the agent accepts the request — not when signals are flowing. Poll agent.signals() or check agent.extensions.info(id).state for "running" if you need to confirm.

Worked example¶

A typical loop — start an extension, watch its signals come online, then stop it cleanly.

import time

ext_id = "<ext-id>"
last = agent.extensions.last_config(ext_id)

# Reuse the previous config if there was one, else start with an empty config
agent.extensions.start(ext_id, last or {})

# Give the extension a moment to publish its signals
time.sleep(1.0)
print(f"{len(agent.signals())} signal(s) live")

# When you're done
agent.extensions.stop(ext_id)

Once running, agent.actions.list() will include any actions the extension publishes — call them via agent.actions.execute(...) (see Run Actions).

Errors and warnings¶

from zelos_sdk.agent import AgentError, ExtensionError

try:
    agent.extensions.start("<ext-id>", {"setting": "value"})
except ExtensionError as exc:
    print("could not start:", exc)
except AgentError as exc:
    print("call failed:", exc, "cause:", exc.cause)

Exception	When
`ValueError`	`id` is empty or not a string.
`ExtensionError`	Extension not installed; lifecycle failure; malformed schema or saved config.
`AgentError`	Catch-all parent. `.cause` has extra detail when available.

When start() or restart() succeeds but the agent attaches a non-fatal warning, the SDK emits a RuntimeWarning. Capture it with the standard warnings module:

import warnings

with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")
    agent.extensions.start("<ext-id>", {"setting": "value"})
    for w in caught:
        if issubclass(w.category, RuntimeWarning):
            print("agent warning:", w.message)

Tips¶

To restart with the previous settings, read them with last_config(id) and pass that dict to restart().
After start(), check info(id).state.value == "running" (or look for new entries in agent.signals()) before relying on the extension's signals or actions.
dev_mode=True flags locally-installed development builds — useful for tooling that should treat them differently from marketplace installs.

What's next¶

Run Actions

Call typed actions exposed by the extensions you've started.
Query Live Data

Time-series queries, latest values, and replay windows.
Develop Extensions

Write your own extension, package it, and publish it to the marketplace.