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 @@ def create_panel(self, content: Text, title: str, style: str = "blue") -> Panel:
             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 @@ def handle_crew_started(self, crew_name: str, source_id: str) -> None:
         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 @@ def handle_flow_created(self, flow_name: str, flow_id: str) -> None:
 
     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 @@ def handle_agent_logs_execution(
 
             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