feat: implement remote build support and archive creation for deployments

- Added a new `create_archive` function to generate a tarball of the project source, including handling of .dockerignore patterns.
- Enhanced the `deploy` function to support remote builds, allowing deployment without local Docker availability.
- Introduced a `--remote` option in the CLI to control remote build behavior.
- Updated the HostBackendClient with methods for requesting upload URLs and updating deployments with source tarballs.

libs/cli/langgraph_cli/archive.py

@@ -0,0 +1,133 @@
+"""Create a tarball of project source for remote builds."""
+
+import os
+import pathlib
+import tarfile
+import tempfile
+
+import click
+
+
+_WARN_SIZE = 50 * 1024 * 1024  # 50 MB
+_MAX_SIZE = 200 * 1024 * 1024  # 200 MB
+
+
+def _tar_filter(tarinfo: tarfile.TarInfo) -> tarfile.TarInfo | None:
+    """Strip symlinks, hardlinks, and traversal paths from archive."""
+    if tarinfo.issym() or tarinfo.islnk():
+        return None
+    if ".." in tarinfo.name.split("/"):
+        return None
+    return tarinfo
+
+
+def _read_ignore_patterns(context_dir: pathlib.Path) -> list[str]:
+    """Read .dockerignore patterns if present."""
+    dockerignore = context_dir / ".dockerignore"
+    if dockerignore.is_file():
+        patterns = []
+        for line in dockerignore.read_text().splitlines():
+            line = line.strip()
+            if line and not line.startswith("#"):
+                patterns.append(line)
+        return patterns
+    return []
+
+
+def _should_ignore(rel_path: str, patterns: list[str]) -> bool:
+    """Check if a relative path matches any dockerignore pattern."""
+    import fnmatch
+
+    rel_path = rel_path.replace(os.sep, "/")
+
+    for pattern in patterns:
+        negate = pattern.startswith("!")
+        if negate:
+            pattern = pattern[1:]
+
+        if fnmatch.fnmatch(rel_path, pattern) or fnmatch.fnmatch(
+            rel_path, f"**/{pattern}"
+        ):
+            if negate:
+                return False
+            return True
+
+        parts = rel_path.split("/")
+        for i in range(len(parts)):
+            partial = "/".join(parts[: i + 1])
+            if fnmatch.fnmatch(partial, pattern):
+                if negate:
+                    return False
+                return True
+
+    return False
+
+
+def create_archive(
+    config_path: pathlib.Path,
+) -> tuple[str, int]:
+    """Create a .tar.gz archive of the project source.
+
+    Returns (archive_path, file_size).
+    The archive root is config.parent (the directory containing langgraph.json).
+    """
+    context_dir = config_path.parent.resolve()
+    config_filename = config_path.name
+    ignore_patterns = _read_ignore_patterns(context_dir)
+
+    tmp_dir = tempfile.mkdtemp(prefix="langgraph-deploy-")
+    archive_path = os.path.join(tmp_dir, "source.tar.gz")
+
+    with tarfile.open(archive_path, "w:gz") as tar:
+        for root, dirs, files in os.walk(context_dir):
+            rel_root = os.path.relpath(root, context_dir)
+            if rel_root == ".":
+                rel_root = ""
+
+            dirs[:] = [
+                d
+                for d in dirs
+                if not _should_ignore(
+                    os.path.join(rel_root, d) if rel_root else d, ignore_patterns
+                )
+            ]
+
+            for f in files:
+                rel_path = os.path.join(rel_root, f) if rel_root else f
+                if _should_ignore(rel_path, ignore_patterns):
+                    continue
+                full_path = os.path.join(root, f)
+                arcname = rel_path.replace(os.sep, "/")
+                info = tar.gettarinfo(full_path, arcname=arcname)
+                filtered = _tar_filter(info)
+                if filtered is None:
+                    continue
+                with open(full_path, "rb") as fobj:
+                    tar.addfile(filtered, fobj)
+
+    file_size = os.path.getsize(archive_path)
+
+    # Validate config file is at archive root
+    with tarfile.open(archive_path, "r:gz") as tar:
+        names = tar.getnames()
+        if config_filename not in names:
+            os.unlink(archive_path)
+            raise click.ClickException(
+                f"Archive validation failed: {config_filename} not found at archive root"
+            )
+
+    if file_size > _MAX_SIZE:
+        os.unlink(archive_path)
+        raise click.ClickException(
+            f"Source archive is {file_size / 1_048_576:.1f} MB, which exceeds the 200 MB limit. "
+            "Check your .dockerignore for large files (model weights, data, node_modules, .venv)."
+        )
+
+    if file_size > _WARN_SIZE:
+        click.secho(
+            f"   Warning: source archive is {file_size / 1_048_576:.1f} MB. "
+            "Consider adding large files to .dockerignore.",
+            fg="yellow",
+        )
+
+    return archive_path, file_size

libs/cli/langgraph_cli/cli.py

@@ -1,10 +1,18 @@
 """CLI entrypoint for LangGraph API server."""
 
+import base64
+import copy
+import json as json_mod
 import os
 import pathlib
+import platform
+import re
 import shutil
 import sys
+import tempfile
+import time
 from collections.abc import Callable, Sequence
+from contextlib import contextmanager
 
 import click
 import click.exceptions
@@ -17,11 +25,147 @@ from langgraph_cli.config import Config
 from langgraph_cli.constants import DEFAULT_CONFIG, DEFAULT_PORT
 from langgraph_cli.docker import DockerCapabilities
 from langgraph_cli.exec import Runner, subp_exec
