Index

Extension discovery and manifest definitions for MADA.

The extensions package provides data structures and discovery utilities for registering MADA functionality through extension manifests.

Modules:

Name	Description
manifest	Dataclasses describing extension manifests and MCP server registrations.
registry	Discovery and validation utilities for manifest-based and legacy extension registrations.
builtins	Built-in extension manifest factory for MCP servers shipped with MADA.

ExtensionManifest dataclass

Describe one installable MADA extension package.

Attributes:

Name	Type	Description
display_name	str	Human-readable name for the extension package.
version	str	Version string reported by the extension package.
provider_package	str	Python package or distribution providing this extension.
mcp_servers	tuple[MCPServerRegistration, ...]	MCP server registrations contributed by the extension.

Source code in mada_tools/extensions/manifest.py

@dataclass(frozen=True)
class ExtensionManifest:
    """Describe one installable MADA extension package.

    Attributes:
        display_name (str):
            Human-readable name for the extension package.
        version (str):
            Version string reported by the extension package.
        provider_package (str):
            Python package or distribution providing this extension.
        mcp_servers (tuple[MCPServerRegistration, ...]):
            MCP server registrations contributed by the extension.
    """

    display_name: str
    version: str
    provider_package: str
    mcp_servers: tuple[MCPServerRegistration, ...] = field(default_factory=tuple)

ExtensionRegistry

Discover extension manifests and expose validated MCP server registrations.

Methods:

Name	Description
discover_extensions	Return all manifest-based and legacy-adapted extensions.
get_available_mcp_servers	Return validated MCP server registrations as a sorted list.
get_mcp_server_index	Return validated MCP server registrations keyed by server name.

Source code in mada_tools/extensions/registry.py

