Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
53 changes: 52 additions & 1 deletion src/mcp_agent/cli/cloud/commands/deploy/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
"""

from pathlib import Path
from datetime import datetime, timezone
from typing import Optional, Union

import typer
Expand Down Expand Up @@ -36,6 +37,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

Expand Down Expand Up @@ -90,6 +92,18 @@ 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",
),
send_metadata: bool = typer.Option(
False,
"--send-metadata/--no-send-metadata",
help="Send deployment metadata (e.g., commit info) to the API if supported",
envvar="MCP_DEPLOY_SEND_METADATA",
),
) -> str:
"""Deploy an MCP agent using the specified configuration.

Expand Down Expand Up @@ -237,6 +251,29 @@ def deploy_config(
)
)

# 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
)
Copy link
Contributor

Choose a reason for hiding this comment

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

Insufficient input sanitization: The app name sanitization only allows alphanumeric characters, hyphens, and underscores, but replaces all other characters with hyphens. This could create collisions (e.g., 'app@name' and 'app#name' both become 'app-name') and doesn't prevent potential issues with git tag naming rules. Git tags have specific restrictions (can't start with '.', can't contain certain sequences like '..', etc.). The sanitization should follow git tag naming conventions more strictly.

Suggested change
# Sanitize app name for tag safety
safe_name = "".join(
c if c.isalnum() or c in "-_" else "-" for c in app_name
)
# Sanitize app name for git tag safety
# Git tags cannot:
# - start with a period (.)
# - contain spaces, ~, ^, :, ?, *, [, \, or control characters
# - contain the sequence '..'
# - end with a slash (/) or .lock
safe_name = app_name.strip()
# Replace problematic characters with hyphens
safe_name = re.sub(r'[~^:?*\[\\\s]', '-', safe_name)
# Handle consecutive dots
safe_name = re.sub(r'\.{2,}', '-', safe_name)
# Remove leading periods
safe_name = re.sub(r'^\.+', '', safe_name)
# Remove trailing .lock or slashes
safe_name = re.sub(r'(/|\.lock)$', '', safe_name)
# Ensure we don't have empty string after sanitization
if not safe_name:
safe_name = "unnamed-app"

Spotted by Diamond

Fix in Graphite


Is this helpful? React 👍 or 👎 to let us know.

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

wrangler_deploy(
app_id=app_id,
api_key=effective_api_key,
Expand All @@ -251,9 +288,23 @@ def deploy_config(

try:
assert isinstance(mcp_app_client, MCPAppClient)
# Optionally include minimal metadata (git only to avoid heavy scans)
metadata = None
if send_metadata:
gm = get_git_metadata(config_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 = run_async(
mcp_app_client.deploy_app(
app_id=app_id,
app_id=app_id, deployment_metadata=metadata
)
)
progress.update(task, description="✅ MCP App deployed successfully!")
Expand Down
93 changes: 91 additions & 2 deletions src/mcp_agent/cli/cloud/commands/deploy/wrangler_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +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.utils.ux import console, print_error, print_warning
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 .constants import (
CLOUDFLARE_ACCOUNT_ID,
Expand Down Expand Up @@ -195,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()

Expand Down
8 changes: 5 additions & 3 deletions src/mcp_agent/cli/mcp_app/api_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand All @@ -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
Copy link
Member

Choose a reason for hiding this comment

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

We'll need to wait for www side to land and get deployed before landing this

Copy link
Member

Choose a reason for hiding this comment

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

Noting that the server-side changes have been deployed so this is good to go


# Use a longer timeout for deployments
deploy_timeout = 300.0
Expand Down
167 changes: 167 additions & 0 deletions src/mcp_agent/cli/utils/git_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,167 @@
"""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 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]:
try:
out = subprocess.check_output(["git", *args], cwd=str(cwd))
return out.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.
"""
# 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,
)


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):
Comment on lines +102 to +107
Copy link

Choose a reason for hiding this comment

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

🛠️ Refactor suggestion

Make directory traversal deterministic to stabilize hashes across platforms

os.walk() doesn’t guarantee directory order; since you hash in traversal order, results can vary. Sort dirnames to ensure stable digests.

Apply this diff in both walkers:

-        dirnames[:] = [
-            d for d in dirnames if d not in ignore_names and not d.startswith(".")
-        ]
+        dirnames[:] = sorted(
+            d for d in dirnames if d not in ignore_names and not d.startswith(".")
+        )

Also applies to: 126-130

🤖 Prompt for AI Agents
In src/mcp_agent/cli/utils/git_utils.py around lines 85-90 and also for the
second walker at 126-130, the directory traversal order is non-deterministic
because os.walk does not guarantee dirnames ordering; after you filter dirnames
in-place, sort dirnames in-place (e.g., dirnames.sort()) so traversal and
hashing are deterministic across platforms, ensuring both walkers apply the same
filter-then-sort step.

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
Comment on lines +148 to +149
Copy link
Contributor

Choose a reason for hiding this comment

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

The current condition has a logic issue with .env files. If .env is included in ignore_names, it will be skipped despite the special case handling. To correctly prioritize the .env exception, consider restructuring the condition like this:

if fname == ".env":
    pass  # Always include .env files
elif fname in ignore_names or fname.startswith("."):
    continue

This ensures .env files are always processed regardless of whether they're in the ignore list.

Suggested change
if fname in ignore_names or (fname.startswith(".") and fname != ".env"):
continue
if fname == ".env":
pass # Always include .env files
elif fname in ignore_names or fname.startswith("."):
continue

Spotted by Diamond

Fix in Graphite


Is this helpful? React 👍 or 👎 to let us know.

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
Loading