ZerdaZerda

Tutorial: Plugin Development

On this page

This tutorial builds a small but complete plugin ecosystem covering all four plugin types, then wires them into a PluginManager. Every API call below matches the actual signatures in the source — see the API Overview for links to full reference pages.

1. Set up your project

Text · 6 lines
my_app/
├── main.py
└── my_plugins/
    ├── search_plugin.py
    ├── summarize_plugin.py
    └── log_action_plugin.py

PluginLoader._scan_directory globs *.py non-recursively and skips files starting with _, so keep all your plugin files directly inside my_plugins/.

2. Build a RetrievalPlugin

my_plugins/search_plugin.py:

Python · 38 lines
from zerda.plugins import RetrievalPlugin, PluginMetadata, SchemaProperty, ExecutionContext, CostTier

class SearchPlugin(RetrievalPlugin):
    METADATA = PluginMetadata(
        name="search",
        version="1.0.0",
        author="tutorial",
        description="Searches a small in-memory document set.",
        capabilities=["search", "retrieval"],
        tags=["docs"],
        use_cases=["Find relevant passages for a user question"],
        cost_tier=CostTier.FREE,
        input_schema=[
            SchemaProperty(name="query", type="string", description="Search text"),
            SchemaProperty(name="top_k", type="integer", description="Max results", required=False, default=3),
        ],
    )

    _DOCS = [
        "RAG combines retrieval with generation.",
        "Plugins extend an agent's capabilities.",
        "Hot reload lets you update plugins without restarting.",
    ]

    async def initialize(self) -> bool:
        return True

    async def execute(self, input_data, context: ExecutionContext):
        q = input_data["query"].lower()
        top_k = input_data.get("top_k", 3)
        matches = [
            {"content": d, "score": 1.0, "source": "tutorial-docs"}
            for d in self._DOCS if q in d.lower()
        ]
        return matches[:top_k]

    async def cleanup(self) -> None:
        pass

3. Build a ProcessingPlugin

my_plugins/summarize_plugin.py:

Python · 21 lines
from zerda.plugins import ProcessingPlugin, PluginMetadata, ExecutionContext

class SummarizePlugin(ProcessingPlugin):
    METADATA = PluginMetadata(
        name="summarize",
        version="1.0.0",
        author="tutorial",
        description="Joins retrieved passages into a single summary string.",
        tags=["processing"],
    )

    async def initialize(self) -> bool:
        return True

    async def execute(self, input_data, context: ExecutionContext):
        items = input_data.get("items", [])
        texts = [i["content"] if isinstance(i, dict) else str(i) for i in items]
        return [" ".join(texts)]

    async def cleanup(self) -> None:
        pass

4. Build an ActionPlugin

my_plugins/log_action_plugin.py:

Python · 24 lines
from zerda.plugins import ActionPlugin, PluginMetadata, ExecutionContext

class LogActionPlugin(ActionPlugin):
    METADATA = PluginMetadata(
        name="log_result",
        version="1.0.0",
        author="tutorial",
        description="Writes a result to the application log.",
        tags=["action"],
    )
    REQUIRES_CONFIRMATION = False   # safe to auto-run; the framework itself does not enforce this flag

    async def can_execute(self, input_data, context: ExecutionContext) -> bool:
        return "text" in input_data

    async def initialize(self) -> bool:
        return True

    async def execute(self, input_data, context: ExecutionContext):
        print(f"[LOG] {input_data['text']}")
        return {"success": True, "message": "logged"}

    async def cleanup(self) -> None:
        pass

5. Wire it together

main.py:

Python · 18 lines
import asyncio
from zerda.plugins import PluginManager, ExecutionContext

async def main():
    manager = PluginManager(plugin_dirs=["./my_plugins"])
    await manager.startup()   # auto-discovers all three plugins above

    print(manager.list_plugins())

    result = await manager.execute_pipeline(
        pipeline=["search", "summarize", "log_result"],
        initial={"query": "plugins"},
    )
    print(result)

    await manager.shutdown()

asyncio.run(main())

Walking through what happens:

  1. search returns [{"content": "Plugins extend an agent's capabilities.", ...}].
  2. execute_pipeline feeds that list directly as input_data to summarize — but summarize expects {"items": [...]}. Since the previous plugin's output is a list, not a dict, execute_pipeline's data if isinstance(data, dict) else {} guard means summarize actually receives {}, not the search results. This is worth calling out explicitly: chaining a RetrievalPlugin (returns List[Dict]) directly into a ProcessingPlugin (expects {"items": [...]}) through execute_pipeline does not work out of the box, because execute_pipeline only passes dict outputs through unchanged.

[!TIP] Fixing the pipeline mismatch Either write a small adapter plugin between them that wraps a list into {"items": [...]}, or don't use execute_pipeline for this case — call manager.execute("search", ...) and manager.execute("summarize", {"items": search_result}) as two separate execute() calls instead.

6. Expose the plugins to an LLM

Python · 2 lines
tools = manager.anthropic_tools()
# [{"name": "search", "description": "...", "input_schema": {...}}, ...]

Pass tools directly as the tools parameter of an Anthropic Messages API call, or manager.tool_descriptors() for the OpenAI equivalent.

7. Add permissions (optional, for safe_mode)

Python · 4 lines
manager = PluginManager(plugin_dirs=["./my_plugins"], safe_mode=True)
await manager.startup()   # does NOT auto-grant permissions when safe_mode=True

manager.grant_permission("log_result", "file_write")

In safe_mode=True, startup() skips the automatic security.grant_all_required() step, so any permission a plugin declares in metadata.permissions must be granted explicitly before that plugin can pass register()'s or execute()'s permission checks.

Next steps