diff --git a/libs/cli/langgraph_cli/deploy.py b/libs/cli/langgraph_cli/deploy.py index 25909338f..17665c807 100644 --- a/libs/cli/langgraph_cli/deploy.py +++ b/libs/cli/langgraph_cli/deploy.py @@ -115,6 +115,172 @@ class BuildResult: show_build_logs_on_failure: bool = False +# --------------------------------------------------------------------------- +# Structured output emitter +# --------------------------------------------------------------------------- + +_emitter: "_Emitter | None" = None +_no_input: bool = False + + +class _Emitter: + """Dual-mode output: JSON-lines (``--json``) or human-readable click text.""" + + def __init__(self, json_mode: bool) -> None: + self._json = json_mode + + @property + def json_mode(self) -> bool: + return self._json + + # -- Structured event helpers ------------------------------------------ + + def step(self, step: int, message: str, **extra: object) -> None: + if self._json: + self._write({"event": "step", "step": step, "message": message, **extra}) + else: + click.secho(f"{step}. {message}", fg="cyan") + + def info(self, message: str, **extra: object) -> None: + if self._json: + self._write({"event": "info", "message": message, **extra}) + else: + click.secho(f" {message}", fg="green") + + def warn(self, message: str, **extra: object) -> None: + """Warning nested under a step. Text mode indents; JSON mode strips leading whitespace.""" + if self._json: + self._write({"event": "warn", "message": message.lstrip(), **extra}) + else: + click.secho(f" {message}", fg="yellow") + + def note(self, message: str, **extra: object) -> None: + """Top-level banner (pre-step). Text mode does not indent.""" + if self._json: + self._write({"event": "note", "message": message, **extra}) + else: + click.secho(message, fg="yellow") + + def error(self, message: str, **extra: object) -> None: + if self._json: + self._write({"event": "error", "message": message, **extra}) + else: + click.secho(f" {message}", fg="red") + + def status_change( + self, + status: str, + elapsed_seconds: float, + finished: bool = False, + ) -> None: + mins, secs = divmod(int(elapsed_seconds), 60) + elapsed_str = f"{mins}m {secs:02d}s" if mins else f"{secs}s" + if self._json: + self._write( + { + "event": "status_change", + "status": status, + "elapsed_seconds": round(elapsed_seconds, 1), + "message": f"{status}... ({elapsed_str})", + } + ) + else: + click.echo(f" {status}... ({elapsed_str})") + + def log(self, message: str) -> None: + if self._json: + self._write({"event": "log", "message": message}) + else: + click.echo(f" | {message}") + + def status_url(self, url: str) -> None: + if self._json: + self._write({"event": "status_url", "url": url}) + else: + click.secho(f" View status: {url}", fg="cyan") + + def result( + self, + status: str, + *, + deployment_id: str, + url: str | None = None, + status_url: str | None = None, + fallback_status_message: str | None = None, + ) -> None: + if self._json: + if status == "succeeded": + message = "Deployment successful!" + elif status == "failed": + message = "Deployment failed" + else: + message = "Timed out waiting for deployment." + payload: dict = { + "event": "result", + "status": status, + "deployment_id": deployment_id, + "message": message, + } + if url: + payload["url"] = url + if status_url: + payload["status_url"] = status_url + self._write(payload) + else: + if status == "succeeded": + click.secho(" Deployment successful!", fg="green") + if url: + click.secho(f" URL: {url}", fg="green") + if status_url: + click.secho(f" View status: {status_url}", fg="green") + elif status == "failed": + click.secho(" Deployment failed", fg="red") + if status_url: + click.secho(f" View status: {status_url}", fg="red") + elif status == "timed_out": + click.secho(" Timed out waiting for deployment.", fg="yellow") + if status_url: + click.secho(f" Check status at: {status_url}", fg="yellow") + elif fallback_status_message: + click.secho(f" {fallback_status_message}", fg="yellow") + + def heartbeat(self, status: str, elapsed_seconds: float) -> None: + if self._json: + mins, secs = divmod(int(elapsed_seconds), 60) + elapsed_str = f"{mins}m {secs:02d}s" if mins else f"{secs}s" + self._write( + { + "event": "heartbeat", + "status": status, + "elapsed_seconds": round(elapsed_seconds, 1), + "message": f"{status}... ({elapsed_str})", + } + ) + + def upload_progress(self, size_mb: float, pct: int) -> None: + if self._json: + self._write( + { + "event": "upload_progress", + "size_mb": round(size_mb, 1), + "pct": pct, + } + ) + else: + click.echo(f"\r Uploading ({size_mb:.1f} MB)... {pct}%", nl=False) + + def _write(self, obj: dict) -> None: + import sys as _sys + + _sys.stdout.write(json_mod.dumps(obj, default=str) + "\n") + _sys.stdout.flush() + + +def _get_emitter() -> _Emitter: + """Return the module-level emitter (falls back to text mode).""" + return _emitter or _Emitter(json_mode=False) + + # --------------------------------------------------------------------------- # Validators # --------------------------------------------------------------------------- @@ -172,15 +338,16 @@ def find_deployment_id_by_name( # --------------------------------------------------------------------------- -def normalize_image_name(value: str | None) -> str: - """Sanitize a deployment/directory name into a valid Docker repository name. +def normalize_name(value: str | None) -> str: + """Sanitize a deployment/directory name into a valid deployment name. - Docker repository names must be lowercase and may only contain - [a-z0-9._-]. Invalid characters are replaced with hyphens. + LangSmith Deployment names only allow lowercase + alphanumeric characters and hyphens ([a-z0-9-]). + Invalid characters are replaced with hyphens. """ if not value: return "app" - slug = re.sub(r"[^a-z0-9._-]+", "-", value.lower()).strip("-.") + slug = re.sub(r"[^a-z0-9-]+", "-", value.lower()).strip("-") return slug or "app" @@ -307,9 +474,8 @@ def _resolve_env_path( 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", + _get_emitter().note( + f"Warning: env file '{env_field}' specified in langgraph.json not found." ) return None return env_path @@ -343,7 +509,7 @@ def _secrets_from_env( 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") + _get_emitter().note(f"Skipping reserved env var: {name}") continue if not value: continue @@ -386,8 +552,8 @@ def _resolve_build_mode( # --------------------------------------------------------------------------- -def _log_deploy_step(step: int, message: str) -> None: - click.secho(f"{step}. {message}", fg="cyan") +def _log_deploy_step(step: int, message: str, **extra: object) -> None: + _get_emitter().step(step, message, **extra) def _resolve_deployment( @@ -411,12 +577,13 @@ def _resolve_deployment( found_id = _call_host_backend_with_optional_tenant( client, lambda c: find_deployment_id_by_name(c, name) ) + em = _get_emitter() if found_id: deployment_id = str(found_id) - click.secho(f" Found existing deployment (ID: {deployment_id})", fg="green") + em.info(f"Found existing deployment (ID: {deployment_id})") else: needs_creation = True - click.secho(not_found_message, fg="yellow") + em.warn(not_found_message) return deployment_id, needs_creation, step + 1 @@ -444,7 +611,7 @@ def _create_deployment( raise HostBackendError( "POST /v2/deployments succeeded but response missing a valid 'id'" ) - click.secho(f" Deployment ID: {created_id}", fg="green") + _get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id) return created_id, step + 1 @@ -458,21 +625,36 @@ def _smith_dashboard_base_url(host_url: str | None) -> str: hostname = parsed.hostname or "" if hostname in ("localhost", "127.0.0.1"): return host_url.rstrip("/") - if hostname.startswith("eu."): - return "https://eu.smith.langchain.com" + + api_host_suffix = "api.host.langchain.com" + if hostname == api_host_suffix: + return "https://smith.langchain.com" + if hostname.endswith(f".{api_host_suffix}"): + prefix = hostname[: -(len(api_host_suffix) + 1)] + return f"https://{prefix}.smith.langchain.com" + return "https://smith.langchain.com" -def _print_deployment_status_url( +def _get_deployment_status_url( updated: object, deployment_id: str, host_url: str | None = None -) -> None: - """Print the deployment status URL when tenant metadata is available.""" +) -> str | None: + """Compute the LangSmith dashboard URL for a deployment, if possible.""" tenant_id = updated.get("tenant_id") if isinstance(updated, dict) else None if not tenant_id: - return + return None base = _smith_dashboard_base_url(host_url) - status_url = f"{base}/o/{tenant_id}/host/deployments/{deployment_id}" - click.secho(f" View status: {status_url}", fg="cyan") + return f"{base}/o/{tenant_id}/host/deployments/{deployment_id}" + + +def _emit_deployment_status_url( + updated: object, deployment_id: str, host_url: str | None = None +) -> str | None: + """Emit the deployment status URL and return it.""" + url = _get_deployment_status_url(updated, deployment_id, host_url) + if url: + _get_emitter().status_url(url) + return url def _poll_revision_status( @@ -486,6 +668,7 @@ def _poll_revision_status( on_interrupt: Callable[[str], None] | None = None, ) -> tuple[str, str | None]: """Poll latest revision status until terminal status or timeout.""" + em = _get_emitter() revisions_resp = client.list_revisions(deployment_id, limit=1) resources = ( revisions_resp.get("resources", []) if isinstance(revisions_resp, dict) else [] @@ -497,7 +680,11 @@ def _poll_revision_status( last_status = "" deadline = time.time() + timeout_seconds start_time = time.monotonic() - with Progress(message=progress_message, elapsed=True) as set_progress: + last_heartbeat = start_time + json_mode = em.json_mode + with Progress( + message=progress_message, elapsed=True, json_mode=json_mode + ) as set_progress: while time.time() < deadline: try: rev = client.get_revision(deployment_id, revision_id) @@ -514,14 +701,15 @@ def _poll_revision_status( if status != last_status: set_progress("") if last_status: - elapsed = time.monotonic() - start_time - mins, secs = divmod(int(elapsed), 60) - elapsed_str = f"{mins}m {secs:02d}s" if mins else f"{secs}s" - click.echo(f" {last_status}... ({elapsed_str})") + em.status_change(last_status, time.monotonic() - start_time) last_status = status if status in _TERMINAL_STATUSES: break set_progress(f"{status}...") + last_heartbeat = time.monotonic() + elif json_mode and time.monotonic() - last_heartbeat > 10: + em.heartbeat(last_status, time.monotonic() - start_time) + last_heartbeat = time.monotonic() if on_poll is not None: on_poll(status, revision_id, set_progress) @@ -538,8 +726,10 @@ def _print_deployment_result( last_status: str, *, dashboard_label: str, + status_url: str | None = None, ) -> None: """Print final deployment status and raise on failure.""" + em = _get_emitter() dep_info = client.get_deployment(deployment_id) custom_url = None if isinstance(dep_info, dict): @@ -548,24 +738,28 @@ def _print_deployment_result( 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") + em.result( + "succeeded", + deployment_id=deployment_id, + url=custom_url, + status_url=status_url, + ) elif last_status in ("BUILD_FAILED", "DEPLOY_FAILED", "CREATE_FAILED"): - click.secho(f" Deployment failed: {last_status}", fg="red") + em.result( + "failed", + deployment_id=deployment_id, + status_url=status_url, + ) raise click.exceptions.Exit(1) else: - click.secho( - f" Timed out waiting for deployment (last status: {last_status}).", - fg="yellow", + em.result( + "timed_out", + deployment_id=deployment_id, + status_url=status_url, + fallback_status_message=( + f"Check status in the LangSmith {dashboard_label}." + ), ) - if custom_url: - click.secho(f" Check status at: {custom_url}", fg="yellow") - else: - click.secho( - f" Check status in the LangSmith {dashboard_label}.", - fg="yellow", - ) # --------------------------------------------------------------------------- @@ -594,14 +788,16 @@ def _docker_config_for_token(registry_host: str, token: str): # --------------------------------------------------------------------------- _UPLOAD_TIMEOUT_SECONDS = 300 +_BYTES_PER_MIB = 1_048_576 class _ProgressReader: - """File-like wrapper that displays upload progress via click.""" + """File-like wrapper that reports upload progress via the emitter.""" - def __init__(self, fobj, file_size: int): + def __init__(self, fobj, file_size: int, emitter: "_Emitter"): self._fobj = fobj self._file_size = file_size + self._emitter = emitter self._uploaded = 0 def read(self, size=-1): @@ -611,10 +807,7 @@ class _ProgressReader: pct = ( int(self._uploaded * 100 / self._file_size) if self._file_size else 100 ) - click.echo( - f"\r Uploading ({self._file_size / 1_048_576:.1f} MB)... {pct}%", - nl=False, - ) + self._emitter.upload_progress(self._file_size / _BYTES_PER_MIB, pct) return data def __len__(self): @@ -626,10 +819,13 @@ def _upload_to_gcs(signed_url: str, file_path: str, file_size: int) -> None: import urllib.error import urllib.request + em = _get_emitter() + with open(file_path, "rb") as f: + reader = _ProgressReader(f, file_size, em) req = urllib.request.Request( signed_url, - data=_ProgressReader(f, file_size), + data=reader, method="PUT", headers={ "Content-Type": "application/gzip", @@ -644,7 +840,8 @@ def _upload_to_gcs(signed_url: str, file_path: str, file_size: int) -> None: raise click.ClickException( f"Upload failed with status {err.code}: {detail}" ) from None - click.echo() + if not em.json_mode: + click.echo() # --------------------------------------------------------------------------- @@ -752,7 +949,7 @@ def _run_local_build( 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) + repo_name = normalize_name(repo_seed) tag_value = normalize_image_tag(tag) remote_image = f"{normalized_registry}/{repo_name}:{tag_value}" @@ -811,9 +1008,8 @@ def _run_local_build( 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", + _get_emitter().warn( + f" Push failed, retrying (attempt {attempt + 2} of {max_push_retries})..." ) else: raise @@ -847,9 +1043,10 @@ def _run_remote_build( """Upload source tarball and trigger a remote build.""" from langgraph_cli.archive import create_archive + em = _get_emitter() _log_deploy_step(step, "Creating source archive") with create_archive(config, config_json) as (archive_path, file_size, config_rel): - click.secho(f" Archive created ({file_size / 1_048_576:.1f} MB)", fg="green") + em.info(f"Archive created ({file_size / _BYTES_PER_MIB:.1f} MB)") step += 1 _log_deploy_step(step, "Requesting upload URL") @@ -897,12 +1094,12 @@ def _run_remote_build( if has_output: set_progress("") if not logs_header_printed: - click.echo(f" {status} (build logs):") + em.info(f"{status} (build logs):") logs_header_printed = True for entry in entries: msg = entry.get("message", "") if msg: - click.echo(f" | {msg}") + em.log(msg) log_offset = logs_resp.get("next_offset") or log_offset if has_output: set_progress(f"{status}...") @@ -910,11 +1107,10 @@ def _run_remote_build( pass def _handle_interrupt(revision_id: str) -> None: - click.secho( - f"\n Interrupted. Deployment ID: {deployment_id}, Revision ID: {revision_id}", - fg="yellow", + em.warn( + f"\nInterrupted. Deployment ID: {deployment_id}, Revision ID: {revision_id}" ) - click.secho(" The build will continue remotely.", fg="yellow") + em.warn("The build will continue remotely.") return BuildResult( updated=updated if isinstance(updated, dict) else {}, @@ -952,12 +1148,20 @@ def _create_host_backend_client( resolved_api_key = val break if not resolved_api_key: + if _no_input: + raise click.ClickException( + "No LangSmith API key found. Set LANGSMITH_API_KEY in the " + "environment or .env file." + ) click.secho( "No LangSmith API key found. Create one at Settings > API Keys in LangSmith.", fg="yellow", ) resolved_api_key = click.prompt("Enter LangSmith API key", hide_input=True) - return HostBackendClient(host_url, resolved_api_key) + tenant_id = env_vars.get("LANGSMITH_TENANT_ID") or os.environ.get( + "LANGSMITH_TENANT_ID" + ) + return HostBackendClient(host_url, resolved_api_key, tenant_id=tenant_id) def _call_host_backend_with_optional_tenant( @@ -982,6 +1186,12 @@ def _call_host_backend_with_optional_tenant( and err.status_code == 403 and "requires workspace specification" in err.message ): + if _no_input: + raise click.ClickException( + "API key is org-scoped and requires a workspace ID. " + "Set LANGSMITH_TENANT_ID in your .env file or " + "use a workspace-scoped API key." + ) from None click.secho( "Your API key is org-scoped and requires a workspace ID.", fg="yellow", @@ -1189,6 +1399,19 @@ def _deploy_base_options( "if Docker is not available locally." ), ), + click.option( + "--json", + "json_output", + is_flag=True, + default=False, + help="Emit structured JSON-lines to stdout instead of human-readable text.", + ), + click.option( + "--no-input", + is_flag=True, + default=False, + help="Never prompt for input; fail with an error if a required value is missing.", + ), ] if include_docker_args: # Only attach build args to the default command; on the group they @@ -1256,36 +1479,50 @@ def _deploy_cmd( no_wait: bool, remote_build_flag: bool | None, docker_build_args: Sequence[str], + json_output: bool, + no_input: bool, ): - click.secho( - "Note: 'langgraph deploy' is in beta. Expect frequent updates and improvements.", - fg="yellow", + global _emitter, _no_input + _emitter = _Emitter(json_mode=json_output) + _no_input = no_input + em = _emitter + + em.note( + "Note: 'langgraph deploy' is in beta. Expect frequent updates and improvements." ) - click.echo() + if not json_output: + click.echo() # -- 1. Preflight -- validate_deploy_commands(install_command, build_command) config_json = langgraph_cli.config.validate_config_file(config) - warn_non_wolfi_distro(config_json) + warn_non_wolfi_distro(config_json, emit=em.note) env_vars = _parse_env_from_config(config_json, config) 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) - env_path = _resolve_env_path(config_json, config) - if env_path is not None: - set_key(str(env_path), _DEPLOYMENT_NAME_ENV, name) - click.echo(f"Saved deployment name to {env_path}") + default_name = normalize_name(pathlib.Path.cwd().name) + if no_input: + name = default_name + else: + name = click.prompt("Deployment name", default=default_name) + if name and not deployment_id: + name = normalize_name(name) + if not no_input: + env_path = _resolve_env_path(config_json, config) + if env_path is not None: + set_key(str(env_path), _DEPLOYMENT_NAME_ENV, name) + em.info(f"Saved deployment name to {env_path}") secrets = _secrets_from_env(_env_without_deployment_name(env_vars)) use_remote_build, local_build_error = _resolve_build_mode(remote_build_flag) if use_remote_build and remote_build_flag is None and local_build_error: - click.secho(f"{local_build_error}\nUsing remote build instead.", fg="yellow") - click.echo() + em.note(f"{local_build_error}\nUsing remote build instead.") + if not json_output: + click.echo() # -- 2. Resolve / create deployment -- client = _create_host_backend_client(host_url, api_key, env_vars=env_vars) @@ -1297,9 +1534,9 @@ def _deploy_cmd( deployment_id, name, not_found_message=( - " No deployment found. Will create." + "No deployment found. Will create." if use_remote_build - else " No deployment found. Will create after build." + else "No deployment found. Will create after build." ), ) @@ -1350,10 +1587,14 @@ def _deploy_cmd( ) # -- 4. Shared wait + result -- - _print_deployment_status_url(build_result.updated, deployment_id, host_url) + dep_status_url = _emit_deployment_status_url( + build_result.updated, + deployment_id, + host_url, + ) if no_wait: - click.secho(f" {build_result.no_result_message}", fg="green") + em.info(build_result.no_result_message) return last_status, revision_id = _poll_revision_status( @@ -1366,7 +1607,7 @@ def _deploy_cmd( on_interrupt=build_result.on_interrupt, ) if not last_status: - click.secho(f" {build_result.no_result_message}", fg="green") + em.info(build_result.no_result_message) return if ( @@ -1375,7 +1616,7 @@ def _deploy_cmd( and not verbose and revision_id is not None ): - click.secho(" Last build log lines:", fg="red") + em.error("Last build log lines:") try: logs_resp = client.get_build_logs( deployment_id, @@ -1387,19 +1628,17 @@ def _deploy_cmd( for entry in entries: msg = entry.get("message", "") if msg: - click.echo(f" | {msg}") + em.log(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", - ) + em.error("(failed to fetch build logs)") + em.warn("Re-run with --verbose to see full build output.") _print_deployment_result( client, deployment_id, last_status, dashboard_label="Deployment dashboard", + status_url=dep_status_url, ) diff --git a/libs/cli/langgraph_cli/progress.py b/libs/cli/langgraph_cli/progress.py index 60b55db2a..f0b60c1a8 100644 --- a/libs/cli/langgraph_cli/progress.py +++ b/libs/cli/langgraph_cli/progress.py @@ -12,10 +12,11 @@ class Progress: while True: yield from "|/-\\" - def __init__(self, *, message="", elapsed: bool = False): + def __init__(self, *, message="", elapsed: bool = False, json_mode: bool = False): self.message = message self._base_message = message self._show_elapsed = elapsed + self._json_mode = json_mode # use this to make sure we don't kill thread when we set msg to "" self._stop = threading.Event() # signalled when the spinner has no text on screen @@ -69,6 +70,9 @@ class Progress: self._line_clear.set() def __enter__(self) -> Callable[[str], None]: + if self._json_mode: + return lambda message: None + if sys.stdout.isatty(): self.thread = threading.Thread(target=self.spinner_task) self.thread.start() @@ -90,6 +94,8 @@ class Progress: return set_message def __exit__(self, exception, value, tb): + if self._json_mode: + return if sys.stdout.isatty(): self.message = "" self._stop.set() diff --git a/libs/cli/langgraph_cli/util.py b/libs/cli/langgraph_cli/util.py index ad9b9c39b..0dae6a5ff 100644 --- a/libs/cli/langgraph_cli/util.py +++ b/libs/cli/langgraph_cli/util.py @@ -1,5 +1,7 @@ """General-purpose utilities shared across the LangGraph CLI.""" +from collections.abc import Callable + import click @@ -7,21 +9,42 @@ def clean_empty_lines(input_str: str): return "\n".join(filter(None, input_str.splitlines())) -def warn_non_wolfi_distro(config_json: dict) -> None: - """Show warning if image_distro is not set to 'wolfi'.""" +def warn_non_wolfi_distro( + config_json: dict, + *, + emit: Callable[[str], None] | None = None, +) -> None: + """Show warning if image_distro is not set to 'wolfi'. + + When ``emit`` is provided, each warning line is sent through it (used by + callers that need JSON-aware output). Otherwise falls back to colored + ``click.secho`` output. + """ image_distro = config_json.get("image_distro", "debian") # Default is debian - if image_distro != "wolfi": - click.secho( - "⚠️ Security Recommendation: Consider switching to Wolfi Linux for enhanced security.", - fg="yellow", - bold=True, + if image_distro == "wolfi": + return + if emit is not None: + emit( + "⚠️ Security Recommendation: Consider switching to Wolfi Linux for enhanced security." ) - click.secho( - " Wolfi is a security-oriented, minimal Linux distribution designed for containers.", - fg="yellow", + emit( + " Wolfi is a security-oriented, minimal Linux distribution designed for containers." ) - click.secho( - ' To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.', - fg="yellow", + emit( + ' To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.' ) - click.secho("") # Empty line for better readability + return + click.secho( + "⚠️ Security Recommendation: Consider switching to Wolfi Linux for enhanced security.", + fg="yellow", + bold=True, + ) + click.secho( + " Wolfi is a security-oriented, minimal Linux distribution designed for containers.", + fg="yellow", + ) + click.secho( + ' To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.', + fg="yellow", + ) + click.secho("") # Empty line for better readability diff --git a/libs/cli/tests/unit_tests/test_deploy_helpers.py b/libs/cli/tests/unit_tests/test_deploy_helpers.py index d7a245074..b4c478f2f 100644 --- a/libs/cli/tests/unit_tests/test_deploy_helpers.py +++ b/libs/cli/tests/unit_tests/test_deploy_helpers.py @@ -1,6 +1,8 @@ import base64 +import io import json import os +import sys import click import httpx @@ -8,12 +10,15 @@ import pytest from langgraph_cli.deploy import ( _call_host_backend_with_optional_tenant, + _create_host_backend_client, _docker_config_for_token, + _Emitter, _env_without_deployment_name, _parse_env_from_config, _resolve_env_path, - normalize_image_name, + _smith_dashboard_base_url, normalize_image_tag, + normalize_name, ) from langgraph_cli.host_backend import HostBackendClient, HostBackendError @@ -40,30 +45,33 @@ class TestDockerConfigForToken: assert "gcr.io" in data["auths"] -class TestNormalizeImageName: +class TestNormalizeName: def test_simple_name(self): - assert normalize_image_name("myapp") == "myapp" + assert normalize_name("myapp") == "myapp" def test_uppercase_lowered(self): - assert normalize_image_name("MyApp") == "myapp" + assert normalize_name("MyApp") == "myapp" def test_special_chars_replaced(self): - assert normalize_image_name("my app!@#v2") == "my-app-v2" + assert normalize_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_dots_replaced_with_hyphens(self): + assert normalize_name("my-app.v2") == "my-app-v2" + + def test_underscores_replaced_with_hyphens(self): + assert normalize_name("simple_graph_name") == "simple-graph-name" def test_leading_trailing_stripped(self): - assert normalize_image_name("--my-app..") == "my-app" + assert normalize_name("--my-app..") == "my-app" def test_empty_string_returns_app(self): - assert normalize_image_name("") == "app" + assert normalize_name("") == "app" def test_none_returns_app(self): - assert normalize_image_name(None) == "app" + assert normalize_name(None) == "app" def test_all_invalid_chars_returns_app(self): - assert normalize_image_name("!!!") == "app" + assert normalize_name("!!!") == "app" class TestNormalizeImageTag: @@ -271,3 +279,256 @@ class TestCallHostBackendWithOptionalTenant: _call_host_backend_with_optional_tenant( client, lambda c: c.list_deployments() ) + + def test_workspace_prompt_blocked_by_no_input(self, monkeypatch): + """With _no_input=True, 403 requiring workspace should raise ClickException.""" + import langgraph_cli.deploy as deploy_mod + + monkeypatch.setattr(deploy_mod, "_no_input", True) + + requires_workspace = '{"detail":"requires workspace specification"}' + client = self._make_client( + lambda req: httpx.Response(403, text=requires_workspace) + ) + with pytest.raises(click.ClickException, match="workspace"): + _call_host_backend_with_optional_tenant( + client, lambda c: c.list_deployments() + ) + + +# --------------------------------------------------------------------------- +# _Emitter JSON mode +# --------------------------------------------------------------------------- + + +class TestEmitterJsonMode: + """Verify that _Emitter in json_mode writes valid JSON-lines to stdout.""" + + def _capture(self, fn): + """Run fn with stdout captured and return parsed JSON objects.""" + buf = io.StringIO() + old = sys.stdout + sys.stdout = buf + try: + fn() + finally: + sys.stdout = old + lines = [line for line in buf.getvalue().splitlines() if line.strip()] + return [json.loads(line) for line in lines] + + def test_step_event(self): + em = _Emitter(json_mode=True) + events = self._capture(lambda: em.step(1, "Building image")) + assert len(events) == 1 + assert events[0]["event"] == "step" + assert events[0]["step"] == 1 + assert events[0]["message"] == "Building image" + + def test_info_event(self): + em = _Emitter(json_mode=True) + events = self._capture(lambda: em.info("All good")) + assert events[0]["event"] == "info" + assert events[0]["message"] == "All good" + + def test_warn_event(self): + em = _Emitter(json_mode=True) + events = self._capture(lambda: em.warn("Careful")) + assert events[0]["event"] == "warn" + + def test_error_event(self): + em = _Emitter(json_mode=True) + events = self._capture(lambda: em.error("Boom")) + assert events[0]["event"] == "error" + assert events[0]["message"] == "Boom" + + def test_status_change_event(self): + em = _Emitter(json_mode=True) + events = self._capture(lambda: em.status_change("building", 12.345)) + assert events[0]["event"] == "status_change" + assert events[0]["status"] == "building" + assert events[0]["elapsed_seconds"] == 12.3 + assert events[0]["message"] == "building... (12s)" + + def test_status_change_event_with_minutes(self): + em = _Emitter(json_mode=True) + events = self._capture(lambda: em.status_change("deploying", 95.0)) + assert events[0]["message"] == "deploying... (1m 35s)" + + def test_log_event(self): + em = _Emitter(json_mode=True) + events = self._capture(lambda: em.log("some output")) + assert events[0] == {"event": "log", "message": "some output"} + + def test_status_url_event(self): + em = _Emitter(json_mode=True) + events = self._capture( + lambda: em.status_url("https://smith.langchain.com/deploy/123") + ) + assert events[0]["event"] == "status_url" + assert events[0]["url"] == "https://smith.langchain.com/deploy/123" + + def test_result_event_full(self): + em = _Emitter(json_mode=True) + events = self._capture( + lambda: em.result( + "succeeded", + deployment_id="dep-1", + url="https://app.example.com", + status_url="https://smith.langchain.com/deploy/dep-1", + ) + ) + assert events[0]["event"] == "result" + assert events[0]["status"] == "succeeded" + assert events[0]["deployment_id"] == "dep-1" + assert events[0]["message"] == "Deployment successful!" + assert events[0]["url"] == "https://app.example.com" + assert events[0]["status_url"] == "https://smith.langchain.com/deploy/dep-1" + + def test_result_event_minimal(self): + em = _Emitter(json_mode=True) + events = self._capture(lambda: em.result("failed", deployment_id="dep-2")) + assert events[0]["event"] == "result" + assert events[0]["status"] == "failed" + assert events[0]["message"] == "Deployment failed" + assert "url" not in events[0] + assert "status_url" not in events[0] + + def test_heartbeat_event(self): + em = _Emitter(json_mode=True) + events = self._capture(lambda: em.heartbeat("building", 30.789)) + assert events[0]["event"] == "heartbeat" + assert events[0]["elapsed_seconds"] == 30.8 + assert events[0]["message"] == "building... (30s)" + + def test_heartbeat_silent_in_text_mode(self, capsys): + em = _Emitter(json_mode=False) + em.heartbeat("building", 10.0) + captured = capsys.readouterr() + assert captured.out == "" + + def test_upload_progress_event(self): + em = _Emitter(json_mode=True) + events = self._capture(lambda: em.upload_progress(5.678, 42)) + assert events[0]["event"] == "upload_progress" + assert events[0]["size_mb"] == 5.7 + assert events[0]["pct"] == 42 + + +# --------------------------------------------------------------------------- +# _Emitter text mode (non-json) +# --------------------------------------------------------------------------- + + +class TestEmitterTextMode: + """Verify that _Emitter in text mode uses click.echo/click.secho.""" + + def test_step_writes_text(self, capsys): + em = _Emitter(json_mode=False) + em.step(1, "Hello") + captured = capsys.readouterr() + assert "1. Hello" in captured.out + + def test_log_writes_text(self, capsys): + em = _Emitter(json_mode=False) + em.log("my line") + captured = capsys.readouterr() + assert "my line" in captured.out + + def test_result_succeeded_text(self, capsys): + em = _Emitter(json_mode=False) + em.result("succeeded", deployment_id="d1", url="https://app.test") + captured = capsys.readouterr() + lines = [line.strip() for line in captured.out.splitlines() if line.strip()] + assert "Deployment successful!" in lines + assert "URL: https://app.test" in lines + + +# --------------------------------------------------------------------------- +# --no-input guard on _create_host_backend_client +# --------------------------------------------------------------------------- + + +class TestCreateHostBackendClientNoInput: + def test_raises_when_no_api_key_and_no_input(self, monkeypatch, tmp_path): + import langgraph_cli.deploy as deploy_mod + + monkeypatch.setattr(deploy_mod, "_no_input", True) + monkeypatch.delenv("LANGSMITH_API_KEY", raising=False) + monkeypatch.delenv("LANGCHAIN_API_KEY", raising=False) + monkeypatch.delenv("LANGGRAPH_HOST_API_KEY", raising=False) + + with pytest.raises(click.ClickException, match="API key"): + _create_host_backend_client( + host_url="https://api.example.com", + api_key=None, + env_vars={}, + ) + + def test_succeeds_with_api_key_in_env(self, monkeypatch, tmp_path): + import langgraph_cli.deploy as deploy_mod + + monkeypatch.setattr(deploy_mod, "_no_input", True) + monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test") + + client = _create_host_backend_client( + host_url="https://api.example.com", + api_key=None, + env_vars={}, + ) + assert client is not None + + +class TestSmithDashboardBaseUrl: + def test_none_returns_default(self): + assert _smith_dashboard_base_url(None) == "https://smith.langchain.com" + + def test_empty_returns_default(self): + assert _smith_dashboard_base_url("") == "https://smith.langchain.com" + + def test_prod_host_url(self): + assert ( + _smith_dashboard_base_url("https://api.host.langchain.com") + == "https://smith.langchain.com" + ) + + def test_dev_host_url(self): + assert ( + _smith_dashboard_base_url("https://dev.api.host.langchain.com") + == "https://dev.smith.langchain.com" + ) + + def test_eu_host_url(self): + assert ( + _smith_dashboard_base_url("https://eu.api.host.langchain.com") + == "https://eu.smith.langchain.com" + ) + + def test_staging_host_url(self): + assert ( + _smith_dashboard_base_url("https://staging.api.host.langchain.com") + == "https://staging.smith.langchain.com" + ) + + def test_localhost(self): + assert ( + _smith_dashboard_base_url("http://localhost:8080") + == "http://localhost:8080" + ) + + def test_localhost_trailing_slash(self): + assert ( + _smith_dashboard_base_url("http://localhost:8080/") + == "http://localhost:8080" + ) + + def test_127_0_0_1(self): + assert ( + _smith_dashboard_base_url("http://127.0.0.1:3000") + == "http://127.0.0.1:3000" + ) + + def test_unknown_domain_returns_default(self): + assert ( + _smith_dashboard_base_url("https://custom.example.com") + == "https://smith.langchain.com" + )