diff --git a/src/fastmcp/cli/cli.py b/src/fastmcp/cli/cli.py index 90c50bc2db..c8a358cea6 100644 --- a/src/fastmcp/cli/cli.py +++ b/src/fastmcp/cli/cli.py @@ -29,6 +29,7 @@ ) 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( diff --git a/src/fastmcp/settings.py b/src/fastmcp/settings.py index ce589834b9..99955375a6 100644 --- a/src/fastmcp/settings.py +++ b/src/fastmcp/settings.py @@ -333,3 +333,18 @@ def normalize_log_level(cls, v): ), ), ] = 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" diff --git a/src/fastmcp/utilities/cli.py b/src/fastmcp/utilities/cli.py index 6b7997b019..13d4965875 100644 --- a/src/fastmcp/utilities/cli.py +++ b/src/fastmcp/utilities/cli.py @@ -17,6 +17,7 @@ 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 @@ def load_and_merge_config( 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)) diff --git a/src/fastmcp/utilities/version_check.py b/src/fastmcp/utilities/version_check.py new file mode 100644 index 0000000000..5c3aba6340 --- /dev/null +++ b/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 diff --git a/tests/utilities/test_version_check.py b/tests/utilities/test_version_check.py new file mode 100644 index 0000000000..9cc9aab3b3 --- /dev/null +++ b/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)