feat: add version check & integrate update notices

lib/crewai/src/crewai/cli/version.py

@@ -1,6 +1,107 @@
+"""Version utilities for CrewAI CLI."""
+
+from collections.abc import Mapping
+from datetime import datetime, timedelta
+from functools import lru_cache
 import importlib.metadata
+import json
+from pathlib import Path
+from typing import Any, cast
+from urllib import request
+from urllib.error import URLError
+
+import appdirs
+from packaging.version import InvalidVersion, parse
+
+
+@lru_cache(maxsize=1)
+def _get_cache_file() -> Path:
+    """Get the path to the version cache file.
+
+    Cached to avoid repeated filesystem operations.
+    """
+    cache_dir = Path(appdirs.user_cache_dir("crewai"))
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    return cache_dir / "version_cache.json"
 
 
 def get_crewai_version() -> str:
-    """Get the version number of CrewAI running the CLI"""
+    """Get the version number of CrewAI running the CLI."""
     return importlib.metadata.version("crewai")
+
+
+def _is_cache_valid(cache_data: Mapping[str, Any]) -> bool:
+    """Check if the cache is still valid, less than 24 hours old."""
+    if "timestamp" not in cache_data:
+        return False
+
+    try:
+        cache_time = datetime.fromisoformat(str(cache_data["timestamp"]))
+        return datetime.now() - cache_time < timedelta(hours=24)
+    except (ValueError, TypeError):
+        return False
+
+
+def get_latest_version_from_pypi(timeout: int = 2) -> str | None:
+    """Get the latest version of CrewAI from PyPI.
+
+    Args:
+        timeout: Request timeout in seconds.
+
+    Returns:
+        Latest version string or None if unable to fetch.
+    """
+    cache_file = _get_cache_file()
+    if cache_file.exists():
+        try:
+            cache_data = json.loads(cache_file.read_text())
+            if _is_cache_valid(cache_data):
+                return cast(str | None, cache_data.get("version"))
+        except (json.JSONDecodeError, OSError):
+            pass
+
+    try:
+        with request.urlopen(
+            "https://pypi.org/pypi/crewai/json", timeout=timeout
+        ) as response:
+            data = json.loads(response.read())
+            latest_version = cast(str, data["info"]["version"])
+
+            cache_data = {
+                "version": latest_version,
+                "timestamp": datetime.now().isoformat(),
+            }
+            cache_file.write_text(json.dumps(cache_data))
+
+            return latest_version
+    except (URLError, json.JSONDecodeError, KeyError, OSError):
+        return None
+
+
+def check_version() -> tuple[str, str | None]:
+    """Check current and latest versions.
+
+    Returns:
+        Tuple of (current_version, latest_version).
+        latest_version is None if unable to fetch from PyPI.
+    """
+    current = get_crewai_version()
+    latest = get_latest_version_from_pypi()
+    return current, latest
+
+
+def is_newer_version_available() -> tuple[bool, str, str | None]:
+    """Check if a newer version is available.
+
+    Returns:
+        Tuple of (is_newer, current_version, latest_version).
+    """
+    current, latest = check_version()
+
+    if latest is None:
+        return False, current, None
+
+    try:
+        return parse(latest) > parse(current), current, latest
+    except (InvalidVersion, TypeError):
+        return False, current, latest

lib/crewai/src/crewai/events/utils/console_formatter.py

@@ -1,11 +1,14 @@
+import os
 import threading
-from typing import Any, ClassVar
+from typing import Any, ClassVar, cast
 
 from rich.console import Console
 from rich.live import Live
 from rich.panel import Panel
 from rich.text import Text
 
+from crewai.cli.version import is_newer_version_available
+
 
 class ConsoleFormatter:
     tool_usage_counts: ClassVar[dict[str, int]] = {}
@@ -35,6 +38,39 @@ class ConsoleFormatter:
             padding=(1, 2),
         )
 
