WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-20 00:28:00 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor(cli): resolve control plane endpoints in one value object

Browse Source
This commit is contained in:
Hugo Durand
2026-09-18 13:36:16 -04:00
parent a0e9bbc93a
commit b11d6572ff
4 changed files with 222 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/cli/langgraph_cli/deploy.py
+20 -57
View File
@@ -23,7 +23,11 @@ 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 Runner, subp_exec
from langgraph_cli.host_backend import HostBackendClient, HostBackendError
from langgraph_cli.host_backend import (
ControlPlaneEndpoints,
HostBackendClient,
HostBackendError,
)
from langgraph_cli.progress import Progress
from langgraph_cli.util import warn_non_wolfi_distro
@@ -658,45 +662,19 @@ def _create_deployment(
return created_id, step + 1
def _smith_dashboard_base_url(host_url: str | None) -> str:
"""Derive the LangSmith dashboard base URL from the API host URL."""
from urllib.parse import urlparse
if not host_url:
return "https://smith.langchain.com"
parsed = urlparse(host_url)
hostname = parsed.hostname or ""
# Self-hosted: host_url is <scheme>://<host>/api-host — return just the root
path = parsed.path.rstrip("/")
if path == "/api-host" or path.endswith("/api-host"):
return f"{parsed.scheme}://{parsed.netloc}"
if hostname in ("localhost", "127.0.0.1"):
return host_url.rstrip("/")
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 _get_deployment_status_url(
updated: object, deployment_id: str, host_url: str | None = None
updated: object, deployment_id: str, host_url: str
) -> 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 None
base = _smith_dashboard_base_url(host_url)
base = ControlPlaneEndpoints.from_control_plane_url(host_url).dashboard_url
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
updated: object, deployment_id: str, host_url: str
) -> str | None:
"""Emit the deployment status URL and return it."""
url = _get_deployment_status_url(updated, deployment_id, host_url)
@@ -1352,30 +1330,13 @@ def _create_host_backend_client(
tenant_id = env_vars.get("LANGSMITH_TENANT_ID") or os.environ.get(
"LANGSMITH_TENANT_ID"
)
# If no explicit host URL was provided, check LANGSMITH_ENDPOINT as a
# fallback so self-hosted customers don't need to know about LANGGRAPH_HOST_URL.
# Self-hosted control plane always lives at <langsmith_endpoint>/api-host.
_cloud_default = "https://api.host.langchain.com"
_cloud_endpoints = {
"https://api.smith.langchain.com",
"https://api.langchain.com",
}
resolved_host = host_url
if not resolved_host or resolved_host == _cloud_default:
langsmith_endpoint = env_vars.get("LANGSMITH_ENDPOINT") or os.environ.get(
"LANGSMITH_ENDPOINT"
)
if (
langsmith_endpoint
and langsmith_endpoint.rstrip("/") not in _cloud_endpoints
):
from urllib.parse import urlparse as _urlparse
_p = _urlparse(langsmith_endpoint)
resolved_host = f"{_p.scheme}://{_p.netloc}/api-host"
else:
resolved_host = _cloud_default
return HostBackendClient(resolved_host, resolved_api_key, tenant_id=tenant_id)
langsmith_endpoint = env_vars.get("LANGSMITH_ENDPOINT") or os.environ.get(
"LANGSMITH_ENDPOINT"
)
endpoints = ControlPlaneEndpoints.resolve(host_url, langsmith_endpoint)
return HostBackendClient(
endpoints.control_plane_url, resolved_api_key, tenant_id=tenant_id
)
def _call_host_backend_with_optional_tenant(
@@ -1418,7 +1379,9 @@ def _call_host_backend_with_optional_tenant(
prompted_for_tenant = True
continue
if err.status_code == 403 and "not enabled" in err.message.lower():
smith_base = _smith_dashboard_base_url(client.base_url)
smith_base = ControlPlaneEndpoints.from_control_plane_url(
client.base_url
).dashboard_url
raise HostBackendError(
"LangSmith Deployment is not enabled for this organization. "
f"Enable it at {smith_base}/host/deployments"
@@ -1454,7 +1417,7 @@ OPT_HOST_DEPLOYMENT_NAME = click.option(
OPT_HOST_URL = click.option(
"--host-url",
envvar="LANGGRAPH_HOST_URL",
default="https://api.host.langchain.com",
default=None,
hidden=True,
)
@@ -2146,7 +2109,7 @@ def deploy_logs(
start_time: str | None,
end_time: str | None,
follow: bool,
host_url: str,
host_url: str | None,
):
env_vars = _parse_env_from_config({}, pathlib.Path.cwd() / DEFAULT_CONFIG)
client = _create_host_backend_client(host_url, api_key, env_vars=env_vars)
libs/cli/langgraph_cli/host_backend.py
+73
View File
@@ -2,11 +2,84 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
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")
@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(CLOUD_API_HOST):
return langsmith_api_host.replace(CLOUD_API_HOST, 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."""