merge conflict

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.31.dev0"
+__version__ = "0.4.32.dev0"

libs/cli/langgraph_cli/exec.py

@@ -1,12 +1,18 @@
 import asyncio
 import signal
 import sys
-from collections.abc import Callable
+from collections.abc import Callable, Coroutine
 from contextlib import contextmanager
-from typing import cast
+from typing import Any, Protocol, TypeVar, cast
 
 import click.exceptions
 
+_T = TypeVar("_T")
+
+
+class CommandRunner(Protocol):
+    def run(self, coro: Coroutine[Any, Any, _T]) -> _T: ...
+
 
 @contextmanager
 def Runner():

libs/cli/langgraph_cli/host_backend.py

@@ -2,18 +2,125 @@
 
 from __future__ import annotations
 
-from typing import Any
+from dataclasses import dataclass
+from typing import Any, Literal
+from urllib.parse import urlparse
 
 import click
 import httpx
 
+CLOUD_CONTROL_PLANE_URL = "https://api.host.langchain.com"
+CLOUD_DASHBOARD_URL = "https://smith.langchain.com"
+CLOUD_DOMAIN = "langchain.com"
+CLOUD_API_HOST = "api.smith.langchain.com"
+CLOUD_CONTROL_PLANE_HOST = "api.host.langchain.com"
+CLOUD_DASHBOARD_HOST = "smith.langchain.com"
+CONTROL_PLANE_PATH = "/api-host"
+LANGSMITH_API_PATHS = ("/api/v1", "/api")
+LOCAL_HOSTNAMES = ("localhost", "127.0.0.1")
+MAX_PAGE_SIZE = 100
+SourceName = Literal["internal_docker", "internal_source", "external_docker"]
+
+
+@dataclass(frozen=True, slots=True)
+class ControlPlaneEndpoints:
+    control_plane_url: str
+    dashboard_url: str
+
+    @classmethod
+    def resolve(
+        cls, host_url: str | None, langsmith_endpoint: str | None
+    ) -> ControlPlaneEndpoints:
+        if host_url:
+            return cls.from_control_plane_url(host_url)
+        if langsmith_endpoint:
+            return cls.from_langsmith_endpoint(langsmith_endpoint)
+        return cls(CLOUD_CONTROL_PLANE_URL, CLOUD_DASHBOARD_URL)
+
+    @property
+    def is_cloud(self) -> bool:
+        hostname = urlparse(self.control_plane_url).hostname or ""
+        return hostname == CLOUD_CONTROL_PLANE_HOST or hostname.endswith(
+            f".{CLOUD_CONTROL_PLANE_HOST}"
+        )
+
+    @classmethod
+    def from_control_plane_url(cls, url: str) -> ControlPlaneEndpoints:
+        control_plane_url = url.rstrip("/")
+        hostname = urlparse(control_plane_url).hostname or ""
+        if control_plane_url.endswith(CONTROL_PLANE_PATH):
+            return cls(control_plane_url, control_plane_url[: -len(CONTROL_PLANE_PATH)])
+        if hostname in LOCAL_HOSTNAMES:
+            return cls(control_plane_url, control_plane_url)
+        return cls(control_plane_url, _cloud_dashboard_for(hostname))
+
+    @classmethod
+    def from_langsmith_endpoint(cls, endpoint: str) -> ControlPlaneEndpoints:
+        parsed = urlparse(endpoint.rstrip("/"))
+        hostname = parsed.hostname or ""
+        if _is_cloud_host(hostname):
+            return cls.from_control_plane_url(
+                f"https://{_cloud_control_plane_host_for(hostname)}"
+            )
+        root = f"{parsed.scheme}://{parsed.netloc}{_without_api_path(parsed.path)}"
+        return cls(f"{root}{CONTROL_PLANE_PATH}", root)
+
+
+def _is_cloud_host(hostname: str) -> bool:
+    return hostname == CLOUD_DOMAIN or hostname.endswith(f".{CLOUD_DOMAIN}")
+
+
+def _cloud_control_plane_host_for(langsmith_api_host: str) -> str:
+    if langsmith_api_host.endswith(f".{CLOUD_API_HOST}"):
+        region = langsmith_api_host[: -len(CLOUD_API_HOST)]
+        return f"{region}{CLOUD_CONTROL_PLANE_HOST}"
+    return CLOUD_CONTROL_PLANE_HOST
+
+
+def _cloud_dashboard_for(control_plane_host: str) -> str:
+    if control_plane_host.endswith(f".{CLOUD_CONTROL_PLANE_HOST}"):
+        region = control_plane_host[: -len(CLOUD_CONTROL_PLANE_HOST) - 1]
+        return f"https://{region}.{CLOUD_DASHBOARD_HOST}"
+    return CLOUD_DASHBOARD_URL
+
+
+def _without_api_path(path: str) -> str:
+    for api_path in LANGSMITH_API_PATHS:
+        if path.endswith(api_path):
+            return path[: -len(api_path)]
+    return path
+
+
+def _resources(payload: object) -> list[dict[str, Any]]:
+    if not isinstance(payload, dict):
+        return []
+    resources = payload.get("resources")
+    if not isinstance(resources, list):
+        return []
+    return [item for item in resources if isinstance(item, dict)]
+
 
 class HostBackendError(click.ClickException):
     """Raised when the host backend returns an error response."""
 