+    def _show_version_update_message_if_needed(self) -> None:
+        """Show version update message if a newer version is available.
+
+        Only displays when verbose mode is enabled and not running in CI/CD.
+        """
+        if not self.verbose:
+            return
+
+        if os.getenv("CI", "").lower() in ("true", "1"):
+            return
+
+        try:
+            is_newer, current, latest = is_newer_version_available()
+            if is_newer and latest:
+                message = f"""A new version of CrewAI is available!
+
+Current version: {current}
+Latest version:  {latest}
+
+To update, run: uv sync --upgrade-package crewai"""
+
+                panel = Panel(
+                    message,
+                    title="✨ Update Available ✨",
+                    border_style="yellow",
+                    padding=(1, 2),
+                )
+                self.console.print(panel)
+                self.console.print()
+        except Exception:  # noqa: S110
+            # Silently ignore errors in version check - it's non-critical
+            pass
+
     def _show_tracing_disabled_message_if_needed(self) -> None:
         """Show tracing disabled message if tracing is not enabled."""
         from crewai.events.listeners.tracing.utils import (
@@ -176,9 +212,10 @@ To enable tracing, do any one of these:
         if not self.verbose:
             return
 
-        # Reset the crew completion event for this new crew execution
         ConsoleFormatter.crew_completion_printed.clear()
 
+        self._show_version_update_message_if_needed()
+
         content = self.create_status_content(
             "Crew Execution Started",
             crew_name,
@@ -237,6 +274,8 @@ To enable tracing, do any one of these:
 
     def handle_flow_started(self, flow_name: str, flow_id: str) -> None:
         """Show flow started panel."""
+        self._show_version_update_message_if_needed()
+
         content = Text()
         content.append("Flow Started\n", style="blue bold")
         content.append("Name: ", style="white")
@@ -885,7 +924,7 @@ To enable tracing, do any one of these:
 
             is_a2a_delegation = False
             try:
-                output_data = json.loads(formatted_answer.output)
+                output_data = json.loads(cast(str, formatted_answer.output))
                 if isinstance(output_data, dict):
                     if output_data.get("is_a2a") is True:
                         is_a2a_delegation = True

lib/crewai/tests/cli/test_version.py

@@ -1,10 +1,20 @@
 """Test for version management."""
 
+from datetime import datetime, timedelta
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
 from crewai import __version__
-from crewai.cli.version import get_crewai_version
+from crewai.cli.version import (
+    _get_cache_file,
+    _is_cache_valid,
+    get_crewai_version,
+    get_latest_version_from_pypi,
+    is_newer_version_available,
+)
 
 
-def test_dynamic_versioning_consistency():
+def test_dynamic_versioning_consistency() -> None:
     """Test that dynamic versioning provides consistent version across all access methods."""
     cli_version = get_crewai_version()
     package_version = __version__
@@ -15,3 +25,186 @@ def test_dynamic_versioning_consistency():
     # Version should not be empty
     assert package_version is not None
     assert len(package_version.strip()) > 0
+
+
+class TestVersionChecking:
+    """Test version checking utilities."""
+
+    def test_get_crewai_version(self) -> None:
+        """Test getting current crewai version."""
+        version = get_crewai_version()
+        assert isinstance(version, str)
+        assert len(version) > 0
+
+    def test_get_cache_file(self) -> None:
+        """Test cache file path generation."""
+        cache_file = _get_cache_file()
+        assert isinstance(cache_file, Path)
+        assert cache_file.name == "version_cache.json"
+
+    def test_is_cache_valid_with_fresh_cache(self) -> None:
+        """Test cache validation with fresh cache."""
+        cache_data = {"timestamp": datetime.now().isoformat(), "version": "1.0.0"}
+        assert _is_cache_valid(cache_data) is True
+
+    def test_is_cache_valid_with_stale_cache(self) -> None:
+        """Test cache validation with stale cache."""
+        old_time = datetime.now() - timedelta(hours=25)
+        cache_data = {"timestamp": old_time.isoformat(), "version": "1.0.0"}
+        assert _is_cache_valid(cache_data) is False
+
+    def test_is_cache_valid_with_missing_timestamp(self) -> None:
+        """Test cache validation with missing timestamp."""
+        cache_data = {"version": "1.0.0"}
+        assert _is_cache_valid(cache_data) is False
+
+    @patch("crewai.cli.version.Path.exists")
+    @patch("crewai.cli.version.request.urlopen")
+    def test_get_latest_version_from_pypi_success(
+        self, mock_urlopen: MagicMock, mock_exists: MagicMock
+    ) -> None:
+        """Test successful PyPI version fetch."""
+        # Mock cache not existing to force fetch from PyPI
+        mock_exists.return_value = False
+
+        mock_response = MagicMock()
+        mock_response.read.return_value = b'{"info": {"version": "2.0.0"}}'
+        mock_urlopen.return_value.__enter__.return_value = mock_response
+
+        version = get_latest_version_from_pypi()
+        assert version == "2.0.0"
+
+    @patch("crewai.cli.version.Path.exists")
+    @patch("crewai.cli.version.request.urlopen")
+    def test_get_latest_version_from_pypi_failure(
+        self, mock_urlopen: MagicMock, mock_exists: MagicMock
+    ) -> None:
+        """Test PyPI version fetch failure."""
+        from urllib.error import URLError
+
+        # Mock cache not existing to force fetch from PyPI
+        mock_exists.return_value = False
+
+        mock_urlopen.side_effect = URLError("Network error")
+
+        version = get_latest_version_from_pypi()
+        assert version is None
+
+    @patch("crewai.cli.version.get_crewai_version")
+    @patch("crewai.cli.version.get_latest_version_from_pypi")
+    def test_is_newer_version_available_true(
+        self, mock_latest: MagicMock, mock_current: MagicMock
+    ) -> None:
+        """Test when newer version is available."""
+        mock_current.return_value = "1.0.0"
+        mock_latest.return_value = "2.0.0"
+
+        is_newer, current, latest = is_newer_version_available()
+        assert is_newer is True
+        assert current == "1.0.0"
+        assert latest == "2.0.0"
+
+    @patch("crewai.cli.version.get_crewai_version")
+    @patch("crewai.cli.version.get_latest_version_from_pypi")
+    def test_is_newer_version_available_false(
+        self, mock_latest: MagicMock, mock_current: MagicMock
+    ) -> None:
+        """Test when no newer version is available."""
+        mock_current.return_value = "2.0.0"
+        mock_latest.return_value = "2.0.0"
+
+        is_newer, current, latest = is_newer_version_available()
+        assert is_newer is False
+        assert current == "2.0.0"
+        assert latest == "2.0.0"
+
+    @patch("crewai.cli.version.get_crewai_version")
+    @patch("crewai.cli.version.get_latest_version_from_pypi")
+    def test_is_newer_version_available_with_none_latest(
+        self, mock_latest: MagicMock, mock_current: MagicMock
+    ) -> None:
+        """Test when PyPI fetch fails."""
+        mock_current.return_value = "1.0.0"
+        mock_latest.return_value = None
+
+        is_newer, current, latest = is_newer_version_available()
+        assert is_newer is False
+        assert current == "1.0.0"
+        assert latest is None
+
+
+class TestConsoleFormatterVersionCheck:
+    """Test version check display in ConsoleFormatter."""
+
+    @patch("crewai.events.utils.console_formatter.is_newer_version_available")
+    @patch.dict("os.environ", {"CI": ""})
+    def test_version_message_shows_when_update_available_and_verbose(
+        self, mock_check: MagicMock
+    ) -> None:
+        """Test version message shows when update available and verbose enabled."""
+        from crewai.events.utils.console_formatter import ConsoleFormatter
+
+        mock_check.return_value = (True, "1.0.0", "2.0.0")
+
+        formatter = ConsoleFormatter(verbose=True)
+        with patch.object(formatter.console, "print") as mock_print:
+            formatter._show_version_update_message_if_needed()
+            assert mock_print.call_count == 2
+
+    @patch("crewai.events.utils.console_formatter.is_newer_version_available")
+    def test_version_message_hides_when_verbose_false(
+        self, mock_check: MagicMock
+    ) -> None:
+        """Test version message hidden when verbose disabled."""
+        from crewai.events.utils.console_formatter import ConsoleFormatter
+
+        mock_check.return_value = (True, "1.0.0", "2.0.0")
+
+        formatter = ConsoleFormatter(verbose=False)
+        with patch.object(formatter.console, "print") as mock_print:
+            formatter._show_version_update_message_if_needed()
+            mock_print.assert_not_called()
+
+    @patch("crewai.events.utils.console_formatter.is_newer_version_available")
+    def test_version_message_hides_when_no_update_available(
+        self, mock_check: MagicMock
+    ) -> None:
+        """Test version message hidden when no update available."""
+        from crewai.events.utils.console_formatter import ConsoleFormatter
+
+        mock_check.return_value = (False, "2.0.0", "2.0.0")
+
+        formatter = ConsoleFormatter(verbose=True)
+        with patch.object(formatter.console, "print") as mock_print:
+            formatter._show_version_update_message_if_needed()
+            mock_print.assert_not_called()
+
+    @patch("crewai.events.utils.console_formatter.is_newer_version_available")
+    @patch.dict("os.environ", {"CI": "true"})
+    def test_version_message_hides_in_ci_environment(
+        self, mock_check: MagicMock
+    ) -> None:
+        """Test version message hidden when running in CI/CD."""
+        from crewai.events.utils.console_formatter import ConsoleFormatter
+
+        mock_check.return_value = (True, "1.0.0", "2.0.0")
+
+        formatter = ConsoleFormatter(verbose=True)
+        with patch.object(formatter.console, "print") as mock_print:
+            formatter._show_version_update_message_if_needed()
+            mock_print.assert_not_called()
+
+    @patch("crewai.events.utils.console_formatter.is_newer_version_available")
+    @patch.dict("os.environ", {"CI": "1"})
+    def test_version_message_hides_in_ci_environment_with_numeric_value(
+        self, mock_check: MagicMock
+    ) -> None:
+        """Test version message hidden when CI=1."""
+        from crewai.events.utils.console_formatter import ConsoleFormatter
+
+        mock_check.return_value = (True, "1.0.0", "2.0.0")
+
+        formatter = ConsoleFormatter(verbose=True)
+        with patch.object(formatter.console, "print") as mock_print:
+            formatter._show_version_update_message_if_needed()
+            mock_print.assert_not_called()