Add list-deployments CLI command.

- Added a default value for the `--host-url` option in the CLI to streamline user experience.
- Removed the explicit error check for `host_url` in the `deploy` function, allowing for a more flexible deployment process.

libs/cli/langgraph_cli/cli.py

@@ -1,14 +1,23 @@
 """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
 from click import secho
+from dotenv import dotenv_values
 
 import langgraph_cli.config
 import langgraph_cli.docker
@@ -17,11 +26,185 @@ 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",
+        "LANGSMITH_DEPLOYMENT_NAME",
+        "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",
+)
+
+_DEPLOYMENT_NAME_ENV = "LANGSMITH_DEPLOYMENT_NAME"
+
+
+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")
+    # validate_config_file will default env to {}
+    if isinstance(env_field, dict) and env_field:
+        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()
+        if not env_path.exists():
+            click.secho(
+                f"Warning: env file '{env_field}' specified in langgraph.json not found.",
+                fg="yellow",
+            )
+            return {}
+    else:
+        env_path = pathlib.Path.cwd() / ".env"
+    return {k: v for k, v in dotenv_values(env_path).items() if v is not None}
+
+
+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
+
+
+def _resolve_host_api_key(
+    api_key: str | None, env_vars: dict[str, str] | None = None
+) -> str | None:
+    """Resolve the host API key from explicit input or supported env vars."""
+    if api_key:
+        return api_key
+    env_vars = env_vars or {}
+    for key_name in _API_KEY_ENV_NAMES:
+        val = env_vars.get(key_name) or os.environ.get(key_name)
+        if val:
+            return val
+    return None
+
+
+def _extract_deployment_url(deployment: dict[str, object]) -> str:
+    """Return the deployment URL exposed by the API response."""
+    source_config = deployment.get("source_config")
+    if isinstance(source_config, dict):
+        for key in ("custom_url", "url"):
+            value = source_config.get(key)
+            if isinstance(value, str) and value:
+                return value
+
+    return "-"
+
+
+def _print_deployments(deployments: Sequence[dict[str, object]]) -> None:
+    """Render deployments in a simple aligned table."""
+    if not deployments:
+        click.secho("No deployments found.", fg="yellow")
+        return
+
+    rows = [
+        (
+            str(deployment.get("id", "-") or "-"),
+            str(deployment.get("name", "-") or "-"),
+            _extract_deployment_url(deployment),
+        )
+        for deployment in deployments
+    ]
+    headers = ("Deployment ID", "Deployment Name", "Deployment URL")
+    widths = [
+        max(len(headers[idx]), max(len(row[idx]) for row in rows))
+        for idx in range(len(headers))
+    ]
+
+    click.secho(
+        "  ".join(headers[idx].ljust(widths[idx]) for idx in range(len(headers))),
+        bold=True,
+    )
+    for row in rows:
+        click.echo("  ".join(row[idx].ljust(widths[idx]) for idx in range(len(row))))
+
+
+_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,6 +487,9 @@ 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:
@@ -312,7 +498,7 @@ def _build(
                 "docker",
                 "pull",
                 langgraph_cli.config.docker_tag(config_json, base_image, api_version),
-                verbose=True,
+                verbose=verbose,
             )
         )
     set("Building...")
@@ -334,7 +520,9 @@ def _build(
     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,
@@ -348,15 +536,16 @@ def _build(
     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,
         )
     )
 
@@ -441,6 +630,519 @@ def build(
         )
 
 
+@click.option(
+    "--api-key",
+    envvar="LANGGRAPH_HOST_API_KEY",
+    help=(
+        "API key. Can also be set via LANGGRAPH_HOST_API_KEY, "
+        "LANGSMITH_API_KEY, or LANGCHAIN_API_KEY environment variable or .env file."
+    ),
+)
+@click.option(
+    "--name",
+    envvar="LANGSMITH_DEPLOYMENT_NAME",
+    help=(
+        "Deployment name. Can also be set via LANGSMITH_DEPLOYMENT_NAME "
+        "environment variable or .env file. 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.argument("docker_build_args", nargs=-1, type=click.UNPROCESSED)
+@cli.command(
+    help=(
+        "[Beta] Build and deploy a LangGraph image to LangSmith Deployments.\n\n"
+        "This command is in beta and under active development. "
+        "Expect frequent updates and improvements.\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,
+    docker_build_args: Sequence[str],
+):
+    click.secho(
+        "Note: 'langgraph deploy' is in beta. Expect frequent updates and improvements.",
+        fg="yellow",
+    )
+    click.echo()
+    config_json = langgraph_cli.config.validate_config_file(config)
+    warn_non_wolfi_distro(config_json)
+
+    env_vars = _parse_env_from_config(config_json, config)
+
+    api_key = _resolve_host_api_key(api_key, env_vars)
+    if not api_key:
+        api_key = click.prompt("Host API key", hide_input=True)
+
+    if not deployment_id and not name:
+        name = env_vars.get(_DEPLOYMENT_NAME_ENV)
+    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)
+
+    # Use buildx to cross-compile for amd64 when running on a non-x86_64 host
+    # (e.g. Apple Silicon). On amd64 hosts, plain docker build is sufficient.
+    needs_buildx = platform.machine() != "x86_64"
+    local_tag = f"langgraph-deploy-tmp:{int(time.time())}"
+
+    with Runner() as runner:
+        if shutil.which("docker") is None:
+            raise click.UsageError(
+                "Docker is required but not installed.\n"
+                "Install Docker Desktop: https://docs.docker.com/get-docker/\n\n"
+                "Remote builds (no Docker required) are coming in a future update."
+            )
+        if needs_buildx:
+            try:
+                runner.run(subp_exec("docker", "buildx", "version", collect=True))
+            except click.exceptions.Exit:
+                raise click.UsageError(
+                    "Docker Buildx is required but not installed.\n"
+                    "Your machine architecture ("
+                    + platform.machine()
+                    + ") requires Buildx to cross-compile images for linux/amd64.\n"
+                    "Install Buildx: https://docs.docker.com/build/install-buildx/\n\n"
+                    "Remote builds (no Docker required) are coming in a future update."
+                ) from None
+
+        def log_step(message: str) -> None:
+            click.secho(message, fg="cyan")
+
+        step = 1
+
+        # -- Step: Build image --
+        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
+
+        # -- Step: Find or create deployment --
+        client = HostBackendClient(host_url, api_key)
+
+        if deployment_id:
+            log_step(f"{step}. Using deployment {deployment_id}")
+            step += 1
+        else:
+            log_step(f"{step}. Looking up deployment '{name}'")
+            try:
+                existing = client.list_deployments(name_contains=name)
+            except HostBackendError as err:
+                if (
+                    err.status_code == 403
+                    and "requires workspace specification" in err.message
+                ):
+                    click.secho(
+                        "Your API key is org-scoped and requires a workspace ID.",
+                        fg="yellow",
+                    )
+                    click.secho(
+                        "Find your workspace ID in LangSmith under Settings > Workspaces.",
+                        fg="yellow",
+                    )
+                    tenant_id = click.prompt("Workspace ID")
+                    client = HostBackendClient(host_url, api_key, tenant_id=tenant_id)
+                    existing = client.list_deployments(name_contains=name)
+                else:
+                    raise
+            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",
+                )
+            else:
+                log_step(f"   Creating deployment '{name}'")
+                payload = {
+                    "name": name,
+                    "source": "internal_docker",
+                    "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")
+            step += 1
+
+        # -- Step: Get push token and authenticate --
+        log_step(f"{step}. Requesting push token")
+        try:
+            push_data = client.request_push_token(deployment_id)
+        except HostBackendError as err:
+            if (
+                err.status_code == 400
+                and "only available for 'internal_docker' source deployments"
+                in err.message
+            ):
+                raise click.ClickException(
+                    f"Deployment '{deployment_id}' was not created by 'langgraph deploy' "
+                    "and cannot be updated with this command.\n"
+                    "Please create a new deployment by running 'langgraph deploy' "
+                    "without --deployment-id, or use a different --name."
+                ) from None
+            raise
+        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]
+
+        # Use a clean Docker config with only the push token so that
+        # system credential helpers (e.g. gcloud) don't interfere.
+        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
+
+            # -- Step: Tag and push --
+            log_step(f"{step}. Pushing image {remote_image}")
+            runner.run(
+                subp_exec(
+                    "docker",
+                    "tag",
+                    local_tag,
+                    remote_image,
+                    verbose=verbose,
+                )
+            )
+            max_push_retries = 3
+            for attempt in range(max_push_retries):
+                try:
+                    with Progress(message="Pushing...", elapsed=not verbose):
+                        runner.run(
+                            subp_exec(
+                                "docker",
+                                "--config",
+                                cfg,
+                                "push",
+                                remote_image,
+                                verbose=verbose,
+                            )
+                        )
+                    break
+                except click.exceptions.Exit:
+                    if attempt < max_push_retries - 1:
+                        click.secho(
+                            f"   Push failed, retrying (attempt {attempt + 2} of {max_push_retries})...",
+                            fg="yellow",
+                        )
+                    else:
+                        raise
+        step += 1
+
+        # -- Step: Update deployment --
+        log_step(f"{step}. Updating deployment {deployment_id}")
+        updated = client.update_deployment(deployment_id, remote_image, secrets=secrets)
+        tenant_id = updated.get("tenant_id") if isinstance(updated, dict) else None
+        if tenant_id:
+            status_url = (
+                f"https://smith.langchain.com/o/{tenant_id}"
+                f"/host/deployments/{deployment_id}"
+            )
+            click.secho(f"   View status: {status_url}", fg="cyan")
+
+        if no_wait:
+            click.secho("   Deployment updated", fg="green")
+            return
+
+        # -- Poll revision status --
+        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 = ""
+
+        deadline = time.time() + 300
+        with Progress(message="Deploying...", elapsed=True) as set_progress:
+            while time.time() < deadline:
+                rev = client.get_revision(deployment_id, revision_id)
+                status = (
+                    rev.get("status", "UNKNOWN") if isinstance(rev, dict) else "UNKNOWN"
+                )
+                if status != last_status:
+                    last_status = status
+                    # pause spinner so we can avoid conflict when writing status
+                    set_progress("")
+                    click.secho(f"   Status: {status}", fg="cyan")
+                    if status in _TERMINAL_STATUSES:
+                        break
+                    set_progress(f"{status}...")
+                time.sleep(1)
+            else:
+                set_progress("")
+
+        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",
+                )
+
+
+@click.option(
+    "--api-key",
+    envvar="LANGGRAPH_HOST_API_KEY",
+    help=(
+        "API key. Can also be set via LANGGRAPH_HOST_API_KEY, "
+        "LANGSMITH_API_KEY, or LANGCHAIN_API_KEY environment variable."
+    ),
+)
+@click.option(
+    "--host-url",
+    envvar="LANGGRAPH_HOST_URL",
+    default="https://api.host.langchain.com",
+    hidden=True,
+)
+@cli.command(
+    "list-deployments",
+    help=(
+        "[Beta] List LangSmith Deployments.\n\n"
+        "This command is in beta and under active development."
+    ),
+)
+@log_command
+def list_deployments(
+    api_key: str | None,
+    host_url: str,
+) -> None:
+    click.secho(
+        "Note: 'langgraph list-deployments' is in beta. Expect frequent updates and improvements.",
+        fg="yellow",
+    )
+    click.echo()
+
+    api_key = _resolve_host_api_key(api_key)
+    if not api_key:
+        api_key = click.prompt("Host API key", hide_input=True)
+
+    client = HostBackendClient(host_url, api_key)
+    try:
+        response = client.list_deployments()
+    except HostBackendError as err:
+        if err.status_code == 403 and "requires workspace specification" in err.message:
+            click.secho(
+                "Your API key is org-scoped and requires a workspace ID.",
+                fg="yellow",
+            )
+            click.secho(
+                "Find your workspace ID in LangSmith under Settings > Workspaces.",
+                fg="yellow",
+            )
+            tenant_id = click.prompt("Workspace ID")
+            client = HostBackendClient(host_url, api_key, tenant_id=tenant_id)
+            response = client.list_deployments()
+        else:
+            raise
+
+    resources = response.get("resources", []) if isinstance(response, dict) else []
+    deployments = [dep for dep in resources if isinstance(dep, dict)]
+    _print_deployments(deployments)
+
+
+def _normalize_image_name(value: str | None) -> str:
+    """Sanitize a deployment/directory name into a valid Docker repository name.
+
+    Docker repository names must be lowercase and may only contain
+    [a-z0-9._-].  Invalid characters are replaced with hyphens.
+    """
+    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:
+    """Validate and return a Docker image tag.
+
+    Tags may only contain [A-Za-z0-9_.-].  Defaults to "latest" when empty.
+    """
+    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/host_backend.py

@@ -0,0 +1,111 @@
+"""HTTP client for LangGraph host backend deployments."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import click
+import httpx
+
+
+class HostBackendError(click.ClickException):
+    """Raised when the host backend returns an error response."""
+
+    def __init__(self, message: str, status_code: int | None = None):
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class HostBackendClient:
+    """Minimal JSON HTTP client for the host backend deployment service."""
+
+    def __init__(self, base_url: str, api_key: str, tenant_id: str | None = None):
+        if not base_url:
+            raise click.UsageError("Host backend URL is required")
+        transport = httpx.HTTPTransport(retries=3)
+        headers: dict[str, str] = {
+            "X-Api-Key": api_key,
+            "Accept": "application/json",
+        }
+        if tenant_id:
+            headers["X-Tenant-ID"] = tenant_id
+        self._base_url = base_url.rstrip("/")
+        self._api_key = api_key
+        self._client = httpx.Client(
+            base_url=self._base_url,
+            headers=headers,
+            transport=transport,
+            timeout=30,
+        )
+
+    def _request(
+        self, method: str, path: str, payload: dict[str, Any] | None = None
+    ) -> Any:
+        try:
+            resp = self._client.request(method, path, json=payload)
+            resp.raise_for_status()
+        except httpx.HTTPStatusError as err:
+            detail = err.response.text or str(err.response.status_code)
+            raise HostBackendError(
+                f"{method} {path} failed with status {err.response.status_code}: {detail}",
+                status_code=err.response.status_code,
+            ) from None
+        except httpx.TransportError as err:
+            raise HostBackendError(str(err)) from None
+
+        if not resp.content:
+            return None
+        try:
+            return resp.json()
+        except ValueError as err:
+            raise HostBackendError(
+                f"Failed to decode response from {path}: {err}"
+            ) 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 | None = None) -> dict[str, Any]:
+        path = "/v2/deployments"
+        if name_contains:
+            query = httpx.QueryParams({"name_contains": name_contains})
+            path = f"{path}?{query}"
+        return self._request("GET", path)
+
+    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}",
+        )

