From cc290b3a2e182608d8b8b2d83f70ab7deb26eda9 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Mon, 20 Apr 2026 10:39:58 -0400 Subject: [PATCH] Make Plugin generic over its Config model (#3983) --- src/fastmcp/server/plugins/base.py | 218 +++++++++++++++++++++++------ tests/server/test_plugins.py | 185 +++++++++++++++++++++--- 2 files changed, 336 insertions(+), 67 deletions(-) diff --git a/src/fastmcp/server/plugins/base.py b/src/fastmcp/server/plugins/base.py index 200498a62..540bae4f5 100644 --- a/src/fastmcp/server/plugins/base.py +++ b/src/fastmcp/server/plugins/base.py @@ -2,8 +2,9 @@ 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. +distributable units. A plugin is a subclass of `Plugin` (optionally +parameterized with a pydantic config model — `Plugin[MyConfig]` — for +typed configuration). See the design document for the full specification. """ @@ -17,7 +18,16 @@ from contextlib import asynccontextmanager from email.message import Message as EmailMessage from importlib import metadata as importlib_metadata from pathlib import Path -from typing import TYPE_CHECKING, Any, ClassVar, cast +from typing import ( + TYPE_CHECKING, + Any, + ClassVar, + Generic, + TypeVar, + cast, + get_args, + get_origin, +) from packaging.requirements import InvalidRequirement, Requirement from packaging.specifiers import InvalidSpecifier, SpecifierSet @@ -243,6 +253,20 @@ class PluginMeta(BaseModel): _DEFAULT_PLUGIN_VERSION = "0.1.0" +class _EmptyConfig(BaseModel): + """Default config for plugins that don't declare their own via the + `Plugin[ConfigType]` generic parameter.""" + + model_config = ConfigDict(extra="forbid") + + +C = TypeVar("C", bound=BaseModel) +"""Type variable for a plugin's config model. Bound to `BaseModel` so +any pydantic model is valid. Plugins without a config omit the generic +parameter; the runtime falls back to `_EmptyConfig` in that case. +""" + + def _derive_plugin_name(cls_name: str) -> str: """Kebab-case a class name, stripping a trailing ``Plugin`` suffix. @@ -259,7 +283,67 @@ def _derive_plugin_name(cls_name: str) -> str: return name -class Plugin: +def _resolve_plugin_config_cls(cls: type) -> type[BaseModel] | None: + """Resolve the config class bound to `Plugin[C]` for a subclass. + + Walks `cls.__orig_bases__`, recursing through intermediate `Plugin` + subclasses and propagating TypeVar substitutions. Returns the bound + `BaseModel` subclass, or `None` if the binding is still a TypeVar + (unresolved — typically an intermediate abstract base). + + Raises `TypeError` if a resolved argument is concrete but not a + `BaseModel` subclass (a misuse of `Plugin[NonPydanticType]`). + """ + + def _resolve(base: Any, substitutions: dict[Any, Any]) -> Any: + origin = get_origin(base) + if origin is None or not ( + isinstance(origin, type) and issubclass(origin, Plugin) + ): + return None + args = get_args(base) + # Apply outer-scope substitutions so a parent's TypeVar bound to + # a concrete type at this level becomes that concrete type here. + resolved_args = tuple(substitutions.get(a, a) for a in args) + + if origin is Plugin: + # We're at the root parameterization. + if not resolved_args: + return None + cfg = resolved_args[0] + # Still a TypeVar: unresolved at this level of the chain. + if isinstance(cfg, TypeVar): + return None + return cfg + + # Intermediate Plugin subclass. Push down its own TypeVar + # substitutions (from its `__parameters__`) and recurse into its + # bases to find the Plugin parameterization. + origin_params = getattr(origin, "__parameters__", ()) + new_subs = { + **substitutions, + **dict(zip(origin_params, resolved_args, strict=False)), + } + for inner in getattr(origin, "__orig_bases__", ()): + found = _resolve(inner, new_subs) + if found is not None: + return found + return None + + for base in getattr(cls, "__orig_bases__", ()): + resolved = _resolve(base, substitutions={}) + if resolved is None: + continue + if not (isinstance(resolved, type) and issubclass(resolved, BaseModel)): + raise TypeError( + f"{cls.__name__}: Plugin[...] generic parameter must be a " + f"pydantic BaseModel subclass, got {resolved!r}" + ) + return resolved + return None + + +class Plugin(Generic[C]): """Base class for FastMCP plugins. Subclass to define a plugin. A subclass may optionally declare a @@ -267,31 +351,29 @@ class Plugin: a default is derived from the class name (kebab-cased, trailing `Plugin` stripped) with version `0.1.0`. Declare `meta` explicitly when publishing or when Horizon/registry-facing metadata matters. - Subclasses may also declare a nested `Config` (subclass of - `pydantic.BaseModel`) describing configuration, and override any of - the lifecycle and contribution hooks. + + **Config typing.** Parameterize `Plugin` with a pydantic model to + give your plugin typed configuration — `self.config.` is then + correctly typed in editors and type checkers, and passing a dict or + model instance to the constructor validates against the model. + Plugins without a config omit the parameter. Example: ```python - from fastmcp.server.plugins import Plugin, PluginMeta from pydantic import BaseModel + from fastmcp.server.plugins import Plugin, PluginMeta - class PIIRedactor(Plugin): - meta = PluginMeta( - name="pii-redactor", - version="0.3.0", - dependencies=[ - "fastmcp-plugin-pii>=0.3.0", - "regex>=2024.0", - ], - ) + class PIIRedactorConfig(BaseModel): + patterns: list[str] = ["ssn", "email"] - class Config(BaseModel): - patterns: list[str] = ["ssn", "email"] + + class PIIRedactor(Plugin[PIIRedactorConfig]): + meta = PluginMeta(name="pii-redactor", version="0.3.0") def middleware(self): - return [PIIMiddleware(self.config)] + # self.config is typed as PIIRedactorConfig + return [PIIMiddleware(self.config.patterns)] ``` """ @@ -303,6 +385,16 @@ class Plugin: Horizon/registry-facing metadata matters. """ + _config_cls: ClassVar[type[BaseModel]] = _EmptyConfig + """Config model class resolved from the `Plugin[C]` generic parameter. + Auto-populated by `__init_subclass__`; falls back to `_EmptyConfig` + for plugins that don't parameterize `Plugin`. + """ + + config: C + """The validated config instance. Typed as `C`, the generic + parameter, so `self.config.` type-checks correctly.""" + def __init_subclass__(cls, **kwargs: Any) -> None: super().__init_subclass__(**kwargs) # Auto-derive meta if the subclass didn't declare its own. We @@ -314,13 +406,17 @@ class Plugin: name=_derive_plugin_name(cls.__name__), version=_DEFAULT_PLUGIN_VERSION, ) - - class Config(BaseModel): - """Default empty configuration. Subclasses override to declare fields.""" - - model_config = ConfigDict(extra="forbid") - - config: BaseModel + # Resolve the Config model from the generic parameter. We walk the + # `__orig_bases__` chain and propagate TypeVar substitutions, so + # both direct parameterization (`class P(Plugin[Cfg])`) and + # deferred binding (`class Abstract(Plugin[_T])` → + # `class P(Abstract[Cfg])`) resolve correctly. Intermediate + # generic bases with their own unrelated TypeVars are unaffected + # because we substitute through each step rather than treating + # `args[0]` as the config unconditionally. + config_cls = _resolve_plugin_config_cls(cls) + if config_cls is not None: + cls._config_cls = config_cls # Framework-internal marker. Set to True by `FastMCP.add_plugin` when # the plugin is added from inside another plugin's setup() (the loader @@ -329,12 +425,7 @@ class Plugin: # 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. + def __init__(self, config: C | dict[str, Any] | None = None) -> None: meta = getattr(type(self), "meta", None) if not isinstance(meta, PluginMeta): raise TypeError( @@ -343,24 +434,54 @@ class Plugin: ) self._validate_meta(meta) - config_cls = type(self).Config + config_cls = type(self)._config_cls + + def _wrap(exc: ValidationError) -> PluginConfigError: + # For unparameterized plugins, pydantic's error string + # includes "1 validation error for _EmptyConfig" — an + # internal class name users shouldn't see. Emit a scoped + # message instead; for parameterized plugins, forward + # pydantic's full diagnostic. + if config_cls is _EmptyConfig: + keys = list(config.keys()) if isinstance(config, dict) else [] + return PluginConfigError( + f"Invalid configuration for {type(self).__name__}: this " + f"plugin declares no config fields but received " + f"{keys}." + ) + return PluginConfigError( + f"Invalid configuration for {type(self).__name__}: {exc}" + ) + if config is None: - value: BaseModel = config_cls() + try: + value: BaseModel = config_cls() + except ValidationError as exc: + # Required config fields with no default: surface the + # failure as PluginConfigError so callers that catch + # the documented exception type behave consistently + # with the dict path below. + raise _wrap(exc) from exc 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 + raise _wrap(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__}" + # `_EmptyConfig` is an internal implementation detail for + # unparameterized plugins. Don't leak its name to authors. + expected = ( + "dict" + if config_cls is _EmptyConfig + else f"{config_cls.__name__} instance or dict" ) - self.config = value + raise PluginConfigError( + f"Config for {type(self).__name__} must be a {expected}, " + f"not {type(config).__name__}" + ) + self.config = cast(C, value) # -- validation ----------------------------------------------------------- @@ -542,11 +663,20 @@ class Plugin: # have produced from a live plugin instance. cls._validate_meta(meta) - config_cls = getattr(cls, "Config", Plugin.Config) + config_cls = cls._config_cls + config_schema = config_cls.model_json_schema() + # `_EmptyConfig` is an internal implementation detail; don't + # leak its name or docstring into the published manifest JSON + # consumed by Horizon, registries, and CI tooling. Pydantic v2 + # emits both `title` (from `__name__`) and `description` (from + # the class docstring) in `model_json_schema()`; strip both. + if config_cls is _EmptyConfig: + config_schema.pop("title", None) + config_schema.pop("description", None) data: dict[str, Any] = { "manifest_version": 1, **meta.model_dump(), - "config_schema": config_cls.model_json_schema(), + "config_schema": config_schema, "entry_point": f"{cls.__module__}:{cls.__qualname__}", } diff --git a/tests/server/test_plugins.py b/tests/server/test_plugins.py index 44384cb33..34ae5d1bd 100644 --- a/tests/server/test_plugins.py +++ b/tests/server/test_plugins.py @@ -8,6 +8,7 @@ from contextlib import 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 pytest from packaging.version import Version @@ -280,50 +281,170 @@ class TestPluginConstruction: 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, Plugin.Config) + assert isinstance(p.config, BaseModel) + # No fields to inspect — the point is that construction works with None. def test_config_accepts_instance(self): - class P(Plugin): + class PConfig(BaseModel): + who: str = "world" + + class P(Plugin[PConfig]): 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) + p = P(PConfig(who="jeremiah")) + assert isinstance(p.config, PConfig) assert p.config.who == "jeremiah" def test_config_accepts_dict(self): - class P(Plugin): + class PConfig(BaseModel): + who: str = "world" + + class P(Plugin[PConfig]): 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) + p = P({"who": "jeremiah"}) + assert isinstance(p.config, PConfig) assert p.config.who == "jeremiah" - def test_invalid_config_raises_plugin_config_error(self): + 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") - class Config(BaseModel): - count: int + # 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) - with pytest.raises(PluginConfigError): - P(config={"count": "not a number"}) + 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(config="not a config") # ty: ignore[invalid-argument-type] + 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: @@ -1189,7 +1310,10 @@ class TestManifest: """manifest() produces a JSON-serializable dict and can write to disk.""" def test_manifest_shape(self): - class P(Plugin): + class PConfig(BaseModel): + who: str = "world" + + class P(Plugin[PConfig]): meta = PluginMeta( name="p", version="0.1.0", @@ -1200,9 +1324,6 @@ class TestManifest: meta={"owning_team": "platform"}, ) - class Config(BaseModel): - who: str = "world" - m = P.manifest() assert m is not None assert m["manifest_version"] == 1 @@ -1218,6 +1339,24 @@ class TestManifest: 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