Add CLI update notifications (#2840)

src/fastmcp/cli/cli.py

@@ -29,6 +29,7 @@ from fastmcp.utilities.inspect import (
 )
 from fastmcp.utilities.logging import get_logger
 from fastmcp.utilities.mcp_server_config import MCPServerConfig
+from fastmcp.utilities.version_check import check_for_newer_version
 
 logger = get_logger("cli")
 console = Console()
@@ -121,6 +122,14 @@ def version(
     else:
         console.print(g)
 
+        # Check for updates (not included in --copy output)
+        if newer_version := check_for_newer_version():
+            console.print()
+            console.print(
+                f"[bold]🎉 FastMCP update available:[/bold] [green]{newer_version}[/green]"
+            )
+            console.print("[dim]Run: pip install --upgrade fastmcp[/dim]")
+
 
 @app.command
 async def dev(

src/fastmcp/settings.py

@@ -333,3 +333,18 @@ class Settings(BaseSettings):
             ),
         ),
     ] = True
+
+    check_for_updates: Annotated[
+        Literal["stable", "prerelease", "off"],
+        Field(
+            description=inspect.cleandoc(
+                """
+                Controls update checking when displaying the CLI banner.
+                - "stable": Check for stable releases only (default)
+                - "prerelease": Also check for pre-release versions (alpha, beta, rc)
+                - "off": Disable update checking entirely
+                Set via FASTMCP_CHECK_FOR_UPDATES environment variable.
+                """
+            ),
+        ),
+    ] = "stable"

src/fastmcp/utilities/cli.py

@@ -17,6 +17,7 @@ from fastmcp.utilities.logging import get_logger
 from fastmcp.utilities.mcp_server_config import MCPServerConfig
 from fastmcp.utilities.mcp_server_config.v1.sources.filesystem import FileSystemSource
 from fastmcp.utilities.types import get_cached_typeadapter
+from fastmcp.utilities.version_check import check_for_newer_version
 
 if TYPE_CHECKING:
     from fastmcp import FastMCP