-    def __init__(self, message: str, status_code: int | None = None):
+    def __init__(
+        self,
+        message: str,
+        status_code: int | None = None,
+        detail: str | None = None,
+    ):
         super().__init__(message)
         self.status_code = status_code
+        self.detail = detail
+
+
+def _error_detail(response: httpx.Response) -> str | None:
+    try:
+        body = response.json()
+    except ValueError:
+        return None
+    detail = body.get("detail") if isinstance(body, dict) else None
+    return detail if isinstance(detail, str) else None
 
 
 class HostBackendClient:
@@ -24,24 +131,37 @@ class HostBackendClient:
         base_url: str,
         api_key: str,
         tenant_id: str | None = None,
+        *,
+        transport: httpx.BaseTransport | 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._endpoints = ControlPlaneEndpoints.from_control_plane_url(base_url)
+        self._base_url = self._endpoints.control_plane_url
         self._client = httpx.Client(
             base_url=self._base_url,
             headers=headers,
-            transport=transport,
+            transport=transport or httpx.HTTPTransport(retries=3),
             timeout=30,
         )
 
+    @property
+    def base_url(self) -> str:
+        return self._base_url
+
+    @property
+    def endpoints(self) -> ControlPlaneEndpoints:
+        return self._endpoints
+
+    def set_tenant(self, tenant_id: str) -> None:
+        self._client.headers["X-Tenant-ID"] = tenant_id
+
     def _request(
         self,
         method: str,
@@ -53,10 +173,12 @@ class HostBackendClient:
             resp = self._client.request(method, path, json=payload, params=params)
             resp.raise_for_status()
         except httpx.HTTPStatusError as err:
-            detail = err.response.text or str(err.response.status_code)
+            detail = _error_detail(err.response)
+            reason = detail or err.response.text or str(err.response.status_code)
             raise HostBackendError(
-                f"{method} {path} failed with status {err.response.status_code}: {detail}",
+                f"{method} {path} failed with status {err.response.status_code}: {reason}",
                 status_code=err.response.status_code,
+                detail=detail,
             ) from None
         except httpx.TransportError as err:
             raise HostBackendError(str(err)) from None
@@ -72,45 +194,52 @@ class HostBackendClient:
 
     def create_deployment(
         self,
+        *,
         name: str | None,
-        deployment_type: str,
-        source: str,
-        config_path: str | None = None,
+        source: SourceName,
+        source_config: dict[str, object],
+        source_revision_config: dict[str, object],
         secrets: list[dict[str, str]] | None = None,
         agent: dict[str, str] | None = None,
     ) -> dict[str, Any]:
-        """Create a deployment."""
         payload: dict[str, Any] = {
             "source": source,
-            "source_config": {"deployment_type": deployment_type},
-            "source_revision_config": {},
+            "source_config": source_config,
+            "source_revision_config": source_revision_config,
         }
         if agent is not None:
             payload["agent"] = agent
         else:
             payload["name"] = name
-        if source == "internal_source" and config_path:
-            payload["source_revision_config"]["langgraph_config_path"] = config_path
         if secrets is not None:
             payload["secrets"] = secrets
         return self._request("POST", "/v2/deployments", payload)
 
     def list_deployments(
         self,
-        name_contains: str = "",
         *,
+        name: str | None = None,
+        name_contains: str | None = None,
+        limit: int | None = None,
         agent_id: str | None = None,
         agent_environment: str | None = None,
-    ) -> dict[str, Any]:
-        params = {"name_contains": name_contains}
-        if agent_id is not None:
-            params["agent_id"] = agent_id
-        if agent_environment is not None:
-            params["agent_environment"] = agent_environment
-        return self._request(
-            "GET",
-            "/v2/deployments",
-            params=params,
+    ) -> list[dict[str, Any]]:
+        given = (
+            ("name", name),
+            ("name_contains", name_contains),
+            ("limit", limit),
+            ("agent_id", agent_id),
+            ("agent_environment", agent_environment),
+        )
+        params = {key: value for key, value in given if value is not None}
+        return _resources(self._request("GET", "/v2/deployments", params=params))
+
+    def get_listener(self, listener_id: str) -> dict[str, Any]:
+        return self._request("GET", f"/v2/listeners/{listener_id}")
+
+    def list_listeners(self) -> list[dict[str, Any]]:
+        return _resources(
+            self._request("GET", "/v2/listeners", params={"limit": MAX_PAGE_SIZE})
         )
 
     def get_deployment(self, deployment_id: str) -> dict[str, Any]:
@@ -136,22 +265,21 @@ class HostBackendClient:
         self,
         deployment_id: str,
         image_uri: str,
+        *,
+        revision_source: SourceName | None,
         secrets: list[dict[str, str]] | None = None,
         tracked_packages: list[str] | None = None,
     ) -> dict[str, Any]:
         payload: dict[str, Any] = {
-            "revision_source": "internal_docker",
             "source_revision_config": {"image_uri": image_uri},
         }