+from langgraph_cli.host_backend import HostBackendClient, HostBackendError
 from langgraph_cli.progress import Progress
 from langgraph_cli.templates import TEMPLATE_HELP_STRING, create_new
 from langgraph_cli.util import warn_non_wolfi_distro
 from langgraph_cli.version import __version__
 
+RESERVED_ENV_VARS = frozenset(
+    [
+        # LANGCHAIN_RESERVED_ENV_VARS from host-backend
+        "LANGCHAIN_TRACING_V2",
+        "LANGSMITH_TRACING_V2",
+        "LANGCHAIN_ENDPOINT",
+        "LANGCHAIN_PROJECT",
+        "LANGSMITH_PROJECT",
+        "LANGSMITH_LANGGRAPH_GIT_REPO",
+        "LANGGRAPH_GIT_REPO_PATH",
+        "LANGCHAIN_API_KEY",
+        "LANGSMITH_CONTROL_PLANE_API_KEY",
+        "POSTGRES_URI",
+        "POSTGRES_PASSWORD",
+        "DATABASE_URI",
+        "LANGSMITH_LANGGRAPH_GIT_REF",
+        "LANGSMITH_LANGGRAPH_GIT_REF_SHA",
+        "LANGGRAPH_AUTH_TYPE",
+        "LANGSMITH_AUTH_ENDPOINT",
+        "LANGSMITH_TENANT_ID",
+        "LANGSMITH_AUTH_VERIFY_TENANT_ID",
+        "LANGSMITH_HOST_PROJECT_ID",
+        "LANGSMITH_HOST_PROJECT_NAME",
+        "LANGSMITH_HOST_REVISION_ID",
+        "LOG_JSON",
+        "LOG_DICT_TRACEBACKS",
+        "REDIS_URI",
+        "LANGCHAIN_CALLBACKS_BACKGROUND",
+        "DD_TRACE_PSYCOPG_ENABLED",
+        "DD_TRACE_REDIS_ENABLED",
+        "LANGGRAPH_CLOUD_LICENSE_KEY",
+        # ALLOWED_SELF_HOSTED_ENV_VARS (rejected for non-self-hosted)
+        "LANGSMITH_API_KEY",
+        "LANGSMITH_ENDPOINT",
+        "POSTGRES_URI_CUSTOM",
+        "REDIS_URI_CUSTOM",
+        "PATH",
+        "PORT",
+        "MOUNT_PREFIX",
+        "LSD_ENV",
+        "LSD_DD_API_KEY",
+        "LSD_DD_ENDPOINT",
+        "LSD_DEPLOYMENT_TYPE",
+    ]
+)
+
+_API_KEY_ENV_NAMES = (
+    "LANGGRAPH_HOST_API_KEY",
+    "LANGSMITH_API_KEY",
+    "LANGCHAIN_API_KEY",
+)
+
+
+def _parse_dotenv_file(path: pathlib.Path) -> dict[str, str]:
+    """Parse a .env file into a dict, skipping comments and blank lines."""
+    result: dict[str, str] = {}
+    if not path.is_file():
+        return result
+    for line in path.read_text(encoding="utf-8").splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        if "=" not in line:
+            continue
+        key, _, value = line.partition("=")
+        key = key.strip()
+        value = value.strip()
+        if (
+            value
+            and len(value) >= 2
+            and value[0] == value[-1]
+            and value[0] in ("'", '"')
+        ):
+            value = value[1:-1]
+        if key:
+            result[key] = value
+    return result
+
+
+def _parse_env_from_config(
+    config_json: dict, config_path: pathlib.Path
+) -> dict[str, str]:
+    """Resolve env vars from langgraph.json 'env' field or a .env fallback."""
+    env_field = config_json.get("env")
+    if isinstance(env_field, dict):
+        return {str(k): str(v) for k, v in env_field.items()}
+    if isinstance(env_field, str):
+        env_path = (config_path.parent / env_field).resolve()
+        return _parse_dotenv_file(env_path)
+    fallback = pathlib.Path.cwd() / ".env"
+    return _parse_dotenv_file(fallback)
+
+
+def _secrets_from_env(
+    env_vars: dict[str, str],
+) -> list[dict[str, str]]:
+    """Convert env dict to secrets list, filtering reserved vars with warnings."""
+    secrets: list[dict[str, str]] = []
+    for name, value in env_vars.items():
+        if name in RESERVED_ENV_VARS:
+            click.secho(f"   Skipping reserved env var: {name}", fg="yellow")
+            continue
+        if not value:
+            continue
+        secrets.append({"name": name, "value": value})
+    return secrets
+
+
+_TERMINAL_STATUSES = frozenset(
+    [
+        "DEPLOYED",
+        "CREATE_FAILED",
+        "BUILD_FAILED",
+        "DEPLOY_FAILED",
+        "SKIPPED",
+    ]
+)
+
+
+@contextmanager
+def _docker_config_for_token(registry_host: str, token: str):
+    """Create a temporary Docker config with only the push token.
+
+    Yields the path to a temporary config directory that can be passed
+    to ``docker --config <path>`` so that system credential helpers
+    (e.g. gcloud) don't interfere with the push token.
+    """
+    auth_b64 = base64.b64encode(f"oauth2accesstoken:{token}".encode()).decode()
+    config_data = {"auths": {registry_host: {"auth": auth_b64}}}
+    with tempfile.TemporaryDirectory() as tmpdir:
+        with open(os.path.join(tmpdir, "config.json"), "w") as f:
+            json_mod.dump(config_data, f)
+        yield tmpdir
+
+
 OPT_DOCKER_COMPOSE = click.option(
     "--docker-compose",
     "-d",
@@ -304,37 +448,37 @@ def _build(
     passthrough: Sequence[str] = (),
     install_command: str | None = None,
     build_command: str | None = None,
+    docker_command: Sequence[str] | None = None,
+    extra_flags: Sequence[str] = (),
+    verbose: bool = True,
 ):
-    # pull latest images
     if pull:
         runner.run(
             subp_exec(
                 "docker",
                 "pull",
                 langgraph_cli.config.docker_tag(config_json, base_image, api_version),
-                verbose=True,
+                verbose=verbose,
             )
         )
     set("Building...")
-    # apply options
     args = [
         "-f",
         "-",  # stdin
         "-t",
         tag,
     ]
-    # determine build context: use current directory for JS projects, config parent for Python
     is_js_project = config_json.get("node_version") and not config_json.get(
         "python_version"
     )
-    # build/install commands only apply to JS projects for now
-    # without install/build command, JS projects will follow the old behavior
     if is_js_project and (build_command or install_command):
         build_context = str(pathlib.Path.cwd())
     else:
         build_context = str(config.parent)
 
-    # apply config
+    # Deep copy to avoid mutating the caller's config (config_to_docker
+    # rewrites graph paths to container-internal paths in place).
+    config_json = copy.deepcopy(config_json)
     stdin, additional_contexts = langgraph_cli.config.config_to_docker(
         config_path=config,
         config=config_json,
@@ -344,19 +488,19 @@ def _build(
         build_command=build_command,
         build_context=build_context,
     )
-    # add additional_contexts
     if additional_contexts:
         for k, v in additional_contexts.items():
             args.extend(["--build-context", f"{k}={v}"])
+    cmd = tuple(docker_command) if docker_command else ("docker", "build")
     runner.run(
         subp_exec(
-            "docker",
-            "build",
+            *cmd,
             *args,
+            *extra_flags,
             *passthrough,
             build_context,
             input=stdin,
-            verbose=True,
+            verbose=verbose,
         )
     )
 
@@ -429,6 +573,662 @@ def build(
         )
 
 
+@click.option(
+    "--api-key",
+    envvar="LANGGRAPH_HOST_API_KEY",
+    help="API key. If omitted, resolved from .env or prompted.",
+)
+@click.option(
+    "--name",
+    help=(
+        "Deployment name. Defaults to current directory name "
+        "if --deployment-id is not provided."
+    ),
+)
+@click.option(
+    "--deployment-id",
+    help=(
+        "ID of an existing deployment to update. If omitted, "
+        "--name is used to find or create the deployment."
+    ),
+)
+@click.option(
+    "--deployment-type",
+    type=click.Choice(["dev", "prod"]),
+    default="dev",
+    show_default=True,
+    help="Deployment type (used when creating a new deployment).",
+)
+@click.option(
+    "--no-wait",
+    is_flag=True,
+    default=False,
+    help="Skip waiting for deployment status.",
+)
+@OPT_VERBOSE
+@click.option(
+    "--host-url",
+    envvar="LANGGRAPH_HOST_URL",
+    default="https://api.host.langchain.com",
+    hidden=True,
+)
+@click.option("--image-name", hidden=True)
+@click.option("--image-tag", default="latest", hidden=True)
+@click.option(
+    "--config",
+    "-c",
+    default=DEFAULT_CONFIG,
+    hidden=True,
+    type=click.Path(
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
+        resolve_path=True,
+        path_type=pathlib.Path,
+    ),
+)
+@click.option("--pull/--no-pull", default=True, hidden=True)
+@click.option("--base-image", hidden=True)
+@click.option("--install-command", hidden=True)
+@click.option("--build-command", hidden=True)
+@click.option("--api-version", type=str, hidden=True)
+@click.option(
+    "--remote/--no-remote",
+    default=None,
+    help=(
+        "Force or disable remote build. Default: auto-detect "
+        "(use remote build when Docker is unavailable)."
+    ),
+)
+@click.argument("docker_build_args", nargs=-1, type=click.UNPROCESSED)
+@cli.command(
+    help=(
+        "Build and deploy a LangGraph image to LangSmith Deployments.\n\n"
+        "Run from the root of your LangGraph project (where langgraph.json "
+        "is located). This command also accepts build flags (--base-image, "
+        "--pull, etc.). See 'langgraph build --help' for details."
+    ),
+    context_settings=dict(ignore_unknown_options=True),
+)
+@log_command
+def deploy(
+    config: pathlib.Path,
+    pull: bool,
+    verbose: bool,
+    api_version: str | None,
+    host_url: str | None,
+    api_key: str | None,
+    deployment_id: str | None,
+    deployment_type: str,
+    name: str | None,
+    image_name: str | None,
+    image_tag: str,
+    base_image: str | None,
+    install_command: str | None,
+    build_command: str | None,
+    no_wait: bool,
+    remote: bool | None,
+    docker_build_args: Sequence[str],
+):
+    config_json = langgraph_cli.config.validate_config_file(config)
+    warn_non_wolfi_distro(config_json)
+
+    env_vars = _parse_env_from_config(config_json, config)
+
+    if not api_key:
+        for key_name in _API_KEY_ENV_NAMES:
+            if key_name in env_vars:
+                api_key = env_vars[key_name]
+                break
+    if not api_key:
+        api_key = click.prompt("Host API key", hide_input=True)
+
+    if not deployment_id and not name:
+        default_name = _normalize_image_name(pathlib.Path.cwd().name)
+        name = click.prompt("Deployment name", default=default_name)
+
+    secrets = _secrets_from_env(env_vars)
+
+    # Determine whether to use remote build
+    use_remote = remote
+    if use_remote is None:
+        docker_available = langgraph_cli.docker.is_docker_available()
+        use_remote = not docker_available
+    elif use_remote is False:
+        pass  # --no-remote: fail if Docker is missing (existing behavior)
+
+    if use_remote:
+        _deploy_remote(
+            config=config,
+            config_json=config_json,
+            verbose=verbose,
+            api_version=api_version,
+            host_url=host_url,
+            api_key=api_key,
+            deployment_id=deployment_id,
+            deployment_type=deployment_type,
+            name=name,
+            base_image=base_image,
+            install_command=install_command,
+            build_command=build_command,
+            no_wait=no_wait,
+            secrets=secrets,
+        )
+    else:
+        _deploy_local(
+            config=config,
+            config_json=config_json,
+            verbose=verbose,
+            api_version=api_version,
+            host_url=host_url,
+            api_key=api_key,
+            deployment_id=deployment_id,
+            deployment_type=deployment_type,
+            name=name,
+            image_name=image_name,
+            image_tag=image_tag,
+            base_image=base_image,
+            install_command=install_command,
+            build_command=build_command,
+            no_wait=no_wait,
+            pull=pull,
+            docker_build_args=docker_build_args,
+            secrets=secrets,
+        )
+
+
+def _deploy_local(
+    *,
+    config: pathlib.Path,
+    config_json: dict,
+    verbose: bool,
+    api_version: str | None,
+    host_url: str | None,
+    api_key: str,
+    deployment_id: str | None,
+    deployment_type: str,
+    name: str | None,
+    image_name: str | None,
+    image_tag: str,
+    base_image: str | None,
+    install_command: str | None,
+    build_command: str | None,
+    no_wait: bool,
+    pull: bool,
+    docker_build_args: Sequence[str],
+    secrets: list[dict[str, str]],
+):
+    """Local Docker build + push deploy path."""
+    needs_buildx = platform.machine() != "x86_64"
+    local_tag = f"langgraph-deploy-tmp:{int(time.time())}"
+
+    with Runner() as runner:
+        langgraph_cli.docker.check_capabilities(
+            runner,
+            require_compose=False,
+            require_buildx=needs_buildx,
+        )
+
+        def log_step(message: str) -> None:
+            click.secho(message, fg="cyan")
+
+        step = 1
+
+        log_step(f"{step}. Building image")
+        if needs_buildx:
+            build_flags: list[str] = [
+                "--platform",
+                "linux/amd64",
+                "--load",
+            ]
+            if not verbose:
+                build_flags.append("--progress=quiet")
+            with Progress(message="Building...", elapsed=not verbose):
+                _build(
+                    runner,
+                    lambda _msg: None,
+                    config,
+                    config_json,
+                    base_image,
+                    api_version,
+                    pull,
+                    local_tag,
+                    docker_build_args,
+                    install_command,
+                    build_command,
+                    docker_command=("docker", "buildx", "build"),
+                    extra_flags=build_flags,
+                    verbose=verbose,
+                )
+        else:
+            with Progress(message="Building...", elapsed=not verbose):
+                _build(
+                    runner,
+                    lambda _msg: None,
+                    config,
+                    config_json,
+                    base_image,
+                    api_version,
+                    pull,
+                    local_tag,
+                    docker_build_args,
+                    install_command,
+                    build_command,
+                    verbose=verbose,
+                )
+        step += 1
+
+        client = HostBackendClient(host_url, api_key)
+
+        deployment_id = _find_or_create_deployment(
+            client, deployment_id, name, deployment_type, secrets, "internal_docker",
+            step, log_step,
+        )
+        step += 1
+
+        log_step(f"{step}. Requesting push token")
+        push_data = client.request_push_token(deployment_id)
+        deployment_token = push_data.get("token")
+        registry_url = push_data.get("registry_url")
+        if not deployment_token or not registry_url:
+            raise click.ClickException(
+                "Push token response missing token or registry_url"
+            )
+        step += 1
+
+        normalized_registry = registry_url.rstrip("/")
+        if "://" in normalized_registry:
+            normalized_registry = normalized_registry.split("//", 1)[1]
+        repo_seed = image_name or name or config.parent.name
+        repo_name = _normalize_image_name(repo_seed)
+        tag_value = _normalize_image_tag(image_tag)
+        remote_image = f"{normalized_registry}/{repo_name}:{tag_value}"
+
+        registry_host = normalized_registry.split("/")[0]
+
+        with _docker_config_for_token(registry_host, deployment_token) as cfg:
+            log_step(f"{step}. Logging into {registry_host}")
+            token_input = (
+                deployment_token
+                if deployment_token.endswith("\n")
+                else f"{deployment_token}\n"
+            )
+            runner.run(
+                subp_exec(
+                    "docker",
+                    "--config",
+                    cfg,
+                    "login",
+                    "-u",
+                    "oauth2accesstoken",
+                    "--password-stdin",
+                    registry_host,
+                    input=token_input,
+                    verbose=verbose,
+                )
+            )
+            step += 1
+
+            log_step(f"{step}. Pushing image {remote_image}")
+            runner.run(
+                subp_exec(
+                    "docker",
+                    "tag",
+                    local_tag,
+                    remote_image,
+                    verbose=verbose,
+                )
+            )
+            with Progress(message="Pushing...", elapsed=not verbose):
+                runner.run(
+                    subp_exec(
+                        "docker",
+                        "--config",
+                        cfg,
+                        "push",
+                        remote_image,
+                        verbose=verbose,
+                    )
+                )
+        step += 1
+
+        log_step(f"{step}. Updating deployment {deployment_id}")
+        client.update_deployment(deployment_id, remote_image, secrets=secrets)
+
+        if no_wait:
+            click.secho("   Deployment updated", fg="green")
+            return
+
+        _poll_revision_status(client, deployment_id, verbose=verbose)
+
+
+def _deploy_remote(
+    *,
+    config: pathlib.Path,
+    config_json: dict,
+    verbose: bool,
+    api_version: str | None,
+    host_url: str | None,
+    api_key: str,
+    deployment_id: str | None,
+    deployment_type: str,
+    name: str | None,
+    base_image: str | None,
+    install_command: str | None,
+    build_command: str | None,
+    no_wait: bool,
+    secrets: list[dict[str, str]],
+):
+    """Remote build deploy path (no local Docker required)."""
+    import urllib.request
+
+    from langgraph_cli.archive import create_archive
+
+    def log_step(message: str) -> None:
+        click.secho(message, fg="cyan")
+
+    step = 1
+    click.secho("Docker not available. Using remote build.", fg="yellow")
+
+    # -- Step: Create tarball --
+    log_step(f"{step}. Creating source archive")
+    try:
+        archive_path, file_size = create_archive(config)
+    except KeyboardInterrupt:
+        click.echo("\nCancelled.")
+        raise click.exceptions.Exit(1)
+    click.secho(
+        f"   Archive created ({file_size / 1_048_576:.1f} MB)", fg="green"
+    )
+    step += 1
+
+    client = HostBackendClient(host_url, api_key)
+
+    # -- Step: Find or create deployment --
+    deployment_id = _find_or_create_deployment(
+        client, deployment_id, name, deployment_type, secrets, "internal_source",
+        step, log_step,
+    )
+    step += 1
+
+    # -- Step: Request upload URL --
+    log_step(f"{step}. Requesting upload URL")
+    upload_data = client.request_upload_url(deployment_id)
+    signed_url = upload_data.get("upload_url")
+    object_path = upload_data.get("object_path")
+    if not signed_url or not object_path:
+        raise click.ClickException("Upload URL response missing required fields")
+    step += 1
+
+    # -- Step: Upload tarball --
+    log_step(f"{step}. Uploading source")
+    try:
+        _upload_to_gcs(signed_url, archive_path, file_size)
+    except KeyboardInterrupt:
+        click.echo("\nUpload cancelled.")
+        raise click.exceptions.Exit(1)
+    finally:
+        try:
+            os.unlink(archive_path)
+        except OSError:
+            pass
+    step += 1
+
+    # -- Step: Update deployment --
+    log_step(f"{step}. Triggering remote build")
+    client.update_deployment_internal_source(
+        deployment_id,
+        source_tarball_path=object_path,
+        secrets=secrets,
+        config_path=config.name,
+        install_command=install_command,
+        build_command=build_command,
+    )
+    step += 1
+
+    if no_wait:
+        click.secho("   Build triggered", fg="green")
+        return
+
+    # -- Poll revision status with optional log streaming --
+    _poll_revision_status(client, deployment_id, verbose=verbose, is_remote_build=True)
+
+
+def _upload_to_gcs(signed_url: str, file_path: str, file_size: int) -> None:
+    """Upload tarball to GCS via signed PUT URL with progress display."""
+    import urllib.request
+
+    uploaded = 0
+
+    with open(file_path, "rb") as f:
+        original_read = f.read
+
+        def tracked_read(size=-1):
+            nonlocal uploaded
+            data = original_read(size)
+            if data:
+                uploaded += len(data)
+                pct = int(uploaded * 100 / file_size) if file_size else 100
+                click.echo(
+                    f"\r   Uploading ({file_size / 1_048_576:.1f} MB)... {pct}%",
+                    nl=False,
+                )
+            return data
+
+        f.read = tracked_read
+
+        req = urllib.request.Request(
+            signed_url,
+            data=f,
+            method="PUT",
+            headers={
+                "Content-Type": "application/gzip",
+                "Content-Length": str(file_size),
+                "X-Goog-Content-Length-Range": "0,209715200",
+            },
+        )
+        try:
+            urllib.request.urlopen(req)
+        except urllib.error.HTTPError as err:
+            detail = err.read().decode("utf-8", errors="ignore")
+            raise click.ClickException(
+                f"Upload failed with status {err.code}: {detail}"
+            ) from None
+    click.echo()
+
+
+def _find_or_create_deployment(
+    client: HostBackendClient,
+    deployment_id: str | None,
+    name: str | None,
+    deployment_type: str,
+    secrets: list[dict[str, str]],
+    source: str,
+    step: int,
+    log_step: Callable[[str], None],
+) -> str:
+    """Find an existing deployment or create a new one. Returns deployment_id."""
+    if deployment_id:
+        log_step(f"{step}. Using deployment {deployment_id}")
+        return deployment_id
+
+    log_step(f"{step}. Looking up deployment '{name}'")
+    existing = client.list_deployments(name_contains=name)
+    found_id = None
+    if isinstance(existing, dict):
+        for dep in existing.get("resources", []):
+            if isinstance(dep, dict) and dep.get("name") == name:
+                found_id = dep.get("id")
+                break
+    if found_id:
+        deployment_id = str(found_id)
+        click.secho(
+            f"   Found existing deployment (ID: {deployment_id})",
+            fg="green",
+        )
+        return deployment_id
+
+    log_step(f"   Creating deployment '{name}'")
+    payload = {
+        "name": name,
+        "source": source,
+        "source_config": {"deployment_type": deployment_type},
+        "source_revision_config": {},
+        "secrets": secrets,
+    }
+    created = client.create_deployment(payload)
+    created_id = created.get("id") if isinstance(created, dict) else None
+    if not isinstance(created_id, str) or not created_id:
+        raise HostBackendError(
+            "POST /v2/deployments succeeded but response missing a valid 'id'"
+        )
+    deployment_id = created_id
+    click.secho(f"   Deployment ID: {deployment_id}", fg="green")
+    return deployment_id
+
+
+def _poll_revision_status(
+    client: HostBackendClient,
+    deployment_id: str,
+    *,
+    verbose: bool = False,
+    is_remote_build: bool = False,
+) -> None:
+    """Poll revision status until terminal, optionally streaming build logs."""
+    revisions_resp = client.list_revisions(deployment_id, limit=1)
+    resources = (
+        revisions_resp.get("resources", [])
+        if isinstance(revisions_resp, dict)
+        else []
+    )
+    if not resources:
+        click.secho("   Deployment updated", fg="green")
+        return
+
+    revision_id = str(resources[0]["id"])
+    last_status = ""
+    log_offset: str | None = None
+
+    deadline = time.time() + 900 if is_remote_build else time.time() + 300
+    with Progress(message="Deploying...", elapsed=True) as set_progress:
+        while time.time() < deadline:
+            try:
+                rev = client.get_revision(deployment_id, revision_id)
+            except KeyboardInterrupt:
+                set_progress("")
+                click.secho(
+                    f"\n   Interrupted. Deployment ID: {deployment_id}, "
+                    f"Revision ID: {revision_id}",
+                    fg="yellow",
+                )
+                click.secho(
+                    "   The build will continue remotely.",
+                    fg="yellow",
+                )
+                raise click.exceptions.Exit(1)
+
+            status = (
+                rev.get("status", "UNKNOWN") if isinstance(rev, dict) else "UNKNOWN"
+            )
+            if status != last_status:
+                last_status = status
+                set_progress("")
+                click.secho(f"   Status: {status}", fg="cyan")
+                if status in _TERMINAL_STATUSES:
+                    break
+                set_progress(f"{status}...")
+
+            # Stream build logs when verbose and building
+            if (
+                is_remote_build
+                and verbose
+                and status in ("AWAITING_BUILD", "BUILDING")
+            ):
+                try:
+                    logs_resp = client.list_build_logs(
+                        deployment_id, revision_id, offset=log_offset
+                    )
+                    if isinstance(logs_resp, dict):
+                        for entry in logs_resp.get("logs", []):
+                            msg = entry.get("message", "")
+                            if msg:
+                                click.echo(f"   | {msg}")
+                        log_offset = logs_resp.get("next_offset") or log_offset
+                except Exception:
+                    pass
+
+            time.sleep(3)
+        else:
+            set_progress("")
+
+    # On BUILD_FAILED, tail the last few build log lines
+    if is_remote_build and last_status == "BUILD_FAILED" and not verbose:
+        click.secho("   Last build log lines:", fg="red")
+        try:
+            logs_resp = client.list_build_logs(
+                deployment_id, revision_id, order="desc", limit=30
+            )
+            if isinstance(logs_resp, dict):
+                entries = list(reversed(logs_resp.get("logs", [])))
+                for entry in entries:
+                    msg = entry.get("message", "")
+                    if msg:
+                        click.echo(f"   | {msg}")
+        except Exception:
+            click.secho("   (failed to fetch build logs)", fg="red")
+        click.secho(
+            "   Re-run with --verbose to see full build output.",
+            fg="yellow",
+        )
+
+    dep_info = client.get_deployment(deployment_id)
+    custom_url = None
+    if isinstance(dep_info, dict):
+        sc = dep_info.get("source_config")
+        if isinstance(sc, dict):
+            custom_url = sc.get("custom_url")
+
+    if last_status == "DEPLOYED":
+        click.secho("   Deployment successful!", fg="green")
+        if custom_url:
+            click.secho(f"   URL: {custom_url}", fg="green")
+    elif last_status in ("BUILD_FAILED", "DEPLOY_FAILED", "CREATE_FAILED"):
+        click.secho(f"   Deployment failed: {last_status}", fg="red")
+        raise click.exceptions.Exit(1)
+    else:
+        click.secho(
+            f"   Timed out waiting for deployment (last status: {last_status}).",
+            fg="yellow",
+        )
+        if custom_url:
+            click.secho(
+                f"   Check status at: {custom_url}",
+                fg="yellow",
+            )
+        else:
+            click.secho(
+                "   Check status in the LangSmith Deployments dashboard.",
+                fg="yellow",
+            )
+
+
+def _normalize_image_name(value: str | None) -> str:
+    if not value:
+        return "app"
+    slug = re.sub(r"[^a-z0-9._-]+", "-", value.lower()).strip("-.")
+    return slug or "app"
+
+
+def _normalize_image_tag(value: str) -> str:
+    if not value:
+        value = "latest"
+    if not re.fullmatch(r"[A-Za-z0-9_.-]+", value):
+        raise click.UsageError(
+            "Image tag may only contain characters A-Z, a-z, 0-9, '_', '-', '.'"
+        )
+    return value
+
+
 def _get_docker_ignore_content() -> str:
     """Return the content of a .dockerignore file.
 

libs/cli/langgraph_cli/docker.py

@@ -24,9 +24,9 @@ DockerComposeType = Literal["plugin", "standalone"]
 
 class DockerCapabilities(NamedTuple):
     version_docker: Version
-    version_compose: Version
+    version_compose: Version | None
     healthcheck_start_interval: bool
-    compose_type: DockerComposeType = "plugin"
+    compose_type: DockerComposeType | None = None
 
 
 def _parse_version(version: str) -> Version:
@@ -45,7 +45,26 @@ def _parse_version(version: str) -> Version:
     )
 
 
-def check_capabilities(runner) -> DockerCapabilities:
+def is_docker_available() -> bool:
+    """Check if Docker is installed and running without raising."""
+    if shutil.which("docker") is None:
+        return False
+    try:
+        import subprocess
+
+        result = subprocess.run(
+            ["docker", "info"],
+            capture_output=True,
+            timeout=10,
+        )
+        return result.returncode == 0
+    except Exception:
+        return False
+
+
+def check_capabilities(
+    runner, *, require_compose: bool = True, require_buildx: bool = False
+) -> DockerCapabilities:
     # check docker available
     if shutil.which("docker") is None:
         raise click.UsageError("Docker not installed") from None
@@ -61,25 +80,33 @@ def check_capabilities(runner) -> DockerCapabilities:
     if not info["ServerVersion"]:
         raise click.UsageError("Docker not running") from None
 
-    compose_type: DockerComposeType
-    try:
-        compose = next(
-            p for p in info["ClientInfo"]["Plugins"] if p["Name"] == "compose"
-        )
-        compose_version_str = compose["Version"]
-        compose_type = "plugin"
-    except (KeyError, StopIteration):
-        if shutil.which("docker-compose") is None:
-            raise click.UsageError("Docker Compose not installed") from None
+    compose_type: DockerComposeType | None = None
+    compose_version: Version | None = None
+    if require_compose:
+        try:
+            compose = next(
+                p for p in info["ClientInfo"]["Plugins"] if p["Name"] == "compose"
+            )
+            compose_version_str = compose["Version"]
+            compose_type = "plugin"
+        except (KeyError, StopIteration):
+            if shutil.which("docker-compose") is None:
+                raise click.UsageError("Docker Compose not installed") from None
 
-        compose_version_str, _ = runner.run(
-            subp_exec("docker-compose", "--version", "--short", collect=True)
-        )
-        compose_type = "standalone"
+            compose_version_str, _ = runner.run(
+                subp_exec("docker-compose", "--version", "--short", collect=True)
+            )
+            compose_type = "standalone"
+        compose_version = _parse_version(compose_version_str)
+
+    if require_buildx:
+        try:
+            runner.run(subp_exec("docker", "buildx", "version", collect=True))
+        except click.exceptions.Exit:
+            raise click.UsageError("Docker Buildx not installed") from None
 
     # parse versions
     docker_version = _parse_version(info["ServerVersion"])
-    compose_version = _parse_version(compose_version_str)
 
     # check capabilities
     return DockerCapabilities(

libs/cli/langgraph_cli/host_backend.py

@@ -0,0 +1,163 @@
+"""HTTP client for LangGraph host backend deployments."""
+
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Any
+
+import click
+
+
+class HostBackendError(click.ClickException):
+    """Raised when the host backend returns an error response."""
+
+
+class HostBackendClient:
+    """Minimal JSON HTTP client for the host backend deployment service."""
+
+    def __init__(self, base_url: str, api_key: str):
+        if not base_url:
+            raise click.UsageError("Host backend URL is required")
+        base_url = base_url.rstrip("/")
+        self._base_url = base_url
+        self._api_key = api_key
+
+    def _request(
+        self, method: str, path: str, payload: dict[str, Any] | None = None
+    ) -> Any:
+        url = f"{self._base_url}{path}"
+        data: bytes | None
+        if payload is not None:
+            data = json.dumps(payload).encode("utf-8")
+        else:
+            data = None
+        headers: dict[str, str] = {
+            "X-Api-Key": self._api_key,
+            "Accept": "application/json",
+        }
+        if data is not None:
+            headers["Content-Type"] = "application/json"
+        req = urllib.request.Request(url, data=data, headers=headers, method=method)
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                body = resp.read()
+        except urllib.error.HTTPError as err:
+            detail = err.read().decode("utf-8", errors="ignore")
+            message = detail or err.reason
+            raise HostBackendError(
+                f"{method} {path} failed with status {err.code}: {message}"
+            ) from None
+        except urllib.error.URLError as err:
+            raise HostBackendError(str(err.reason)) from None
+
+        if not body:
+            return None
+        try:
+            return json.loads(body)
+        except json.JSONDecodeError as err:
+            raise HostBackendError(
+                f"Failed to decode response from {path}: {err.msg}"
+            ) from None
+
+    def create_deployment(self, payload: dict[str, Any]) -> dict[str, Any]:
+        return self._request("POST", "/v2/deployments", payload)
+
+    def list_deployments(self, name_contains: str) -> dict[str, Any]:
+        encoded = urllib.parse.quote(name_contains, safe="")
+        return self._request("GET", f"/v2/deployments?name_contains={encoded}")
+
+    def get_deployment(self, deployment_id: str) -> dict[str, Any]:
+        return self._request("GET", f"/v2/deployments/{deployment_id}")
+
+    def request_push_token(self, deployment_id: str) -> dict[str, Any]:
+        return self._request(
+            "POST",
+            f"/v2/deployments/{deployment_id}/push-token",
+        )
+
+    def update_deployment(
+        self,
+        deployment_id: str,
+        image_uri: str,
+        secrets: list[dict[str, str]] | None = None,
+    ) -> dict[str, Any]:
+        payload: dict[str, Any] = {
+            "source_revision_config": {"image_uri": image_uri},
+        }
+        if secrets is not None:
+            payload["secrets"] = secrets
+        return self._request(
+            "PATCH",
+            f"/v2/deployments/{deployment_id}",
+            payload,
+        )
+
+    def list_revisions(self, deployment_id: str, limit: int = 1) -> dict[str, Any]:
+        return self._request(
+            "GET",
+            f"/v2/deployments/{deployment_id}/revisions?limit={limit}",
+        )
+
+    def get_revision(self, deployment_id: str, revision_id: str) -> dict[str, Any]:
+        return self._request(
+            "GET",
+            f"/v2/deployments/{deployment_id}/revisions/{revision_id}",
+        )
+
+    def request_upload_url(self, deployment_id: str) -> dict[str, Any]:
+        """Get a signed GCS URL for uploading the source tarball."""
+        return self._request(
+            "POST",
+            f"/v2/deployments/{deployment_id}/upload-url",
+        )
+
+    def update_deployment_internal_source(
+        self,
+        deployment_id: str,
+        source_tarball_path: str,
+        secrets: list[dict[str, str]] | None = None,
+        config_path: str | None = None,
+        install_command: str | None = None,
+        build_command: str | None = None,
+    ) -> dict[str, Any]:
+        """Trigger a remote build revision with the uploaded tarball."""
+        src_config: dict[str, Any] = {
+            "source_tarball_path": source_tarball_path,
+        }
+        if config_path is not None:
+            src_config["langgraph_config_path"] = config_path
+
+        payload: dict[str, Any] = {"source_revision_config": src_config}
+
+        source_config: dict[str, Any] = {}
+        if install_command is not None:
+            source_config["install_command"] = install_command
+        if build_command is not None:
+            source_config["build_command"] = build_command
+        if source_config:
+            payload["source_config"] = source_config
+
+        if secrets is not None:
+            payload["secrets"] = secrets
+        return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload)
+
+    def list_build_logs(
+        self,
+        deployment_id: str,
+        revision_id: str,
+        order: str = "asc",
+        limit: int = 50,
+        offset: str | None = None,
+    ) -> dict[str, Any]:
+        """Fetch build logs for a revision."""
+        payload: dict[str, Any] = {"order": order, "limit": limit}
+        if offset:
+            payload["offset"] = offset
+        return self._request(
+            "POST",
+            f"/v1/projects/{deployment_id}/revisions/{revision_id}/build_logs",
+            payload,
+        )

libs/cli/langgraph_cli/progress.py

@@ -12,8 +12,11 @@ class Progress:
         while True:
             yield from "|/-\\"
 
-    def __init__(self, *, message=""):
+    def __init__(self, *, message="", elapsed: bool = False):
         self.message = message
+        self._base_message = message
+        self._show_elapsed = elapsed
+        self._stop = threading.Event()
         self.spinner_generator = self.spinning_cursor()
 
     def spinner_iteration(self):
@@ -21,7 +24,6 @@ class Progress:
         sys.stdout.write(next(self.spinner_generator) + " " + message)
         sys.stdout.flush()
         time.sleep(self.delay)
-        # clear the spinner and message
         sys.stdout.write(
             "\b" * (len(message) + 2)
             + " " * (len(message) + 2)
@@ -29,13 +31,26 @@ class Progress:
         )
         sys.stdout.flush()
 
+    def _format_elapsed(self, seconds: float) -> str:
+        mins, secs = divmod(int(seconds), 60)
+        if mins:
+            return f"{self._base_message} ({mins}m {secs:02d}s)"
+        return f"{self._base_message} ({secs}s)"
+
     def spinner_task(self):
-        while self.message:
+        start = time.monotonic()
+        while not self._stop.is_set():
+            if not self.message:
+                time.sleep(self.delay)
+                continue
+            if self._show_elapsed:
+                self.message = self._format_elapsed(time.monotonic() - start)
             message = self.message
+            if not message:
+                continue
             sys.stdout.write(next(self.spinner_generator) + " " + message)
             sys.stdout.flush()
             time.sleep(self.delay)
-            # clear the spinner and message
             sys.stdout.write(
                 "\b" * (len(message) + 2)
                 + " " * (len(message) + 2)
@@ -50,21 +65,22 @@ class Progress:
 
             def set_message(message):
                 self.message = message
-                if not message:
-                    self.thread.join()
+                self._base_message = message or self._base_message
 
             return set_message
         else:
 
             def set_message(message):
-                sys.stderr.write(message + "\n")
-                sys.stderr.flush()
+                if message:
+                    sys.stderr.write(message + "\n")
+                    sys.stderr.flush()
 
             return set_message
 
     def __exit__(self, exception, value, tb):
         if sys.stdout.isatty():
             self.message = ""
+            self._stop.set()
             try:
                 self.thread.join()
             finally: