Plugin Discovery Patterns Agent Skill

def load_plugin(name: str) -> object | None:
    """Load a plugin by name via entry_points."""
    try:
        eps = entry_points(group="my_tool.plugins")
        for ep in eps:
            if ep.name == name:
                plugin_class = ep.load()
                return plugin_class()
    except Exception:
        logger.debug("Failed to discover plugin %r", name, exc_info=True)
    return None

def _read_plugins() -> list[str]:
    """Read plugin specs from the config file."""
    path = _get_plugins_config_path()
    if not path.exists():
        return []
    lines = path.read_text().splitlines()
    return [line.strip() for line in lines if line.strip() and not line.strip().startswith("#")]

def plugin_list():
    configured = _read_plugins()
    active_eps = list(entry_points(group="my_tool.plugins"))
    active_ep_names = [ep.name for ep in active_eps]

    for spec in configured:
        pkg_name = _extract_package_name(spec)
        is_active = any(pkg_name in ep_name or ep_name in pkg_name
                        for ep_name in active_ep_names)
        if is_active:
            print(f"    {ok_mark} {spec}")
        else:
            print(f"    {warn_mark} {spec}")
            print(f"         (configured but not active — run: my-tool upgrade --force)")

def plugin_add(spec: str):
    name = _extract_package_name(spec)
    specs = _read_plugins()
    # Dedup: replace existing entry with same package name
    existing_names = [_extract_package_name(s) for s in specs]
    if name in existing_names:
        idx = existing_names.index(name)
        specs[idx] = spec                   # allows upgrading a pinned spec
    else:
        specs.append(spec)
    _write_plugins(specs)
    _reinstall_with_plugins(specs)          # uv tool install --with ...

def _extract_package_name(spec: str) -> str:
    """Extract the bare package name from a PEP 508 spec string.
    'my-plugin @ git+https://...' -> 'my-plugin'
    'my-pkg>=1.0' -> 'my-pkg'
    """
    return re.split(r"\s*[@>=<!~]", spec)[0].strip()

@dataclass(frozen=True)
class ServiceProvider:
    """Provider-agnostic service configuration. Instances are immutable."""
    kind: str = "default"
    host: str = "api.example.com"
    token_env: str = "API_TOKEN"
    api_base: str = ""            # auto-derived when empty
    scheme: str = "https"

def __post_init__(self) -> None:
    # Parse scheme from host if present (e.g. "http://localhost:10110")
    parsed_scheme, bare_host = self._parse_host(self.host)
    if parsed_scheme:
        object.__setattr__(self, "host", bare_host)
        if self.scheme == "https":
            object.__setattr__(self, "scheme", parsed_scheme)

def _resolve_value(value_ref, context: dict):
    """Resolve a value reference against the context.
    Value references use {"path": "/field_name"} format.
    """
    if isinstance(value_ref, dict) and "path" in value_ref:
        path = value_ref["path"].lstrip("/")
        parts = path.split("/")
        current = context
        for part in parts:
            if isinstance(current, dict):
                current = current.get(part)
            else:
                return None
        return current
    return value_ref

def _evaluate_function_call(fc: dict, context: dict) -> bool:
    func_name = fc.get("call")
    args = fc.get("args") or {}
    # Normalize positional list args → named dict args
    if isinstance(args, list):
        args = {"value": args[0]} if args else {}

    if func_name == "required":
        value = _resolve_value(args.get("value"), context)
        return value is not None and value != ""
    if func_name == "regex":
        value = _resolve_value(args.get("value"), context)
        return bool(re.match(args.get("pattern", ""), str(value)))
    # ... length, numeric, email, and, or, not ...

# plugins.py — two-tier plugin discovery
import re
from importlib.metadata import entry_points
from pathlib import Path

PLUGIN_GROUP = "my_tool.plugins"
PLUGINS_CONFIG = Path.home() / ".config" / "my-tool" / "plugins"

def load_plugin(name: str):
    """Load a plugin by name via entry points."""
    for ep in entry_points(group=PLUGIN_GROUP):
        if ep.name == name:
            return ep.load()()
    return None

def configured_plugins() -> list[str]:
    if not PLUGINS_CONFIG.exists():
        return []
    return [l.strip() for l in PLUGINS_CONFIG.read_text().splitlines()
            if l.strip() and not l.strip().startswith("#")]

def active_plugins() -> list[str]:
    return [ep.name for ep in entry_points(group=PLUGIN_GROUP)]

def check_discrepancies():
    configured = {extract_name(s) for s in configured_plugins()}
    active = set(active_plugins())
    missing = configured - active    # configured but not installed
    orphaned = active - configured   # installed but not in config
    return missing, orphaned

def extract_name(spec: str) -> str:
    return re.split(r"\s*[@>=<!~]", spec)[0].strip()

# provider.py — frozen dataclass provider abstraction
from dataclasses import dataclass

@dataclass(frozen=True)
class Provider:
    kind: str = "default"
    host: str = "api.example.com"
    token_env: str = "API_TOKEN"
    scheme: str = "https"

    @property
    def base_url(self) -> str:
        return f"{self.scheme}://{self.host}"

    @classmethod
    def from_env(cls, prefix: str = "MY_TOOL") -> "Provider":
        import os
        return cls(
            kind=os.environ.get(f"{prefix}_PROVIDER", "default"),
            host=os.environ.get(f"{prefix}_HOST", "api.example.com"),
            token_env=os.environ.get(f"{prefix}_TOKEN_ENV", "API_TOKEN"),
        )

# What a plugin looks like (the interface it must implement):
from typing import Protocol

class MyPlugin(Protocol):
    @property
    def name(self) -> str: ...
    def run(self, params: dict) -> dict: ...