Add git tag/commit breadcrumbs to deploy flow

src/mcp_agent/cli/cloud/commands/app/status/main.py

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

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:

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,7 @@
     print_info,
     print_success,
 )
+from mcp_agent.cli.utils.git_utils import get_git_metadata, create_git_tag
 
 from .wrangler_wrapper import wrangler_deploy
 
@@ -85,6 +87,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 +137,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 +186,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 +257,38 @@ 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 tag safety
+                safe_name = "".join(
+                    c if c.isalnum() or c in "-_" else "-" for c in 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 +351,35 @@ 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,
+                    }
+                app = await mcp_app_client.deploy_app(
+                    app_id=app_id, deployment_metadata=metadata
+                )
+                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 +396,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 +415,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

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()
 

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,7 +50,7 @@ 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,
         )
 
 
@@ -100,7 +103,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

src/mcp_agent/cli/cloud/commands/workflows/cancel/main.py

@@ -41,7 +41,10 @@ async def _cancel_workflow_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(

src/mcp_agent/cli/cloud/commands/workflows/describe/main.py

@@ -46,7 +46,10 @@ async def _describe_workflow_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(

src/mcp_agent/cli/cloud/commands/workflows/list/main.py

@@ -43,7 +43,10 @@ async def _list_workflows_async(server_id_or_url: str, format: str = "text") ->
     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(

src/mcp_agent/cli/cloud/commands/workflows/resume/main.py

@@ -45,7 +45,10 @@ async def _signal_workflow_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(

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(