diff --git a/libs/cli/langgraph_cli/cli.py b/libs/cli/langgraph_cli/cli.py index 21715d50c..4734a59f2 100644 --- a/libs/cli/langgraph_cli/cli.py +++ b/libs/cli/langgraph_cli/cli.py @@ -13,6 +13,7 @@ import tempfile import time from collections.abc import Callable, Sequence from contextlib import contextmanager +from datetime import datetime, timezone import click import click.exceptions @@ -26,6 +27,7 @@ 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.helpers import format_log_entry, level_fg, resolve_deployment_id from langgraph_cli.host_backend import HostBackendClient, HostBackendError from langgraph_cli.progress import Progress from langgraph_cli.templates import TEMPLATE_HELP_STRING, create_new @@ -296,6 +298,17 @@ OPT_HOST_API_KEY = click.option( ), ) + +OPT_HOST_DEPLOYMENT_NAME = click.option( + "--name", + envvar=_DEPLOYMENT_NAME_ENV, + 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." + ), +) + OPT_HOST_URL = click.option( "--host-url", envvar="LANGGRAPH_HOST_URL", @@ -684,15 +697,7 @@ def _deploy_base_options( def _apply(target: Callable) -> Callable: decorators = [ OPT_HOST_API_KEY, - 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." - ), - ), + OPT_HOST_DEPLOYMENT_NAME, click.option( "--deployment-id", help=( @@ -1294,6 +1299,175 @@ def _normalize_image_tag(value: str) -> str: return value +@OPT_HOST_API_KEY +@OPT_HOST_DEPLOYMENT_NAME +@click.option( + "--deployment-id", + help="Deployment ID. If omitted, --name is used to find the deployment.", +) +@click.option( + "--type", + "log_type", + type=click.Choice(["deploy", "build"]), + default="deploy", + show_default=True, + help=( + "Log stream to fetch: 'deploy' shows agent server runtime logs; " + "'build' shows build logs (for deployments built remotely)." + ), +) +@click.option( + "--revision-id", + help="Specific revision ID. For build logs, defaults to latest revision.", +) +@click.option( + "--level", + type=click.Choice( + ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False + ), + help="Filter by log level.", +) +@click.option( + "--limit", + type=int, + default=100, + show_default=True, + help="Max log entries to fetch.", +) +@click.option( + "--query", + "-q", + help="Search string filter.", +) +@click.option( + "--start-time", + help="ISO8601 start time (e.g. 2026-03-08T00:00:00Z).", +) +@click.option( + "--end-time", + help="ISO8601 end time. (e.g. 2026-03-08T00:00:00Z)", +) +@click.option( + "--follow", + "-f", + is_flag=True, + default=False, + help="Continuously poll for new logs.", +) +@OPT_HOST_URL +@deploy.command( + "logs", + help=( + "[Beta] Fetch LangSmith Deployment logs. Use 'deploy' for agent runtime " + "logs, or 'build' for remote build logs." + ), +) +@log_command +def deploy_logs( + api_key: str | None, + name: str | None, + deployment_id: str | None, + log_type: str, + revision_id: str | None, + level: str | None, + limit: int, + query: str | None, + start_time: str | None, + end_time: str | None, + follow: bool, + host_url: str, +): + client = _create_host_backend_client(host_url, api_key) + dep_id = resolve_deployment_id(client, deployment_id, name) + + if log_type == "build" and not revision_id: + revisions_resp = client.list_revisions(dep_id, limit=1) + resources = ( + revisions_resp.get("resources", []) + if isinstance(revisions_resp, dict) + else [] + ) + if not resources: + raise click.ClickException( + "No revisions found for this deployment. Cannot fetch build logs." + ) + revision_id = str(resources[0]["id"]) + click.secho(f"Using latest revision: {revision_id}", fg="cyan") + + payload: dict = {"limit": limit, "order": "desc"} + if level: + payload["level"] = level.upper() + if query: + payload["query"] = query + if start_time: + payload["start_time"] = start_time + if end_time: + payload["end_time"] = end_time + + def _fetch(request_payload: dict) -> list[dict]: + if log_type == "build": + resp = client.get_build_logs(dep_id, revision_id, request_payload) + else: + resp = client.get_deploy_logs(dep_id, request_payload, revision_id) + + if isinstance(resp, dict): + return resp.get("logs", []) + return [] + + def _print_entries(entries: list[dict], *, reverse: bool = False) -> None: + iterable = reversed(entries) if reverse else entries + for entry in iterable: + line = format_log_entry(entry) + fg = level_fg(entry.get("level", "")) + click.secho(line, fg=fg) + + def _fetch_and_print(request_payload: dict, *, reverse: bool = False) -> list[dict]: + entries = _fetch(request_payload) + _print_entries(entries, reverse=reverse) + return entries + + def _fetch_and_print_new(request_payload: dict, seen_ids: set[str]) -> list[dict]: + entries = _fetch(request_payload) + new = [e for e in entries if e.get("id", "") not in seen_ids] + if new: + _print_entries(new) + seen_ids.update(e.get("id", "") for e in new) + return new + + # initial log fetch will be newest -> oldest, so we need to reverse + entries = _fetch_and_print(payload, reverse=True) + + if not follow: + if not entries: + click.secho("No log entries found.", fg="yellow") + return + + payload["order"] = "asc" + seen_ids: set[str] = {e.get("id", "") for e in entries if e.get("id")} + + def _update_start_time(ts) -> None: + if ts is None: + return + if isinstance(ts, (int, float)): + dt = datetime.fromtimestamp(ts / 1000, tz=timezone.utc) + payload["start_time"] = dt.isoformat() + else: + payload["start_time"] = str(ts) + + if entries: + # entries are in descending order here, so index 0 is the newest log + _update_start_time(entries[0].get("timestamp")) + + try: + while True: + time.sleep(2) + new_entries = _fetch_and_print_new(payload, seen_ids) + if new_entries: + _update_start_time(new_entries[-1].get("timestamp")) + except KeyboardInterrupt: + click.echo("\nStopped.") + + def _get_docker_ignore_content() -> str: """Return the content of a .dockerignore file. diff --git a/libs/cli/langgraph_cli/helpers.py b/libs/cli/langgraph_cli/helpers.py new file mode 100644 index 000000000..efca64d4b --- /dev/null +++ b/libs/cli/langgraph_cli/helpers.py @@ -0,0 +1,59 @@ +"""Helpers for the ``langgraph logs`` CLI command.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import click + +from langgraph_cli.host_backend import HostBackendClient + + +def resolve_deployment_id( + client: HostBackendClient, + deployment_id: str | None, + name: str | None, +) -> str: + """Resolve a deployment ID from --deployment-id or --name.""" + if deployment_id: + return deployment_id + if not name: + raise click.UsageError("Either --deployment-id or --name is required.") + existing = client.list_deployments(name_contains=name) + if isinstance(existing, dict): + for dep in existing.get("resources", []): + if isinstance(dep, dict) and dep.get("name") == name: + found_id = dep.get("id") + if found_id: + return str(found_id) + raise click.ClickException(f"Deployment '{name}' not found.") + + +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 diff --git a/libs/cli/langgraph_cli/host_backend.py b/libs/cli/langgraph_cli/host_backend.py index 3e8be1a69..b34f38dc2 100644 --- a/libs/cli/langgraph_cli/host_backend.py +++ b/libs/cli/langgraph_cli/host_backend.py @@ -19,7 +19,12 @@ class HostBackendError(click.ClickException): 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): + 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) @@ -30,7 +35,6 @@ class HostBackendClient: 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, @@ -116,3 +120,24 @@ class HostBackendClient: "GET", f"/v2/deployments/{deployment_id}/revisions/{revision_id}", ) + + def get_build_logs( + self, project_id: str, revision_id: str, payload: dict[str, Any] + ) -> Any: + return self._request( + "POST", + f"/v1/projects/{project_id}/revisions/{revision_id}/build_logs", + payload, + ) + + def get_deploy_logs( + self, + project_id: str, + payload: dict[str, Any], + revision_id: str | None = None, + ) -> Any: + if revision_id: + path = f"/v1/projects/{project_id}/revisions/{revision_id}/deploy_logs" + else: + path = f"/v1/projects/{project_id}/deploy_logs" + return self._request("POST", path, payload) diff --git a/libs/cli/tests/unit_tests/test_host_backend.py b/libs/cli/tests/unit_tests/test_host_backend.py index e4d41f4c8..d31289579 100644 --- a/libs/cli/tests/unit_tests/test_host_backend.py +++ b/libs/cli/tests/unit_tests/test_host_backend.py @@ -182,3 +182,41 @@ def test_list_revisions(client): def test_get_revision(client): result = client.get_revision("dep-123", "rev-456") assert result == {"ok": True} + + +def test_get_build_logs(client): + result = client.get_build_logs("proj-1", "rev-1", {"limit": 10}) + assert result == {"ok": True} + + +def test_get_deploy_logs_all_revisions(): + def handler(req: httpx.Request) -> httpx.Response: + assert "/v1/projects/proj-1/deploy_logs" in str(req.url) + assert "/revisions/" not in str(req.url) + return httpx.Response(200, json={"logs": [{"message": "running"}]}) + + c = HostBackendClient("https://api.example.com", "key") + c._client = httpx.Client( + base_url="https://api.example.com", + transport=httpx.MockTransport(handler), + headers={"X-Api-Key": "key", "Accept": "application/json"}, + timeout=30, + ) + result = c.get_deploy_logs("proj-1", {"limit": 10}) + assert result == {"logs": [{"message": "running"}]} + + +def test_get_deploy_logs_specific_revision(): + def handler(req: httpx.Request) -> httpx.Response: + assert "/v1/projects/proj-1/revisions/rev-2/deploy_logs" in str(req.url) + return httpx.Response(200, json={"logs": []}) + + c = HostBackendClient("https://api.example.com", "key") + c._client = httpx.Client( + base_url="https://api.example.com", + transport=httpx.MockTransport(handler), + headers={"X-Api-Key": "key", "Accept": "application/json"}, + timeout=30, + ) + result = c.get_deploy_logs("proj-1", {"limit": 10}, revision_id="rev-2") + assert result == {"logs": []} diff --git a/libs/cli/tests/unit_tests/test_logs_helpers.py b/libs/cli/tests/unit_tests/test_logs_helpers.py new file mode 100644 index 000000000..76b30cdaf --- /dev/null +++ b/libs/cli/tests/unit_tests/test_logs_helpers.py @@ -0,0 +1,58 @@ +from langgraph_cli.helpers import format_log_entry, format_timestamp, level_fg + + +class TestFormatTimestamp: + def test_epoch_ms(self): + assert format_timestamp(1773119644012) == "2026-03-10 05:14:04" + + def test_string_passthrough(self): + assert format_timestamp("2026-03-08T00:00:00Z") == "2026-03-08T00:00:00Z" + + def test_empty(self): + assert format_timestamp("") == "" + + def test_none(self): + assert format_timestamp(None) == "" + + +class TestFormatLogEntry: + def test_full_entry_epoch(self): + entry = {"timestamp": 1773119644012, "level": "ERROR", "message": "boom"} + result = format_log_entry(entry) + assert result == "[2026-03-10 05:14:04] [ERROR] boom" + + def test_full_entry_string(self): + entry = { + "timestamp": "2026-03-08T12:00:00Z", + "level": "ERROR", + "message": "boom", + } + assert format_log_entry(entry) == "[2026-03-08T12:00:00Z] [ERROR] boom" + + def test_no_level(self): + entry = {"timestamp": "2026-03-08T12:00:00Z", "message": "hello"} + assert format_log_entry(entry) == "[2026-03-08T12:00:00Z] hello" + + def test_no_timestamp(self): + entry = {"message": "bare message"} + assert format_log_entry(entry) == "bare message" + + def test_empty_entry(self): + assert format_log_entry({}) == "" + + +class TestLevelFg: + def test_error(self): + assert level_fg("ERROR") == "red" + + def test_error_lowercase(self): + assert level_fg("error") == "red" + + def test_warning(self): + assert level_fg("WARNING") == "yellow" + + def test_info_returns_none(self): + assert level_fg("INFO") is None + + def test_empty_returns_none(self): + assert level_fg("") is None