fastmcp/tests/server/test_plugins.py

"""Tests for the FastMCP plugin primitive."""

from __future__ import annotations

import asyncio
import json
from contextlib import asynccontextmanager, suppress
from importlib import metadata as importlib_metadata
from importlib.metadata import version as dist_version
from pathlib import Path
from typing import Generic, TypeVar

import pydantic
import pytest
from packaging.version import Version
from pydantic import BaseModel, ValidationError

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 _TestPlugin(Plugin):
    """Base for test plugins. Relies on Plugin's auto-derived meta —
    subclasses override `meta` only when a test asserts on a specific
    name or version."""


class TestPluginMeta:
    """PluginMeta is the source-of-truth metadata model."""

    def test_required_fields(self):
        meta = PluginMeta(name="x")
        assert meta.name == "x"
        # `version` is optional — bundled plugins don't track a separate
        # release cadence from their container.
        assert meta.version is None
        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"

    def test_version_is_optional_and_defaults_to_none(self):
        """Bundled plugins don't have an independent version; `None` is
        the honest answer and avoids both lockstep lies (phantom bumps)
        and sentinel strings like "bundled" that break semver consumers."""
        meta = PluginMeta(name="bundled")
        assert meta.version is None
        # Manifest emission keeps the field — consumers see `null` and
        # can render "bundled" or similar at the presentation layer.
        assert meta.model_dump()["version"] is None

    def test_explicit_version_still_accepted(self):
        """Published plugins set a real semver, typically via
        `PluginMeta.from_package(...)`; the field still accepts any string."""
        meta = PluginMeta(name="published", version="1.2.3")
        assert meta.version == "1.2.3"


class TestFromPackage:
    """PluginMeta.from_package() derives metadata from importlib.metadata."""

    # pydantic is a hard dependency of fastmcp, so it's always installed
    # in the test environment and has well-formed metadata we can read.
    # We deliberately don't use fastmcp itself as the smoke-test
    # distribution because from_package() refuses to pin fastmcp (see
    # test_fastmcp_as_distribution_is_rejected).

    def test_derives_version_description_from_real_package(self):
        meta = PluginMeta.from_package("pydantic", name="pydantic-smoke-test")

        assert meta.name == "pydantic-smoke-test"
        assert meta.version == dist_version("pydantic")
        # Description is whatever pydantic itself declares; only assert
        # that the field is populated.
        assert meta.description is not None
        # Dep pin uses `Version.public` (strips only local segment;
        # preserves pre/dev/post, which ARE valid with `>=` per PEP 440).
        public = Version(dist_version("pydantic")).public
        assert meta.dependencies == [f"pydantic>={public}"]

    def test_overrides_take_precedence(self):
        meta = PluginMeta.from_package(
            "pydantic",
            name="override-test",
            version="99.0.0",
            description="I override the derived description",
            tags=["security"],
        )
        assert meta.version == "99.0.0"
        assert meta.description == "I override the derived description"
        assert meta.tags == ["security"]

    def test_overriding_dependencies_replaces_the_pin(self):
        """If a plugin author passes dependencies explicitly, the containing
        distribution pin isn't re-added — author owns the list."""
        meta = PluginMeta.from_package(
            "pydantic",
            name="custom-deps",
            dependencies=["regex>=2024.0"],
        )
        assert meta.dependencies == ["regex>=2024.0"]

    def test_missing_distribution_raises_plugin_error(self):
        with pytest.raises(PluginError, match="not installed"):
            PluginMeta.from_package(
                "this-package-definitely-does-not-exist-1234abcd",
                name="missing",
            )

    def test_fastmcp_as_distribution_is_rejected(self):
        """`fastmcp` is implicit per the primitive contract; pinning it
        in `dependencies` would produce a manifest that fails validation."""
        with pytest.raises(PluginError, match="implicit"):
            PluginMeta.from_package("fastmcp", name="would-be-fastmcp-plugin")

    def test_fastmcp_rejection_is_case_insensitive(self):
        """PEP 503 canonicalization lowercases the distribution name, so
        `FastMCP` and `FASTMCP` both canonicalize to `fastmcp` and must
        be rejected. `fast-mcp` / `fast_mcp` canonicalize to `fast-mcp`
        — a different distribution — and are not rejected here."""
        for variant in ("FastMCP", "FASTMCP", "fAsTmCp"):
            with pytest.raises(PluginError, match="implicit"):
                PluginMeta.from_package(variant, name="x")

    @pytest.mark.parametrize(
        "dist_version_str, expected_pin",
        [
            # Pre/dev/post segments are valid with `>=` and must be
            # preserved so the pin tracks prerelease channels accurately.
            ("1.2.3.dev0", "synthetic-pin>=1.2.3.dev0"),
            ("1.2.3rc1", "synthetic-pin>=1.2.3rc1"),
            ("1.2.3.post1", "synthetic-pin>=1.2.3.post1"),
            # Local versions are NOT valid with `>=` per PEP 440; we
            # strip only that segment via Version.public.
            ("1.2.3+abc.def", "synthetic-pin>=1.2.3"),
            # Dev build with a local segment: strip just the local.
            ("1.2.3.dev5+abc123", "synthetic-pin>=1.2.3.dev5"),
            # Plain release — unchanged.
            ("2.0.0", "synthetic-pin>=2.0.0"),
        ],
    )
    def test_pin_preserves_pre_dev_post_but_strips_local(
        self, monkeypatch, dist_version_str, expected_pin
    ):
        """PEP 440 restricts only local versions from `>=` / `<=`;
        prereleases, dev, and post segments remain valid. The pin uses
        `Version.public` (strips only the local segment) so development
        channels keep their identity in the generated pin."""
        real_distribution = importlib_metadata.distribution

        class FakeDist:
            version = dist_version_str

            def __init__(self):
                self.metadata = real_distribution("pydantic").metadata

        def fake_distribution(name):
            if name == "synthetic-pin":
                return FakeDist()
            return real_distribution(name)

        # `from_package` reaches `importlib_metadata.distribution` through
        # the `plugins.base` module's alias; patch there.
        from fastmcp.server.plugins import base as plugins_base

        monkeypatch.setattr(
            plugins_base.importlib_metadata, "distribution", fake_distribution
        )

        meta = PluginMeta.from_package("synthetic-pin", name="pin-test")
        assert meta.dependencies == [expected_pin]

        # Resulting meta round-trips through _validate_meta cleanly.
        Plugin._validate_meta(meta)

    def test_whitespace_only_author_header_falls_back_to_email(self, monkeypatch):
        """A METADATA file with `Author: ` (whitespace only) must not
        block the `Author-email` fallback. Similar for `Home-page`
        falling back to Project-URL."""
        real_distribution = importlib_metadata.distribution
        pydantic_metadata = real_distribution("pydantic").metadata

        class FakeMessage:
            def items(self):
                return [
                    ("Metadata-Version", "2.1"),
                    ("Name", "whitespace-test"),
                    ("Version", "1.0.0"),
                    ("Author", "   "),  # whitespace only
                    ("Author-email", "real@example.com"),
                    ("Home-page", ""),  # empty
                    ("Project-URL", "Homepage, https://example.com"),
                ]

        class FakeDist:
            version = "1.0.0"
            metadata = FakeMessage()

        def fake_distribution(name):
            if name == "whitespace-test":
                return FakeDist()
            return real_distribution(name)

        from fastmcp.server.plugins import base as plugins_base

        monkeypatch.setattr(
            plugins_base.importlib_metadata, "distribution", fake_distribution
        )

        meta = PluginMeta.from_package("whitespace-test", name="ws-test")
        # Whitespace Author didn't block Author-email.
        assert meta.author == "real@example.com"
        # Empty Home-page fell through to the Project-URL label match.
        assert meta.homepage == "https://example.com"

        # Avoid "pydantic_metadata unused" lint noise.
        _ = pydantic_metadata

    def test_name_override_required_if_not_provided(self):
        """`name` is required on PluginMeta; from_package doesn't default
        it from the distribution name (plugin name and distribution name
        serve different purposes)."""
        with pytest.raises(ValidationError):
            PluginMeta.from_package("pydantic")


