Repository navigation
Expand file tree
/
Copy pathdeploy.py
More file actions
2805 lines (2461 loc) · 94.3 KB
/
Copy pathdeploy.py
File metadata and controls
2805 lines (2461 loc) · 94.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""Deploy command and subcommands for the LangGraph CLI."""
import base64
import json as json_mod
import os
import pathlib
import platform
import re
import tempfile
import time
from collections.abc import Callable, Mapping, Sequence
from contextlib import contextmanager
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from typing import Protocol, TypeVar
import click
import click.exceptions
from dotenv import dotenv_values, set_key
import langgraph_cli.config
from langgraph_cli.analytics import log_command
from langgraph_cli.config import Config
from langgraph_cli.constants import DEFAULT_CONFIG
from langgraph_cli.dependency_tracking import find_tracked_packages
from langgraph_cli.docker import build_docker_image, can_build_locally
from langgraph_cli.exec import CommandRunner, Runner, subp_exec
from langgraph_cli.host_backend import (
MAX_PAGE_SIZE,
ControlPlaneEndpoints,
HostBackendClient,
HostBackendError,
SourceName,
)
from langgraph_cli.image_reference import DIGEST_SEPARATOR, ImageReference
from langgraph_cli.progress import Progress
from langgraph_cli.util import warn_non_wolfi_distro
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
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",
)
_T = TypeVar("_T")
_DEPLOYMENT_NAME_ENV = "LANGSMITH_DEPLOYMENT_NAME"
_DEFAULT_IMAGE_TAG = "latest"
_DEPLOYMENT_PLATFORM = "linux/amd64"
_NATIVE_AMD64_MACHINE = "x86_64"
_PUSH_ATTEMPTS = 3
_LOCAL_BUILD_TAG_PREFIX = "langgraph-deploy-tmp"
_OPERATOR_DEFAULT_RESOURCE_SPEC: Mapping[str, object] = {}
_CUSTOMER_REGISTRY_SOURCE: SourceName = "external_docker"
_LISTENER_REQUIRED_MARKER = "listener_id' is required"
_LISTENERS_SHOWN = 10
_LISTENER_NOT_FOUND_STATUSES = frozenset({404, 422})
_LISTENERS_DOCS_URL = "https://docs.langchain.com/langsmith/control-plane#listeners"
_NO_LISTENERS = (
"This workspace has no listeners, so --listener-id and --k8s-namespace "
"do not apply."
)
_TERMINAL_STATUSES = frozenset(
[
"DEPLOYED",
"CREATE_FAILED",
"BUILD_FAILED",
"DEPLOY_FAILED",
"SKIPPED",
]
)
# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------
@dataclass
class BuildResult:
"""Captures the outcome of a build stage so the shared wait tail can be
parameterized identically for both local and remote builds."""
updated: dict = field(default_factory=dict)
progress_message: str = ""
timeout_seconds: int = 300
poll_interval_seconds: int = 1
no_result_message: str = "Deployment updated"
on_poll: Callable[[str, str, Callable[[str], None]], None] | None = None
on_interrupt: Callable[[str], None] | None = None
show_build_logs_on_failure: bool = False
@dataclass(frozen=True, slots=True)
class ById:
deployment_id: str
@dataclass(frozen=True, slots=True)
class ByName:
name: str
@dataclass(frozen=True, slots=True)
class ByAgent:
agent_id: str
environment: str
DeploymentSelector = ById | ByName | ByAgent
@dataclass(frozen=True, slots=True)
class Listener:
id: str
compute_id: str
namespaces: tuple[str, ...]
@classmethod
def from_resource(cls, resource: Mapping[str, object]) -> "Listener":
identifier = str(resource.get("id") or "")
if not identifier:
raise HostBackendError(
"The control plane returned a listener without an id."
)
compute_config = resource.get("compute_config")
namespaces = (
compute_config.get("k8s_namespaces")
if isinstance(compute_config, Mapping)
else None
)
return cls(
identifier,
str(resource.get("compute_id", "")),
tuple(str(namespace) for namespace in namespaces)
if isinstance(namespaces, list)
else (),
)
@dataclass(frozen=True, slots=True)
class Unplaced:
@property
def summary(self) -> str:
return ""
def source_config(self) -> dict[str, object]:
return {}
@dataclass(frozen=True, slots=True)
class OnListener:
listener_id: str
k8s_namespace: str
@property
def summary(self) -> str:
return (
f"Deploying through listener {self.listener_id} "
f"in namespace {self.k8s_namespace}"
)
def source_config(self) -> dict[str, object]:
return {
"listener_id": self.listener_id,
"listener_config": {"k8s_namespace": self.k8s_namespace},
}
Placement = Unplaced | OnListener
@dataclass(frozen=True, slots=True)
class RequestedPlacement:
listener_id: str | None = None
k8s_namespace: str | None = None
@property
def requested(self) -> bool:
return self.listener_id is not None or self.k8s_namespace is not None
def ensure_not_requested(self, deployment_id: str) -> None:
if self.requested:
raise click.UsageError(
"Listener and namespace are fixed when a deployment is created. "
f"Deployment {deployment_id} already exists, so drop --listener-id "
"and --k8s-namespace, or create a new deployment with a different "
"--name."
)
def on(self, listener: Listener) -> Placement:
return OnListener(listener.id, self._namespace(listener))
def among(self, listeners: Sequence[Listener]) -> Placement:
if not listeners:
if self.requested:
raise click.UsageError(_NO_LISTENERS)
return Unplaced()
if len(listeners) > 1:
raise click.UsageError(
"This workspace has several listeners. Choose one with "
f"--listener-id:\n{_describe_listeners(listeners)}"
)
return self.on(listeners[0])
def _namespace(self, listener: Listener) -> str:
if not listener.namespaces:
raise click.UsageError(
f"Listener {listener.id} serves no namespaces. Check its configuration."
)
if self.k8s_namespace is None:
if len(listener.namespaces) == 1:
return listener.namespaces[0]
raise click.UsageError(
f"Listener {listener.id} serves several namespaces. Choose one with "
f"--k8s-namespace: {', '.join(listener.namespaces)}"
)
if self.k8s_namespace not in listener.namespaces:
raise click.UsageError(
f"Listener {listener.id} does not serve namespace "
f"'{self.k8s_namespace}'. Choose one of: "
f"{', '.join(listener.namespaces)}"
)
return self.k8s_namespace
def _describe_listeners(listeners: Sequence[Listener]) -> str:
shown = listeners[:_LISTENERS_SHOWN]
lines = [
f" {listener.id} cluster {listener.compute_id} "
f"namespaces: {', '.join(listener.namespaces)}"
for listener in shown
]
if len(listeners) > len(shown):
lines.append(f" ... and {len(listeners) - len(shown)} more")
if len(listeners) == MAX_PAGE_SIZE:
lines.append(f" (only the first {MAX_PAGE_SIZE} listeners were read)")
return "\n".join(lines)
@dataclass(frozen=True, slots=True)
class ExistingDeployment:
id: str
source: str | None
# ---------------------------------------------------------------------------
# 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
# ---------------------------------------------------------------------------
def deployment_selector(deployment_id: str | None, name: str | None) -> ById | ByName:
if deployment_id:
return ById(deployment_id)
if name:
return ByName(name)
raise click.UsageError("Either --deployment-id or --name is required.")
def validate_deploy_commands(
install_command: str | None, build_command: str | None
) -> None:
"""Validate optional deploy commands for disallowed content."""
if install_command and langgraph_cli.config.has_disallowed_build_command_content(
install_command
):
raise click.UsageError(
"install_command contains disallowed characters or patterns."
)
if build_command and langgraph_cli.config.has_disallowed_build_command_content(
build_command
):
raise click.UsageError(
"build_command contains disallowed characters or patterns."
)
# ---------------------------------------------------------------------------
# Deployment lookup
# ---------------------------------------------------------------------------
def _source_of(resource: object) -> str | None:
if not isinstance(resource, dict):
return None
source = resource.get("source")
return source if isinstance(source, str) else None
def find_deployment_by_name(
client: HostBackendClient, name: str
) -> ExistingDeployment | None:
listed = client.list_deployments(name=name, name_contains=name, limit=MAX_PAGE_SIZE)
for resource in listed:
if resource.get("name") == name and resource.get("id"):
return ExistingDeployment(str(resource["id"]), _source_of(resource))
if len(listed) >= MAX_PAGE_SIZE:
raise click.ClickException(
"This workspace has more deployments than the CLI can search, so it "
f"cannot tell whether '{name}' already exists. Pass --deployment-id to "
"update an existing deployment."
)
return None
# ---------------------------------------------------------------------------
# Formatting helpers
# ---------------------------------------------------------------------------
def normalize_name(value: str | None) -> str:
"""Sanitize a deployment/directory name into a valid deployment name.
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("-")
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 = _DEFAULT_IMAGE_TAG
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 _validate_prebuilt_image(
runner: CommandRunner, image: str, *, verbose: bool
) -> None:
"""Ensure a prebuilt image exists locally for linux/amd64."""
try:
stdout, _ = runner.run(
subp_exec(
"docker",
"image",
"inspect",
"--format",
"{{.Os}}/{{.Architecture}}",
image,
verbose=verbose,
collect=True,
)
)
except FileNotFoundError:
raise click.ClickException(
"Docker is required but not installed.\n"
"Install Docker Desktop: https://docs.docker.com/get-docker/"
) from None
except click.exceptions.Exit:
raise click.ClickException(
f"Docker image '{image}' was not found locally. Build or pull the image "
"before deploying with --image."
) from None
image_platform = (stdout or "").strip()
if image_platform != _DEPLOYMENT_PLATFORM:
detected = image_platform or "unknown"
raise click.ClickException(
f"Docker image '{image}' targets {detected}, but LangSmith Deployment "
f"requires {_DEPLOYMENT_PLATFORM}. Rebuild or pull the image for "
f"{_DEPLOYMENT_PLATFORM} before deploying with --image."
)
_get_emitter().info(f"Image is available for {_DEPLOYMENT_PLATFORM}")
def _extract_deployment_url(deployment: dict[str, object]) -> str:
source_config = deployment.get("source_config")
if isinstance(source_config, dict):
custom_url = source_config.get("custom_url")
if isinstance(custom_url, str) and custom_url:
return custom_url
return "-"
def format_deployments_table(deployments: Sequence[dict[str, object]]) -> str:
headers = ("Deployment ID", "Deployment Name", "Deployment URL")
rows = [
(
str(deployment.get("id", "-") or "-"),
str(deployment.get("name", "-") or "-"),
_extract_deployment_url(deployment),
)
for deployment in deployments
]
widths = [
max(len(headers[index]), *(len(row[index]) for row in rows))
for index in range(len(headers))
]
def format_row(row: Sequence[str]) -> str:
return " ".join(value.ljust(widths[index]) for index, value in enumerate(row))
lines = [format_row(headers), format_row(tuple("-" * width for width in widths))]
lines.extend(format_row(row) for row in rows)
return "\n".join(lines)
def _extract_listener_namespaces(listener: dict[str, object]) -> str:
compute_config = listener.get("compute_config")
namespaces = (
compute_config.get("k8s_namespaces")
if isinstance(compute_config, dict)
else None
)
if isinstance(namespaces, list) and namespaces:
return ", ".join(str(namespace) for namespace in namespaces)
return "-"
def format_listeners_table(listeners: Sequence[dict[str, object]]) -> str:
headers = ("Listener ID", "Compute ID", "Namespaces")
rows = [
(
str(listener.get("id", "-") or "-"),
str(listener.get("compute_id", "-") or "-"),
_extract_listener_namespaces(listener),
)
for listener in listeners
]
widths = [
max(len(headers[index]), *(len(row[index]) for row in rows))
for index in range(len(headers))
]
def format_row(row: Sequence[str]) -> str:
return " ".join(value.ljust(widths[index]) for index, value in enumerate(row))
lines = [format_row(headers), format_row(tuple("-" * width for width in widths))]
lines.extend(format_row(row) for row in rows)
return "\n".join(lines)
def format_revisions_table(revisions: Sequence[dict[str, object]]) -> str:
headers = ("Revision ID", "Status", "Created At")
latest_deployed_seen = False
rows = []
for revision in revisions:
status = str(revision.get("status", "-") or "-")
if status == "DEPLOYED":
if latest_deployed_seen:
status = "REPLACED"
else:
latest_deployed_seen = True
rows.append(
(
str(revision.get("id", "-") or "-"),
status,
str(revision.get("created_at", "-") or "-"),
)
)
widths = [
max(len(headers[index]), *(len(row[index]) for row in rows))
for index in range(len(headers))
]
def format_row(row: Sequence[str]) -> str:
return " ".join(value.ljust(widths[index]) for index, value in enumerate(row))
lines = [format_row(headers), format_row(tuple("-" * width for width in widths))]
lines.extend(format_row(row) for row in rows)
return "\n".join(lines)
def format_timestamp(ts) -> str:
"""Convert a timestamp (epoch ms or string) to a readable string."""
if isinstance(ts, (int, float)):
dt = datetime.fromtimestamp(ts / 1000, tz=timezone.utc)
return dt.strftime("%Y-%m-%d %H:%M:%S")
return str(ts) if ts else ""
def format_log_entry(entry: dict) -> str:
"""Format a single log entry for display."""
ts = format_timestamp(entry.get("timestamp", ""))
level = entry.get("level", "")
message = entry.get("message", "")
if ts and level:
return f"[{ts}] [{level}] {message}"
elif ts:
return f"[{ts}] {message}"
return message
def level_fg(level: str) -> str | None:
"""Return click color for a log level."""
level_upper = level.upper() if level else ""
if level_upper in {"ERROR", "CRITICAL"}:
return "red"
if level_upper == "WARNING":
return "yellow"
return None
# ---------------------------------------------------------------------------
# Env / secrets helpers
# ---------------------------------------------------------------------------
def _resolve_env_path(
config_json: dict, config_path: pathlib.Path
) -> pathlib.Path | None:
"""Return the .env file path implied by the config, or None for inline dicts."""
env_field = config_json.get("env")
if isinstance(env_field, dict) and env_field:
return None
if isinstance(env_field, str):
env_path = (config_path.parent / env_field).resolve()
if not env_path.exists():
_get_emitter().note(
f"Warning: env file '{env_field}' specified in langgraph.json not found."
)
return None
return env_path
return pathlib.Path.cwd() / ".env"
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) and env_field:
return {str(k): str(v) for k, v in env_field.items()}
env_path = _resolve_env_path(config_json, config_path)
if env_path is None:
return {}
return {k: v for k, v in dotenv_values(env_path).items() if v is not None}
def _env_without_deployment_name(env_vars: dict[str, str]) -> dict[str, str]:
"""Return env vars copy with deployment-name key removed."""
filtered = dict(env_vars)
filtered.pop(_DEPLOYMENT_NAME_ENV, None)
return filtered
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:
_get_emitter().note(f"Skipping reserved env var: {name}")
continue
if not value:
continue
secrets.append({"name": name, "value": value})
return secrets
# ---------------------------------------------------------------------------
# Build mode resolution
# ---------------------------------------------------------------------------
def _resolve_build_mode(
remote_build_flag: bool | None,
*,
force_local: bool = False,
) -> tuple[bool, str | None]:
"""Determine whether to use a remote build.
Returns (use_remote_build, local_build_error). Raises UsageError when
--no-remote is set but the machine cannot build locally. When
`force_local` is set, the function short-circuits and always selects a
local build.
"""
if force_local:
return False, None
local_build_supported, local_build_error = can_build_locally()
if remote_build_flag is True:
return True, local_build_error
if remote_build_flag is False:
if not local_build_supported:
details = "\n\nOr re-run with --remote to use remote builds."
raise click.UsageError(
f"{local_build_error or 'Unable to build locally.'}{details}"
)
return False, None
# auto-detect
return not local_build_supported, local_build_error
# ---------------------------------------------------------------------------
# Deployment orchestration helpers
# ---------------------------------------------------------------------------
def _log_deploy_step(step: int, message: str, **extra: object) -> None:
_get_emitter().step(step, message, **extra)
def _fetch_deployment(
client: HostBackendClient, step: int, selector: ById
) -> tuple[ExistingDeployment, int]:
_log_deploy_step(step, f"Using deployment {selector.deployment_id}")
resource = _call_host_backend_with_optional_tenant(
client, lambda c: c.get_deployment(selector.deployment_id)
)
return ExistingDeployment(selector.deployment_id, _source_of(resource)), step + 1
def _find_deployment(
client: HostBackendClient,
step: int,
selector: ByName | ByAgent,
*,
not_found_message: str,
) -> tuple[ExistingDeployment | None, int]:
if isinstance(selector, ByAgent):
_log_deploy_step(
step, f"Looking up agent '{selector.agent_id}' in {selector.environment}"
)
existing = _call_host_backend_with_optional_tenant(
client,
lambda c: c.list_deployments(
agent_id=selector.agent_id,
agent_environment=selector.environment,
limit=MAX_PAGE_SIZE,
),
)
if len(existing) > 1:
raise click.ClickException(
"This control plane does not filter deployments by agent, so the "
f"CLI cannot tell which one belongs to '{selector.agent_id}' in "
f"{selector.environment}. Deploy by --name instead."
)
found = next(
(
ExistingDeployment(str(dep["id"]), _source_of(dep))
for dep in existing
if dep.get("id") and not dep.get("is_preview")
),
None,
)
else:
_log_deploy_step(step, f"Looking up deployment '{selector.name}'")
found = _call_host_backend_with_optional_tenant(
client, lambda c: find_deployment_by_name(c, selector.name)
)
em = _get_emitter()
if found is None:
em.warn(not_found_message)
else:
em.info(f"Found existing deployment (ID: {found.id})")
return found, step + 1
@dataclass(frozen=True, slots=True)
class CreatedDeployment:
id: str
resource: dict[str, object]
def _create_deployment(
client: HostBackendClient,
step: int,
*,
name: str | None,
source: str,
source_config: dict[str, object],
source_revision_config: dict[str, object],
secrets: list[dict[str, str]],
agent: dict[str, str] | None = None,
) -> tuple[CreatedDeployment, int]:
_log_deploy_step(
step,
f"Creating deployment for agent '{agent['agent_id']}' in {agent['environment']}"
if agent is not None
else f"Creating deployment '{name}'",
)
try:
created = client.create_deployment(
name=name,
source=source,
source_config=source_config,
source_revision_config=source_revision_config,
secrets=secrets,
agent=agent,
)
except HostBackendError as err:
if agent is not None and err.status_code == 409:
raise HostBackendError(
"This agent already has a deployment in this environment.",
status_code=409,
) from None
raise
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'"
)
if agent is not None:
_get_emitter().info(f"Deployment name: {created.get('name')}")
_get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id)
return CreatedDeployment(created_id, created), step + 1
def _get_deployment_status_url(
updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
) -> str | None:
tenant_id = updated.get("tenant_id") if isinstance(updated, dict) else None
if not tenant_id:
return None
return f"{endpoints.dashboard_url}/o/{tenant_id}/host/deployments/{deployment_id}"
def _emit_deployment_status_url(
updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
) -> str | None:
url = _get_deployment_status_url(updated, deployment_id, endpoints)
if url:
_get_emitter().status_url(url)
return url
def _poll_revision_status(
client: HostBackendClient,
deployment_id: str,
*,
progress_message: str,
timeout_seconds: int,
poll_interval_seconds: int,
on_poll: Callable[[str, str, Callable[[str], None]], None] | None = None,
on_interrupt: Callable[[str], None] | None = None,
) -> tuple[str, str | None]:
"""Poll latest revision status until terminal status or timeout."""
em = _get_emitter()
revisions = client.list_revisions(deployment_id, limit=1)
if not revisions:
return "", None
revision_id = str(revisions[0]["id"])
last_status = ""
deadline = time.time() + timeout_seconds
start_time = time.monotonic()
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)
except KeyboardInterrupt:
set_progress("")
if on_interrupt is not None:
on_interrupt(revision_id)
raise click.exceptions.Exit(1) from None
raise
status = (
rev.get("status", "UNKNOWN") if isinstance(rev, dict) else "UNKNOWN"
)
if status != last_status:
set_progress("")
if last_status:
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()