class ExtensionRegistry:
    """Discover extension manifests and expose validated MCP server registrations.

    Methods:
        discover_extensions:
            Return all manifest-based and legacy-adapted extensions.
        get_available_mcp_servers:
            Return validated MCP server registrations as a sorted list.
        get_mcp_server_index:
            Return validated MCP server registrations keyed by server name.
    """

    def discover_extensions(self) -> List[ExtensionManifest]:
        """Discover manifest-based and legacy extensions.

        Returns:
            List[ExtensionManifest]:
                Discovered extension manifests after manifest loading and legacy
                entry-point adaptation.
        """
        manifests = self._discover_manifest_extensions()
        manifests.extend(
            self._discover_legacy_server_extensions(
                existing_server_names={server.name for manifest in manifests for server in manifest.mcp_servers}
            )
        )
        return manifests

    def get_available_mcp_servers(self) -> List[MCPServerRegistration]:
        """Return validated MCP server registrations.

        Returns:
            List[MCPServerRegistration]:
                MCP server registrations sorted by provider package and server
                name.
        """
        return sorted(
            self.get_mcp_server_index().values(),
            key=lambda server: (server.package.lower(), server.name.lower()),
        )

    def get_mcp_server_index(self) -> Dict[str, MCPServerRegistration]:
        """Return validated MCP servers indexed by server name.

        Returns:
            Dict[str, MCPServerRegistration]:
                Mapping of server names to validated MCP server registrations.
        """
        available: Dict[str, MCPServerRegistration] = {}

        for manifest in self.discover_extensions():
            for server in manifest.mcp_servers:
                if server.name in available:
                    LOG.warning(
                        "Plugin server name collision for '%s', already discovered. Skipping server from package '%s'.",
                        server.name,
                        manifest.provider_package,
                    )
                    continue

                available[server.name] = server

        return available

    def _discover_manifest_extensions(self) -> List[ExtensionManifest]:
        """Discover extensions registered under the manifest entry point group.

        Returns:
            List[ExtensionManifest]:
                Valid manifest-based extensions discovered from the
                `mada_tools.extensions` entry point group.
        """
        manifests: List[ExtensionManifest] = []
        seen_provider_packages: set[str] = set()

        for entry_point in self._load_entry_points("mada_tools.extensions"):
            loaded = self._load_manifest_factory(entry_point)
            if loaded is None:
                continue

            manifest = self._call_manifest_factory(entry_point.name, loaded)
            if manifest is None:
                continue

            if not self._validate_extension_manifest(manifest):
                continue

            if manifest.provider_package in seen_provider_packages:
                LOG.warning(
                    "Duplicate extension package '%s' discovered. Skipping later registration from '%s'.",
                    manifest.provider_package,
                    entry_point.value,
                )
                continue

            seen_provider_packages.add(manifest.provider_package)
            manifests.append(manifest)

        return manifests

    def _discover_legacy_server_extensions(self, existing_server_names: set[str]) -> List[ExtensionManifest]:
        """Discover legacy server entry points and adapt them into manifests.

        Args:
            existing_server_names (set[str]):
                Server names already registered by manifest-based extensions.
                Legacy registrations using these names are skipped so manifest
                registrations win per server.

        Returns:
            List[ExtensionManifest]:
                Synthetic manifests created from legacy
                `mada_tools.servers` entry points.
        """
        grouped_servers: dict[str, list[MCPServerRegistration]] = defaultdict(list)

        for entry_point in self._load_entry_points("mada_tools.servers"):
            module_path = entry_point.value.split(":")[0].strip()
            provider = getattr(entry_point, "dist", None)
            provider_name = getattr(provider, "name", None) or "unknown"

            # Prefer manifest-based registrations on a per-server basis so
            # providers can migrate incrementally.
            if entry_point.name in existing_server_names:
                continue

            server = MCPServerRegistration(
                name=entry_point.name,
                module_path=module_path,
                package=provider_name,
            )

            if not self._validate_mcp_server_registration(server):
                continue

            grouped_servers[provider_name].append(server)

        manifests: List[ExtensionManifest] = []
        for provider_name, servers in grouped_servers.items():
            manifest = ExtensionManifest(
                display_name=f"Legacy MCP servers from {provider_name}",
                version="legacy",
                provider_package=provider_name,
                mcp_servers=tuple(servers),
            )

            if self._validate_extension_manifest(manifest):
                manifests.append(manifest)

        return manifests

    def _load_entry_points(self, group: str) -> list[Any]:
        """Load entry points for one group across modern and legacy APIs.

        Args:
            group (str):
                Entry point group name to load.

        Returns:
            list[Any]:
                Entry point objects returned by the active metadata API.
        """
        try:
            from importlib.metadata import entry_points
        except Exception:
            from importlib_metadata import entry_points  # type: ignore

        try:
            eps = entry_points()
            return list(eps.select(group=group) if hasattr(eps, "select") else eps.get(group, []))
        except Exception as e:
            LOG.warning("Failed to read entry points for '%s': %s", group, e)
            return []

    def _load_manifest_factory(self, entry_point: Any) -> Any | None:
        """Load one manifest factory from an entry point.

        Args:
            entry_point (Any):
                Entry point object to load.

        Returns:
            Any | None:
                Loaded factory object, or `None` if loading fails.
        """
        try:
            return entry_point.load()
        except Exception as e:
            LOG.warning("Failed to load extension entry point '%s': %s", entry_point.value, e)
            return None

    def _call_manifest_factory(self, entry_point_name: str, loaded: Any) -> Optional[ExtensionManifest]:
        """Call a loaded manifest factory and return its manifest.

        Args:
            entry_point_name (str):
                Name of the entry point being invoked.
            loaded (Any):
                Loaded factory object returned from the entry point.

        Returns:
            Optional[ExtensionManifest]:
                The produced manifest, or `None` if validation fails.
        """
        if not callable(loaded):
            LOG.warning("Extension entry point '%s' did not load a callable factory", entry_point_name)
            return None

        try:
            manifest = loaded()
        except Exception as e:
            LOG.warning("Extension manifest factory '%s' failed: %s", entry_point_name, e)
            return None

        if not isinstance(manifest, ExtensionManifest):
            LOG.warning(
                "Extension manifest factory '%s' returned %s instead of ExtensionManifest",
                entry_point_name,
                type(manifest).__name__,
            )
            return None

        return manifest

    def _validate_extension_manifest(self, manifest: ExtensionManifest) -> bool:
        """Validate one manifest and its registered MCP servers.

        Args:
            manifest (ExtensionManifest):
                Manifest to validate.

        Returns:
            bool:
                `True` when the manifest is structurally valid, otherwise
                `False`.
        """
        if not manifest.provider_package:
            LOG.warning("Encountered extension manifest with empty provider_package")
            return False

        if not manifest.display_name:
            LOG.warning("Extension '%s' is missing display_name", manifest.provider_package)
            return False

        seen_server_names: set[str] = set()
        for server in manifest.mcp_servers:
            if not self._validate_mcp_server_registration(server):
                return False

            if server.name in seen_server_names:
                LOG.warning(
                    "Extension '%s' registers duplicate MCP server name '%s'",
                    manifest.provider_package,
                    server.name,
                )
                return False

            seen_server_names.add(server.name)

        return True

    def _validate_mcp_server_registration(self, server: MCPServerRegistration) -> bool:
        """Validate one MCP server registration, including runtime importability.

        Args:
            server (MCPServerRegistration):
                Server registration to validate.

        Returns:
            bool:
                `True` when the registration is usable, otherwise `False`.
        """
        if not server.name:
            LOG.warning("Encountered MCP server registration with empty name")
            return False

        if not server.module_path:
            LOG.warning("MCP server '%s' is missing module_path", server.name)
            return False

        if not server.package:
            LOG.warning("MCP server '%s' is missing package", server.name)
            return False

        try:
            mod = importlib.import_module(server.module_path)
        except Exception as e:
            LOG.warning(
                "Could not import server module for '%s' from '%s': %s",
                server.name,
                server.module_path,
                e,
            )
            return False

        if not hasattr(mod, "main") or not callable(getattr(mod, "main")):
            LOG.warning(
                "Server module '%s' for '%s' does not expose callable main()",
                server.module_path,
                server.name,
            )
            return False

        return True