libs/cli/langgraph_cli/progress.py

@@ -12,8 +12,12 @@ 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
+        # use this to make sure we don't kill thread when we set msg to ""
+        self._stop = threading.Event()
         self.spinner_generator = self.spinning_cursor()
 
     def spinner_iteration(self):
@@ -29,9 +33,23 @@ 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)
@@ -50,21 +68,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:

libs/cli/pyproject.toml

@@ -13,7 +13,9 @@ license = "MIT"
 license-files = ['LICENSE']
 dependencies = [
     "click>=8.1.7",
+    "httpx>=0.24.0",
     "langgraph-sdk>=0.1.0 ; python_version >= '3.11'",
+    "python-dotenv>=0.8.0",
 ]
 [tool.hatch.version]
 path = "langgraph_cli/__init__.py"
@@ -21,7 +23,6 @@ path = "langgraph_cli/__init__.py"
 inmem = [
     "langgraph-api>=0.5.35,<0.8.0 ; python_version >= '3.11'",
     "langgraph-runtime-inmem>=0.7 ; python_version >= '3.11'",
-    "python-dotenv>=0.8.0",
 ]
 
 [project.urls]

libs/cli/tests/unit_tests/cli/test_cli.py

@@ -7,6 +7,7 @@ import textwrap
 from contextlib import contextmanager
 from pathlib import Path
 
