Add PluginMeta.from_package() helper (#3974)

This commit is contained in:
Jeremiah Lowin 2026-04-19 08:56:42 -04:00 committed by GitHub
commit e2e49f77d2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 319 additions and 2 deletions

View file

@ -13,12 +13,17 @@ from __future__ import annotations
import json
from collections.abc import AsyncIterator
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
from typing import TYPE_CHECKING, Any, ClassVar, cast
from packaging.requirements import InvalidRequirement, Requirement
from packaging.specifiers import InvalidSpecifier, SpecifierSet
from packaging.utils import canonicalize_name
from packaging.version import InvalidVersion, Version
from pydantic import BaseModel, ConfigDict, ValidationError
from typing_extensions import Self
import fastmcp
from fastmcp.exceptions import FastMCPError
@ -92,6 +97,147 @@ class PluginMeta(BaseModel):
model_config = ConfigDict(extra="forbid")
@classmethod
def from_package(cls, distribution: str, /, **overrides: Any) -> Self:
"""Derive plugin metadata from an installed Python distribution.
Reads `version`, `description`, `author`, and `homepage` from the
distribution's metadata (as recorded in its `pyproject.toml` and
exposed via `importlib.metadata`), and pins the distribution
itself as the sole entry in `dependencies` so the manifest
automatically reflects the containing package and stays in sync
with every new release. Runtime dependencies declared in the
distribution's `Requires-Dist` are NOT harvested; plugin authors
pass additional runtime deps via the `dependencies` override.
Any keyword argument overrides the derived value.
Example:
```python
class MyPiiRedactor(Plugin):
meta = PluginMeta.from_package(
"fastmcp-plugin-my-pii", # distribution name on PyPI
name="my-pii", # plugin identifier
tags=["security"],
)
```
Args:
distribution: The installed distribution name to read from
(e.g. `"fastmcp-plugin-my-pii"`). Must be importable via
`importlib.metadata`. Cannot be `fastmcp` itself use
`fastmcp_version` for core compatibility.
**overrides: Any `PluginMeta` field. Overrides take precedence
over the derived value. `name` is required unless a
`name` override is supplied; the distribution name is not
used as the plugin name by default since the two serve
different purposes (distribution = wheel identity, plugin
name = runtime identifier shown to Horizon / CLI users).
Raises:
PluginError: If the distribution is not installed in the
current environment, if `distribution` is `fastmcp`
(which would produce an invalid manifest), or if the
distribution's version cannot be parsed.
"""
# FastMCP itself is implicit; pinning it would produce a manifest
# that Plugin._validate_meta rejects. Plugin authors expressing
# core compatibility should use the `fastmcp_version` field.
if canonicalize_name(distribution) == "fastmcp":
raise PluginError(
f"PluginMeta.from_package({distribution!r}): "
f"`fastmcp` is implicit and must not be used as the "
f"containing distribution. Use the `fastmcp_version` "
f"field on PluginMeta to express core compatibility."
)
try:
dist = importlib_metadata.distribution(distribution)
except importlib_metadata.PackageNotFoundError as exc:
raise PluginError(
f"PluginMeta.from_package({distribution!r}): distribution "
f"is not installed in the current environment. Install it "
f"(e.g. via `uv pip install {distribution}`) before "
f"calling from_package."
) from exc
# `dist.metadata` is an email.message.Message at runtime, but
# `importlib.metadata.PackageMetadata`'s stubs don't expose that
# interface. Cast to email.message.Message to flatten header
# access (item lookup returns None on miss; `items()` yields one
# entry per header, including repeated keys like Project-URL).
raw = cast(EmailMessage, dist.metadata)
headers: dict[str, str] = {}
all_project_urls: list[str] = []
for key, value in raw.items():
if key == "Project-URL":
all_project_urls.append(value)
else:
# For repeated headers we only need one; first-wins.
headers.setdefault(key, value)
def _first_non_blank(*values: str | None) -> str | None:
"""Return the first value whose `.strip()` is truthy, or None.
Guards against whitespace-only headers silently blocking the
fallback chain (e.g. a METADATA file with `Author: ` would
otherwise make the `Author-email` fallback unreachable).
"""
for v in values:
if v is not None and v.strip():
return v.strip()
return None
derived: dict[str, Any] = {"version": dist.version}
# description ← Summary header
summary = _first_non_blank(headers.get("Summary"))
if summary:
derived["description"] = summary
# author ← Author, falling back to Author-email
author = _first_non_blank(headers.get("Author"), headers.get("Author-email"))
if author:
derived["author"] = author
# homepage ← Home-page, falling back to the first Project-URL
# whose label looks like a canonical homepage reference
homepage = _first_non_blank(headers.get("Home-page"))
if not homepage:
for entry in all_project_urls:
# Project-URL values are `"Label, URL"` pairs.
label, _, url = entry.partition(",")
if label.strip().lower() in {
"homepage",
"home",
"repository",
"source",
}:
homepage = _first_non_blank(url)
if homepage:
break
if homepage:
derived["homepage"] = homepage
# dependencies — pin the containing distribution at its current
# version, minus the local segment. PEP 440 only restricts local
# versions (`+abc.def`) from use with `>=` / `<=`; prereleases
# (`rc1`), dev (`.dev0`), and post segments are all valid there,
# so we preserve them to keep the pin meaningful for actively
# developed distributions. `Version.public` strips exactly the
# local segment.
try:
public = Version(dist.version).public
except InvalidVersion as exc:
raise PluginError(
f"PluginMeta.from_package({distribution!r}): could not "
f"parse distribution version {dist.version!r}: {exc}"
) from exc
derived["dependencies"] = [f"{distribution}>={public}"]
derived.update(overrides)
return cls(**derived)
class Plugin:
"""Base class for FastMCP plugins.

View file

@ -5,10 +5,13 @@ from __future__ import annotations
import asyncio
import json
from contextlib import suppress
from importlib import metadata as importlib_metadata
from importlib.metadata import version as dist_version
from pathlib import Path
import pytest
from pydantic import BaseModel
from packaging.version import Version
from pydantic import BaseModel, ValidationError
import fastmcp
from fastmcp import Client, FastMCP
@ -68,6 +71,174 @@ class TestPluginMeta:
assert meta.owning_team == "platform"
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."""