+        if revision_source is not None:
+            payload["revision_source"] = revision_source
         if tracked_packages:
             payload["tracked_packages"] = tracked_packages
         if secrets is not None:
             payload["secrets"] = secrets
-        return self._request(
-            "PATCH",
-            f"/v2/deployments/{deployment_id}",
-            payload,
-        )
+        return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload)
 
     def update_deployment_internal_source(
         self,
@@ -186,10 +314,15 @@ class HostBackendClient:
             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 list_revisions(
+        self, deployment_id: str, limit: int = 1
+    ) -> list[dict[str, Any]]:
+        return _resources(
+            self._request(
+                "GET",
+                f"/v2/deployments/{deployment_id}/revisions",
+                params={"limit": limit},
+            )
         )
 
     def get_revision(self, deployment_id: str, revision_id: str) -> dict[str, Any]:

libs/cli/langgraph_cli/image_reference.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+
+DIGEST_SEPARATOR = "@sha256:"
+DIGEST_MARKER = "@"
+TAG_SEPARATOR = ":"
+PATH_SEPARATOR = "/"
+
+
+@dataclass(frozen=True, slots=True)
+class ImageReference:
+    repository: str
+    tag: str | None = None
+
+    @classmethod
+    def parse(cls, reference: str) -> ImageReference:
+        if DIGEST_MARKER in reference:
+            raise ValueError(f"{reference!r} carries a digest and cannot be tagged")
+        path_start = reference.rfind(PATH_SEPARATOR) + 1
+        name, separator, tag = reference[path_start:].partition(TAG_SEPARATOR)
+        if not separator:
+            return cls(reference)
+        return cls(reference[:path_start] + name, tag)
+
+    def with_tag(self, tag: str) -> ImageReference:
+        return replace(self, tag=tag)
+
+    def matches_digest(self, repo_digest: str) -> bool:
+        return repo_digest.startswith(f"{self.repository}{DIGEST_SEPARATOR}")
+
+    def __str__(self) -> str:
+        if self.tag is None:
+            return self.repository
+        return f"{self.repository}{TAG_SEPARATOR}{self.tag}"