+import pytest
 from click.testing import CliRunner
 
 from langgraph_cli.cli import cli, prepare_args_and_stdin
@@ -287,6 +288,65 @@ def test_version_option() -> None:
     )
 
 
+def test_list_deployments_command_formats_output(monkeypatch: pytest.MonkeyPatch):
+    class FakeHostBackendClient:
+        def __init__(self, base_url, api_key, tenant_id=None):
+            self.base_url = base_url
+            self.api_key = api_key
+            self.tenant_id = tenant_id
+
+        def list_deployments(self):
+            return {
+                "resources": [
+                    {
+                        "id": "dep_123",
+                        "name": "alpha",
+                        "source_config": {"custom_url": "https://alpha.example.com"},
+                    },
+                    {
+                        "id": "dep_456",
+                        "name": "beta",
+                        "source_config": {"custom_url": "https://beta.example.com"},
+                    },
+                ]
+            }
+
+    monkeypatch.setattr("langgraph_cli.cli.HostBackendClient", FakeHostBackendClient)
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["list-deployments", "--api-key", "test-key"])
+
+    assert result.exit_code == 0, result.output
+    assert "Deployment ID" in result.output
+    assert "Deployment Name" in result.output
+    assert "Deployment URL" in result.output
+    assert "dep_123" in result.output
+    assert "alpha" in result.output
+    assert "https://alpha.example.com" in result.output
+    assert "dep_456" in result.output
+    assert "beta" in result.output
+    assert "https://beta.example.com" in result.output
+
+
+def test_list_deployments_command_empty_result(monkeypatch: pytest.MonkeyPatch):
+    class FakeHostBackendClient:
+        def __init__(self, base_url, api_key, tenant_id=None):
+            self.base_url = base_url
+            self.api_key = api_key
+            self.tenant_id = tenant_id
+
+        def list_deployments(self):
+            return {"resources": []}
+
+    monkeypatch.setattr("langgraph_cli.cli.HostBackendClient", FakeHostBackendClient)
+
+    runner = CliRunner()
+    result = runner.invoke(cli, ["list-deployments", "--api-key", "test-key"])
+
+    assert result.exit_code == 0, result.output
+    assert "No deployments found." in result.output
+
+
 def test_dockerfile_command_basic() -> None:
     """Test the 'dockerfile' command with basic configuration."""
     runner = CliRunner()

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -0,0 +1,134 @@
+import base64
+import json
+import os
+
+import click
+import pytest
+
+from langgraph_cli.cli import (
+    _docker_config_for_token,
+    _normalize_image_name,
+    _normalize_image_tag,
+    _parse_env_from_config,
+)
+
+
+class TestDockerConfigForToken:
+    def test_creates_config_json(self):
+        with _docker_config_for_token("us-docker.pkg.dev", "my-token") as cfg:
+            config_path = os.path.join(cfg, "config.json")
+            assert os.path.isfile(config_path)
+            with open(config_path) as f:
+                data = json.load(f)
+            expected_auth = base64.b64encode(b"oauth2accesstoken:my-token").decode()
+            assert data == {"auths": {"us-docker.pkg.dev": {"auth": expected_auth}}}
+
+    def test_tempdir_cleaned_up(self):
+        with _docker_config_for_token("registry.example.com", "tok") as cfg:
+            assert os.path.isdir(cfg)
+        assert not os.path.exists(cfg)
+
+    def test_different_registries(self):
+        with _docker_config_for_token("gcr.io", "token123") as cfg:
+            with open(os.path.join(cfg, "config.json")) as f:
+                data = json.load(f)
+            assert "gcr.io" in data["auths"]
+
+
+class TestNormalizeImageName:
+    def test_simple_name(self):
+        assert _normalize_image_name("myapp") == "myapp"
+
+    def test_uppercase_lowered(self):
+        assert _normalize_image_name("MyApp") == "myapp"
+
+    def test_special_chars_replaced(self):
+        assert _normalize_image_name("my app!@#v2") == "my-app-v2"
+
+    def test_dots_and_hyphens_kept(self):
+        assert _normalize_image_name("my-app.v2") == "my-app.v2"
+
+    def test_leading_trailing_stripped(self):
+        assert _normalize_image_name("--my-app..") == "my-app"
+
+    def test_empty_string_returns_app(self):
+        assert _normalize_image_name("") == "app"
+
+    def test_none_returns_app(self):
+        assert _normalize_image_name(None) == "app"
+
+    def test_all_invalid_chars_returns_app(self):
+        assert _normalize_image_name("!!!") == "app"
+
+
+class TestNormalizeImageTag:
+    def test_valid_tag(self):
+        assert _normalize_image_tag("v1.2.3") == "v1.2.3"
+
+    def test_empty_defaults_to_latest(self):
+        assert _normalize_image_tag("") == "latest"
+
+    def test_alphanumeric_and_special(self):
+        assert _normalize_image_tag("my_tag-1.0") == "my_tag-1.0"
+
+    def test_invalid_chars_raises(self):
+        with pytest.raises(click.UsageError, match="Image tag may only contain"):
+            _normalize_image_tag("v1.0:bad")
+
+    def test_spaces_raises(self):
+        with pytest.raises(click.UsageError, match="Image tag may only contain"):
+            _normalize_image_tag("has space")
+
+
+class TestParseEnvFromConfig:
+    def test_env_dict(self, tmp_path):
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({"env": {"FOO": "bar", "NUM": 42}}, config_path)
+        assert result == {"FOO": "bar", "NUM": "42"}
+
+    def test_env_string_dotenv_file(self, tmp_path):
+        env_file = tmp_path / "my.env"
+        env_file.write_text("KEY1=val1\nKEY2=val2\n")
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({"env": "my.env"}, config_path)
+        assert result == {"KEY1": "val1", "KEY2": "val2"}
+
+    def test_env_missing_falls_back_to_dotenv(self, tmp_path, monkeypatch):
+        env_file = tmp_path / ".env"
+        env_file.write_text("DEFAULT_KEY=default_val\n")
+        monkeypatch.chdir(tmp_path)
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({}, config_path)
+        assert result == {"DEFAULT_KEY": "default_val"}
+
+    def test_env_empty_dict_falls_back_to_dotenv(self, tmp_path, monkeypatch):
+        """validate_config defaults env to {}, should still fall back to .env."""
+        env_file = tmp_path / ".env"
+        env_file.write_text("MY_KEY=my_val\n")
+        monkeypatch.chdir(tmp_path)
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({"env": {}}, config_path)
+        assert result == {"MY_KEY": "my_val"}
+
+    def test_env_missing_no_dotenv_returns_empty(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({}, config_path)
+        assert result == {}
+
+    def test_env_dotenv_filters_none_values(self, tmp_path):
+        # Lines like "KEY=" produce empty string, lines like "KEY" produce None
+        env_file = tmp_path / "test.env"
+        env_file.write_text("GOOD=value\nEMPTY=\n")
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({"env": "test.env"}, config_path)
+        assert "GOOD" in result
+        assert result["GOOD"] == "value"
+        # EMPTY= gives empty string, not None, so it should be present
+        assert result["EMPTY"] == ""

libs/cli/tests/unit_tests/test_host_backend.py

@@ -0,0 +1,178 @@
+import httpx
+import pytest
+
+from langgraph_cli.host_backend import HostBackendClient, HostBackendError
+
+
+@pytest.fixture
+def mock_transport():
+    return httpx.MockTransport(lambda req: httpx.Response(200, json={"ok": True}))
+
+
+@pytest.fixture
+def client(mock_transport):
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=mock_transport,
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    return c
+
+
+def test_constructor_strips_trailing_slash():
+    c = HostBackendClient("https://api.example.com/", "key")
+    assert str(c._client.base_url) == "https://api.example.com"
+
+
+def test_constructor_empty_url_raises():
+    with pytest.raises(Exception, match="Host backend URL is required"):
+        HostBackendClient("", "key")
+
+
+def test_request_sends_headers():
+    def handler(req: httpx.Request) -> httpx.Response:
+        assert req.headers["x-api-key"] == "test-key"
+        assert req.headers["accept"] == "application/json"
+        return httpx.Response(200, json={"ok": True})
+
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    result = c._request("GET", "/test")
+    assert result == {"ok": True}
+
+
+def test_request_sends_json_payload():
+    def handler(req: httpx.Request) -> httpx.Response:
+        assert req.headers["content-type"] == "application/json"
+        assert req.content == b'{"key":"value"}'
+        return httpx.Response(200, json={"created": True})
+
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    result = c._request("POST", "/test", {"key": "value"})
+    assert result == {"created": True}
+
+
+def test_request_empty_body_returns_none():
+    transport = httpx.MockTransport(lambda req: httpx.Response(200, content=b""))
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=transport,
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    assert c._request("DELETE", "/test") is None
+
+
+def test_request_http_error_raises():
+    transport = httpx.MockTransport(lambda req: httpx.Response(404, text="not found"))
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=transport,
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    with pytest.raises(HostBackendError, match="404"):
+        c._request("GET", "/missing")
+
+
+def test_request_invalid_json_raises():
+    transport = httpx.MockTransport(
+        lambda req: httpx.Response(200, content=b"not json")
+    )
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=transport,
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    with pytest.raises(HostBackendError, match="Failed to decode"):
+        c._request("GET", "/bad-json")
+
+
+def test_request_transport_error_raises():
+    def handler(req: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("connection refused")
+
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    with pytest.raises(HostBackendError, match="connection refused"):
+        c._request("GET", "/test")
+
+
+def test_create_deployment(client):
+    result = client.create_deployment({"name": "my-deploy"})
+    assert result == {"ok": True}
+
+
+def test_get_deployment(client):
+    result = client.get_deployment("dep-123")
+    assert result == {"ok": True}
+
+
+def test_list_deployments(client):
+    result = client.list_deployments("my-app")
+    assert result == {"ok": True}
+
+
+def test_list_deployments_without_filter_uses_base_path():
+    def handler(req: httpx.Request) -> httpx.Response:
+        assert str(req.url) == "https://api.example.com/v2/deployments"
+        return httpx.Response(200, json={"ok": True})
+
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    result = c.list_deployments()
+    assert result == {"ok": True}
+
+
+def test_request_push_token(client):
+    result = client.request_push_token("dep-123")
+    assert result == {"ok": True}
+
+
+def test_update_deployment(client):
+    result = client.update_deployment(
+        "dep-123", "image:latest", secrets=[{"name": "KEY", "value": "val"}]
+    )
+    assert result == {"ok": True}
+
+
+def test_update_deployment_no_secrets(client):
+    result = client.update_deployment("dep-123", "image:latest")
+    assert result == {"ok": True}
+
+
+def test_list_revisions(client):
+    result = client.list_revisions("dep-123", limit=5)
+    assert result == {"ok": True}
+
+
+def test_get_revision(client):
+    result = client.get_revision("dep-123", "rev-456")
+    assert result == {"ok": True}

libs/cli/uv.lock

@@ -983,7 +983,9 @@ name = "langgraph-cli"
 source = { editable = "." }
 dependencies = [
     { name = "click" },
+    { name = "httpx" },
     { name = "langgraph-sdk", marker = "python_full_version >= '3.11'" },
+    { name = "python-dotenv" },
 ]
 
 [package.optional-dependencies]
@@ -1021,10 +1023,11 @@ test = [
 [package.metadata]
 requires-dist = [
     { name = "click", specifier = ">=8.1.7" },
+    { name = "httpx", specifier = ">=0.24.0" },
     { name = "langgraph-api", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.5.35,<0.8.0" },
     { name = "langgraph-runtime-inmem", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.7" },
     { name = "langgraph-sdk", marker = "python_full_version >= '3.11'", specifier = ">=0.1.0" },
-    { name = "python-dotenv", marker = "extra == 'inmem'", specifier = ">=0.8.0" },
+    { name = "python-dotenv", specifier = ">=0.8.0" },
 ]
 provides-extras = ["inmem"]
 

libs/langgraph/uv.lock

@@ -1689,23 +1689,25 @@ name = "langgraph-cli"
 source = { editable = "../cli" }
 dependencies = [
     { name = "click", marker = "python_full_version < '3.14'" },
+    { name = "httpx", marker = "python_full_version < '3.14'" },
     { name = "langgraph-sdk", marker = "python_full_version >= '3.11' and python_full_version < '3.14'" },
+    { name = "python-dotenv", marker = "python_full_version < '3.14'" },
 ]
 
 [package.optional-dependencies]
 inmem = [
     { name = "langgraph-api", marker = "python_full_version >= '3.11' and python_full_version < '3.14'" },
     { name = "langgraph-runtime-inmem", marker = "python_full_version >= '3.11' and python_full_version < '3.14'" },
-    { name = "python-dotenv", marker = "python_full_version < '3.14'" },
 ]
 
 [package.metadata]
 requires-dist = [
     { name = "click", specifier = ">=8.1.7" },
+    { name = "httpx", specifier = ">=0.24.0" },
     { name = "langgraph-api", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.5.35,<0.8.0" },
     { name = "langgraph-runtime-inmem", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.7" },
     { name = "langgraph-sdk", marker = "python_full_version >= '3.11'", specifier = ">=0.1.0" },
-    { name = "python-dotenv", marker = "extra == 'inmem'", specifier = ">=0.8.0" },
+    { name = "python-dotenv", specifier = ">=0.8.0" },
 ]
 provides-extras = ["inmem"]