Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions src/fastmcp/cli/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,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()
Expand Down Expand Up @@ -122,6 +123,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: uv pip install --upgrade fastmcp[/dim]")


@app.command
async def dev(
Expand Down
15 changes: 15 additions & 0 deletions src/fastmcp/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -392,6 +392,21 @@ 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"

@property
def server_auth_class(self) -> AuthProvider | None:
from fastmcp.utilities.types import get_cached_typeadapter
Expand Down
35 changes: 31 additions & 4 deletions src/fastmcp/utilities/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)),
Expand All @@ -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: uv 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))
153 changes: 153 additions & 0 deletions src/fastmcp/utilities/version_check.py
Original file line number Diff line number Diff line change
@@ -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
Comment on lines +9 to +12
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Avoid circular import from module-level fastmcp import

Importing fastmcp.utilities.version_check directly now executes import fastmcp at module load time. That triggers fastmcp.__init__fastmcp.server.server fastmcp.utilities.cli, which does from fastmcp.utilities.version_check import check_for_newer_version while version_check is still initializing. In that scenario, the attribute is not defined yet and Python raises ImportError from a partially initialized module. This breaks direct imports (including the new tests) and any downstream usage that wants version-check utilities without importing fastmcp first. Move the import fastmcp into the functions that need it or otherwise break the module-level dependency to avoid the cycle.

Useful? React with 👍 / 👎.


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
Loading
Loading