Lab: Plugin system

Add a plugin system to a CLI by defining a Processor Protocol, writing two built-in processors, registering them via entry points, and chaining them at runtime.

This lab guides you through adding a real plugin system to a minimal CLI tool. You will define the interface, write two built-in processors, register them as entry points, and verify the discovery-and-chaining pipeline works end-to-end. By the end, any third-party package that declares the right entry point will automatically be picked up — without touching your tool's source code.

The runnable cells below simulate the discovery step without actual package installation. The pyproject.toml snippets are the configuration you would use in a real project — copy them when you build this locally.

Step 1 — Define the Processor Protocol

The Protocol is the public contract. Keep it minimal: one method and one versioning attribute.

Python — editable, runs in your browser

@runtime_checkable is what makes isinstance() work here. Without it, Protocol is only used for static type-checking by tools like mypy or pyright and cannot be checked at runtime.

Step 2 — Write the built-in processors

The two built-in processors live in the tool's own package and are registered as entry points alongside any third-party ones.

Python — editable, runs in your browser

Step 3 — Register via pyproject.toml

In your real project, add this section to pyproject.toml:

[project.entry-points."mytool.processors"]
filter-empty = "mytool.processors:FilterEmpty"
to-uppercase = "mytool.processors:ToUppercase"

After pip install -e . (or pip install mytool), these two processors will appear when importlib.metadata.entry_points(group="mytool.processors") is called.

Third-party authors add their own packages with the same declaration:

[project.entry-points."mytool.processors"]
deduplicate = "myplugin.processors:DeduplicateProcessor"

No changes to mytool's source code — just installing the third-party package is enough.

Step 4 — Discovery and chaining

Python — editable, runs in your browser

# Full pipeline: define Protocol, processors, discovery, and chaining.

from typing import Protocol, runtime_checkable

CURRENT_API_VERSION = 2

@runtime_checkable
class Processor(Protocol):
  api_version: int
  def process(self, lines: list) -> list: ...

class FilterEmpty:
  api_version = 2
  def process(self, lines: list) -> list:
      return [line for line in lines if line.strip()]

class ToUppercase:
  api_version = 2
  def process(self, lines: list) -> list:
      return [line.upper() for line in lines]

# --- Simulate entry point discovery ---
class FakeEntryPoint:
  def __init__(self, name, cls):
      self.name = name
      self._cls = cls
  def load(self):
      return self._cls

REGISTERED = [
  FakeEntryPoint("filter-empty", FilterEmpty),
  FakeEntryPoint("to-uppercase", ToUppercase),
]

def load_processors(entry_point_list):
  """Discover, version-check, and instantiate all registered processors."""
  processors = []
  for ep in entry_point_list:
      try:
          cls = ep.load()
      except Exception as exc:
          print(f"  WARNING: could not load '{ep.name}': {exc}")
          continue
      version = getattr(cls, "api_version", None)
      if version != CURRENT_API_VERSION:
          print(f"  WARNING: skipping '{ep.name}' (api_version={version})")
          continue
      processors.append(cls())
      print(f"  Loaded: {ep.name}")
  return processors

def run_pipeline(lines: list, processors: list) -> list:
  """Chain processors in order; each receives the previous output."""
  for proc in processors:
      lines = proc.process(lines)
  return lines

# --- Test ---
pipeline = load_processors(REGISTERED)
print()

input_lines = ["hello", "", "world", "  ", "foo"]
result = run_pipeline(input_lines, pipeline)
print("Input: ", input_lines)
print("Output:", result)

assert result == ["HELLO", "WORLD", "FOO"], f"Unexpected: {result}"
print("Pipeline test passed.")

The output should be ["HELLO", "WORLD", "FOO"]. Blank and whitespace-only lines are removed first, then every remaining line is uppercased.

Step 5 — Verify a third-party processor integrates cleanly

Add a processor that is not in the built-in list and confirm it slots in without any changes to the discovery code:

Python — editable, runs in your browser

# Third-party processor (would live in a separate installed package)
class PrefixLineNumber:
  """Add a line number prefix to each line."""
  api_version = 2

def process(self, lines: list) -> list:
      return [f"{i+1:03d}: {line}" for i, line in enumerate(lines)]

# Re-use the previous infrastructure (collapsed for brevity)
CURRENT_API_VERSION = 2

class FakeEntryPoint:
  def __init__(self, name, cls):
      self.name = name
      self._cls = cls
  def load(self):
      return self._cls

class FilterEmpty:
  api_version = 2
  def process(self, lines):
      return [l for l in lines if l.strip()]

class ToUppercase:
  api_version = 2
  def process(self, lines):
      return [l.upper() for l in lines]

ALL_REGISTERED = [
  FakeEntryPoint("filter-empty", FilterEmpty),
  FakeEntryPoint("to-uppercase", ToUppercase),
  FakeEntryPoint("prefix-line-number", PrefixLineNumber),  # third-party
]

def load_processors(eps):
  procs = []
  for ep in eps:
      try:
          cls = ep.load()
      except Exception as exc:
          print(f"  WARNING: {ep.name}: {exc}")
          continue
      if getattr(cls, "api_version", None) != CURRENT_API_VERSION:
          print(f"  WARNING: skipping {ep.name}")
          continue
      procs.append(cls())
  return procs

def run_pipeline(lines, processors):
  for p in processors:
      lines = p.process(lines)
  return lines

pipeline = load_processors(ALL_REGISTERED)
result = run_pipeline(["alpha", "", "beta", "gamma"], pipeline)
for line in result:
  print(line)

Three processors, three transformations, zero changes to discovery code. The PrefixLineNumber processor is treated identically to the built-in ones because it satisfies the same Protocol and declares the same api_version.

What you built

You assembled a plugin system from its constituent parts:

A Protocol that is the complete, versioned public contract.
Two built-in processors that satisfy it.
A pyproject.toml entry point declaration that registers them.
A discovery function that loads, version-checks, and instantiates everything.
A pipeline runner that chains processors in order.

The same pattern scales from two built-in processors to dozens of community- maintained ones. The only coordination required between plugin authors and the tool is the Protocol definition and the entry point group name.

Where to go next

Next module: Cross-Platform — making Python tools work correctly on Linux, macOS, and Windows without platform-specific branches.

Finished reading? Mark it complete to track your progress.

Step 1 — Define the Processor Protocol

Step 2 — Write the built-in processors

Step 3 — Register via pyproject.toml

Step 4 — Discovery and chaining

Step 5 — Verify a third-party processor integrates cleanly

What you built

Where to go next

On this page