langgraph/libs/cli/langgraph_cli/host_backend.py

"""HTTP client for LangGraph host backend deployments."""

from __future__ import annotations

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")
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)

    @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


class HostBackendError(click.ClickException):
    """Raised when the host backend returns an error response."""

    def __init__(self, message: str, status_code: int | None = None):
        super().__init__(message)
        self.status_code = status_code


class HostBackendClient:
    """Minimal JSON HTTP client for the host backend deployment service."""

    def __init__(
        self,
        base_url: str,
        api_key: str,
        tenant_id: str | None = None,
        *,
        transport: httpx.BaseTransport | None = None,
    ):
        if not base_url:
            raise click.UsageError("Host backend URL is required")
        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._client = httpx.Client(
            base_url=self._base_url,
            headers=headers,
            transport=transport or httpx.HTTPTransport(retries=3),
            timeout=30,
        )

    @property
    def base_url(self) -> str:
        return self._base_url

    def set_tenant(self, tenant_id: str) -> None:
        self._client.headers["X-Tenant-ID"] = tenant_id

    def _request(
        self,
        method: str,
        path: str,
        payload: dict[str, Any] | None = None,
        params: dict[str, Any] | None = None,
    ) -> Any:
        try:
            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)
            raise HostBackendError(
                f"{method} {path} failed with status {err.response.status_code}: {detail}",
                status_code=err.response.status_code,
            ) from None
        except httpx.TransportError as err:
            raise HostBackendError(str(err)) from None

        if not resp.content:
            return None
        try:
            return resp.json()
        except ValueError as err:
            raise HostBackendError(
                f"Failed to decode response from {path}: {err}"
            ) from None

    def create_deployment(
        self,
        *,
        name: str,
        source: SourceName,
        source_config: dict[str, object],
        source_revision_config: dict[str, object],
        secrets: list[dict[str, str]] | None = None,
    ) -> dict[str, Any]:
        payload: dict[str, Any] = {
            "name": name,
            "source": source,
            "source_config": source_config,
            "source_revision_config": source_revision_config,
        }
        if secrets is not None:
            payload["secrets"] = secrets
        return self._request("POST", "/v2/deployments", payload)

    def list_deployments(self, name_contains: str = "") -> dict[str, Any]:
        return self._request(
            "GET",
            "/v2/deployments",
            params={"name_contains": name_contains},
        )

    def get_deployment(self, deployment_id: str) -> dict[str, Any]:
        return self._request("GET", f"/v2/deployments/{deployment_id}")

    def delete_deployment(self, deployment_id: str) -> None:
        return self._request("DELETE", f"/v2/deployments/{deployment_id}")

    def request_push_token(self, deployment_id: str) -> dict[str, Any]:
        return self._request(
            "POST",
            f"/v2/deployments/{deployment_id}/push-token",
        )

    def request_upload_url(self, deployment_id: str) -> dict[str, Any]:
        """Get a signed GCS URL for uploading the source tarball."""
        return self._request(
            "POST",
            f"/v2/deployments/{deployment_id}/upload-url",
        )

    def update_deployment(
        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] = {
            "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)

    def update_deployment_internal_source(
        self,
        deployment_id: str,
        source_tarball_path: str,
        config_path: str,
        secrets: list[dict[str, str]] | None = None,
        install_command: str | None = None,
        build_command: str | None = None,
        tracked_packages: list[str] | None = None,
    ) -> dict[str, Any]:
        """Trigger a remote build revision with the uploaded tarball."""
        payload: dict[str, Any] = {
            "revision_source": "internal_source",
            "source_revision_config": {
                "source_tarball_path": source_tarball_path,
                "langgraph_config_path": config_path,
            },
        }
        if tracked_packages:
            payload["tracked_packages"] = tracked_packages

        source_config: dict[str, Any] = {}
        if install_command is not None:
            source_config["install_command"] = install_command
        if build_command is not None:
            source_config["build_command"] = build_command
        if source_config:
            payload["source_config"] = source_config

        if secrets is not None:
            payload["secrets"] = secrets
        return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload)

    def list_revisions(self, deployment_id: str, limit: int = 1) -> dict[str, Any]:
        return self._request(
            "GET",
            f"/v2/deployments/{deployment_id}/revisions?limit={limit}",
        )

    def get_revision(self, deployment_id: str, revision_id: str) -> dict[str, Any]:
        return self._request(
            "GET",
            f"/v2/deployments/{deployment_id}/revisions/{revision_id}",
        )

    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)