discover_extensions()

Discover manifest-based and legacy extensions.

Returns:

Type	Description
List[ExtensionManifest]	List[ExtensionManifest]: Discovered extension manifests after manifest loading and legacy entry-point adaptation.

Source code in mada_tools/extensions/registry.py

def discover_extensions(self) -> List[ExtensionManifest]:
    """Discover manifest-based and legacy extensions.

    Returns:
        List[ExtensionManifest]:
            Discovered extension manifests after manifest loading and legacy
            entry-point adaptation.
    """
    manifests = self._discover_manifest_extensions()
    manifests.extend(
        self._discover_legacy_server_extensions(
            existing_server_names={server.name for manifest in manifests for server in manifest.mcp_servers}
        )
    )
    return manifests

get_available_mcp_servers()

Return validated MCP server registrations.

Returns:

Type	Description
List[MCPServerRegistration]	List[MCPServerRegistration]: MCP server registrations sorted by provider package and server name.

Source code in mada_tools/extensions/registry.py

def get_available_mcp_servers(self) -> List[MCPServerRegistration]:
    """Return validated MCP server registrations.

    Returns:
        List[MCPServerRegistration]:
            MCP server registrations sorted by provider package and server
            name.
    """
    return sorted(
        self.get_mcp_server_index().values(),
        key=lambda server: (server.package.lower(), server.name.lower()),
    )

get_mcp_server_index()

Return validated MCP servers indexed by server name.

Returns:

Type	Description
Dict[str, MCPServerRegistration]	Dict[str, MCPServerRegistration]: Mapping of server names to validated MCP server registrations.

Source code in mada_tools/extensions/registry.py

def get_mcp_server_index(self) -> Dict[str, MCPServerRegistration]:
    """Return validated MCP servers indexed by server name.

    Returns:
        Dict[str, MCPServerRegistration]:
            Mapping of server names to validated MCP server registrations.
    """
    available: Dict[str, MCPServerRegistration] = {}

    for manifest in self.discover_extensions():
        for server in manifest.mcp_servers:
            if server.name in available:
                LOG.warning(
                    "Plugin server name collision for '%s', already discovered. Skipping server from package '%s'.",
                    server.name,
                    manifest.provider_package,
                )
                continue

            available[server.name] = server

    return available

MCPServerRegistration dataclass

Describe one MCP server contributed by an extension.

Attributes:

Name	Type	Description
name	str	The server name used by MADA configuration and discovery output.
module_path	str	Importable Python module path for the server implementation.
package	str	Provider package name associated with the registration.
description	str | None	Optional human-readable description of the server registration.

Source code in mada_tools/extensions/manifest.py

@dataclass(frozen=True)
class MCPServerRegistration:
    """Describe one MCP server contributed by an extension.

    Attributes:
        name (str):
            The server name used by MADA configuration and discovery output.
        module_path (str):
            Importable Python module path for the server implementation.
        package (str):
            Provider package name associated with the registration.
        description (str | None):
            Optional human-readable description of the server registration.
    """

    name: str
    module_path: str
    package: str
    description: str | None = None