ZerdaZerda

types — Plugin Type Hierarchy

On this page

Overview

Four canonical plugin archetypes for a RAG ecosystem, each extending BasePlugin with a PLUGIN_TYPE class marker (used by PluginRegistry's type index) and, in some cases, a narrower execute() signature or extra convenience methods.

RetrievalPlugin

Python · 10 lines
class RetrievalPlugin(BasePlugin):
    PLUGIN_TYPE = "retrieval"

    @abstractmethod
    async def execute(self, input_data: Dict[str, Any], context: ExecutionContext) -> List[Dict[str, Any]]:
        """Return ranked list of retrieved documents."""

    async def retrieve(self, query: str, top_k: int = 5, context: ExecutionContext = None) -> List[Dict[str, Any]]:
        ctx = context or ExecutionContext(query=query)
        return await self.safe_execute({"query": query, "top_k": top_k}, ctx)

Output contract (documented in the class docstring, not enforced at runtime): each returned dict should contain at least content: str, score: float (0–1 relevance), and source: str.

retrieve() is a convenience wrapper — it builds a minimal ExecutionContext if none is given and calls safe_execute() (so it still goes through validation, timeout, retries-if-configured-by-caller, and caching).

Example

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

class InMemorySearch(RetrievalPlugin):
    METADATA = PluginMetadata(
        name="inmem_search", version="1.0.0", author="docs",
        description="Searches an in-memory list of documents.",
        input_schema=[SchemaProperty(name="query", type="string", description="Search text")],
    )

    def __init__(self, docs, config=None):
        super().__init__(config)
        self._docs = docs

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

    async def execute(self, input_data, context: ExecutionContext):
        q = input_data["query"].lower()
        return [
            {"content": d, "score": 1.0 if q in d.lower() else 0.0, "source": "memory"}
            for d in self._docs if q in d.lower()
        ]

    async def cleanup(self) -> None:
        pass

plugin = InMemorySearch(["The sky is blue.", "RAG stands for Retrieval-Augmented Generation."])
await plugin.initialize()
results = await plugin.retrieve("RAG", top_k=3)

ToolPlugin

Python · 6 lines
class ToolPlugin(BasePlugin):
    PLUGIN_TYPE = "tool"

    @abstractmethod
    async def execute(self, input_data: Dict[str, Any], context: ExecutionContext) -> Dict[str, Any]:
        """Run the tool and return a structured result."""

The simplest of the four types — adds only the PLUGIN_TYPE marker and a narrowed return-type annotation on execute(). Per its docstring, this is "the primary target for LLM function-calling," i.e. the type you'd typically expose via PluginManager.tool_descriptors() / anthropic_tools().

ActionPlugin

Python · 15 lines
class ActionPlugin(BasePlugin):
    PLUGIN_TYPE = "action"
    REQUIRES_CONFIRMATION: bool = True

    async def can_execute(self, input_data, context) -> bool:
        return True  # override for pre-flight checks

    async def safe_execute(self, input_data, context) -> Any:
        if not await self.can_execute(input_data, context):
            raise PluginPermissionError(f"[{self.name}] Pre-flight check failed; execution blocked.")
        return await super().safe_execute(input_data, context)

    @abstractmethod
    async def execute(self, input_data, context) -> Dict[str, Any]:
        """Execute the action. Return dict with at least "success": bool and "message": str."""

The only type in this package that overrides safe_execute() rather than only execute(). It inserts a can_execute() pre-flight gate before delegating to BasePlugin.safe_execute(). This means the pre-flight check runs before caching/timeout logic — a rejected can_execute() never touches the cache or the timeout wrapper.

REQUIRES_CONFIRMATION is a class attribute intended as a signal to calling code (e.g. an application's UI layer) that this action should be confirmed before running; the framework itself does not read or enforce this flag anywhere in plugin_manager.py or elsewhere in the package — enforcing it is left to the application.

Raises

  • PluginPermissionError — if can_execute() returns False.
  • Everything BasePlugin.safe_execute() can raise (see base_plugin).

Example

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

class SendEmailPlugin(ActionPlugin):
    METADATA = PluginMetadata(
        name="send_email", version="1.0.0", author="docs",
        description="Sends an email.",
        input_schema=[SchemaProperty(name="to", type="string", description="Recipient")],
        permissions=["network_access"],
    )

    async def can_execute(self, input_data, context) -> bool:
        return "@" in input_data.get("to", "")

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

    async def execute(self, input_data, context: ExecutionContext) -> dict:
        # ... actually send the email ...
        return {"success": True, "message": f"Sent to {input_data['to']}"}

    async def cleanup(self) -> None:
        pass

ProcessingPlugin

Python · 10 lines
class ProcessingPlugin(BasePlugin):
    PLUGIN_TYPE = "processing"

    @abstractmethod
    async def execute(self, input_data: Dict[str, Any], context: ExecutionContext) -> List[Any]:
        """Process and return transformed items."""

    async def process(self, items: List[Any], context: ExecutionContext = None, **kwargs) -> List[Any]:
        ctx = context or ExecutionContext()
        return await self.safe_execute({"items": items, **kwargs}, ctx)

Input contract: {"items": List[Any], ...extra}. Output contract: List[Any] — same or transformed items. Like RetrievalPlugin.retrieve(), process() is a thin convenience wrapper around safe_execute().

Example

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

class UppercasePlugin(ProcessingPlugin):
    METADATA = PluginMetadata(
        name="uppercase", version="1.0.0", author="docs",
        description="Uppercases a list of strings.",
    )

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

    async def execute(self, input_data, context: ExecutionContext) -> list:
        return [s.upper() for s in input_data["items"]]

    async def cleanup(self) -> None:
        pass

plugin = UppercasePlugin()
await plugin.initialize()
out = await plugin.process(["hello", "world"])   # ["HELLO", "WORLD"]

PLUGIN_TYPE_MAP

Python · 6 lines
PLUGIN_TYPE_MAP = {
    "retrieval":  RetrievalPlugin,
    "tool":       ToolPlugin,
    "action":     ActionPlugin,
    "processing": ProcessingPlugin,
}

A plain dict constant exported from plugins/types/__init__ (and re-exported from the top-level plugins package). Maps each PLUGIN_TYPE string to its corresponding abstract base class. Not referenced elsewhere within this source tree's other modules — provided as a lookup table for application code that needs to resolve a type name back to a class (e.g. for validating a manifest before dynamic subclassing).

Choosing a type

If your plugin...Use
Fetches/searches documents, passages, or recordsRetrievalPlugin
Calls an external API or utility and returns a resultToolPlugin
Mutates state — sends something, writes something, triggers somethingActionPlugin
Transforms, filters, reranks, or summarizes a list of itemsProcessingPlugin
Doesn't fit any of the aboveSubclass BasePlugin directly

Related pages