class TestPluginConstruction:
    """Plugin construction validates meta and config at instantiation time."""

    def test_plugin_without_meta_auto_derives_from_class_name(self):
        class ChannelPlugin(Plugin):
            pass

        p = ChannelPlugin()
        # Class name is kebab-cased and the trailing "Plugin" suffix stripped.
        assert p.meta.name == "channel"
        # Bundled plugins have no independent version.
        assert p.meta.version is None

    def test_plugin_meta_auto_derivation_handles_acronyms(self):
        class PIIRedactor(Plugin):
            pass

        class CodeMode(Plugin):
            pass

        class HTTPServerPlugin(Plugin):
            pass

        assert PIIRedactor.meta.name == "pii-redactor"
        assert CodeMode.meta.name == "code-mode"
        # Trailing "-plugin" stripped, internal acronym preserved.
        assert HTTPServerPlugin.meta.name == "http-server"

    def test_explicit_meta_is_not_overridden(self):
        class P(Plugin):
            meta = PluginMeta(name="custom", version="2.0.0")

        assert P.meta.name == "custom"
        assert P.meta.version == "2.0.0"

    def test_plugin_with_default_config(self):
        """A Plugin without a generic parameter gets an empty default config."""

        class P(Plugin):
            meta = PluginMeta(name="p", version="0.1.0")

        p = P()
        assert isinstance(p.config, BaseModel)
        # No fields to inspect — the point is that construction works with None.

    def test_config_accepts_instance(self):
        class PConfig(BaseModel):
            who: str = "world"

        class P(Plugin[PConfig]):
            meta = PluginMeta(name="p", version="0.1.0")

        p = P(PConfig(who="jeremiah"))
        assert isinstance(p.config, PConfig)
        assert p.config.who == "jeremiah"

    def test_config_accepts_dict(self):
        class PConfig(BaseModel):
            who: str = "world"

        class P(Plugin[PConfig]):
            meta = PluginMeta(name="p", version="0.1.0")

        p = P({"who": "jeremiah"})
        assert isinstance(p.config, PConfig)
        assert p.config.who == "jeremiah"

    def test_generic_parameter_binds_config_cls(self):
        """`Plugin[ConfigType]` stashes the Config on the subclass so dict
        validation, manifest generation, and runtime introspection all use
        the author-declared model."""

        class PConfig(BaseModel):
            who: str = "world"

        class P(Plugin[PConfig]):
            meta = PluginMeta(name="p", version="0.1.0")

        assert P._config_cls is PConfig

    def test_unparameterized_plugin_uses_empty_default_config(self):
        """A Plugin without a generic parameter gets an empty default that
        rejects unknown keys (extra='forbid')."""

        class P(Plugin):
            meta = PluginMeta(name="p", version="0.1.0")

        # No-arg construction works.
        P()
        # Unknown config keys are rejected by the empty default.
        with pytest.raises(PluginConfigError) as exc_info:
            P({"who": "jeremiah"})
        # The error message must not leak the `_EmptyConfig` implementation
        # class name; users shouldn't see private framework detail.
        assert "_EmptyConfig" not in str(exc_info.value)
        assert "no config fields" in str(exc_info.value)

    def test_invalid_config_raises_plugin_config_error(self):
        """Wrong-typed value for a declared field wraps ValidationError
        into PluginConfigError — exercising the generic Plugin[C] path."""

        class PConfig(BaseModel):
            count: int

        class P(Plugin[PConfig]):
            meta = PluginMeta(name="p", version="0.1.0")

        with pytest.raises(PluginConfigError, match="count"):
            P({"count": "not a number"})

    def test_required_field_missing_raises_plugin_config_error_on_no_args(self):
        """Required config field with no default must surface as
        PluginConfigError (not a raw pydantic.ValidationError) when the
        plugin is constructed with no arguments."""

        class PConfig(BaseModel):
            api_key: str  # required, no default

        class P(Plugin[PConfig]):
            meta = PluginMeta(name="p", version="0.1.0")

        with pytest.raises(PluginConfigError, match="api_key"):
            P()

    def test_bad_config_type_raises(self):
        class P(Plugin):
            meta = PluginMeta(name="p", version="0.1.0")

        with pytest.raises(PluginConfigError):
            P("not a config")  # type: ignore[arg-type]

    def test_non_basemodel_generic_arg_raises_at_class_creation(self):
        """`Plugin[T]` where T is not a pydantic BaseModel must fail loudly.

        The `# ty: ignore` tells the static checker that violating the type
        bound is intentional here — we're exercising the *runtime* guard.
        """

        class NotAModel:
            pass

        with pytest.raises(TypeError, match="BaseModel subclass"):

            class _Bad(Plugin[NotAModel]):  # ty: ignore[invalid-type-arguments]
                meta = PluginMeta(name="bad", version="0.1.0")

    def test_intermediate_generic_subclass_parameterization_is_not_misread_as_config(
        self,
    ):
        """A concrete subclass of an intermediate Plugin base with its
        own generic parameter must not have its generic arg misread as
        the plugin's config type.

        Given `class Intermediate(Plugin[Cfg], Generic[T])` and
        `class Concrete(Intermediate[int])`, `int` is the intermediate's
        own TypeVar substitution, NOT the plugin config. `Concrete`
        should inherit `Cfg` through the intermediate, not raise because
        `int` isn't a `BaseModel`.
        """
        _T = TypeVar("_T")

        class Cfg(BaseModel):
            value: int = 0

        class Intermediate(Plugin[Cfg], Generic[_T]):
            meta = PluginMeta(name="intermediate", version="0.1.0")

        class Concrete(Intermediate[int]):
            meta = PluginMeta(name="concrete", version="0.1.0")

        assert Intermediate._config_cls is Cfg
        assert Concrete._config_cls is Cfg
        assert isinstance(Concrete().config, Cfg)

    def test_deferred_config_binding_resolves_in_concrete_subclass(self):
        """Abstract plugin bases declare `Plugin[_T]` with an unbound
        TypeVar; concrete subclasses bind `_T` via `AbstractBase[Cfg]`.
        The resolver must propagate the substitution through the chain.
        """
        _T = TypeVar("_T", bound=BaseModel)

        class MyConfig(BaseModel):
            api_key: str = "default"

        class AbstractPlugin(Plugin[_T]):
            meta = PluginMeta(name="abstract", version="0.1.0")

        class ConcretePlugin(AbstractPlugin[MyConfig]):
            meta = PluginMeta(name="concrete", version="0.1.0")

        # Abstract base can't resolve (TypeVar still unbound).
        assert AbstractPlugin._config_cls is not MyConfig
        # Concrete leaf resolves through the intermediate.
        assert ConcretePlugin._config_cls is MyConfig
        assert isinstance(ConcretePlugin().config, MyConfig)
        assert ConcretePlugin({"api_key": "secret"}).config.api_key == "secret"
        # Manifest reflects the concrete config, not the empty default.
        m = ConcretePlugin.manifest()
        assert m is not None
        assert "api_key" in m["config_schema"]["properties"]


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 TestConfigJsonSerializable:
    """Plugin configs must be JSON-serializable — a hard rule. Configs
    are loaded from JSON/YAML, rendered into Horizon/registry forms,
    and published in manifests; any field that can't round-trip through
    JSON breaks the distribution story."""

    def test_arbitrary_type_without_pydantic_hooks_rejected(self):
        """A raw Python class with no pydantic hooks can't be described
        in JSON — `model_json_schema()` fails and we surface the
        failure as PluginError."""

        class Arbitrary:
            pass

        class BadConfig(BaseModel):
            model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
            thing: Arbitrary = Arbitrary()

        with pytest.raises(PluginError, match="not JSON"):

            class Bad(Plugin[BadConfig]):
                meta = PluginMeta(name="bad", version="0.1.0")

    def test_arbitrary_type_with_pydantic_hooks_accepted(self):
        """A custom type with `__get_pydantic_core_schema__` and a
        JSON-safe serializer IS JSON-round-trippable, even alongside
        `arbitrary_types_allowed=True`. Plugin authors can bring their
        own types as long as they provide the hooks."""
        from pydantic_core import core_schema

        class JsonSafe:
            def __init__(self, value: str):
                self.value = value

            @classmethod
            def __get_pydantic_core_schema__(cls, source, handler):
                return core_schema.no_info_after_validator_function(
                    cls,
                    handler(str),
                    serialization=core_schema.plain_serializer_function_ser_schema(
                        lambda v: v.value, return_schema=core_schema.str_schema()
                    ),
                )

        class GoodConfig(BaseModel):
            model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
            thing: JsonSafe = JsonSafe("hi")

        class Good(Plugin[GoodConfig]):
            meta = PluginMeta(name="good", version="0.1.0")

        # The config dumps cleanly to JSON because of the plugin-
        # author-provided hooks.
        assert Good().config.model_dump(mode="json") == {"thing": "hi"}

    def test_callable_field_rejected(self):
        """Callable fields can't be round-tripped through JSON — reject."""
        from collections.abc import Callable

        class BadConfig(BaseModel):
            handler: Callable[[str], str]

        with pytest.raises(PluginError, match="not JSON"):

            class Bad(Plugin[BadConfig]):
                meta = PluginMeta(name="bad", version="0.1.0")

    @pytest.mark.parametrize(
        "field_type, default",
        [
            (pydantic.SecretStr, pydantic.SecretStr("s3cret")),
            (int, 42),
            (str, "hello"),
            (list[str], ["a"]),
        ],
    )
    def test_common_json_serializable_fields_accepted(self, field_type, default):
        """Common pydantic-supported types round-trip through JSON and
        must pass validation."""

        class GoodConfig(BaseModel):
            value: field_type = default  # type: ignore[valid-type]

        class Good(Plugin[GoodConfig]):
            meta = PluginMeta(name="good", version="0.1.0")

        # Construction and config access both work.
        plugin = Good()
        assert plugin.config.value == default

    def test_nested_basemodel_field_accepted(self):
        """Nested pydantic models are fully JSON-serializable."""

        class Inner(BaseModel):
            name: str = "x"
            count: int = 0

        class OuterConfig(BaseModel):
            inner: Inner = Inner()

        class Outer(Plugin[OuterConfig]):
            meta = PluginMeta(name="outer", version="0.1.0")

        assert Outer().config.inner.name == "x"

    def test_datetime_and_path_fields_accepted(self):
        """datetime and Path are pydantic-supported JSON types."""
        from datetime import datetime
        from pathlib import Path as PathlibPath

        class GoodConfig(BaseModel):
            when: datetime = datetime(2026, 1, 1)
            where: PathlibPath = PathlibPath("/tmp")

        class Good(Plugin[GoodConfig]):
            meta = PluginMeta(name="good", version="0.1.0")

        assert isinstance(Good().config.when, datetime)

    def test_partial_hooks_without_serializer_rejected(self):
        """A custom type with `__get_pydantic_json_schema__` but no
        matching serializer would pass a schema-generation-only check
        while failing at runtime `model_dump(mode='json')`. The
        validator exercises the real serialization path when the
        config is buildable without args, so the break surfaces at
        class creation (for configs with defaults) rather than at
        publish/serialize time."""
        from pydantic_core import core_schema

        class Tricky:
            @classmethod
            def __get_pydantic_core_schema__(cls, source, handler):
                # Validator only — no serialization= argument.
                return core_schema.no_info_plain_validator_function(lambda v: cls())

            @classmethod
            def __get_pydantic_json_schema__(cls, schema, handler):
                return {"type": "string"}

        class PartialConfig(BaseModel):
            model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
            x: Tricky = Tricky()

        with pytest.raises(PluginError, match="cannot be serialized"):

            class Bad(Plugin[PartialConfig]):
                meta = PluginMeta(name="bad", version="0.1.0")

    def test_default_value_failing_validator_raises_plugin_error(self):
        """A ValidationError from a field validator rejecting its own
        default isn't the 'required field, skip' case — surface it as
        PluginError at class creation rather than silently accepting a
        class that can never be constructed without args."""
        from pydantic import field_validator

        class BadDefaultConfig(BaseModel):
            model_config = pydantic.ConfigDict(validate_default=True)
            x: int = -1

            @field_validator("x")
            @classmethod
            def must_be_positive(cls, v: int) -> int:
                if v <= 0:
                    raise ValueError("x must be positive")
                return v

        with pytest.raises(PluginError, match="invalid default"):

            class Bad(Plugin[BadDefaultConfig]):
                meta = PluginMeta(name="bad", version="0.1.0")

    def test_non_validation_exception_from_default_wrapped_as_plugin_error(self):
        """Non-ValidationError exceptions during default-construction
        (TypeError from a broken default_factory, etc.) are wrapped as
        PluginError so the message carries plugin attribution."""
        from pydantic import Field

        def broken_factory() -> int:
            raise TypeError("factory is busted")

        class BadFactoryConfig(BaseModel):
            x: int = Field(default_factory=broken_factory)

        with pytest.raises(PluginError, match="could not be instantiated"):

            class Bad(Plugin[BadFactoryConfig]):
                meta = PluginMeta(name="bad", version="0.1.0")

    def test_required_field_partial_hooks_not_caught_at_class_creation(self):
        """Documented trade-off: a partial-hooks type on a required
        field slips past class-creation validation because the dump
        test only runs on a buildable instance. The violation surfaces
        at first real instantiation / serialization instead."""
        from pydantic_core import core_schema

        class Tricky:
            @classmethod
            def __get_pydantic_core_schema__(cls, source, handler):
                return core_schema.no_info_plain_validator_function(lambda v: cls())

            @classmethod
            def __get_pydantic_json_schema__(cls, schema, handler):
                return {"type": "string"}

        class RequiredBadConfig(BaseModel):
            model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
            x: Tricky  # required, no default — class creation can't build

        # Plugin class is created without raising. Manifest also succeeds
        # because the schema itself is emittable.
        class Required(Plugin[RequiredBadConfig]):
            meta = PluginMeta(name="req", version="0.1.0")

        m = Required.manifest()
        assert m is not None
        assert "x" in m["config_schema"]["properties"]

    def test_manifest_revalidates_rebuilt_forward_reference_config(self):
        """A config with a forward reference skips validation at class
        creation, then gets validated at manifest() time once the
        reference is resolved and the model is rebuilt. If the rebuilt
        config has a partial-hooks default that the dump check catches,
        manifest() must raise."""
        from typing import Optional

        from pydantic_core import core_schema

        class Tricky:
            @classmethod
            def __get_pydantic_core_schema__(cls, source, handler):
                return core_schema.no_info_plain_validator_function(lambda v: cls())

            @classmethod
            def __get_pydantic_json_schema__(cls, schema, handler):
                return {"type": "string"}

        class UnfinishedConfig(BaseModel):
            model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
            child: Optional[Child] = None  # noqa: F821, UP045
            broken: Tricky = Tricky()

        # Plugin class creation defers validation (forward ref unresolved).
        class Deferred(Plugin[UnfinishedConfig]):
            meta = PluginMeta(name="deferred", version="0.1.0")

        class Child(BaseModel):
            pass

        UnfinishedConfig.model_rebuild()

        # manifest() re-runs _validate_config_cls against the
        # now-complete model and catches the partial-hooks violation
        # via the dump test.
        with pytest.raises(PluginError, match="cannot be serialized"):
            Deferred.manifest()

    def test_forward_reference_config_skips_validation(self):
        """Configs with unresolved forward references can't be
        schema-checked at class creation; validation skips so the
        plugin class itself can be defined. The author is expected
        to run the check later (manifest generation will trip any
        real problems)."""

        class UnfinishedConfig(BaseModel):
            child: NotYetDefined  # noqa: F821  # ty: ignore[unresolved-reference]

        # This should NOT raise even though UnfinishedConfig isn't
        # fully defined — validation defers until the model is
        # rebuildable.
        class P(Plugin[UnfinishedConfig]):
            meta = PluginMeta(name="p", version="0.1.0")

        assert P._config_cls is UnfinishedConfig

    def test_empty_default_config_passes_validation(self):
        """The framework's internal `_EmptyConfig` must pass its own
        JSON-serializable check (regression: it's used as the fallback
        for every unparameterized plugin)."""

        class P(Plugin):
            meta = PluginMeta(name="p", version="0.1.0")

        # If _EmptyConfig failed validation, class creation above would
        # have raised. This test is explicit documentation of the
        # requirement.
        assert P().config is not None


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_from_on_install(self):
        """A plugin's `on_install(server)` can call `server.add_plugin()`
        to register child plugins. The loader runs at registration time,
        so child plugins are installed before any lifespan starts."""
        recorder = _Recorder()

        class ChildA(Plugin):
            meta = PluginMeta(name="child-a", version="0.1.0")

            async def setup(self, server):
                recorder.events.append(("setup", "child-a"))

        class ChildB(Plugin):
            meta = PluginMeta(name="child-b", version="0.1.0")

            async def setup(self, server):
                recorder.events.append(("setup", "child-b"))

        class Loader(Plugin):
            meta = PluginMeta(name="loader", version="0.1.0")

            def on_install(self, server):
                server.add_plugin(ChildA())
                server.add_plugin(ChildB())

        mcp = FastMCP("t", plugins=[Loader()])
        # Children are registered at construction; no lifespan needed.
        assert [p.meta.name for p in mcp.plugins] == [
            "loader",
            "child-a",
            "child-b",
        ]

        async with Client(mcp) as c:
            await c.ping()

        # Children stay registered across the lifespan — they're permanent
        # (registered during construction), not ephemeral.
        assert [p.meta.name for p in mcp.plugins] == [
            "loader",
            "child-a",
            "child-b",
        ]
        assert ("setup", "child-a") in recorder.events
        assert ("setup", "child-b") in recorder.events

    def test_loader_child_contributions_install_during_on_install(self):
        """Children registered by on_install install immediately.

        The plugin list preserves parent-before-child order. Contribution
        side effects happen when each plugin is added; FastMCP does not
        defer or transact loader installs.
        """

        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")

            def on_install(self, server):
                server.add_plugin(Child())

            def middleware(self):
                return [_TraceMiddleware("loader")]

        mcp = FastMCP("t", plugins=[Loader()])

        assert [p.meta.name for p in mcp.plugins] == ["loader", "child"]
        tags = [m.tag for m in mcp.middleware if isinstance(m, _TraceMiddleware)]
        assert tags == ["child", "loader"]

    def test_same_instance_registered_twice_raises(self):
        """Plugin instances are single-server: a second registration of
        the same instance (on any server) raises at `install()`."""

        class P(Plugin):
            meta = PluginMeta(name="p", version="0.1.0")

        p = P()
        mcp = FastMCP("t")
        mcp.add_plugin(p)

        with pytest.raises(PluginError, match="already installed"):
            mcp.add_plugin(p)

    def test_instance_on_second_server_raises(self):
        """Sharing a plugin instance across two servers is not supported."""

        class P(Plugin):
            meta = PluginMeta(name="p", version="0.1.0")

        p = P()
        FastMCP("a", plugins=[p])

        with pytest.raises(PluginError, match="already installed"):
            FastMCP("b", plugins=[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

    def test_add_plugin_raises_when_later_hook_raises(self):
        """Contribution hook errors fail loudly.

        Plugin installation is not transactional: plugin hooks can mutate
        arbitrary server state, so FastMCP does not pretend to recover a
        partially configured server. Callers should discard the server
        after a failed plugin install.
        """

        class Flaky(Plugin):
            meta = PluginMeta(name="flaky", version="0.1.0")

            def middleware(self):
                return [_TraceMiddleware("flaky")]

            def transforms(self):
                raise RuntimeError("transforms exploded")

        mcp = FastMCP("t")

        flaky = Flaky()
        with pytest.raises(RuntimeError, match="transforms exploded"):
            mcp.add_plugin(flaky)

        assert flaky in mcp.plugins
        assert flaky._installed_on is mcp

    async def test_add_plugin_raises_when_routes_raises(self):
        """Route hook errors fail loudly without best-effort recovery."""

        class RoutesBoom(Plugin):
            meta = PluginMeta(name="routes-boom", version="0.1.0")

            def routes(self):
                raise RuntimeError("routes exploded")

        mcp = FastMCP("t")
        plugin = RoutesBoom()

        with pytest.raises(RuntimeError, match="routes exploded"):
            mcp.add_plugin(plugin)

        assert plugin in mcp.plugins
        assert plugin._installed_on is mcp

    async def test_on_install_loader_contributions_persist_across_cycles(self):
        """Plugins registered via a loader's `on_install(server)` are permanent
        just like plugins passed to `FastMCP(plugins=[...])`. Their
        contributions are installed once and survive across lifespan
        cycles without re-registration or cleanup."""

        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")

            def on_install(self, server):
                server.add_plugin(Child())

        mcp = FastMCP("t", plugins=[Loader()])

        # `on_install` ran during construction; child is registered.
        assert [p.meta.name for p in mcp.plugins] == ["loader", "child"]

        # Contributions installed once, persist across multiple cycles.
        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 == ["child"]


class TestRunHook:
    """Plugins that override `run()` directly (the long-running pattern)."""

    async def test_run_override_wraps_server_lifetime(self):
        """A plugin overriding run() sees the server live between setup and teardown."""

        recorder = _Recorder()

        class Long(Plugin):
            meta = PluginMeta(name="long", version="0.1.0")

            @asynccontextmanager
            async def run(self, server):
                recorder.events.append(("enter", "long"))
                try:
                    yield
                finally:
                    recorder.events.append(("exit", "long"))

        mcp = FastMCP("t", plugins=[Long()])
        async with Client(mcp) as c:
            await c.ping()
            # Mid-cycle: enter fired, exit hasn't.
            assert ("enter", "long") in recorder.events
            assert ("exit", "long") not in recorder.events

        # After teardown: both fired.
        assert recorder.events == [("enter", "long"), ("exit", "long")]

    async def test_run_override_can_use_async_with(self):
        """A plugin's run() can acquire an async-context resource and release it on exit."""

        recorder = _Recorder()

        @asynccontextmanager
        async def fake_resource():
            recorder.events.append(("acquire", "resource"))
            try:
                yield "handle"
            finally:
                recorder.events.append(("release", "resource"))

        class WithResource(Plugin):
            meta = PluginMeta(name="with-resource", version="0.1.0")

            @asynccontextmanager
            async def run(self, server):
                async with fake_resource() as handle:
                    self.handle = handle
                    yield

        p = WithResource()
        mcp = FastMCP("t", plugins=[p])
        async with Client(mcp) as c:
            await c.ping()
            assert p.handle == "handle"

        # async with cleanup fired on exit path
        assert recorder.events == [
            ("acquire", "resource"),
            ("release", "resource"),
        ]

    async def test_run_override_cancellation_propagates_into_background_task(self):
        """A long-running background task inside run() is cancelled on shutdown."""

        recorder = _Recorder()

        class Background(Plugin):
            meta = PluginMeta(name="background", version="0.1.0")

            @asynccontextmanager
            async def run(self, server):
                async def worker():
                    try:
                        await asyncio.Event().wait()
                    except asyncio.CancelledError:
                        recorder.events.append(("cancelled", "worker"))
                        raise

                task = asyncio.create_task(worker())
                try:
                    yield
                finally:
                    task.cancel()
                    with suppress(asyncio.CancelledError):
                        await task

        mcp = FastMCP("t", plugins=[Background()])
        async with Client(mcp) as c:
            await c.ping()

        assert recorder.events == [("cancelled", "worker")]

    async def test_run_override_raising_before_yield_aborts_startup(self):
        """If a plugin's run() raises before yielding, startup fails cleanly."""

        class BadStart(Plugin):
            meta = PluginMeta(name="bad-start", version="0.1.0")

            @asynccontextmanager
            async def run(self, server):
                raise RuntimeError("cannot start")
                yield  # unreachable

        mcp = FastMCP("t", plugins=[BadStart()])
        with pytest.raises(RuntimeError, match="cannot start"):
            async with Client(mcp) as c:
                await c.ping()

    async def test_run_override_composes_with_simple_setup_teardown_plugins(self):
        """A server can mix run-override plugins with setup/teardown plugins."""

        recorder = _Recorder()

        class Simple(Plugin):
            meta = PluginMeta(name="simple", version="0.1.0")

            async def setup(self, server):
                recorder.events.append(("setup", "simple"))

            async def teardown(self):
                recorder.events.append(("teardown", "simple"))

        class LongRunning(Plugin):
            meta = PluginMeta(name="long-running", version="0.1.0")

            @asynccontextmanager
            async def run(self, server):
                recorder.events.append(("enter", "long-running"))
                try:
                    yield
                finally:
                    recorder.events.append(("exit", "long-running"))

        mcp = FastMCP("t", plugins=[Simple(), LongRunning()])
        async with Client(mcp) as c:
            await c.ping()

        # Enter order follows registration; exit order is reversed.
        assert recorder.events == [
            ("setup", "simple"),
            ("enter", "long-running"),
            ("exit", "long-running"),
            ("teardown", "simple"),
        ]


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 PConfig(BaseModel):
            who: str = "world"

        class P(Plugin[PConfig]):
            meta = PluginMeta(
                name="p",
                version="0.1.0",
                description="demo",
                tags=["x"],
                dependencies=["demo>=0.1"],
                fastmcp_version=">=3.0",
                meta={"owning_team": "platform"},
            )

        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_omits_empty_config_internal_name_and_docstring(self):
        """For plugins without a Config, the manifest's `config_schema`
        must not leak `_EmptyConfig` — neither as `title` nor as
        `description` (pydantic emits both by default)."""

        class P(Plugin):
            meta = PluginMeta(name="p", version="0.1.0")

        m = P.manifest()
        assert m is not None
        schema = m["config_schema"]
        assert "_EmptyConfig" not in schema.get("title", "")
        # Pydantic v2 emits the class docstring as `description`; strip it too.
        assert (
            "description" not in schema or "_EmptyConfig" not in schema["description"]
        )
        assert "Plugin[ConfigType]" not in schema.get("description", "")

    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()


class TestPluginCapabilities:
    """Plugins contribute partial ServerCapabilities dicts via `capabilities()`."""

    def test_default_returns_empty(self):
        """Plugin with no override contributes nothing."""
        assert _TestPlugin().capabilities() == {}

    async def test_capabilities_are_collected_at_registration_time(self):
        """Capability hooks are sync install-time contributions, not
        lifespan-time callbacks."""
        calls = 0

        class P(_TestPlugin):
            def capabilities(self):
                nonlocal calls
                calls += 1
                return {"experimental": {"registered": {}}}

        mcp = FastMCP("t", plugins=[P()])
        assert calls == 1

        async with Client(mcp) as c:
            result = c.initialize_result
            assert result is not None
            experimental = result.capabilities.experimental or {}
            assert experimental.get("registered") == {}

        assert calls == 1

    async def test_experimental_contribution_reaches_initialize_response(self):
        """An experimental capability entry flows through to the client."""

        class P(_TestPlugin):
            def capabilities(self):
                return {"experimental": {"my/ext": {}}}

        mcp = FastMCP("t", plugins=[P()])

        async with Client(mcp) as c:
            result = c.initialize_result
            assert result is not None
            experimental = result.capabilities.experimental or {}
            assert experimental.get("my/ext") == {}

    async def test_multiple_plugins_merge_into_same_field(self):
        """Contributions to the same top-level field are deep-merged."""

        class A(_TestPlugin):
            def capabilities(self):
                return {"experimental": {"alpha": {"version": 1}}}

        class B(_TestPlugin):
            def capabilities(self):
                return {"experimental": {"beta": {}}}

        mcp = FastMCP("t", plugins=[A(), B()])

        async with Client(mcp) as c:
            result = c.initialize_result
            assert result is not None
            experimental = result.capabilities.experimental or {}
            assert experimental.get("alpha") == {"version": 1}
            assert experimental.get("beta") == {}

    async def test_later_plugin_overrides_earlier_on_same_key(self):
        """Plugins run in sequence; later contributions override earlier ones.

        Plugin order is a user-facing configuration knob — same as
        middleware order — so overriding a built-in or earlier plugin's
        capability is intentional, not an error.
        """

        class Earlier(_TestPlugin):
            def capabilities(self):
                return {"experimental": {"shared": {"owner": "earlier"}}}

        class Later(_TestPlugin):
            def capabilities(self):
                return {"experimental": {"shared": {"owner": "later"}}}

        mcp = FastMCP("t", plugins=[Earlier(), Later()])

        async with Client(mcp) as c:
            result = c.initialize_result
            assert result is not None
            experimental = result.capabilities.experimental or {}
            assert experimental.get("shared") == {"owner": "later"}

    async def test_plugin_can_add_non_experimental_field(self):
        """Plugins can advertise top-level capability fields the server didn't set.

        `logging` is off by default on a FastMCP server; a plugin turning
        it on must surface in the initialize response.
        """

        class P(_TestPlugin):
            def capabilities(self):
                return {"logging": {}}

        mcp = FastMCP("t", plugins=[P()])

        async with Client(mcp) as c:
            result = c.initialize_result
            assert result is not None
            assert result.capabilities.logging is not None

    async def test_plugin_can_override_built_in_subfield(self):
        """Deep-merge applies to typed sub-fields of pre-populated capability objects.

        FastMCP already advertises `tools.listChanged=True` by default; a
        plugin flipping it to `False` exercises the merge path through a
        pydantic sub-model (not just the experimental dict).
        """

        class P(_TestPlugin):
            def capabilities(self):
                return {"tools": {"listChanged": False}}

        mcp = FastMCP("t", plugins=[P()])

        async with Client(mcp) as c:
            result = c.initialize_result
            assert result is not None
            assert result.capabilities.tools is not None
            assert result.capabilities.tools.listChanged is False

    async def test_plugin_owned_capability_dict_is_not_mutated_across_plugins(self):
        """Plugin-returned dicts must not be mutated by the merge.

        A plugin that returns a cached/class-level dict from
        `capabilities()` gets the same object back on subsequent calls.
        If the merge wrote that dict into `merged` by reference, a later
        plugin's contribution would add keys to the earlier plugin's
        dict, leaking state across initializations.
        """

        class A(_TestPlugin):
            _caps = {"experimental": {"alpha": {}}}

            def capabilities(self):
                return self._caps

        class B(_TestPlugin):
            def capabilities(self):
                return {"experimental": {"beta": {}}}

        a = A()
        mcp = FastMCP("t", plugins=[a, B()])

        async with Client(mcp) as c:
            result = c.initialize_result
            assert result is not None
            experimental = result.capabilities.experimental or {}
            assert "alpha" in experimental
            assert "beta" in experimental

        # A's cached dict must not have been mutated to contain B's entry.
        assert a._caps == {"experimental": {"alpha": {}}}

    async def test_on_install_loader_plugin_capabilities_contribute(self):
        """Plugins added via the on_install loader pattern contribute
        capabilities just like plugins passed to `plugins=[...]`."""

        class Loaded(_TestPlugin):
            def capabilities(self):
                return {"experimental": {"loaded": {}}}

        class Loader(_TestPlugin):
            def on_install(self, server):
                server.add_plugin(Loaded())

        mcp = FastMCP("t", plugins=[Loader()])

        async with Client(mcp) as c:
            result = c.initialize_result
            assert result is not None
            experimental = result.capabilities.experimental or {}
            assert experimental.get("loaded") == {}

    async def test_on_install_loader_capabilities_follow_plugin_order(self):
        """Capabilities merge in plugin-list order even though child
        plugins install while the parent's `on_install()` hook is running."""

        class Loaded(_TestPlugin):
            def capabilities(self):
                return {"experimental": {"shared": {"owner": "child"}}}

        class Loader(_TestPlugin):
            def on_install(self, server):
                server.add_plugin(Loaded())

            def capabilities(self):
                return {"experimental": {"shared": {"owner": "loader"}}}

        mcp = FastMCP("t", plugins=[Loader()])
        assert [type(plugin).__name__ for plugin in mcp.plugins] == [
            "Loader",
            "Loaded",
        ]

        async with Client(mcp) as c:
            result = c.initialize_result
            assert result is not None
            experimental = result.capabilities.experimental or {}
            assert experimental.get("shared") == {"owner": "child"}