diff --git a/src/fastmcp/cli/cli.py b/src/fastmcp/cli/cli.py
index 3c11fa35f..d412afc65 100644
--- a/src/fastmcp/cli/cli.py
+++ b/src/fastmcp/cli/cli.py
@@ -23,6 +23,7 @@ from fastmcp.cli.auth import auth_app
 from fastmcp.cli.client import call_command, discover_command, list_command
 from fastmcp.cli.generate import generate_cli_command
 from fastmcp.cli.install import install_app
+from fastmcp.cli.plugin import plugin_app
 from fastmcp.cli.tasks import tasks_app
 from fastmcp.utilities.cli import is_already_in_uv_subprocess, load_and_merge_config
 from fastmcp.utilities.inspect import (
@@ -1102,6 +1103,9 @@ app.command(install_app)
 # Add tasks subcommand group
 app.command(tasks_app)
 
+# Add plugin subcommand group
+app.command(plugin_app)
+
 # Add client query commands
 app.command(list_command, name="list")
 app.command(call_command, name="call")
diff --git a/src/fastmcp/cli/plugin.py b/src/fastmcp/cli/plugin.py
new file mode 100644
index 000000000..4d7dda8d6
--- /dev/null
+++ b/src/fastmcp/cli/plugin.py
@@ -0,0 +1,102 @@
+"""CLI commands for working with FastMCP plugins.
+
+Currently exposes a single verb, `fastmcp plugin manifest`, which imports
+a plugin class and emits its manifest (metadata + config schema + entry
+point) as JSON. The manifest is the artifact downstream consumers
+(Horizon, registries, CI tooling) ingest to discover and configure the
+plugin without importing its module themselves.
+"""
+
+from __future__ import annotations
+
+import importlib
+import json
+import sys
+from pathlib import Path
+from typing import Annotated
+
+import cyclopts
+from cyclopts import Parameter
+
+from fastmcp.server.plugins import Plugin
+from fastmcp.server.plugins.base import PluginError
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger("cli.plugin")
+
+plugin_app = cyclopts.App(
+    name="plugin",
+    help="Work with FastMCP plugins.",
+    default_parameter=Parameter(negative=()),
+)
+
+
+def _resolve_plugin_class(entry_point: str) -> type[Plugin]:
+    """Import a plugin class from a `module.path:ClassName` spec.
+
+    The class portion may be dotted (e.g. `module:Outer.MyPlugin`) to
+    resolve a nested class, matching the `entry_point` format
+    `Plugin.manifest()` emits via `__qualname__`.
+    """
+    if ":" not in entry_point:
+        raise ValueError(
+            f"Invalid plugin reference {entry_point!r}: "
+            f"expected 'module.path:ClassName'"
+        )
+    module_path, class_name = entry_point.split(":", 1)
+    try:
+        module = importlib.import_module(module_path)
+    except ImportError as exc:
+        raise ImportError(f"Could not import module {module_path!r}: {exc}") from exc
+
+    cls: object = module
+    for part in class_name.split("."):
+        try:
+            cls = getattr(cls, part)
+        except AttributeError as exc:
+            raise AttributeError(
+                f"Module {module_path!r} has no attribute {class_name!r}"
+            ) from exc
+
+    if not isinstance(cls, type) or not issubclass(cls, Plugin):
+        raise TypeError(f"{entry_point!r} does not refer to a fastmcp.Plugin subclass")
+    return cls
+
+
+@plugin_app.command(name="manifest")
+def manifest_command(
+    entry_point: Annotated[
+        str,
+        Parameter(help="Plugin reference in 'module.path:ClassName' form."),
+    ],
+    output: Annotated[
+        Path | None,
+        Parameter(
+            name=["--output", "-o"],
+            help="Write manifest JSON to this path instead of stdout.",
+        ),
+    ] = None,
+) -> None:
+    """Emit a plugin's manifest as JSON.
+
+    Imports the referenced plugin class and prints its manifest to stdout,
+    or writes it to the path given by `-o/--output`.
+    """
+    try:
+        cls = _resolve_plugin_class(entry_point)
+    except (ImportError, AttributeError, TypeError, ValueError) as exc:
+        logger.error(str(exc))
+        sys.exit(1)
+
+    try:
+        manifest = cls.manifest()
+    except (PluginError, TypeError) as exc:
+        logger.error(str(exc))
+        sys.exit(1)
+
+    if output is None:
+        print(json.dumps(manifest, indent=2, sort_keys=False))
+        return
+
+    output.write_text(json.dumps(manifest, indent=2, sort_keys=False))
+    print(f"Wrote manifest for {cls.meta.name} to {output}")
diff --git a/src/fastmcp/server/mixins/lifespan.py b/src/fastmcp/server/mixins/lifespan.py
index b64d725ed..6ee9e99e7 100644
--- a/src/fastmcp/server/mixins/lifespan.py
+++ b/src/fastmcp/server/mixins/lifespan.py
@@ -171,6 +171,15 @@ class LifespanMixin:
             self._lifespan_result = user_lifespan_result
             self._lifespan_result_set = True
 
+            # Plugin setup pass: runs before provider lifespans and _started.
+            # Plugins may contribute providers, so this must happen before
+            # we start their lifespans below. Register teardown BEFORE the
+            # setup pass so that a partial-setup failure (one plugin's
+            # setup() raises after earlier plugins already initialized)
+            # still triggers teardown for the plugins that completed.
+            stack.push_async_callback(self._run_plugin_teardown)
+            await self._run_plugin_setup_pass()
+
             # Start lifespans for all providers
             for provider in self.providers:
                 await stack.enter_async_context(provider.lifespan())
diff --git a/src/fastmcp/server/plugins/__init__.py b/src/fastmcp/server/plugins/__init__.py
new file mode 100644
index 000000000..71af6dd1b
--- /dev/null
+++ b/src/fastmcp/server/plugins/__init__.py
@@ -0,0 +1,15 @@
+"""FastMCP plugin primitive.
+
+Plugins are reusable, configurable units that contribute middleware,
+transforms, providers, and custom HTTP routes to a FastMCP server. See
+the design document for the full specification.
+
+Only the two user-facing primitives are re-exported here: `Plugin`
+(subclass to define a plugin) and `PluginMeta` (the metadata model
+plugins instantiate). Error classes live in `fastmcp.server.plugins.base`
+and can be imported from there if needed.
+"""
+
+from fastmcp.server.plugins.base import Plugin, PluginMeta
+
+__all__ = ["Plugin", "PluginMeta"]
diff --git a/src/fastmcp/server/plugins/base.py b/src/fastmcp/server/plugins/base.py
new file mode 100644
index 000000000..de78d6aac
--- /dev/null
+++ b/src/fastmcp/server/plugins/base.py
@@ -0,0 +1,317 @@
+"""Plugin primitive for FastMCP.
+
+Plugins package server-side behavior — middleware, component transforms,
+providers, and custom HTTP routes — into reusable, configurable,
+distributable units. A plugin is a subclass of `Plugin` with a
+class-level `PluginMeta` and an optional nested `Config` model.
+
+See the design document for the full specification.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, ClassVar
+
+from packaging.requirements import InvalidRequirement, Requirement
+from packaging.specifiers import InvalidSpecifier, SpecifierSet
+from pydantic import BaseModel, ConfigDict, ValidationError
+
+import fastmcp
+from fastmcp.exceptions import FastMCPError
+from fastmcp.server.middleware import Middleware
+from fastmcp.server.providers import Provider
+from fastmcp.server.transforms import Transform
+
+if TYPE_CHECKING:
+    from starlette.routing import BaseRoute
+
+    from fastmcp.server.server import FastMCP
+
+
+class PluginError(FastMCPError):
+    """Base class for plugin-related errors."""
+
+
+class PluginConfigError(PluginError):
+    """Raised when a plugin's configuration fails validation."""
+
+
+class PluginCompatibilityError(PluginError):
+    """Raised when a plugin declares a FastMCP version it is not compatible with."""
+
+
+class PluginMeta(BaseModel):
+    """Descriptive metadata for a plugin.
+
+    Users who want typed custom fields subclass this model. Users who want
+    to attach ad-hoc fields without defining a model put them in the
+    `meta` dict. Unknown top-level fields are rejected to prevent future
+    collisions with standard fields.
+    """
+
+    name: str
+    """Plugin name. Required. Must be unique within a server."""
+
+    version: str
+    """Plugin version (plugin's own semver, independent of fastmcp)."""
+
+    description: str | None = None
+    """Short human-readable description."""
+
+    tags: list[str] = []
+    """Free-form tags for discovery and filtering."""
+
+    author: str | None = None
+    """Author identifier (person, team, or org)."""
+
+    homepage: str | None = None
+    """Homepage URL."""
+
+    dependencies: list[str] = []
+    """PEP 508 requirement specifiers for packages required to import and
+    run the plugin. Includes the plugin's own containing package plus any
+    runtime extras. FastMCP itself is implicit and must not be listed.
+    """
+
+    fastmcp_version: str | None = None
+    """Optional PEP 440 specifier expressing compatibility with FastMCP
+    core (e.g. `">=3.0"`). Verified at registration time.
+    """
+
+    meta: dict[str, Any] = {}
+    """Free-form bag for custom fields that have not been standardized.
+    Namespaced to prevent collisions with future standard fields.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+
+class Plugin:
+    """Base class for FastMCP plugins.
+
+    Subclass to define a plugin. A subclass must declare a class-level
+    `meta` attribute (a `PluginMeta` instance). It may optionally
+    declare a nested `Config` (subclass of `pydantic.BaseModel`)
+    describing its configuration schema, and override any of the lifecycle
+    and contribution hooks.
+
+    Example:
+        ```python
+        from fastmcp.server.plugins import Plugin, PluginMeta
+        from pydantic import BaseModel
+
+
+        class PIIRedactor(Plugin):
+            meta = PluginMeta(
+                name="pii-redactor",
+                version="0.3.0",
+                dependencies=[
+                    "fastmcp-plugin-pii>=0.3.0",
+                    "regex>=2024.0",
+                ],
+            )
+
+            class Config(BaseModel):
+                patterns: list[str] = ["ssn", "email"]
+
+            def middleware(self):
+                return [PIIMiddleware(self.config)]
+        ```
+    """
+
+    meta: ClassVar[PluginMeta]
+    """Class-level metadata. Required on every subclass."""
+
+    class Config(BaseModel):
+        """Default empty configuration. Subclasses override to declare fields."""
+
+        model_config = ConfigDict(extra="forbid")
+
+    config: BaseModel
+
+    # Framework-internal marker. Set to True by `FastMCP.add_plugin` when
+    # the plugin is added from inside another plugin's setup() (the loader
+    # pattern). The server removes ephemeral plugins and their
+    # contributions on teardown so loaders don't accumulate duplicates
+    # across lifespan cycles.
+    _fastmcp_ephemeral: bool = False
+
+    def __init__(self, config: BaseModel | dict[str, Any] | None = None) -> None:
+        # A subclass's nested Config is a distinct class from Plugin.Config;
+        # we accept any BaseModel instance here and validate at runtime that
+        # it's (or coerces to) the subclass's own Config type. This is why
+        # `config` is typed as BaseModel rather than the nested Config — the
+        # nested declaration does not imply subclass relationship.
+        meta = getattr(type(self), "meta", None)
+        if not isinstance(meta, PluginMeta):
+            raise TypeError(
+                f"{type(self).__name__} must declare a class-level "
+                f"'meta' attribute of type PluginMeta"
+            )
+        self._validate_meta(meta)
+
+        config_cls = type(self).Config
+        if config is None:
+            value: BaseModel = config_cls()
+        elif isinstance(config, config_cls):
+            value = config
+        elif isinstance(config, dict):
+            try:
+                value = config_cls(**config)
+            except ValidationError as exc:
+                raise PluginConfigError(
+                    f"Invalid configuration for {type(self).__name__}: {exc}"
+                ) from exc
+        else:
+            raise PluginConfigError(
+                f"Config for {type(self).__name__} must be a {config_cls.__name__} "
+                f"instance or dict, not {type(config).__name__}"
+            )
+        self.config = value
+
+    # -- validation -----------------------------------------------------------
+
+    @staticmethod
+    def _validate_meta(meta: PluginMeta) -> None:
+        """Check that the plugin's declared metadata is internally consistent."""
+        for dep in meta.dependencies:
+            try:
+                req = Requirement(dep)
+            except InvalidRequirement as exc:
+                raise PluginError(
+                    f"Plugin {meta.name!r}: invalid PEP 508 requirement {dep!r}: {exc}"
+                ) from exc
+            if req.name.lower().replace("_", "-") == "fastmcp":
+                raise PluginError(
+                    f"Plugin {meta.name!r}: 'fastmcp' must not appear in "
+                    f"dependencies. Use the 'fastmcp_version' field instead."
+                )
+
+        if meta.fastmcp_version is not None:
+            try:
+                SpecifierSet(meta.fastmcp_version)
+            except InvalidSpecifier as exc:
+                raise PluginError(
+                    f"Plugin {meta.name!r}: invalid fastmcp_version "
+                    f"specifier {meta.fastmcp_version!r}: {exc}"
+                ) from exc
+
+    def check_fastmcp_compatibility(self) -> None:
+        """Raise if the declared `fastmcp_version` excludes the running FastMCP."""
+        spec_str = self.meta.fastmcp_version
+        if spec_str is None:
+            return
+        spec = SpecifierSet(spec_str)
+        current = fastmcp.__version__
+        if current not in spec:
+            raise PluginCompatibilityError(
+                f"Plugin {self.meta.name!r} requires fastmcp {spec_str}, "
+                f"but running fastmcp is {current}."
+            )
+
+    # -- lifecycle ------------------------------------------------------------
+
+    async def setup(self, server: FastMCP) -> None:
+        """Called on each lifespan cycle during the server's setup pass, before the server binds.
+
+        Receives the server it's attaching to; may call
+        `server.add_plugin()` to register additional plugins (used by
+        loader plugins). Async so that plugins can open database
+        connections, warm HTTP clients, or otherwise perform
+        `await`-able initialization. Plugins must not assume other
+        plugins are present during their own `setup()` — the full list
+        may not yet be populated.
+        """
+
+    async def teardown(self) -> None:
+        """Called on each lifespan cycle when the server shuts down, in reverse registration order.
+
+        Async so that plugins can close connections, flush buffers, or
+        otherwise perform `await`-able cleanup.
+        """
+
+    # -- contribution hooks ---------------------------------------------------
+
+    def middleware(self) -> list[Middleware]:
+        """Return MCP-layer middleware to install on the server."""
+        return []
+
+    def transforms(self) -> list[Transform]:
+        """Return component transforms (tools, resources, prompts)."""
+        return []
+
+    def providers(self) -> list[Provider]:
+        """Return component providers."""
+        return []
+
+    def routes(self) -> list[BaseRoute]:
+        """Return custom HTTP routes to mount on the server's ASGI app.
+
+        Routes contributed here are **not authenticated by the framework**
+        — the MCP auth provider does not gate them. They are appropriate
+        for webhook endpoints whose callers carry their own authentication
+        scheme (e.g. an HMAC-signed header), and the plugin is responsible
+        for verifying inbound requests inside the handler.
+
+        Routes otherwise receive the full incoming HTTP request unchanged,
+        including all headers the client sent. If a caller has provided
+        the same credentials it would use for an authenticated MCP call,
+        those headers are available on `request.headers` for the handler
+        to inspect — the plugin chooses whether and how to validate them.
+        """
+        return []
+
+    # -- introspection --------------------------------------------------------
+
+    @classmethod
+    def manifest(
+        cls,
+        path: str | Path | None = None,
+    ) -> dict[str, Any] | None:
+        """Return the plugin's manifest as a dict, or write it to `path` as JSON.
+
+        Does not instantiate the plugin. The manifest is a JSON-serializable
+        dict that combines the plugin's metadata, its config schema, and an
+        importable entry point. Downstream consumers (Horizon, registries,
+        CI tooling) read the manifest to discover plugins and render
+        configuration forms without installing the plugin's dependencies.
+        """
+        meta = getattr(cls, "meta", None)
+        if not isinstance(meta, PluginMeta):
+            raise TypeError(
+                f"{cls.__name__} must declare a class-level "
+                f"'meta' attribute of type PluginMeta"
+            )
+
+        # Validate meta the same way instance construction does, so
+        # `fastmcp plugin manifest` can't emit an artifact (malformed
+        # PEP 508 deps, bad fastmcp_version specifier, fastmcp declared
+        # as a dep, ...) that downstream tooling couldn't otherwise
+        # have produced from a live plugin instance.
+        cls._validate_meta(meta)
+
+        config_cls = getattr(cls, "Config", Plugin.Config)
+        data: dict[str, Any] = {
+            "manifest_version": 1,
+            **meta.model_dump(),
+            "config_schema": config_cls.model_json_schema(),
+            "entry_point": f"{cls.__module__}:{cls.__qualname__}",
+        }
+
+        if path is None:
+            return data
+
+        target = Path(path)
+        target.write_text(json.dumps(data, indent=2, sort_keys=False))
+        return None
+
+
+__all__ = [
+    "Plugin",
+    "PluginCompatibilityError",
+    "PluginConfigError",
+    "PluginError",
+    "PluginMeta",
+]
diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py
index 2f00e2017..f9550dc66 100644
--- a/src/fastmcp/server/server.py
+++ b/src/fastmcp/server/server.py
@@ -68,6 +68,8 @@ from fastmcp.server.lifespan import Lifespan
 from fastmcp.server.low_level import LowLevelServer
 from fastmcp.server.middleware import Middleware, MiddlewareContext
 from fastmcp.server.mixins import LifespanMixin, MCPOperationsMixin, TransportMixin
+from fastmcp.server.plugins import Plugin
+from fastmcp.server.plugins.base import PluginError
 from fastmcp.server.providers import LocalProvider, Provider
 from fastmcp.server.providers.aggregate import AggregateProvider
 from fastmcp.server.tasks.config import TaskConfig, TaskMeta
@@ -185,16 +187,16 @@ def _get_auth_context() -> tuple[bool, Any]:
 def _is_model_visible(tool: Tool) -> bool:
     """Check whether a tool should be visible to the model.
 
-    Tools registered via ``@app.tool()`` (without ``model=True``) have
-    ``meta["ui"]["visibility"] == ["app"]`` — they are callable by app UIs
+    Tools registered via `@app.tool()` (without `model=True`) have
+    `meta["ui"]["visibility"] == ["app"]` — they are callable by app UIs
     but should not appear in the model's tool list.
 
     Returns True (visible) when:
-    - The tool has no ``meta.ui.visibility`` (normal tools).
-    - ``"model"`` is in the visibility list (e.g. ``["model"]`` or ``["app", "model"]``).
+    - The tool has no `meta.ui.visibility` (normal tools).
+    - `"model"` is in the visibility list (e.g. `["model"]` or `["app", "model"]`).
 
-    Returns False when the visibility list exists and does not contain ``"model"``
-    (e.g. ``["app"]``).
+    Returns False when the visibility list exists and does not contain `"model"`
+    (e.g. `["app"]`).
     """
     meta = tool.meta
     if not meta:
@@ -212,8 +214,8 @@ def _is_app_visible(tool: Tool) -> bool:
     """Check whether a tool has explicitly opted into app-callable visibility.
 
     Gates the dispatcher's hashed-name routing path: only tools whose
-    ``meta.ui.visibility`` list contains ``"app"`` can be reached via
-    ``<hash>_<local_name>`` calls. Tools without an explicit visibility
+    `meta.ui.visibility` list contains `"app"` can be reached via
+    `<hash>_<local_name>` calls. Tools without an explicit visibility
     declaration are NOT app-callable — they must be reached by their
     display name through the normal transform-aware resolution path.
 
@@ -295,6 +297,7 @@ class FastMCP(
         middleware: Sequence[Middleware] | None = None,
         providers: Sequence[Provider] | None = None,
         transforms: Sequence[Transform] | None = None,
+        plugins: Sequence[Plugin] | None = None,
         lifespan: LifespanCallable | Lifespan | None = None,
         tools: Sequence[Tool | Callable[..., Any]] | None = None,
         on_duplicate: DuplicateBehavior | None = None,
@@ -414,6 +417,33 @@ class FastMCP(
 
             self.middleware.append(DereferenceRefsMiddleware())
 
+        # Plugin registry: an ordered list, populated by `add_plugin()` and
+        # `plugins=[...]`. Setup and contribution collection happen during
+        # the server's lifespan startup (see `_run_plugin_setup_pass`).
+        self.plugins: list[Plugin] = []
+        # Plugins whose contributions (middleware/transforms/providers/routes)
+        # have been collected onto the server. Server contributions persist
+        # across lifespan cycles like server-authored middleware, so we
+        # don't re-install them on re-entry.
+        self._plugins_contributed: set[int] = set()
+        # Plugins whose setup() has completed successfully in the current
+        # lifespan cycle. Reset on teardown. Used to scope teardown to
+        # plugins that actually ran setup, so a partial-setup failure
+        # still triggers teardown for everything that was initialized.
+        self._plugins_set_up: set[int] = set()
+        # Per-plugin record of contributions we installed, stored as
+        # (container, item) tuples so we can reverse them when an
+        # ephemeral plugin is torn down.
+        self._plugin_contributions: dict[int, list[tuple[list[Any], Any]]] = {}
+        # True while the setup pass is executing. `add_plugin()` uses this
+        # to flag ephemeral plugins — plugins added from inside another
+        # plugin's setup() (the loader pattern). Ephemeral plugins are
+        # removed on teardown so loaders can freshly re-hydrate children
+        # on the next lifespan cycle without accumulating duplicates.
+        self._in_plugin_setup_pass: bool = False
+        for p in plugins or []:
+            self.add_plugin(p)
+
         # Set up MCP protocol handlers
         self._setup_handlers()
 
@@ -478,6 +508,182 @@ class FastMCP(
     def add_middleware(self, middleware: Middleware) -> None:
         self.middleware.append(middleware)
 
+    def add_plugin(self, plugin: Plugin) -> None:
+        """Register a plugin with this server.
+
+        Appends the plugin to the server's ordered plugin list and
+        synchronously collects its HTTP routes (see below). Middleware,
+        transforms, and providers are collected later, during the server's
+        startup sequence, because those hooks may reference state the
+        plugin populates during `setup()`.
+
+        HTTP routes are collected eagerly because HTTP transports snapshot
+        the server's route list when they construct the Starlette app —
+        which happens before the lifespan runs. A route returned by
+        `plugin.routes()` after the app is built would sit in
+        `_additional_http_routes` but never be mounted and would always
+        404. Collecting at registration time keeps non-loader plugins
+        working for HTTP transports.
+
+        Loader caveat: plugins added from inside another plugin's
+        `setup()` (the loader pattern) can still contribute middleware,
+        transforms, and providers, but their routes may not be reachable
+        over HTTP/SSE transports — those transports' route lists are
+        already fixed by the time `setup()` runs. Loaders that need to
+        contribute routes should use the stdio transport or expose the
+        routes via a non-loader plugin registered at construction time.
+
+        Raises:
+            PluginError: If called after the server has started, or if the
+                plugin's `fastmcp_version` compatibility check fails.
+
+        Args:
+            plugin: A :class:`Plugin` instance. Plugins are registered in
+                the order they are added; middleware is a stack.
+        """
+        if self._started.is_set():
+            raise PluginError(
+                f"Cannot add plugin {plugin.meta.name!r}: the server has "
+                "already started. Register plugins before the server binds."
+            )
+        plugin.check_fastmcp_compatibility()
+        # Compute routes up front so a failure inside plugin.routes() does
+        # not leave a half-registered plugin in self.plugins.
+        routes = list(plugin.routes())
+        self.plugins.append(plugin)
+        # Flag loader-added plugins as ephemeral so teardown can remove
+        # them along with their contributions. Written unconditionally so
+        # re-registering an instance that was previously marked ephemeral
+        # (added inside a setup pass and then cleaned up) as a permanent
+        # plugin clears the stale marker rather than inheriting it.
+        plugin._fastmcp_ephemeral = self._in_plugin_setup_pass
+        records = self._plugin_contributions.setdefault(id(plugin), [])
+        for route in routes:
+            self._additional_http_routes.append(route)
+            records.append((self._additional_http_routes, route))
+
+    async def _run_plugin_setup_pass(self) -> None:
+        """Run setup() on every registered plugin and collect contributions.
+
+        Called during server startup (from `_lifespan_manager`), before
+        the server binds. Iterates the plugin list in order, awaiting
+        `setup(server)` on each; plugins added during another plugin's
+        setup (the loader pattern) are picked up by the same loop because
+        the iteration advances against a live index.
+
+        Setup runs every lifespan cycle — plugins expect a matching
+        `setup`/`teardown` pair. Contribution collection is one-shot
+        per plugin: once a plugin's middleware/transforms/providers/routes
+        have been installed on the server they persist across cycles,
+        matching how server-authored middleware behaves.
+        """
+        # Setup pass: mutating-list iteration. New plugins appended by a
+        # plugin's setup() are picked up on subsequent iterations. We mark
+        # each plugin as "set up" only after setup() returns so that a
+        # partial-setup failure scopes teardown to plugins that actually
+        # completed initialization. The _in_plugin_setup_pass flag lets
+        # add_plugin() mark new plugins as ephemeral (see add_plugin).
+        self._in_plugin_setup_pass = True
+        try:
+            i = 0
+            while i < len(self.plugins):
+                plugin = self.plugins[i]
+                await plugin.setup(self)
+                self._plugins_set_up.add(id(plugin))
+                i += 1
+
+            # Contribution collection: run in registration order. Guarded
+            # per-plugin because contributions persist across lifespan
+            # cycles; new plugins (e.g. added by a loader during this
+            # cycle's setup) still get their contributions collected. Note:
+            # routes are collected synchronously at add_plugin() time
+            # because HTTP transports snapshot the route list before the
+            # lifespan runs.
+            for plugin in self.plugins:
+                if id(plugin) in self._plugins_contributed:
+                    continue
+                # Gather everything first: any hook that raises aborts
+                # before any server state is mutated, so a retry on the
+                # next lifespan cycle starts clean rather than appending
+                # duplicate middleware on top of partial contributions.
+                mws = list(plugin.middleware())
+                transforms_ = list(plugin.transforms())
+                providers_ = list(plugin.providers())
+
+                records = self._plugin_contributions.setdefault(id(plugin), [])
+                for mw in mws:
+                    self.add_middleware(mw)
+                    records.append((self.middleware, mw))
+                for transform in transforms_:
+                    self.add_transform(transform)
+                    records.append((self._transforms, transform))
+                for provider in providers_:
+                    # add_provider may wrap the value (for example a
+                    # FastMCP is wrapped in FastMCPProvider). Record
+                    # whatever actually landed in self.providers so
+                    # teardown can find it by identity.
+                    before = len(self.providers)
+                    self.add_provider(provider)
+                    for stored in self.providers[before:]:
+                        records.append((self.providers, stored))
+                self._plugins_contributed.add(id(plugin))
+        finally:
+            self._in_plugin_setup_pass = False
+
+    async def _run_plugin_teardown(self) -> None:
+        """Await `teardown()` on plugins whose setup() completed this cycle.
+
+        Iterates the plugin list in reverse registration order and only
+        tears down plugins that were successfully set up. This makes
+        teardown safe to register before `_run_plugin_setup_pass` — a
+        failure partway through setup still runs teardown for the plugins
+        that had already initialized.
+
+        Ephemeral plugins (those added from inside another plugin's
+        setup() by a loader) are removed from the server after teardown
+        runs, along with the contributions they installed. This keeps
+        loader plugins from accumulating duplicate children across
+        repeated lifespan cycles.
+        """
+        # Snapshot + clear first so that a slow teardown combined with a
+        # re-entry cannot double-count the set.
+        set_up = self._plugins_set_up
+        self._plugins_set_up = set()
+        for plugin in reversed(self.plugins):
+            if id(plugin) not in set_up:
+                continue
+            # Discard first so a plugin that appears twice in the list
+            # (same instance registered twice — see test_duplicates_allowed)
+            # only gets torn down once, even though its setup() ran once
+            # per list entry.
+            set_up.discard(id(plugin))
+            try:
+                await plugin.teardown()
+            except Exception:
+                logger.exception("Plugin %r raised during teardown", plugin.meta.name)
+
+        # Remove ephemeral plugins and their contributions. On the next
+        # lifespan cycle, the loader that produced them will freshly
+        # re-hydrate its children — without this cleanup, the plugin list
+        # and contribution registries would grow on every cycle.
+        ephemeral = [p for p in self.plugins if getattr(p, "_fastmcp_ephemeral", False)]
+        for plugin in ephemeral:
+            records = self._plugin_contributions.pop(id(plugin), [])
+            for container, item in reversed(records):
+                # Remove by identity rather than equality so a permanent
+                # contribution that happens to compare equal to `item`
+                # (e.g. a dataclass-style middleware with value-based
+                # `__eq__`) is not accidentally stripped. list.remove()
+                # uses `==`, which is the wrong matcher here.
+                for i, entry in enumerate(container):
+                    if entry is item:
+                        del container[i]
+                        break
+            self._plugins_contributed.discard(id(plugin))
+        self.plugins = [
+            p for p in self.plugins if not getattr(p, "_fastmcp_ephemeral", False)
+        ]
+
     def add_provider(self, provider: Provider, *, namespace: str = "") -> None:
         """Add a provider for dynamic tools, resources, and prompts.
 
@@ -497,10 +703,10 @@ class FastMCP(
     def _rewrite_prefab_uris(self, tools: list[Tool]) -> list[Tool]:
         """Replace placeholder Prefab URIs with per-tool hashed ones.
 
-        For each tool whose ``meta.ui.resourceUri`` is the placeholder,
-        reads the tool's stored hash from ``meta.fastmcp._tool_hash``
+        For each tool whose `meta.ui.resourceUri` is the placeholder,
+        reads the tool's stored hash from `meta.fastmcp._tool_hash`
         and rewrites the URI to the per-tool form. Also strips CSP from
-        tool meta (it belongs on the resource). Produces ``model_copy``
+        tool meta (it belongs on the resource). Produces `model_copy`
         views — originals are untouched.
         """
         from fastmcp.server.providers.prefab_synthesis import (
@@ -573,7 +779,7 @@ class FastMCP(
         """Add a tool transformation.
 
         .. deprecated::
-            Use ``add_transform(ToolTransform({...}))`` instead.
+            Use `add_transform(ToolTransform({...}))` instead.
         """
         if fastmcp.settings.deprecation_warnings:
             warnings.warn(
@@ -1569,7 +1775,7 @@ class FastMCP(
         """Remove tool(s) from the server.
 
         .. deprecated::
-            Use ``mcp.local_provider.remove_tool(name)`` instead.
+            Use `mcp.local_provider.remove_tool(name)` instead.
 
         Args:
             name: The name of the tool to remove.
@@ -2137,7 +2343,7 @@ class FastMCP(
         optionally with a given prefix.
 
         .. deprecated::
-            Use :meth:`mount` instead. ``import_server`` will be removed in a
+            Use :meth:`mount` instead. `import_server` will be removed in a
             future version.
 
         Note that when a server is *imported*, its objects are immediately
diff --git a/tests/cli/test_plugin_cli.py b/tests/cli/test_plugin_cli.py
new file mode 100644
index 000000000..553499a77
--- /dev/null
+++ b/tests/cli/test_plugin_cli.py
@@ -0,0 +1,95 @@
+"""Tests for the `fastmcp plugin` CLI."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import textwrap
+from pathlib import Path
+
+
+def _run_fastmcp(
+    *args: str, cwd: Path, env_extra: dict[str, str] | None = None
+) -> subprocess.CompletedProcess[str]:
+    """Invoke the `fastmcp` CLI as a subprocess."""
+    env = os.environ.copy()
+    env["PYTHONPATH"] = str(cwd)
+    if env_extra:
+        env.update(env_extra)
+    return subprocess.run(
+        [sys.executable, "-m", "fastmcp.cli", *args],
+        cwd=cwd,
+        capture_output=True,
+        text=True,
+        env=env,
+    )
+
+
+class TestManifestCLI:
+    def test_manifest_for_top_level_class(self, tmp_path: Path):
+        (tmp_path / "demo.py").write_text(
+            textwrap.dedent(
+                """
+                from fastmcp.server.plugins import Plugin, PluginMeta
+
+                class Demo(Plugin):
+                    meta = PluginMeta(name="demo", version="0.1.0")
+                """
+            )
+        )
+        result = _run_fastmcp("plugin", "manifest", "demo:Demo", cwd=tmp_path)
+        assert result.returncode == 0, result.stderr
+        manifest = json.loads(result.stdout)
+        assert manifest["name"] == "demo"
+        assert manifest["entry_point"] == "demo:Demo"
+
+    def test_manifest_for_nested_class(self, tmp_path: Path):
+        """`__qualname__` produces dotted paths for nested classes; the CLI
+        must traverse the dots to resolve the inner class."""
+        (tmp_path / "demo.py").write_text(
+            textwrap.dedent(
+                """
+                from fastmcp.server.plugins import Plugin, PluginMeta
+
+                class Outer:
+                    class Inner(Plugin):
+                        meta = PluginMeta(name="inner", version="0.1.0")
+                """
+            )
+        )
+        result = _run_fastmcp("plugin", "manifest", "demo:Outer.Inner", cwd=tmp_path)
+        assert result.returncode == 0, result.stderr
+        manifest = json.loads(result.stdout)
+        assert manifest["name"] == "inner"
+        assert manifest["entry_point"] == "demo:Outer.Inner"
+
+    def test_manifest_emits_clean_error_for_invalid_meta(self, tmp_path: Path):
+        """A plugin with invalid meta must produce a clean error, not a traceback."""
+        (tmp_path / "bad.py").write_text(
+            textwrap.dedent(
+                """
+                from fastmcp.server.plugins import Plugin, PluginMeta
+
+                class Bad(Plugin):
+                    meta = PluginMeta(
+                        name="bad",
+                        version="0.1.0",
+                        dependencies=["not a valid pep508 spec!!"],
+                    )
+                """
+            )
+        )
+        result = _run_fastmcp("plugin", "manifest", "bad:Bad", cwd=tmp_path)
+        assert result.returncode == 1
+        # Error goes through logger.error, not as a Python traceback.
+        assert "Traceback" not in result.stderr
+        assert "PEP 508" in result.stderr
+
+    def test_manifest_emits_clean_error_for_missing_module(self, tmp_path: Path):
+        result = _run_fastmcp(
+            "plugin", "manifest", "nonexistent_module:Thing", cwd=tmp_path
+        )
+        assert result.returncode == 1
+        assert "Traceback" not in result.stderr
diff --git a/tests/server/test_plugins.py b/tests/server/test_plugins.py
new file mode 100644
index 000000000..ae70dcf53
--- /dev/null
+++ b/tests/server/test_plugins.py
@@ -0,0 +1,881 @@
+"""Tests for the FastMCP plugin primitive."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from pydantic import BaseModel
+
+import fastmcp
+from fastmcp import Client, FastMCP
+from fastmcp.server.middleware import Middleware
+from fastmcp.server.plugins import Plugin, PluginMeta
+from fastmcp.server.plugins.base import (
+    PluginCompatibilityError,
+    PluginConfigError,
+    PluginError,
+)
+
+
+class _TraceMiddleware(Middleware):
+    """Tiny identity middleware tagged by name so we can see it in a stack."""
+
+    def __init__(self, tag: str) -> None:
+        self.tag = tag
+
+
+class _Recorder:
+    """Shared record of plugin lifecycle events for assertions in tests."""
+
+    def __init__(self) -> None:
+        self.events: list[tuple[str, str]] = []
+
+
+class TestPluginMeta:
+    """PluginMeta is the source-of-truth metadata model."""
+
+    def test_required_fields(self):
+        meta = PluginMeta(name="x", version="0.1.0")
+        assert meta.name == "x"
+        assert meta.version == "0.1.0"
+        assert meta.description is None
+        assert meta.tags == []
+        assert meta.dependencies == []
+        assert meta.fastmcp_version is None
+        assert meta.meta == {}
+
+    def test_unknown_top_level_field_rejected(self):
+        with pytest.raises(Exception):
+            PluginMeta(name="x", version="0.1.0", owning_team="platform")  # ty: ignore[unknown-argument]
+
+    def test_custom_fields_allowed_under_meta_dict(self):
+        meta = PluginMeta(
+            name="x",
+            version="0.1.0",
+            meta={"owning_team": "platform", "maintainer": "jlowin"},
+        )
+        assert meta.meta["owning_team"] == "platform"
+
+    def test_subclass_can_add_typed_fields(self):
+        class AcmeMeta(PluginMeta):
+            owning_team: str
+
+        meta = AcmeMeta(name="x", version="0.1.0", owning_team="platform")
+        assert meta.owning_team == "platform"
+
+
+class TestPluginConstruction:
+    """Plugin construction validates meta and config at instantiation time."""
+
+    def test_plugin_without_meta_raises(self):
+        class NoMeta(Plugin):
+            pass
+
+        with pytest.raises(TypeError, match="meta"):
+            NoMeta()
+
+    def test_plugin_with_default_config(self):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+        p = P()
+        assert isinstance(p.config, Plugin.Config)
+
+    def test_config_accepts_instance(self):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            class Config(BaseModel):
+                who: str = "world"
+
+        p = P(config=P.Config(who="jeremiah"))
+        assert isinstance(p.config, P.Config)
+        assert p.config.who == "jeremiah"
+
+    def test_config_accepts_dict(self):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            class Config(BaseModel):
+                who: str = "world"
+
+        p = P(config={"who": "jeremiah"})
+        assert isinstance(p.config, P.Config)
+        assert p.config.who == "jeremiah"
+
+    def test_invalid_config_raises_plugin_config_error(self):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            class Config(BaseModel):
+                count: int
+
+        with pytest.raises(PluginConfigError):
+            P(config={"count": "not a number"})
+
+    def test_bad_config_type_raises(self):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+        with pytest.raises(PluginConfigError):
+            P(config="not a config")  # ty: ignore[invalid-argument-type]
+
+
+class TestPluginValidation:
+    """Meta validation rejects malformed values eagerly."""
+
+    def test_fastmcp_in_dependencies_rejected(self):
+        class Bad(Plugin):
+            meta = PluginMeta(
+                name="bad",
+                version="0.1.0",
+                dependencies=["fastmcp>=3.0"],
+            )
+
+        with pytest.raises(PluginError, match="fastmcp"):
+            Bad()
+
+    def test_invalid_dependency_spec_rejected(self):
+        class Bad(Plugin):
+            meta = PluginMeta(
+                name="bad",
+                version="0.1.0",
+                dependencies=["not a valid pep508 spec!!"],
+            )
+
+        with pytest.raises(PluginError, match="PEP 508"):
+            Bad()
+
+    def test_invalid_fastmcp_version_spec_rejected(self):
+        class Bad(Plugin):
+            meta = PluginMeta(
+                name="bad",
+                version="0.1.0",
+                fastmcp_version="not-a-specifier",
+            )
+
+        with pytest.raises(PluginError, match="fastmcp_version"):
+            Bad()
+
+    def test_incompatible_fastmcp_version_raises(self, monkeypatch):
+        # Pin the version we're checking against so the test doesn't depend
+        # on whatever build-time version the running interpreter has (CI
+        # builds can resolve to "0.0.0" via uv-dynamic-versioning's
+        # fallback, which would match specifiers like "<0.1").
+        monkeypatch.setattr(fastmcp, "__version__", "3.0.0")
+
+        class Incompat(Plugin):
+            meta = PluginMeta(
+                name="incompat",
+                version="0.1.0",
+                fastmcp_version=">=100.0.0",
+            )
+
+        with pytest.raises(PluginCompatibilityError):
+            Incompat().check_fastmcp_compatibility()
+
+
+class TestRegistration:
+    """Plugins register before startup; add_plugin is a list append."""
+
+    def test_plugins_kwarg_registers(self):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+        mcp = FastMCP("t", plugins=[P(), P()])
+        assert [p.meta.name for p in mcp.plugins] == ["p", "p"]
+
+    def test_add_plugin_appends(self):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+        mcp = FastMCP("t")
+        mcp.add_plugin(P())
+        mcp.add_plugin(P())
+        assert len(mcp.plugins) == 2
+
+    def test_duplicates_allowed(self):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+        mcp = FastMCP("t")
+        mcp.add_plugin(P())
+        mcp.add_plugin(P())
+        # No dedup, no warn, no raise.
+        assert len(mcp.plugins) == 2
+
+    def test_add_plugin_checks_fastmcp_version_at_registration(self, monkeypatch):
+        monkeypatch.setattr(fastmcp, "__version__", "3.0.0")
+
+        class Incompat(Plugin):
+            meta = PluginMeta(
+                name="incompat",
+                version="0.1.0",
+                fastmcp_version=">=100.0.0",
+            )
+
+        mcp = FastMCP("t")
+        with pytest.raises(PluginCompatibilityError):
+            mcp.add_plugin(Incompat())
+
+    def test_add_plugin_does_not_call_setup(self):
+        """setup() runs during startup, not at add_plugin."""
+
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            async def setup(self, server):
+                raise AssertionError("setup should not run at registration time")
+
+        mcp = FastMCP("t")
+        mcp.add_plugin(P())  # must not raise
+
+
+class TestLifecycle:
+    """Setup and teardown run during the server's lifespan."""
+
+    async def test_setup_runs_during_startup(self):
+        recorder = _Recorder()
+
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            async def setup(self, server):
+                recorder.events.append(("setup", "p"))
+
+            async def teardown(self):
+                recorder.events.append(("teardown", "p"))
+
+        mcp = FastMCP("t", plugins=[P()])
+        async with Client(mcp) as c:
+            await c.ping()
+        assert recorder.events == [("setup", "p"), ("teardown", "p")]
+
+    async def test_setup_order_follows_registration(self):
+        recorder = _Recorder()
+
+        def make(name: str) -> type[Plugin]:
+            class _P(Plugin):
+                meta = PluginMeta(name=name, version="0.1.0")
+
+                async def setup(self, server):
+                    recorder.events.append(("setup", name))
+
+                async def teardown(self):
+                    recorder.events.append(("teardown", name))
+
+            return _P
+
+        A, B, C = make("a"), make("b"), make("c")
+        mcp = FastMCP("t", plugins=[A(), B()])
+        mcp.add_plugin(C())
+
+        async with Client(mcp) as c:
+            await c.ping()
+
+        # Setup in registration order; teardown reversed.
+        assert [e for e in recorder.events if e[0] == "setup"] == [
+            ("setup", "a"),
+            ("setup", "b"),
+            ("setup", "c"),
+        ]
+        assert [e for e in recorder.events if e[0] == "teardown"] == [
+            ("teardown", "c"),
+            ("teardown", "b"),
+            ("teardown", "a"),
+        ]
+
+    async def test_loader_pattern_adds_plugins_during_setup(self):
+        """A plugin's setup() can call server.add_plugin() and the setup pass sees it.
+
+        Mid-cycle the loader-added children are present; after teardown
+        they're removed (ephemeral cleanup), so the loader can freshly
+        re-hydrate them on the next cycle.
+        """
+        recorder = _Recorder()
+
+        class Child(Plugin):
+            meta = PluginMeta(name="child", version="0.1.0")
+
+            async def setup(self, server):
+                recorder.events.append(("setup", "child"))
+
+        class Loader(Plugin):
+            meta = PluginMeta(name="loader", version="0.1.0")
+
+            async def setup(self, server):
+                recorder.events.append(("setup", "loader"))
+                server.add_plugin(Child())
+                server.add_plugin(Child())
+
+        mcp = FastMCP("t", plugins=[Loader()])
+        async with Client(mcp) as c:
+            await c.ping()
+            # Mid-cycle, the loader's children are registered.
+            assert [p.meta.name for p in mcp.plugins] == [
+                "loader",
+                "child",
+                "child",
+            ]
+
+        assert recorder.events == [
+            ("setup", "loader"),
+            ("setup", "child"),
+            ("setup", "child"),
+        ]
+        # After teardown, ephemeral children have been removed.
+        assert [p.meta.name for p in mcp.plugins] == ["loader"]
+
+    async def test_add_plugin_after_startup_raises(self):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+        mcp = FastMCP("t")
+        async with Client(mcp) as c:
+            await c.ping()
+            with pytest.raises(PluginError, match="already started"):
+                mcp.add_plugin(P())
+
+    async def test_duplicate_registration_tears_down_once(self):
+        """Registering the same instance twice must only call teardown() once.
+
+        setup() runs per list entry (so the plugin receives both entries),
+        but teardown() is an idempotent cleanup — a second call on a
+        plugin that has closed its resources would likely raise on an
+        already-closed connection.
+        """
+        recorder = _Recorder()
+
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            async def teardown(self):
+                recorder.events.append(("teardown", "p"))
+
+        p = P()
+        mcp = FastMCP("t")
+        mcp.add_plugin(p)
+        mcp.add_plugin(p)
+
+        async with Client(mcp) as c:
+            await c.ping()
+
+        assert [e for e in recorder.events if e[0] == "teardown"] == [
+            ("teardown", "p"),
+        ]
+
+    async def test_teardown_exception_is_logged_not_raised(self):
+        class Boom(Plugin):
+            meta = PluginMeta(name="boom", version="0.1.0")
+
+            async def teardown(self):
+                raise RuntimeError("boom")
+
+        mcp = FastMCP("t", plugins=[Boom()])
+        # Should not raise out of the client context manager.
+        async with Client(mcp) as c:
+            await c.ping()
+
+    async def test_setup_and_teardown_run_on_every_lifespan_cycle(self):
+        """A server reused across multiple lifespan cycles re-runs setup/teardown."""
+        recorder = _Recorder()
+
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            async def setup(self, server):
+                recorder.events.append(("setup", "p"))
+
+            async def teardown(self):
+                recorder.events.append(("teardown", "p"))
+
+        mcp = FastMCP("t", plugins=[P()])
+
+        async with Client(mcp) as c:
+            await c.ping()
+        async with Client(mcp) as c:
+            await c.ping()
+
+        # Both cycles run setup and teardown; a one-shot guard would have
+        # skipped the second cycle.
+        assert recorder.events == [
+            ("setup", "p"),
+            ("teardown", "p"),
+            ("setup", "p"),
+            ("teardown", "p"),
+        ]
+
+    async def test_contributions_not_doubled_across_lifespan_cycles(self):
+        """Contribution hooks are collected once per plugin, not per cycle."""
+
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            def middleware(self):
+                return [_TraceMiddleware("p")]
+
+        mcp = FastMCP("t", plugins=[P()])
+
+        async with Client(mcp) as c:
+            await c.ping()
+        async with Client(mcp) as c:
+            await c.ping()
+
+        tags = [m.tag for m in mcp.middleware if isinstance(m, _TraceMiddleware)]
+        assert tags == ["p"]
+
+    async def test_teardown_runs_for_plugins_that_set_up_when_later_plugin_fails(self):
+        """Partial-setup failure still triggers teardown on already-initialized plugins."""
+        recorder = _Recorder()
+
+        class Good(Plugin):
+            meta = PluginMeta(name="good", version="0.1.0")
+
+            async def setup(self, server):
+                recorder.events.append(("setup", "good"))
+
+            async def teardown(self):
+                recorder.events.append(("teardown", "good"))
+
+        class BadSetup(Plugin):
+            meta = PluginMeta(name="bad", version="0.1.0")
+
+            async def setup(self, server):
+                recorder.events.append(("setup", "bad"))
+                raise RuntimeError("setup failed")
+
+            async def teardown(self):
+                # Must not be called — setup() never completed.
+                recorder.events.append(("teardown", "bad"))
+
+        mcp = FastMCP("t", plugins=[Good(), BadSetup()])
+
+        with pytest.raises(RuntimeError, match="setup failed"):
+            async with Client(mcp) as c:
+                await c.ping()
+
+        assert ("setup", "good") in recorder.events
+        assert ("setup", "bad") in recorder.events
+        assert ("teardown", "good") in recorder.events
+        # BadSetup never completed setup(); its teardown must not run.
+        assert ("teardown", "bad") not in recorder.events
+
+    async def test_contribution_collection_is_atomic_when_later_hook_raises(self):
+        """A failing hook on one plugin must not leave partial contributions behind.
+
+        If a plugin's ``middleware()`` succeeds but ``transforms()``
+        raises, the middleware must not have been installed — otherwise a
+        retry on the next lifespan attempt would pick up the plugin
+        again (because we never marked it contributed) and append
+        duplicate middleware on top of the partial prior state.
+        """
+
+        class Flaky(Plugin):
+            meta = PluginMeta(name="flaky", version="0.1.0")
+            _fail: bool = True
+
+            def middleware(self):
+                return [_TraceMiddleware("flaky")]
+
+            def transforms(self):
+                if Flaky._fail:
+                    raise RuntimeError("transforms exploded")
+                return []
+
+        mcp = FastMCP("t", plugins=[Flaky()])
+        baseline = list(mcp.middleware)
+
+        with pytest.raises(RuntimeError, match="transforms exploded"):
+            async with Client(mcp) as c:
+                await c.ping()
+
+        # Partial state from the failed cycle must not have landed.
+        assert mcp.middleware == baseline
+
+        # Retry succeeds; middleware is installed exactly once.
+        Flaky._fail = False
+        async with Client(mcp) as c:
+            await c.ping()
+
+        tags = [m.tag for m in mcp.middleware if isinstance(m, _TraceMiddleware)]
+        assert tags == ["flaky"]
+
+    async def test_add_plugin_is_atomic_when_routes_raises(self):
+        """If plugin.routes() raises, the plugin must not be left in the server's list.
+
+        Otherwise a later startup would run the half-registered plugin's
+        lifecycle even though registration reported an error.
+        """
+
+        class RoutesBoom(Plugin):
+            meta = PluginMeta(name="routes-boom", version="0.1.0")
+
+            def routes(self):
+                raise RuntimeError("routes exploded")
+
+        mcp = FastMCP("t")
+        with pytest.raises(RuntimeError, match="routes exploded"):
+            mcp.add_plugin(RoutesBoom())
+
+        assert mcp.plugins == []
+        # Contribution book-keeping for the failed plugin was never created.
+        # This is a weaker assertion — we just care the plugin isn't linger.
+        assert not any(isinstance(p, RoutesBoom) for p in mcp.plugins)
+
+    async def test_ephemeral_fastmcp_provider_is_removed_on_teardown(self):
+        """Loader-added FastMCP providers are auto-wrapped; teardown must still remove them.
+
+        ``add_provider`` wraps a FastMCP in a FastMCPProvider before it
+        lands in ``self.providers``. Recording the pre-wrap object would
+        cause teardown to miss the wrapped provider and leak it across
+        cycles.
+        """
+
+        class ProviderPlugin(Plugin):
+            meta = PluginMeta(name="wrapper", version="0.1.0")
+
+            def __init__(self, config=None):
+                super().__init__(config)
+                self._child = FastMCP("child")
+
+            def providers(self):
+                return [self._child]
+
+        class Loader(Plugin):
+            meta = PluginMeta(name="loader", version="0.1.0")
+
+            async def setup(self, server):
+                server.add_plugin(ProviderPlugin())
+
+        mcp = FastMCP("t", plugins=[Loader()])
+        baseline_providers = list(mcp.providers)
+
+        async with Client(mcp) as c:
+            await c.ping()
+        async with Client(mcp) as c:
+            await c.ping()
+
+        assert [p.meta.name for p in mcp.plugins] == ["loader"]
+        # The wrapped provider that was added on each cycle was removed
+        # on each teardown — the provider list is back to baseline.
+        assert mcp.providers == baseline_providers
+
+    async def test_ephemeral_cleanup_removes_by_identity_not_equality(self):
+        """A permanent contribution that compares equal to an ephemeral one is preserved.
+
+        list.remove() uses `==`, which is the wrong matcher when a
+        middleware defines value-based equality. A loader-added middleware
+        that happens to `==` a user-registered middleware must not cause
+        the user's to be removed during ephemeral cleanup.
+        """
+
+        class EqMiddleware(Middleware):
+            """Middleware that compares equal to any other EqMiddleware."""
+
+            def __eq__(self, other):
+                return isinstance(other, EqMiddleware)
+
+            def __hash__(self):
+                return 0
+
+        permanent = EqMiddleware()
+
+        class Child(Plugin):
+            meta = PluginMeta(name="child", version="0.1.0")
+
+            def middleware(self):
+                # A distinct instance, but equal to `permanent` by __eq__.
+                return [EqMiddleware()]
+
+        class Loader(Plugin):
+            meta = PluginMeta(name="loader", version="0.1.0")
+
+            async def setup(self, server):
+                server.add_plugin(Child())
+
+        mcp = FastMCP("t", middleware=[permanent], plugins=[Loader()])
+        assert permanent in mcp.middleware
+
+        async with Client(mcp) as c:
+            await c.ping()
+
+        # The ephemeral child's middleware was removed; the permanent
+        # user-registered one (which was `==` to it) is still installed.
+        assert any(m is permanent for m in mcp.middleware)
+
+    async def test_reregistering_ephemeral_instance_as_permanent_clears_marker(self):
+        """A previously-ephemeral instance re-registered by the user is permanent.
+
+        Without clearing the marker on normal `add_plugin`, the second
+        registration would inherit `_fastmcp_ephemeral = True` from the
+        first (loader-added) cycle and get deleted during teardown, losing
+        its contributions.
+        """
+        leaked: list[Plugin] = []
+
+        class Child(Plugin):
+            meta = PluginMeta(name="child", version="0.1.0")
+
+            def middleware(self):
+                return [_TraceMiddleware("child")]
+
+        class Loader(Plugin):
+            meta = PluginMeta(name="loader", version="0.1.0")
+
+            async def setup(self, server):
+                # The loader is in control of the instance, so we can
+                # hand it back to the test via a closure.
+                child = Child()
+                leaked.append(child)
+                server.add_plugin(child)
+
+        mcp = FastMCP("t", plugins=[Loader()])
+
+        async with Client(mcp) as c:
+            await c.ping()
+
+        # Ephemeral cleanup ran — child is no longer in the plugin list,
+        # and its middleware is gone.
+        assert [p.meta.name for p in mcp.plugins] == ["loader"]
+        child_instance = leaked[0]
+        assert child_instance._fastmcp_ephemeral is True
+
+        # User re-registers the same instance as a permanent plugin.
+        mcp.add_plugin(child_instance)
+        assert child_instance._fastmcp_ephemeral is False
+
+        async with Client(mcp) as c:
+            await c.ping()
+
+        # After a second cycle, the permanent registration survives and
+        # its middleware is installed exactly once.
+        assert child_instance in mcp.plugins
+        tags = [m.tag for m in mcp.middleware if isinstance(m, _TraceMiddleware)]
+        assert tags == ["child"]
+
+    async def test_loader_plugins_do_not_accumulate_across_cycles(self):
+        """Loader-added (ephemeral) plugins and their contributions are removed on teardown.
+
+        Without this, a loader that adds children in setup() causes the
+        plugin list — and every contribution those children install — to
+        grow on every lifespan cycle.
+        """
+
+        class Child(Plugin):
+            meta = PluginMeta(name="child", version="0.1.0")
+
+            def middleware(self):
+                return [_TraceMiddleware("child")]
+
+        class Loader(Plugin):
+            meta = PluginMeta(name="loader", version="0.1.0")
+
+            async def setup(self, server):
+                server.add_plugin(Child())
+
+        mcp = FastMCP("t", plugins=[Loader()])
+        baseline_middleware = list(mcp.middleware)
+
+        async with Client(mcp) as c:
+            await c.ping()
+        async with Client(mcp) as c:
+            await c.ping()
+        async with Client(mcp) as c:
+            await c.ping()
+
+        # After three cycles: the loader remains, the ephemeral child has
+        # been removed, and the middleware it installed was reversed out
+        # each time so nothing has accumulated.
+        assert [p.meta.name for p in mcp.plugins] == ["loader"]
+        assert mcp.middleware == baseline_middleware
+
+
+class TestContributions:
+    """Plugin contributions are installed during the setup pass."""
+
+    async def test_middleware_contribution(self):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            def middleware(self):
+                return [_TraceMiddleware("p")]
+
+        mcp = FastMCP("t", plugins=[P()])
+        async with Client(mcp) as c:
+            await c.ping()
+
+        tags = [m.tag for m in mcp.middleware if isinstance(m, _TraceMiddleware)]
+        assert tags == ["p"]
+
+    async def test_contribution_order_follows_registration(self):
+        class P(Plugin):
+            def __init__(self, name: str) -> None:
+                super().__init__()
+                self._name = name
+
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            def middleware(self):
+                return [_TraceMiddleware(self._name)]
+
+        a, b = P("a"), P("b")
+        mcp = FastMCP("t", plugins=[a, b])
+        async with Client(mcp) as c:
+            await c.ping()
+
+        tags = [m.tag for m in mcp.middleware if isinstance(m, _TraceMiddleware)]
+        assert tags == ["a", "b"]
+
+    async def test_custom_route_contribution(self):
+        from starlette.responses import JSONResponse
+        from starlette.routing import Route
+
+        async def health(request):
+            return JSONResponse({"ok": True})
+
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            def routes(self):
+                return [Route("/healthz", endpoint=health, methods=["GET"])]
+
+        mcp = FastMCP("t", plugins=[P()])
+        async with Client(mcp) as c:
+            await c.ping()
+
+        assert any(
+            getattr(r, "path", None) == "/healthz" for r in mcp._additional_http_routes
+        )
+
+    def test_plugin_route_mounted_on_http_app(self):
+        """Plugin routes must be in place before http_app() snapshots routes.
+
+        Regression test for collecting routes at ``add_plugin()`` time
+        rather than during the lifespan's setup pass. HTTP transports
+        call ``_get_additional_http_routes()`` at app construction, which
+        happens before the lifespan runs; routes added during setup would
+        sit in ``_additional_http_routes`` but never be mounted and would
+        always 404.
+        """
+
+        def _walk_paths(routes):
+            for route in routes:
+                path = getattr(route, "path", None)
+                if path is not None:
+                    yield path
+                inner = getattr(route, "routes", None)
+                if inner:
+                    yield from _walk_paths(inner)
+
+        from starlette.responses import JSONResponse
+        from starlette.routing import Route
+
+        async def health(request):
+            return JSONResponse({"ok": True})
+
+        class Health(Plugin):
+            meta = PluginMeta(name="health", version="0.1.0")
+
+            def routes(self):
+                return [Route("/healthz", endpoint=health, methods=["GET"])]
+
+        mcp = FastMCP("t", plugins=[Health()])
+        app = mcp.http_app()
+
+        paths = set(_walk_paths(app.router.routes))
+        assert "/healthz" in paths
+
+
+class TestManifest:
+    """manifest() produces a JSON-serializable dict and can write to disk."""
+
+    def test_manifest_shape(self):
+        class P(Plugin):
+            meta = PluginMeta(
+                name="p",
+                version="0.1.0",
+                description="demo",
+                tags=["x"],
+                dependencies=["demo>=0.1"],
+                fastmcp_version=">=3.0",
+                meta={"owning_team": "platform"},
+            )
+
+            class Config(BaseModel):
+                who: str = "world"
+
+        m = P.manifest()
+        assert m is not None
+        assert m["manifest_version"] == 1
+        assert m["name"] == "p"
+        assert m["version"] == "0.1.0"
+        assert m["description"] == "demo"
+        assert m["tags"] == ["x"]
+        assert m["dependencies"] == ["demo>=0.1"]
+        assert m["fastmcp_version"] == ">=3.0"
+        assert m["meta"] == {"owning_team": "platform"}
+        assert ":" in m["entry_point"]
+        assert m["entry_point"].endswith(".P")
+        assert m["config_schema"]["type"] == "object"
+        assert "who" in m["config_schema"]["properties"]
+
+    def test_manifest_custom_fields_subclass(self):
+        class AcmeMeta(PluginMeta):
+            owning_team: str
+
+        class P(Plugin):
+            meta = AcmeMeta(name="p", version="0.1.0", owning_team="platform")
+
+        m = P.manifest()
+        assert m is not None
+        assert m["owning_team"] == "platform"
+
+    def test_manifest_write_to_path(self, tmp_path: Path):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+        out = tmp_path / "plugin.json"
+        result = P.manifest(path=out)
+        assert result is None
+        data = json.loads(out.read_text())
+        assert data["name"] == "p"
+
+    def test_manifest_does_not_instantiate(self):
+        class P(Plugin):
+            meta = PluginMeta(name="p", version="0.1.0")
+
+            def __init__(self, *_args, **_kwargs):  # type: ignore[no-untyped-def]
+                raise AssertionError("manifest() must not instantiate the plugin")
+
+        # Should succeed without calling __init__.
+        assert P.manifest() is not None
+
+    def test_manifest_validates_meta(self):
+        """Invalid meta (e.g. malformed deps) must not emit a manifest.
+
+        Otherwise `fastmcp plugin manifest` could publish artifacts with
+        malformed PEP 508 dep strings or bad fastmcp_version specifiers —
+        artifacts that downstream tooling can't parse consistently.
+        """
+
+        class BadDeps(Plugin):
+            meta = PluginMeta(
+                name="bad-deps",
+                version="0.1.0",
+                dependencies=["not a valid pep508 spec!!"],
+            )
+
+        with pytest.raises(PluginError, match="PEP 508"):
+            BadDeps.manifest()
+
+        class FastmcpInDeps(Plugin):
+            meta = PluginMeta(
+                name="fastmcp-in-deps",
+                version="0.1.0",
+                dependencies=["fastmcp>=3.0"],
+            )
+
+        with pytest.raises(PluginError, match="fastmcp"):
+            FastmcpInDeps.manifest()