@@ -200,6 +201,9 @@ LOGO_ASCII_4 = (
 def log_server_banner(server: FastMCP[Any]) -> None:
     """Creates and logs a formatted banner with server information and logo."""
 
+    # Check for updates (non-blocking, fails silently)
+    newer_version = check_for_newer_version()
+
     # Create the logo text
     # Use Text with no_wrap and markup disabled to preserve ANSI escape codes
     logo_text = Text.from_ansi(LOGO_ASCII_4, no_wrap=True)
@@ -231,8 +235,10 @@ def log_server_banner(server: FastMCP[Any]) -> None:
 
     # v3 notice banner (shown below main panel)
     v3_line1 = Text("✨ FastMCP 3.0 is coming!", style="bold")
-    v3_line2 = Text(
-        "Pin fastmcp<3 in production, then upgrade when you're ready.", style="dim"
+    v3_line2 = Text.assemble(
+        ("Pin ", "dim"),
+        ("`fastmcp < 3`", "dim bold"),
+        (" in production, then upgrade when you're ready.", "dim"),
     )
     v3_notice = Panel(
         Group(Align.center(v3_line1), Align.center(v3_line2)),
@@ -250,5 +256,26 @@ def log_server_banner(server: FastMCP[Any]) -> None:
     )
 
     console = Console(stderr=True)
-    # Center both panels
-    console.print(Group("\n", Align.center(panel), Align.center(v3_notice), "\n"))
+
+    # Build output elements
+    output_elements: list[Align | Panel | str] = ["\n", Align.center(panel)]
+    output_elements.append(Align.center(v3_notice))
+
+    # Add update notice if a newer version is available (shown last for visibility)
+    if newer_version:
+        update_line1 = Text.assemble(
+            ("🎉 Update available: ", "bold"),
+            (newer_version, "bold green"),
+        )
+        update_line2 = Text("Run: pip install --upgrade fastmcp", style="dim")
+        update_notice = Panel(
+            Group(Align.center(update_line1), Align.center(update_line2)),
+            border_style="blue",
+            padding=(0, 2),
+            width=80,
+        )
+        output_elements.append(Align.center(update_notice))
+
+    output_elements.append("\n")
+
+    console.print(Group(*output_elements))

src/fastmcp/utilities/version_check.py

@@ -0,0 +1,153 @@
+"""Version checking utilities for FastMCP."""
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+
+import httpx
+from packaging.version import Version
+
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+PYPI_URL = "https://pypi.org/pypi/fastmcp/json"
+CACHE_TTL_SECONDS = 60 * 60 * 12  # 12 hours
+REQUEST_TIMEOUT_SECONDS = 2.0
+
+
+def _get_cache_path(include_prereleases: bool = False) -> Path:
+    """Get the path to the version cache file."""
+    import fastmcp
+
+    suffix = "_prerelease" if include_prereleases else ""
+    return fastmcp.settings.home / f"version_cache{suffix}.json"
+
+
+def _read_cache(include_prereleases: bool = False) -> tuple[str | None, float]:
+    """Read cached version info.
+
+    Returns:
+        Tuple of (cached_version, cache_timestamp) or (None, 0) if no cache.
+    """
+    cache_path = _get_cache_path(include_prereleases)
+    if not cache_path.exists():
+        return None, 0
+
+    try:
+        data = json.loads(cache_path.read_text())
+        return data.get("latest_version"), data.get("timestamp", 0)
+    except (json.JSONDecodeError, OSError):
+        return None, 0
+
+
+def _write_cache(latest_version: str, include_prereleases: bool = False) -> None:
+    """Write version info to cache."""
+    cache_path = _get_cache_path(include_prereleases)
+    try:
+        cache_path.parent.mkdir(parents=True, exist_ok=True)
+        cache_path.write_text(
+            json.dumps({"latest_version": latest_version, "timestamp": time.time()})
+        )
+    except OSError:
+        # Silently ignore cache write failures
+        pass
+
+
+def _fetch_latest_version(include_prereleases: bool = False) -> str | None:
+    """Fetch the latest version from PyPI.
+
+    Args:
+        include_prereleases: If True, include pre-release versions (alpha, beta, rc).
+
+    Returns:
+        The latest version string, or None if the fetch failed.
+    """
+    try:
+        response = httpx.get(PYPI_URL, timeout=REQUEST_TIMEOUT_SECONDS)
+        response.raise_for_status()
+        data = response.json()
+
+        releases = data.get("releases", {})
+        if not releases:
+            return None
+
+        versions = []
+        for version_str in releases:
+            try:
+                v = Version(version_str)
+                # Skip prereleases if not requested
+                if not include_prereleases and v.is_prerelease:
+                    continue
+                versions.append(v)
+            except ValueError:
+                logger.debug(f"Skipping invalid version string: {version_str}")
+                continue
+
+        if not versions:
+            return None
+
+        return str(max(versions))
+
+    except (httpx.HTTPError, json.JSONDecodeError, KeyError):
+        return None
+
+
+def get_latest_version(include_prereleases: bool = False) -> str | None:
+    """Get the latest version of FastMCP from PyPI, using cache when available.
+
+    Args:
+        include_prereleases: If True, include pre-release versions.
+
+    Returns:
+        The latest version string, or None if unavailable.
+    """
+    # Check cache first
+    cached_version, cache_timestamp = _read_cache(include_prereleases)
+    if cached_version and (time.time() - cache_timestamp) < CACHE_TTL_SECONDS:
+        return cached_version
+
+    # Fetch from PyPI
+    latest_version = _fetch_latest_version(include_prereleases)
+
+    # Update cache if we got a valid version
+    if latest_version:
+        _write_cache(latest_version, include_prereleases)
+        return latest_version
+
+    # Return stale cache if available
+    return cached_version
+
+
+def check_for_newer_version() -> str | None:
+    """Check if a newer version of FastMCP is available.
+
+    Returns:
+        The latest version string if newer than current, None otherwise.
+    """
+    import fastmcp
+
+    setting = fastmcp.settings.check_for_updates
+    if setting == "off":
+        return None
+
+    include_prereleases = setting == "prerelease"
+    latest_version = get_latest_version(include_prereleases)
+    if not latest_version:
+        return None
+
+    try:
+        current = Version(fastmcp.__version__)
+        latest = Version(latest_version)
+
+        if latest > current:
+            return latest_version
+    except ValueError:
+        logger.debug(
+            f"Could not compare versions: current={fastmcp.__version__!r}, "
+            f"latest={latest_version!r}"
+        )
+
+    return None

tests/utilities/test_version_check.py

@@ -0,0 +1,315 @@
+"""Tests for version checking utilities."""
+
+import json
+import time
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import httpx
+import pytest
+
+from fastmcp.utilities.version_check import (
+    CACHE_TTL_SECONDS,
+    _fetch_latest_version,
+    _get_cache_path,
+    _read_cache,
+    _write_cache,
+    check_for_newer_version,
+    get_latest_version,
+)
+
+
+class TestCachePath:
+    def test_cache_path_in_home_directory(self):
+        """Cache file should be in fastmcp home directory."""
+        cache_path = _get_cache_path()
+        assert cache_path.name == "version_cache.json"
+        assert "fastmcp" in str(cache_path).lower()
+
+    def test_cache_path_prerelease_suffix(self):
+        """Prerelease cache uses different file."""
+        cache_path = _get_cache_path(include_prereleases=True)
+        assert cache_path.name == "version_cache_prerelease.json"
+
+
+class TestReadCache:
+    def test_read_cache_no_file(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        """Reading non-existent cache returns None."""
+        monkeypatch.setattr(
+            "fastmcp.utilities.version_check._get_cache_path",
+            lambda include_prereleases=False: tmp_path / "nonexistent.json",
+        )
+        version, timestamp = _read_cache()
+        assert version is None
+        assert timestamp == 0
+
+    def test_read_cache_valid(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+        """Reading valid cache returns version and timestamp."""
+        cache_file = tmp_path / "version_cache.json"
+        cache_file.write_text(
+            json.dumps({"latest_version": "2.5.0", "timestamp": 1000})
+        )
+        monkeypatch.setattr(
+            "fastmcp.utilities.version_check._get_cache_path",
+            lambda include_prereleases=False: cache_file,
+        )
+
+        version, timestamp = _read_cache()
+        assert version == "2.5.0"
+        assert timestamp == 1000
+
+    def test_read_cache_invalid_json(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Reading invalid JSON returns None."""
+        cache_file = tmp_path / "version_cache.json"
+        cache_file.write_text("not valid json")
+        monkeypatch.setattr(
+            "fastmcp.utilities.version_check._get_cache_path",
+            lambda include_prereleases=False: cache_file,
+        )
+
+        version, timestamp = _read_cache()
+        assert version is None
+        assert timestamp == 0
+
+
+class TestWriteCache:
+    def test_write_cache_creates_file(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Writing cache creates the cache file."""
+        cache_file = tmp_path / "subdir" / "version_cache.json"
+        monkeypatch.setattr(
+            "fastmcp.utilities.version_check._get_cache_path",
+            lambda include_prereleases=False: cache_file,
+        )
+
+        _write_cache("2.6.0")
+
+        assert cache_file.exists()
+        data = json.loads(cache_file.read_text())
+        assert data["latest_version"] == "2.6.0"
+        assert "timestamp" in data
+
+
+class TestFetchLatestVersion:
+    def test_fetch_success(self):
+        """Successful fetch returns highest stable version."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "releases": {
+                "2.5.0": [],
+                "2.4.0": [],
+                "2.6.0b1": [],  # prerelease should be skipped
+            }
+        }
+
+        with patch("httpx.get", return_value=mock_response) as mock_get:
+            version = _fetch_latest_version()
+            assert version == "2.5.0"
+            mock_get.assert_called_once()
+
+    def test_fetch_network_error(self):
+        """Network error returns None."""
+        with patch("httpx.get", side_effect=httpx.HTTPError("Network error")):
+            version = _fetch_latest_version()
+            assert version is None
+
+    def test_fetch_invalid_response(self):
+        """Invalid response returns None."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"unexpected": "format"}
+
+        with patch("httpx.get", return_value=mock_response):
+            version = _fetch_latest_version()
+            assert version is None
+
+    def test_fetch_prereleases(self):
+        """Fetching with prereleases returns highest version."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "info": {"version": "2.5.0"},
+            "releases": {
+                "2.5.0": [],
+                "2.6.0b1": [],
+                "2.6.0b2": [],
+                "2.4.0": [],
+            },
+        }
+
+        with patch("httpx.get", return_value=mock_response):
+            version = _fetch_latest_version(include_prereleases=True)
+            assert version == "2.6.0b2"
+
+    def test_fetch_prereleases_stable_is_highest(self):
+        """Prerelease mode still returns stable if it's highest."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "info": {"version": "2.5.0"},
+            "releases": {
+                "2.5.0": [],
+                "2.5.0b1": [],
+                "2.4.0": [],
+            },
+        }
+
+        with patch("httpx.get", return_value=mock_response):
+            version = _fetch_latest_version(include_prereleases=True)
+            assert version == "2.5.0"
+
+
+class TestGetLatestVersion:
+    def test_returns_cached_version_if_fresh(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Uses cached version if cache is fresh."""
+        cache_file = tmp_path / "version_cache.json"
+        cache_file.write_text(
+            json.dumps({"latest_version": "2.5.0", "timestamp": time.time()})
+        )
+        monkeypatch.setattr(
+            "fastmcp.utilities.version_check._get_cache_path",
+            lambda include_prereleases=False: cache_file,
+        )
+
+        with patch(
+            "fastmcp.utilities.version_check._fetch_latest_version"
+        ) as mock_fetch:
+            version = get_latest_version()
+            assert version == "2.5.0"
+            mock_fetch.assert_not_called()
+
+    def test_fetches_if_cache_stale(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Fetches from PyPI if cache is stale."""
+        cache_file = tmp_path / "version_cache.json"
+        old_timestamp = time.time() - CACHE_TTL_SECONDS - 100
+        cache_file.write_text(
+            json.dumps({"latest_version": "2.4.0", "timestamp": old_timestamp})
+        )
+        monkeypatch.setattr(
+            "fastmcp.utilities.version_check._get_cache_path",
+            lambda include_prereleases=False: cache_file,
+        )
+
+        with patch(
+            "fastmcp.utilities.version_check._fetch_latest_version",
+            return_value="2.5.0",
+        ) as mock_fetch:
+            version = get_latest_version()
+            assert version == "2.5.0"
+            mock_fetch.assert_called_once()
+
+    def test_returns_stale_cache_if_fetch_fails(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Returns stale cache if fetch fails."""
+        cache_file = tmp_path / "version_cache.json"
+        old_timestamp = time.time() - CACHE_TTL_SECONDS - 100
+        cache_file.write_text(
+            json.dumps({"latest_version": "2.4.0", "timestamp": old_timestamp})
+        )
+        monkeypatch.setattr(
+            "fastmcp.utilities.version_check._get_cache_path",
+            lambda include_prereleases=False: cache_file,
+        )
+
+        with patch(
+            "fastmcp.utilities.version_check._fetch_latest_version", return_value=None
+        ):
+            version = get_latest_version()
+            assert version == "2.4.0"
+
+
+class TestCheckForNewerVersion:
+    def test_returns_none_if_disabled(self, monkeypatch: pytest.MonkeyPatch):
+        """Returns None if check_for_updates is off."""
+        import fastmcp
+
+        monkeypatch.setattr(fastmcp.settings, "check_for_updates", "off")
+
+        result = check_for_newer_version()
+        assert result is None
+
+    def test_returns_none_if_current(self, monkeypatch: pytest.MonkeyPatch):
+        """Returns None if current version is latest."""
+        import fastmcp
+
+        monkeypatch.setattr(fastmcp.settings, "check_for_updates", "stable")
+        monkeypatch.setattr(fastmcp, "__version__", "2.5.0")
+
+        with patch(
+            "fastmcp.utilities.version_check.get_latest_version", return_value="2.5.0"
+        ):
+            result = check_for_newer_version()
+            assert result is None
+
+    def test_returns_version_if_newer(self, monkeypatch: pytest.MonkeyPatch):
+        """Returns new version if available."""
+        import fastmcp
+
+        monkeypatch.setattr(fastmcp.settings, "check_for_updates", "stable")
+        monkeypatch.setattr(fastmcp, "__version__", "2.4.0")
+
+        with patch(
+            "fastmcp.utilities.version_check.get_latest_version", return_value="2.5.0"
+        ):
+            result = check_for_newer_version()
+            assert result == "2.5.0"
+
+    def test_returns_none_if_older_available(self, monkeypatch: pytest.MonkeyPatch):
+        """Returns None if pypi version is older than current (dev version)."""
+        import fastmcp
+
+        monkeypatch.setattr(fastmcp.settings, "check_for_updates", "stable")
+        monkeypatch.setattr(fastmcp, "__version__", "3.0.0.dev1")
+
+        with patch(
+            "fastmcp.utilities.version_check.get_latest_version", return_value="2.5.0"
+        ):
+            result = check_for_newer_version()
+            assert result is None
+
+    def test_handles_invalid_versions(self, monkeypatch: pytest.MonkeyPatch):
+        """Handles invalid version strings gracefully."""
+        import fastmcp
+
+        monkeypatch.setattr(fastmcp.settings, "check_for_updates", "stable")
+        monkeypatch.setattr(fastmcp, "__version__", "invalid")
+
+        with patch(
+            "fastmcp.utilities.version_check.get_latest_version",
+            return_value="also-invalid",
+        ):
+            result = check_for_newer_version()
+            assert result is None
+
+    def test_prerelease_setting(self, monkeypatch: pytest.MonkeyPatch):
+        """Prerelease setting passes include_prereleases=True."""
+        import fastmcp
+
+        monkeypatch.setattr(fastmcp.settings, "check_for_updates", "prerelease")
+        monkeypatch.setattr(fastmcp, "__version__", "2.5.0")
+
+        with patch(
+            "fastmcp.utilities.version_check.get_latest_version", return_value="2.6.0b1"
+        ) as mock_get:
+            result = check_for_newer_version()
+            assert result == "2.6.0b1"
+            mock_get.assert_called_once_with(True)
+
+    def test_stable_setting(self, monkeypatch: pytest.MonkeyPatch):
+        """Stable setting passes include_prereleases=False."""
+        import fastmcp
+
+        monkeypatch.setattr(fastmcp.settings, "check_for_updates", "stable")
+        monkeypatch.setattr(fastmcp, "__version__", "2.4.0")
+
+        with patch(
+            "fastmcp.utilities.version_check.get_latest_version", return_value="2.5.0"
+        ) as mock_get:
+            result = check_for_newer_version()
+            assert result == "2.5.0"
+            mock_get.assert_called_once_with(False)