diff --git a/src/mcp_agent/cli/cloud/commands/app/status/main.py b/src/mcp_agent/cli/cloud/commands/app/status/main.py
index 463e8beb9..0c3c69812 100644
--- a/src/mcp_agent/cli/cloud/commands/app/status/main.py
+++ b/src/mcp_agent/cli/cloud/commands/app/status/main.py
@@ -37,7 +37,7 @@ def get_app_status(
         None,
         "--id",
         "-i",
-        help="ID or server URL of the app or app configuration to get details for.",
+        help="ID, server URL, or name of the app to get details for.",
     ),
     api_url: Optional[str] = typer.Option(
         settings.API_BASE_URL,
@@ -58,7 +58,7 @@ def get_app_status(
     if not effective_api_key:
         raise CLIError(
             "Must be logged in to get app status. Run 'mcp-agent login', set MCP_API_KEY environment variable or specify --api-key option.",
-            retriable=False
+            retriable=False,
         )
 
     client = MCPAppClient(
@@ -94,7 +94,7 @@ def get_app_status(
     except UnauthenticatedError as e:
         raise CLIError(
             "Invalid API key. Run 'mcp-agent login' or set MCP_API_KEY environment variable with new API key.",
-            retriable=False
+            retriable=False,
         ) from e
     except Exception as e:
         # Re-raise with more context - top-level CLI handler will show clean message
diff --git a/src/mcp_agent/cli/cloud/commands/auth/whoami/main.py b/src/mcp_agent/cli/cloud/commands/auth/whoami/main.py
index bd4306e9c..64f26dfa0 100644
--- a/src/mcp_agent/cli/cloud/commands/auth/whoami/main.py
+++ b/src/mcp_agent/cli/cloud/commands/auth/whoami/main.py
@@ -29,7 +29,9 @@ def whoami() -> None:
         )
     if not credentials:
         raise CLIError(
-            "Not authenticated. Set MCP_API_KEY or run 'mcp-agent login'.", exit_code=4, retriable=False
+            "Not authenticated. Set MCP_API_KEY or run 'mcp-agent login'.",
+            exit_code=4,
+            retriable=False,
         )
 
     if credentials.is_token_expired:
diff --git a/src/mcp_agent/cli/cloud/commands/deploy/main.py b/src/mcp_agent/cli/cloud/commands/deploy/main.py
index 4810837dd..19a7f6e37 100644
--- a/src/mcp_agent/cli/cloud/commands/deploy/main.py
+++ b/src/mcp_agent/cli/cloud/commands/deploy/main.py
@@ -5,6 +5,7 @@
 """
 
 from pathlib import Path
+from datetime import datetime, timezone
 from typing import Optional
 
 import typer
@@ -35,6 +36,11 @@
     print_info,
     print_success,
 )
+from mcp_agent.cli.utils.git_utils import (
+    get_git_metadata,
+    create_git_tag,
+    sanitize_git_ref_component,
+)
 
 from .wrangler_wrapper import wrangler_deploy
 
@@ -85,6 +91,12 @@ def deploy_config(
         help="API key for authentication. Defaults to MCP_API_KEY environment variable.",
         envvar=ENV_API_KEY,
     ),
+    git_tag: bool = typer.Option(
+        False,
+        "--git-tag/--no-git-tag",
+        help="Create a local git tag for this deploy (if in a git repo)",
+        envvar="MCP_DEPLOY_GIT_TAG",
+    ),
     retry_count: int = typer.Option(
         3,
         "--retry-count",
@@ -129,12 +141,12 @@ def deploy_config(
         if not effective_api_url:
             raise CLIError(
                 "MCP_API_BASE_URL environment variable or --api-url option must be set.",
-                retriable=False
+                retriable=False,
             )
         if not effective_api_key:
             raise CLIError(
                 "Must be logged in to deploy. Run 'mcp-agent login', set MCP_API_KEY environment variable or specify --api-key option.",
-                retriable=False
+                retriable=False,
             )
         print_info(f"Using API at {effective_api_url}")
 
@@ -178,7 +190,7 @@ def deploy_config(
         except UnauthenticatedError as e:
             raise CLIError(
                 "Invalid API key for deployment. Run 'mcp-agent login' or set MCP_API_KEY environment variable with new API key.",
-                retriable=False
+                retriable=False,
             ) from e
         except Exception as e:
             raise CLIError(f"Error checking or creating app: {str(e)}") from e
@@ -249,13 +261,36 @@ def deploy_config(
             )
         )
 
-        app = run_async(_deploy_with_retry(
-            app_id=app_id,
-            api_key=effective_api_key,
-            project_dir=config_dir,
-            mcp_app_client=mcp_app_client,
-            retry_count=retry_count,
-        ))
+        # Optionally create a local git tag as a breadcrumb of this deployment
+        if git_tag:
+            git_meta = get_git_metadata(config_dir)
+            if git_meta:
+                # Sanitize app name for git tag safety
+                safe_name = sanitize_git_ref_component(app_name)
+                ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
+                tag_name = f"mcp-deploy/{safe_name}/{ts}-{git_meta.short_sha}"
+                msg = (
+                    f"MCP Agent deploy for app '{app_name}' (id {app_id})\n"
+                    f"Commit: {git_meta.commit_sha}\n"
+                    f"Branch: {git_meta.branch or ''}\n"
+                    f"Dirty: {git_meta.dirty}"
+                )
+                if create_git_tag(config_dir, tag_name, msg):
+                    print_success(f"Created local git tag: {tag_name}")
+                else:
+                    print_info("Skipping git tag (not a repo or tag failed)")
+            else:
+                print_info("Skipping git tag (not a git repository)")
+
+        app = run_async(
+            _deploy_with_retry(
+                app_id=app_id,
+                api_key=effective_api_key,
+                project_dir=config_dir,
+                mcp_app_client=mcp_app_client,
+                retry_count=retry_count,
+            )
+        )
 
         print_info(f"App ID: {app_id}")
         if app.appServerInfo:
@@ -318,13 +353,45 @@ async def _perform_api_deployment():
             SpinnerColumn(spinner_name="arrow3"),
             TextColumn("[progress.description]{task.description}"),
         ) as progress:
-            deploy_task = progress.add_task(f"Deploying MCP App bundle{attempt_suffix}...", total=None)
+            deploy_task = progress.add_task(
+                f"Deploying MCP App bundle{attempt_suffix}...", total=None
+            )
             try:
-                app = await mcp_app_client.deploy_app(app_id=app_id)
-                progress.update(deploy_task, description=f"✅ MCP App deployed successfully{attempt_suffix}!")
+                # Optionally include minimal metadata (git only to avoid heavy scans)
+                metadata = None
+                gm = get_git_metadata(project_dir)
+                if gm:
+                    metadata = {
+                        "source": "git",
+                        "commit": gm.commit_sha,
+                        "short": gm.short_sha,
+                        "branch": gm.branch,
+                        "dirty": gm.dirty,
+                        "tag": gm.tag,
+                        "message": gm.commit_message,
+                    }
+
+                try:
+                    app = await mcp_app_client.deploy_app(
+                        app_id=app_id, deployment_metadata=metadata
+                    )
+                except Exception as e:
+                    # Fallback: if API rejects deploymentMetadata, retry once without it
+                    try:
+                        app = await mcp_app_client.deploy_app(
+                            app_id=app_id, deployment_metadata=None
+                        )
+                    except Exception:
+                        raise e
+                progress.update(
+                    deploy_task,
+                    description=f"✅ MCP App deployed successfully{attempt_suffix}!",
+                )
                 return app
             except Exception:
-                progress.update(deploy_task, description=f"❌ Deployment failed{attempt_suffix}")
+                progress.update(
+                    deploy_task, description=f"❌ Deployment failed{attempt_suffix}"
+                )
                 raise
 
     if retry_count > 1:
@@ -341,7 +408,9 @@ async def _perform_api_deployment():
     except RetryError as e:
         attempts_text = "attempts" if retry_count > 1 else "attempt"
         print_error(f"Deployment failed after {retry_count} {attempts_text}")
-        raise CLIError(f"Deployment failed after {retry_count} {attempts_text}. Last error: {e.original_error}") from e.original_error
+        raise CLIError(
+            f"Deployment failed after {retry_count} {attempts_text}. Last error: {e.original_error}"
+        ) from e.original_error
 
 
 def get_config_files(config_dir: Path) -> tuple[Path, Optional[Path], Optional[Path]]:
@@ -358,7 +427,7 @@ def get_config_files(config_dir: Path) -> tuple[Path, Optional[Path], Optional[P
     if not config_file.exists():
         raise CLIError(
             f"Configuration file '{MCP_CONFIG_FILENAME}' not found in {config_dir}",
-            retriable=False
+            retriable=False,
         )
 
     secrets_file: Optional[Path] = None
diff --git a/src/mcp_agent/cli/cloud/commands/deploy/wrangler_wrapper.py b/src/mcp_agent/cli/cloud/commands/deploy/wrangler_wrapper.py
index 3e6094683..229d947c5 100644
--- a/src/mcp_agent/cli/cloud/commands/deploy/wrangler_wrapper.py
+++ b/src/mcp_agent/cli/cloud/commands/deploy/wrangler_wrapper.py
@@ -5,14 +5,18 @@
 import tempfile
 import textwrap
 from pathlib import Path
+import json
 
 from rich.progress import Progress, SpinnerColumn, TextColumn
 
 from mcp_agent.cli.config import settings
-from mcp_agent.cli.core.constants import (
-    MCP_SECRETS_FILENAME,
+from mcp_agent.cli.core.constants import MCP_SECRETS_FILENAME
+from mcp_agent.cli.utils.ux import console, print_error, print_warning, print_info
+from mcp_agent.cli.utils.git_utils import (
+    get_git_metadata,
+    compute_directory_fingerprint,
+    utc_iso_now,
 )
-from mcp_agent.cli.utils.ux import console, print_error, print_warning
 
 from .constants import (
     CLOUDFLARE_ACCOUNT_ID,
@@ -197,13 +201,96 @@ def ignore_patterns(_path, names):
                 # Rename in place
                 file_path.rename(py_path)
 
-        # Create temporary wrangler.toml
+        # Collect deployment metadata (git if available, else workspace hash)
+        git_meta = get_git_metadata(project_dir)
+        deploy_source = "git" if git_meta else "workspace"
+        meta_vars = {
+            "MCP_DEPLOY_SOURCE": deploy_source,
+            "MCP_DEPLOY_TIME_UTC": utc_iso_now(),
+        }
+        if git_meta:
+            meta_vars.update(
+                {
+                    "MCP_DEPLOY_GIT_COMMIT": git_meta.commit_sha,
+                    "MCP_DEPLOY_GIT_SHORT": git_meta.short_sha,
+                    "MCP_DEPLOY_GIT_BRANCH": git_meta.branch or "",
+                    "MCP_DEPLOY_GIT_DIRTY": "true" if git_meta.dirty else "false",
+                }
+            )
+            # Friendly console hint
+            dirty_mark = "*" if git_meta.dirty else ""
+            print_info(
+                f"Deploying from git commit {git_meta.short_sha}{dirty_mark} on branch {git_meta.branch or '?'}"
+            )
+        else:
+            # Compute a cheap fingerprint (metadata-based) of the prepared project
+            bundle_hash = compute_directory_fingerprint(
+                temp_project_dir,
+                ignore_names={
+                    ".git",
+                    "logs",
+                    "__pycache__",
+                    "node_modules",
+                    "venv",
+                    MCP_SECRETS_FILENAME,
+                },
+            )
+            meta_vars.update({"MCP_DEPLOY_WORKSPACE_HASH": bundle_hash})
+            print_info(f"Deploying from non-git workspace (hash {bundle_hash[:12]}…)")
+
+        # Write a breadcrumb file into the project so it ships with the bundle.
+        # Use a Python file for guaranteed inclusion without renaming.
+        breadcrumb = {
+            "version": 1,
+            "app_id": app_id,
+            "deploy_time_utc": meta_vars["MCP_DEPLOY_TIME_UTC"],
+            "source": meta_vars["MCP_DEPLOY_SOURCE"],
+        }
+        if git_meta:
+            breadcrumb.update(
+                {
+                    "git": {
+                        "commit": git_meta.commit_sha,
+                        "short": git_meta.short_sha,
+                        "branch": git_meta.branch,
+                        "dirty": git_meta.dirty,
+                        "tag": git_meta.tag,
+                        "message": git_meta.commit_message,
+                    }
+                }
+            )
+        else:
+            breadcrumb.update(
+                {"workspace_fingerprint": meta_vars["MCP_DEPLOY_WORKSPACE_HASH"]}
+            )
+
+        breadcrumb_py = textwrap.dedent(
+            """
+            # Auto-generated by mcp-agent deploy. Do not edit.
+            # Contains deployment metadata for traceability.
+            import json as _json
+            BREADCRUMB = %s
+            BREADCRUMB_JSON = _json.dumps(BREADCRUMB, separators=(",", ":"))
+            __all__ = ["BREADCRUMB", "BREADCRUMB_JSON"]
+            """
+        ).strip() % (json.dumps(breadcrumb, indent=2))
+
+        (temp_project_dir / "mcp_deploy_breadcrumb.py").write_text(breadcrumb_py)
+
+        # Create temporary wrangler.toml with [vars] carrying deploy metadata
+        # Use TOML strings and keep values simple/escaped; also include a compact JSON blob
+        meta_json = json.dumps(meta_vars, separators=(",", ":"))
+        vars_lines = ["[vars]"] + [f'{k} = "{v}"' for k, v in meta_vars.items()]
+        vars_lines.append(f'MCP_DEPLOY_META = """{meta_json}"""')
+
         wrangler_toml_content = textwrap.dedent(
             f"""
             name = "{app_id}"
             main = "{main_py}"
             compatibility_flags = ["python_workers"]
             compatibility_date = "2025-06-26"
+
+            {os.linesep.join(vars_lines)}
         """
         ).strip()
 
diff --git a/src/mcp_agent/cli/cloud/commands/servers/delete/main.py b/src/mcp_agent/cli/cloud/commands/servers/delete/main.py
index 9afcaf6ab..c53de327d 100644
--- a/src/mcp_agent/cli/cloud/commands/servers/delete/main.py
+++ b/src/mcp_agent/cli/cloud/commands/servers/delete/main.py
@@ -17,7 +17,7 @@
 @handle_server_api_errors
 def delete_server(
     id_or_url: str = typer.Argument(
-        ..., help="Server ID or app configuration ID to delete"
+        ..., help="App ID, server URL, or app name to delete"
     ),
     force: bool = typer.Option(
         False, "--force", "-f", help="Force deletion without confirmation prompt"
diff --git a/src/mcp_agent/cli/cloud/commands/servers/describe/main.py b/src/mcp_agent/cli/cloud/commands/servers/describe/main.py
index 50ca0051c..cd0a0f4f3 100644
--- a/src/mcp_agent/cli/cloud/commands/servers/describe/main.py
+++ b/src/mcp_agent/cli/cloud/commands/servers/describe/main.py
@@ -20,7 +20,7 @@
 @handle_server_api_errors
 def describe_server(
     id_or_url: str = typer.Argument(
-        ..., help="Server ID or app configuration ID to describe"
+        ..., help="App ID, server URL, or app name to describe"
     ),
     format: Optional[str] = typer.Option(
         "text", "--format", help="Output format (text|json|yaml)"
diff --git a/src/mcp_agent/cli/cloud/commands/utils.py b/src/mcp_agent/cli/cloud/commands/utils.py
index 0a2c28fb9..c05a24fb7 100644
--- a/src/mcp_agent/cli/cloud/commands/utils.py
+++ b/src/mcp_agent/cli/cloud/commands/utils.py
@@ -29,7 +29,10 @@ def setup_authenticated_client() -> MCPAppClient:
     effective_api_key = settings.API_KEY or load_api_key_credentials()
 
     if not effective_api_key:
-        raise CLIError("Must be authenticated. Set MCP_API_KEY or run 'mcp-agent login'.", retriable=False)
+        raise CLIError(
+            "Must be authenticated. Set MCP_API_KEY or run 'mcp-agent login'.",
+            retriable=False,
+        )
 
     return MCPAppClient(api_url=DEFAULT_API_BASE_URL, api_key=effective_api_key)
 
@@ -47,18 +50,23 @@ def validate_output_format(format: str) -> None:
     if format not in valid_formats:
         raise CLIError(
             f"Invalid format '{format}'. Valid options are: {', '.join(valid_formats)}",
-            retriable=False
+            retriable=False,
         )
 
 
 async def resolve_server_async(
-    client: MCPAppClient, id_or_url: str
+    client: MCPAppClient, id_or_url_or_name: str
 ) -> Union[MCPApp, MCPAppConfiguration]:
-    """Resolve server from ID or URL (async).
+    """Resolve server from ID, server URL, app configuration ID, or app name (async).
+
+    Resolution order:
+    1) Treat as ID or server URL via get_app_or_config
+    2) Treat as app name -> lookup app ID -> get_app
 
     Args:
         client: Authenticated MCP App client
-        id_or_url: Server identifier (app ID, app config ID, or server URL)
+        id_or_url_or_name: Identifier that may be an app ID, app config ID,
+            server URL, or app name
 
     Returns:
         Server object (MCPApp or MCPAppConfiguration)
@@ -66,21 +74,30 @@ async def resolve_server_async(
     Raises:
         CLIError: If server resolution fails
     """
+    # First try as ID or server URL
+    try:
+        return await client.get_app_or_config(id_or_url_or_name)
+    except Exception:
+        pass
+
+    # Fallback: try as app name -> map to app ID
     try:
-        return await client.get_app_or_config(id_or_url)
-    except Exception as e:
-        raise CLIError(f"Failed to resolve server '{id_or_url}': {str(e)}") from e
+        app_id = await client.get_app_id_by_name(id_or_url_or_name)
+        if app_id:
+            return await client.get_app(app_id=app_id)
+    except Exception:
+        pass
+
+    raise CLIError(
+        f"Failed to resolve server '{id_or_url_or_name}' as an ID, server URL, or app name"
+    )
 
 
 def resolve_server(
-    client: MCPAppClient, id_or_url: str
+    client: MCPAppClient, id_or_url_or_name: str
 ) -> Union[MCPApp, MCPAppConfiguration]:
-    """Resolve server from ID or URL (sync wrapper).
-
-    Safe for synchronous CLI contexts. For async code paths, prefer
-    using resolve_server_async to avoid nested event loops.
-    """
-    return run_async(resolve_server_async(client, id_or_url))
+    """Resolve server from ID, server URL, app config ID, or app name (sync wrapper)."""
+    return run_async(resolve_server_async(client, id_or_url_or_name))
 
 
 def handle_server_api_errors(func):
@@ -100,7 +117,7 @@ def wrapper(*args, **kwargs):
         except UnauthenticatedError as e:
             raise CLIError(
                 "Invalid API key. Run 'mcp-agent login' or set MCP_API_KEY environment variable with new API key.",
-                retriable=False
+                retriable=False,
             ) from e
         except CLIError:
             # Re-raise CLIErrors as-is
diff --git a/src/mcp_agent/cli/cloud/commands/workflows/cancel/main.py b/src/mcp_agent/cli/cloud/commands/workflows/cancel/main.py
index 5dae3172b..858001e31 100644
--- a/src/mcp_agent/cli/cloud/commands/workflows/cancel/main.py
+++ b/src/mcp_agent/cli/cloud/commands/workflows/cancel/main.py
@@ -17,31 +17,36 @@
 
 
 async def _cancel_workflow_async(
-    server_id_or_url: str, run_id: str, reason: Optional[str] = None
+    server_id_or_url_or_name: str, run_id: str, reason: Optional[str] = None
 ) -> None:
     """Cancel a workflow using MCP tool calls to a deployed server."""
-    if server_id_or_url.startswith(("http://", "https://")):
-        server_url = server_id_or_url
+    if server_id_or_url_or_name.startswith(("http://", "https://")):
+        server_url = server_id_or_url_or_name
     else:
         client = setup_authenticated_client()
-        server = await resolve_server_async(client, server_id_or_url)
+        server = await resolve_server_async(client, server_id_or_url_or_name)
 
         if hasattr(server, "appServerInfo") and server.appServerInfo:
             server_url = server.appServerInfo.serverUrl
         else:
             raise CLIError(
-                f"Server '{server_id_or_url}' is not deployed or has no server URL"
+                f"Server '{server_id_or_url_or_name}' is not deployed or has no server URL"
             )
 
         if not server_url:
-            raise CLIError(f"No server URL found for server '{server_id_or_url}'")
+            raise CLIError(
+                f"No server URL found for server '{server_id_or_url_or_name}'"
+            )
 
     from mcp_agent.cli.config import settings as _settings
 
     effective_api_key = _settings.API_KEY or load_api_key_credentials()
 
     if not effective_api_key:
-        raise CLIError("Must be logged in to access server. Run 'mcp-agent login'.", retriable=False)
+        raise CLIError(
+            "Must be logged in to access server. Run 'mcp-agent login'.",
+            retriable=False,
+        )
 
     try:
         async with mcp_connection_session(
@@ -72,8 +77,8 @@ async def _cancel_workflow_async(
 
 @handle_server_api_errors
 def cancel_workflow(
-    server_id_or_url: str = typer.Argument(
-        ..., help="Server ID or URL hosting the workflow"
+    server_id_or_url_or_name: str = typer.Argument(
+        ..., help="App ID, server URL, or app name hosting the workflow"
     ),
     run_id: str = typer.Argument(..., help="Run ID of the workflow to cancel"),
     reason: Optional[str] = typer.Option(
@@ -91,4 +96,4 @@ def cancel_workflow(
 
         mcp-agent cloud workflows cancel app_abc123 run_xyz789 --reason "User requested"
     """
-    run_async(_cancel_workflow_async(server_id_or_url, run_id, reason))
+    run_async(_cancel_workflow_async(server_id_or_url_or_name, run_id, reason))
diff --git a/src/mcp_agent/cli/cloud/commands/workflows/describe/main.py b/src/mcp_agent/cli/cloud/commands/workflows/describe/main.py
index e4b31b4bb..73aeb99be 100644
--- a/src/mcp_agent/cli/cloud/commands/workflows/describe/main.py
+++ b/src/mcp_agent/cli/cloud/commands/workflows/describe/main.py
@@ -22,31 +22,36 @@
 
 
 async def _describe_workflow_async(
-    server_id_or_url: str, run_id: str, format: str = "text"
+    server_id_or_url_or_name: str, run_id: str, format: str = "text"
 ) -> None:
     """Describe a workflow using MCP tool calls to a deployed server."""
-    if server_id_or_url.startswith(("http://", "https://")):
-        server_url = server_id_or_url
+    if server_id_or_url_or_name.startswith(("http://", "https://")):
+        server_url = server_id_or_url_or_name
     else:
         client = setup_authenticated_client()
-        server = await resolve_server_async(client, server_id_or_url)
+        server = await resolve_server_async(client, server_id_or_url_or_name)
 
         if hasattr(server, "appServerInfo") and server.appServerInfo:
             server_url = server.appServerInfo.serverUrl
         else:
             raise CLIError(
-                f"Server '{server_id_or_url}' is not deployed or has no server URL"
+                f"Server '{server_id_or_url_or_name}' is not deployed or has no server URL"
             )
 
         if not server_url:
-            raise CLIError(f"No server URL found for server '{server_id_or_url}'")
+            raise CLIError(
+                f"No server URL found for server '{server_id_or_url_or_name}'"
+            )
 
     from mcp_agent.cli.config import settings as _settings
 
     effective_api_key = _settings.API_KEY or load_api_key_credentials()
 
     if not effective_api_key:
-        raise CLIError("Must be logged in to access server. Run 'mcp-agent login'.", retriable=False)
+        raise CLIError(
+            "Must be logged in to access server. Run 'mcp-agent login'.",
+            retriable=False,
+        )
 
     try:
         async with mcp_connection_session(
@@ -70,8 +75,8 @@ async def _describe_workflow_async(
 
 @handle_server_api_errors
 def describe_workflow(
-    server_id_or_url: str = typer.Argument(
-        ..., help="Server ID or URL hosting the workflow"
+    server_id_or_url_or_name: str = typer.Argument(
+        ..., help="App ID, server URL, or app name hosting the workflow"
     ),
     run_id: str = typer.Argument(..., help="Run ID of the workflow to describe"),
     format: Optional[str] = typer.Option(
@@ -93,7 +98,7 @@ def describe_workflow(
         console.print("[red]Error: --format must be 'text', 'json', or 'yaml'[/red]")
         raise typer.Exit(6)
 
-    run_async(_describe_workflow_async(server_id_or_url, run_id, format))
+    run_async(_describe_workflow_async(server_id_or_url_or_name, run_id, format))
 
 
 def print_workflow_status(workflow_status: WorkflowRun, format: str = "text") -> None:
diff --git a/src/mcp_agent/cli/cloud/commands/workflows/list/main.py b/src/mcp_agent/cli/cloud/commands/workflows/list/main.py
index 862c20a60..e3dbc2eda 100644
--- a/src/mcp_agent/cli/cloud/commands/workflows/list/main.py
+++ b/src/mcp_agent/cli/cloud/commands/workflows/list/main.py
@@ -20,30 +20,37 @@
 )
 
 
-async def _list_workflows_async(server_id_or_url: str, format: str = "text") -> None:
+async def _list_workflows_async(
+    server_id_or_url_or_name: str, format: str = "text"
+) -> None:
     """List available workflows using MCP tool calls to a deployed server."""
-    if server_id_or_url.startswith(("http://", "https://")):
-        server_url = server_id_or_url
+    if server_id_or_url_or_name.startswith(("http://", "https://")):
+        server_url = server_id_or_url_or_name
     else:
         client = setup_authenticated_client()
-        server = await resolve_server_async(client, server_id_or_url)
+        server = await resolve_server_async(client, server_id_or_url_or_name)
 
         if hasattr(server, "appServerInfo") and server.appServerInfo:
             server_url = server.appServerInfo.serverUrl
         else:
             raise CLIError(
-                f"Server '{server_id_or_url}' is not deployed or has no server URL"
+                f"Server '{server_id_or_url_or_name}' is not deployed or has no server URL"
             )
 
         if not server_url:
-            raise CLIError(f"No server URL found for server '{server_id_or_url}'")
+            raise CLIError(
+                f"No server URL found for server '{server_id_or_url_or_name}'"
+            )
 
     from mcp_agent.cli.config import settings as _settings
 
     effective_api_key = _settings.API_KEY or load_api_key_credentials()
 
     if not effective_api_key:
-        raise CLIError("Must be logged in to access server. Run 'mcp-agent login'.", retriable=False)
+        raise CLIError(
+            "Must be logged in to access server. Run 'mcp-agent login'.",
+            retriable=False,
+        )
 
     try:
         async with mcp_connection_session(
@@ -73,19 +80,19 @@ async def _list_workflows_async(server_id_or_url: str, format: str = "text") ->
                     print_workflows(workflows)
             except Exception as e:
                 print_error(
-                    f"Error listing workflows for server {server_id_or_url}: {str(e)}"
+                    f"Error listing workflows for server {server_id_or_url_or_name}: {str(e)}"
                 )
 
     except Exception as e:
         raise CLIError(
-            f"Error listing workflows for server {server_id_or_url}: {str(e)}"
+            f"Error listing workflows for server {server_id_or_url_or_name}: {str(e)}"
         ) from e
 
 
 @handle_server_api_errors
 def list_workflows(
-    server_id_or_url: str = typer.Argument(
-        ..., help="Server ID or URL to list workflows for"
+    server_id_or_url_or_name: str = typer.Argument(
+        ..., help="App ID, server URL, or app name to list workflows for"
     ),
     format: Optional[str] = typer.Option(
         "text", "--format", help="Output format (text|json|yaml)"
@@ -103,4 +110,4 @@ def list_workflows(
         mcp-agent cloud workflows list https://server.example.com --format json
     """
     validate_output_format(format)
-    run_async(_list_workflows_async(server_id_or_url, format))
+    run_async(_list_workflows_async(server_id_or_url_or_name, format))
diff --git a/src/mcp_agent/cli/cloud/commands/workflows/resume/main.py b/src/mcp_agent/cli/cloud/commands/workflows/resume/main.py
index 5a828ccbd..0a3e414f7 100644
--- a/src/mcp_agent/cli/cloud/commands/workflows/resume/main.py
+++ b/src/mcp_agent/cli/cloud/commands/workflows/resume/main.py
@@ -18,34 +18,39 @@
 
 
 async def _signal_workflow_async(
-    server_id_or_url: str,
+    server_id_or_url_or_name: str,
     run_id: str,
     signal_name: str = "resume",
     payload: Optional[Dict[str, Any]] = None,
 ) -> None:
     """Send a signal to a workflow using MCP tool calls to a deployed server."""
-    if server_id_or_url.startswith(("http://", "https://")):
-        server_url = server_id_or_url
+    if server_id_or_url_or_name.startswith(("http://", "https://")):
+        server_url = server_id_or_url_or_name
     else:
         client = setup_authenticated_client()
-        server = await resolve_server_async(client, server_id_or_url)
+        server = await resolve_server_async(client, server_id_or_url_or_name)
 
         if hasattr(server, "appServerInfo") and server.appServerInfo:
             server_url = server.appServerInfo.serverUrl
         else:
             raise CLIError(
-                f"Server '{server_id_or_url}' is not deployed or has no server URL"
+                f"Server '{server_id_or_url_or_name}' is not deployed or has no server URL"
             )
 
         if not server_url:
-            raise CLIError(f"No server URL found for server '{server_id_or_url}'")
+            raise CLIError(
+                f"No server URL found for server '{server_id_or_url_or_name}'"
+            )
 
     from mcp_agent.cli.config import settings as _settings
 
     effective_api_key = _settings.API_KEY or load_api_key_credentials()
 
     if not effective_api_key:
-        raise CLIError("Must be logged in to access server. Run 'mcp-agent login'.", retriable=False)
+        raise CLIError(
+            "Must be logged in to access server. Run 'mcp-agent login'.",
+            retriable=False,
+        )
 
     try:
         async with mcp_connection_session(
@@ -112,8 +117,8 @@ async def _signal_workflow_async(
 
 @handle_server_api_errors
 def resume_workflow(
-    server_id_or_url: str = typer.Argument(
-        ..., help="Server ID or URL hosting the workflow"
+    server_id_or_url_or_name: str = typer.Argument(
+        ..., help="App ID, server URL, or app name hosting the workflow"
     ),
     run_id: str = typer.Argument(..., help="Run ID of the workflow to resume"),
     signal_name: Optional[str] = "resume",
@@ -144,15 +149,15 @@ def resume_workflow(
 
     run_async(
         _signal_workflow_async(
-            server_id_or_url, run_id, signal_name or "resume", payload
+            server_id_or_url_or_name, run_id, signal_name or "resume", payload
         )
     )
 
 
 @handle_server_api_errors
 def suspend_workflow(
-    server_id_or_url: str = typer.Argument(
-        ..., help="Server ID or URL hosting the workflow"
+    server_id_or_url_or_name: str = typer.Argument(
+        ..., help="App ID, server URL, or app name hosting the workflow"
     ),
     run_id: str = typer.Argument(..., help="Run ID of the workflow to suspend"),
     payload: Optional[str] = typer.Option(
@@ -174,4 +179,6 @@ def suspend_workflow(
         except json.JSONDecodeError as e:
             raise typer.BadParameter(f"Invalid JSON payload: {str(e)}") from e
 
-    run_async(_signal_workflow_async(server_id_or_url, run_id, "suspend", payload))
+    run_async(
+        _signal_workflow_async(server_id_or_url_or_name, run_id, "suspend", payload)
+    )
diff --git a/src/mcp_agent/cli/cloud/commands/workflows/runs/main.py b/src/mcp_agent/cli/cloud/commands/workflows/runs/main.py
index df89e3a3f..3dc6e3c1d 100644
--- a/src/mcp_agent/cli/cloud/commands/workflows/runs/main.py
+++ b/src/mcp_agent/cli/cloud/commands/workflows/runs/main.py
@@ -47,7 +47,10 @@ async def _list_workflow_runs_async(
     effective_api_key = _settings.API_KEY or load_api_key_credentials()
 
     if not effective_api_key:
-        raise CLIError("Must be logged in to access server. Run 'mcp-agent login'.", retriable=False)
+        raise CLIError(
+            "Must be logged in to access server. Run 'mcp-agent login'.",
+            retriable=False,
+        )
 
     try:
         async with mcp_connection_session(
@@ -88,7 +91,7 @@ async def _list_workflow_runs_async(
 
 def list_workflow_runs(
     server_id_or_url: str = typer.Argument(
-        ..., help="Server ID, app config ID, or server URL to list workflow runs for"
+        ..., help="App ID, server URL, or app name to list workflow runs for"
     ),
     limit: Optional[int] = typer.Option(
         None, "--limit", help="Maximum number of results to return"
diff --git a/src/mcp_agent/cli/cloud/main.py b/src/mcp_agent/cli/cloud/main.py
index 6ac0ae6cb..cd532d4d8 100644
--- a/src/mcp_agent/cli/cloud/main.py
+++ b/src/mcp_agent/cli/cloud/main.py
@@ -37,6 +37,7 @@
 )
 from mcp_agent.cli.utils.typer_utils import HelpfulTyperGroup
 from mcp_agent.cli.utils.ux import print_error
+from mcp_agent.cli.utils.version_check import maybe_warn_newer_version
 
 # Setup file logging
 LOG_DIR = Path.home() / ".mcp-agent" / "logs"
@@ -174,6 +175,11 @@ def callback(
     ),
 ) -> None:
     """MCP Agent Cloud CLI."""
+    # Best-effort version check (5s timeout, non-fatal). Guard to run once.
+    try:
+        maybe_warn_newer_version()
+    except Exception:
+        pass
     if version:
         v = metadata_version("mcp-agent")
         typer.echo(f"MCP Agent Cloud CLI version: {v}")
@@ -183,6 +189,11 @@ def callback(
 def run() -> None:
     """Run the CLI application."""
     try:
+        # Run best-effort version check before Typer may early-exit on --help
+        try:
+            maybe_warn_newer_version()
+        except Exception:
+            pass
         app()
     except Exception as e:
         # Unexpected errors - log full exception and show clean error to user
diff --git a/src/mcp_agent/cli/main.py b/src/mcp_agent/cli/main.py
index 599836d0c..0db8e80e5 100644
--- a/src/mcp_agent/cli/main.py
+++ b/src/mcp_agent/cli/main.py
@@ -16,6 +16,7 @@
 from rich.console import Console
 
 from mcp_agent.cli.utils.ux import print_error
+from mcp_agent.cli.utils.version_check import maybe_warn_newer_version
 
 # Mount existing cloud CLI
 try:
@@ -142,6 +143,12 @@ def main(
         console.print("mcp-agent - Model Context Protocol agent CLI\n")
         console.print("Run 'mcp-agent --help' to see all commands.")
 
+    # Best-effort version check (5s timeout, non-fatal)
+    try:
+        maybe_warn_newer_version()
+    except Exception:
+        pass
+
 
 # Mount non-cloud command groups (top-level, curated)
 app.add_typer(init_cmd.app, name="init", help="Scaffold a new mcp-agent project")
@@ -189,6 +196,11 @@ def main(
 def run() -> None:
     """Run the CLI application."""
     try:
+        # Run best-effort version check before Typer may early-exit on --help
+        try:
+            maybe_warn_newer_version()
+        except Exception:
+            pass
         app()
     except Exception as e:
         # Unexpected errors - log full exception and show clean error to user
diff --git a/src/mcp_agent/cli/mcp_app/api_client.py b/src/mcp_agent/cli/mcp_app/api_client.py
index 9631c2d3d..0609fc552 100644
--- a/src/mcp_agent/cli/mcp_app/api_client.py
+++ b/src/mcp_agent/cli/mcp_app/api_client.py
@@ -309,6 +309,7 @@ async def get_app_id_by_name(self, name: str) -> Optional[str]:
     async def deploy_app(
         self,
         app_id: str,
+        deployment_metadata: Optional[Dict[str, Any]] = None,
     ) -> MCPApp:
         """Deploy an MCP App via the API.
 
@@ -326,9 +327,10 @@ async def deploy_app(
         if not app_id or not is_valid_app_id_format(app_id):
             raise ValueError(f"Invalid app ID format: {app_id}")
 
-        payload = {
-            "appId": app_id,
-        }
+        payload: Dict[str, Any] = {"appId": app_id}
+        if deployment_metadata:
+            # Tentative field; include only when requested
+            payload["deploymentMetadata"] = deployment_metadata
 
         # Use a longer timeout for deployments
         deploy_timeout = 300.0
diff --git a/src/mcp_agent/cli/secrets/processor.py b/src/mcp_agent/cli/secrets/processor.py
index c94fb4221..21b36d4d0 100644
--- a/src/mcp_agent/cli/secrets/processor.py
+++ b/src/mcp_agent/cli/secrets/processor.py
@@ -94,7 +94,7 @@ async def process_config_secrets(
         if not effective_api_key:
             raise CLIError(
                 "Must have API key to process secrets. Login via 'mcp-agent login'.",
-                retriable=False
+                retriable=False,
             )
 
         # Create a new client
@@ -201,7 +201,9 @@ async def process_secrets_in_config_str(
             existing_config = load_yaml_with_secrets(existing_secrets_content)
             print_info("Loaded existing secrets configuration for reuse")
         except Exception as e:
-            raise CLIError(f"Failed to parse existing secrets YAML: {str(e)}", retriable=False) from e
+            raise CLIError(
+                f"Failed to parse existing secrets YAML: {str(e)}", retriable=False
+            ) from e
 
     # Make sure the existing config secrets are actually valid for the user
     if existing_config:
diff --git a/src/mcp_agent/cli/utils/git_utils.py b/src/mcp_agent/cli/utils/git_utils.py
new file mode 100644
index 000000000..c6be0c152
--- /dev/null
+++ b/src/mcp_agent/cli/utils/git_utils.py
@@ -0,0 +1,219 @@
+"""Lightweight git helpers for deployment metadata and tagging.
+
+These helpers avoid third-party dependencies and use subprocess to query git.
+All functions are safe to call outside a git repo (they return None/fallbacks).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import re
+import os
+import subprocess
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+
+
+@dataclass
+class GitMetadata:
+    """Key git details about the working copy to embed with deployments."""
+
+    commit_sha: str
+    short_sha: str
+    branch: Optional[str]
+    dirty: bool
+    tag: Optional[str]
+    commit_message: Optional[str]
+
+
+def _run_git(args: list[str], cwd: Path) -> Optional[str]:
+    """Run a git command and return stdout, suppressing all stderr noise.
+
+    Returns None on any error or non-zero exit to avoid leaking git messages
+    like "fatal: no tag exactly matches" to the console.
+    """
+    try:
+        proc = subprocess.run(
+            ["git", *args],
+            cwd=str(cwd),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            check=False,
+        )
+        if proc.returncode != 0:
+            return None
+        return proc.stdout.decode("utf-8", errors="replace").strip()
+    except Exception:
+        return None
+
+
+def get_git_metadata(project_dir: Path) -> Optional[GitMetadata]:
+    """Return GitMetadata for the repo containing project_dir, if any.
+
+    Returns None if git is unavailable or project_dir is not inside a repo.
+    """
+    try:
+        # Fast probe: are we inside a work-tree?
+        inside = _run_git(["rev-parse", "--is-inside-work-tree"], project_dir)
+        if inside is None or inside != "true":
+            return None
+
+        commit_sha = _run_git(["rev-parse", "HEAD"], project_dir)
+        if not commit_sha:
+            return None
+
+        short_sha = (
+            _run_git(["rev-parse", "--short", "HEAD"], project_dir) or commit_sha[:7]
+        )
+        branch = _run_git(["rev-parse", "--abbrev-ref", "HEAD"], project_dir)
+        status = _run_git(["status", "--porcelain"], project_dir)
+        dirty = bool(status)
+        tag = _run_git(["describe", "--tags", "--exact-match"], project_dir)
+        commit_message = _run_git(["log", "-1", "--pretty=%s"], project_dir)
+
+        return GitMetadata(
+            commit_sha=commit_sha,
+            short_sha=short_sha,
+            branch=branch,
+            dirty=dirty,
+            tag=tag,
+            commit_message=commit_message,
+        )
+    except Exception:
+        return None
+
+
+def utc_iso_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def compute_directory_hash(root: Path, *, ignore_names: set[str] | None = None) -> str:
+    """Compute SHA256 over file names and contents under root.
+
+    NOTE: This reads file contents and can be expensive for very large trees.
+    Prefer `compute_directory_fingerprint` below for fast fingerprints.
+    """
+    if ignore_names is None:
+        ignore_names = set()
+
+    h = hashlib.sha256()
+    for dirpath, dirnames, filenames in os.walk(root):
+        # Filter dirnames in-place to prune traversal
+        dirnames[:] = [
+            d for d in dirnames if d not in ignore_names and not d.startswith(".")
+        ]
+        for fname in sorted(filenames):
+            if fname in ignore_names or fname.startswith("."):
+                # Allow .env explicitly
+                if fname == ".env":
+                    pass
+                else:
+                    continue
+            fpath = Path(dirpath) / fname
+            if fpath.is_symlink():
+                continue
+            rel = fpath.relative_to(root).as_posix()
+            try:
+                with open(fpath, "rb") as f:
+                    data = f.read()
+            except Exception:
+                data = b""
+            h.update(rel.encode("utf-8"))
+            h.update(b"\0")
+            h.update(data)
+            h.update(b"\n")
+    return h.hexdigest()
+
+
+def compute_directory_fingerprint(
+    root: Path, *, ignore_names: set[str] | None = None
+) -> str:
+    """Compute a cheap, stable SHA256 over file metadata under root.
+
+    This avoids reading file contents. The hash includes the relative path,
+    file size and modification time for each included file. Hidden files/dirs
+    and any names in `ignore_names` are skipped, as are symlinks.
+    """
+    if ignore_names is None:
+        ignore_names = set()
+
+    h = hashlib.sha256()
+    for dirpath, dirnames, filenames in os.walk(root):
+        dirnames[:] = [
+            d for d in dirnames if d not in ignore_names and not d.startswith(".")
+        ]
+        for fname in sorted(filenames):
+            if fname in ignore_names or (fname.startswith(".") and fname != ".env"):
+                continue
+            fpath = Path(dirpath) / fname
+            if fpath.is_symlink():
+                continue
+            rel = fpath.relative_to(root).as_posix()
+            try:
+                st = fpath.stat()
+                size = st.st_size
+                mtime = int(st.st_mtime)
+            except Exception:
+                size = -1
+                mtime = 0
+            h.update(rel.encode("utf-8"))
+            h.update(b"\0")
+            h.update(str(size).encode("utf-8"))
+            h.update(b"\0")
+            h.update(str(mtime).encode("utf-8"))
+            h.update(b"\n")
+    return h.hexdigest()
+
+
+def create_git_tag(project_dir: Path, tag_name: str, message: str) -> bool:
+    """Create an annotated git tag at HEAD. Returns True on success.
+
+    Does nothing and returns False if not a repo or git fails.
+    """
+    inside = _run_git(["rev-parse", "--is-inside-work-tree"], project_dir)
+    if inside is None or inside != "true":
+        return False
+    try:
+        subprocess.check_call(
+            ["git", "tag", "-a", tag_name, "-m", message], cwd=str(project_dir)
+        )
+        return True
+    except Exception:
+        return False
+
+
+_INVALID_REF_CHARS = re.compile(r"[~^:?*\[\\\s]")
+
+
+def sanitize_git_ref_component(name: str) -> str:
+    """Sanitize a string to be safe as a single refname component.
+
+    Rules (aligned with `git check-ref-format` constraints and our usage):
+    - Disallow spaces and special characters: ~ ^ : ? * [ \ (replace with '-')
+    - Replace '/' to avoid creating nested namespaces from user input
+    - Collapse consecutive dots '..' into '-'
+    - Remove leading dots '.' (cannot start with '.')
+    - Remove trailing '.lock' and trailing dots
+    - Disallow '@{' sequence
+    - Ensure non-empty; fallback to 'unnamed'
+    """
+    s = name.strip()
+    # Replace disallowed characters and whitespace
+    s = _INVALID_REF_CHARS.sub("-", s)
+    # Replace slashes to avoid extra path segments
+    s = s.replace("/", "-")
+    # Collapse consecutive dots
+    s = re.sub(r"\.{2,}", "-", s)
+    # Remove '@{'
+    s = s.replace("@{", "-{")
+    # Remove leading dots and hyphens (avoid CLI option-like names)
+    s = re.sub(r"^[\.-]+", "", s)
+    # Remove trailing .lock
+    s = re.sub(r"\.lock$", "", s, flags=re.IGNORECASE)
+    # Remove trailing dots
+    s = re.sub(r"\.+$", "", s)
+    if not s:
+        s = "unnamed"
+    return s
diff --git a/src/mcp_agent/cli/utils/retry.py b/src/mcp_agent/cli/utils/retry.py
index af551b13c..498c9c9cd 100644
--- a/src/mcp_agent/cli/utils/retry.py
+++ b/src/mcp_agent/cli/utils/retry.py
@@ -15,7 +15,9 @@ class RetryError(Exception):
     def __init__(self, original_error: Exception, attempts: int):
         self.original_error = original_error
         self.attempts = attempts
-        super().__init__(f"Failed after {attempts} attempts. Last error: {original_error}")
+        super().__init__(
+            f"Failed after {attempts} attempts. Last error: {original_error}"
+        )
 
 
 def is_retryable_error(error: Exception) -> bool:
@@ -44,7 +46,7 @@ def retry_with_exponential_backoff(
     max_delay: float = 60.0,
     retryable_check: Optional[Callable[[Exception], bool]] = None,
     *args,
-    **kwargs
+    **kwargs,
 ) -> Any:
     """Retry a function with exponential backoff.
 
@@ -80,7 +82,9 @@ def retry_with_exponential_backoff(
             if attempt == max_attempts or not retryable_check(e):
                 break
 
-            print_warning(f"Attempt {attempt}/{max_attempts} failed: {e}. Retrying in {delay:.1f}s...")
+            print_warning(
+                f"Attempt {attempt}/{max_attempts} failed: {e}. Retrying in {delay:.1f}s..."
+            )
 
             time.sleep(delay)
             delay = min(delay * backoff_multiplier, max_delay)
@@ -102,7 +106,7 @@ async def retry_async_with_exponential_backoff(
     max_delay: float = 60.0,
     retryable_check: Optional[Callable[[Exception], bool]] = None,
     *args,
-    **kwargs
+    **kwargs,
 ) -> Any:
     """Async version of retry with exponential backoff.
 
@@ -141,7 +145,9 @@ async def retry_async_with_exponential_backoff(
             if attempt == max_attempts or not retryable_check(e):
                 break
 
-            print_warning(f"Attempt {attempt}/{max_attempts} failed: {e}. Retrying in {delay:.1f}s...")
+            print_warning(
+                f"Attempt {attempt}/{max_attempts} failed: {e}. Retrying in {delay:.1f}s..."
+            )
 
             await asyncio.sleep(delay)
             delay = min(delay * backoff_multiplier, max_delay)
@@ -152,4 +158,4 @@ async def retry_async_with_exponential_backoff(
         else:
             raise last_exception
 
-    raise RuntimeError("Unexpected error in retry logic")
\ No newline at end of file
+    raise RuntimeError("Unexpected error in retry logic")
diff --git a/src/mcp_agent/cli/utils/version_check.py b/src/mcp_agent/cli/utils/version_check.py
new file mode 100644
index 000000000..496996ba6
--- /dev/null
+++ b/src/mcp_agent/cli/utils/version_check.py
@@ -0,0 +1,109 @@
+"""Best-effort PyPI version check for mcp-agent.
+
+- Contacts PyPI JSON API for latest version
+- Compares with installed version
+- Prints an info hint if an update is available
+- Times out after 5 seconds and never raises
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Optional
+
+from mcp_agent.cli.utils.ux import print_info
+
+
+def _get_installed_version() -> Optional[str]:
+    try:
+        import importlib.metadata as _im  # py3.8+
+
+        return _im.version("mcp-agent")
+    except Exception:
+        return None
+
+
+def _parse_version(s: str):
+    # Prefer packaging if available
+    try:
+        from packaging.version import parse as _vparse  # type: ignore
+
+        return _vparse(s)
+    except Exception:
+        # Fallback: simple tuple of ints (non-PEP440 safe)
+        return _simple_version_tuple(s)
+
+
+def _simple_version_tuple(s: str):
+    parts = s.split(".")
+    out = []
+    for p in parts:
+        num = ""
+        for ch in p:
+            if ch.isdigit():
+                num += ch
+            else:
+                break
+        if num:
+            out.append(int(num))
+        else:
+            break
+    return tuple(out)
+
+
+def _is_outdated(current: str, latest: str) -> bool:
+    try:
+        return _parse_version(latest) > _parse_version(current)
+    except Exception:
+        # Best-effort: if comparison fails, only warn when strings differ
+        return latest != current
+
+
+def _fetch_latest_version(timeout_seconds: float = 5.0) -> Optional[str]:
+    try:
+        import httpx
+
+        url = "https://pypi.org/pypi/mcp-agent/json"
+        timeout = httpx.Timeout(timeout_seconds)
+        with httpx.Client(timeout=timeout) as client:
+            resp = client.get(url)
+            if resp.status_code == 200:
+                data = resp.json()
+                version = (data or {}).get("info", {}).get("version")
+                if isinstance(version, str) and version:
+                    return version
+    except Exception:
+        pass
+    return None
+
+
+def maybe_warn_newer_version() -> None:
+    """Best-effort, once-per-process, 5s-timeout version check.
+
+    Honors env var MCP_AGENT_DISABLE_VERSION_CHECK=true/1/yes to skip.
+    """
+    if os.environ.get("MCP_AGENT_DISABLE_VERSION_CHECK", "").lower() in {
+        "1",
+        "true",
+        "yes",
+    }:
+        return
+
+    # Ensure we run at most once per process
+    if os.environ.get("MCP_AGENT_VERSION_CHECKED"):
+        return
+    os.environ["MCP_AGENT_VERSION_CHECKED"] = "1"
+
+    current = _get_installed_version()
+    if not current:
+        return
+
+    latest = _fetch_latest_version(timeout_seconds=5.0)
+    if not latest:
+        return
+
+    if _is_outdated(current, latest):
+        print_info(
+            f"A new version of mcp-agent is available: {current} -> {latest}. Update with: 'uv tool update mcp-agent'",
+            console_output=True,
+        )