From 16bc576b1bc006b0be3962c1e5149245a9812419 Mon Sep 17 00:00:00 2001 From: Hugo Durand Date: Tue, 22 Sep 2026 14:42:58 -0400 Subject: [PATCH] feat(cli): place new self-hosted deployments on a listener --- libs/cli/langgraph_cli/deploy.py | 80 +++++--- .../unit_tests/cli/test_deploy_command.py | 173 ++++++++++++++++-- .../tests/unit_tests/test_deploy_helpers.py | 14 ++ 3 files changed, 220 insertions(+), 47 deletions(-) diff --git a/libs/cli/langgraph_cli/deploy.py b/libs/cli/langgraph_cli/deploy.py index 73d12f802..892046647 100644 --- a/libs/cli/langgraph_cli/deploy.py +++ b/libs/cli/langgraph_cli/deploy.py @@ -101,14 +101,6 @@ _NATIVE_AMD64_MACHINE = "x86_64" _PUSH_ATTEMPTS = 3 _LOCAL_BUILD_TAG_PREFIX = "langgraph-deploy-tmp" _OPERATOR_DEFAULT_RESOURCE_SPEC: Mapping[str, object] = {} -_LISTENER_REQUIRED_MARKER = "listener_id' is required" -_HYBRID_LISTENER_GUIDANCE = ( - "This workspace deploys through a listener in your own cluster, and the " - "control plane needs a listener ID to create a deployment. Create the " - "deployment once in the LangSmith UI, choosing the listener and namespace, " - "then re-run with --deployment-id ." -) - _CUSTOMER_REGISTRY_SOURCE: SourceName = "external_docker" @@ -823,21 +815,20 @@ def _create_deployment( def _get_deployment_status_url( - updated: object, deployment_id: str, host_url: str + updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints ) -> 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 = ControlPlaneEndpoints.from_control_plane_url(host_url).dashboard_url - return f"{base}/o/{tenant_id}/host/deployments/{deployment_id}" + return f"{endpoints.dashboard_url}/o/{tenant_id}/host/deployments/{deployment_id}" def _emit_deployment_status_url( - updated: object, deployment_id: str, host_url: str + updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints ) -> str | None: """Emit the deployment status URL and return it.""" - url = _get_deployment_status_url(updated, deployment_id, host_url) + url = _get_deployment_status_url(updated, deployment_id, endpoints) if url: _get_emitter().status_url(url) return url @@ -1380,6 +1371,7 @@ def _run_remote_build( @dataclass(frozen=True, slots=True) class DeployContext: client: HostBackendClient + endpoints: ControlPlaneEndpoints spec: BuildSpec verbose: bool selector: DeploymentSelector @@ -1421,6 +1413,13 @@ def _resolve_or_create( return created.id, step +def _available_listeners(client: HostBackendClient) -> tuple[Listener, ...]: + resources = _call_host_backend_with_optional_tenant( + client, lambda c: c.list_listeners() + ) + return tuple(Listener.from_resource(resource) for resource in resources) + + def _ensure_customer_registry_source(existing: ExistingDeployment) -> None: if existing.source != _CUSTOMER_REGISTRY_SOURCE: raise click.UsageError( @@ -1483,6 +1482,7 @@ class RemoteBuildSource: class CustomerRegistrySource: reference: ImageReference prebuilt_image: str | None + placement: RequestedPlacement = RequestedPlacement() def run(self, ctx: DeployContext) -> DeployOutcome: if isinstance(ctx.selector, ById): @@ -1516,21 +1516,22 @@ class CustomerRegistrySource: ) def _create(self, ctx: DeployContext, name: str, step: int) -> DeployOutcome: + placement = self.placement.resolve( + _available_listeners(ctx.client), required=ctx.endpoints.is_cloud + ) image_uri, step = self._publish(ctx, step) - try: - created, _ = _create_deployment( - ctx.client, - step, - name=name, - source=_CUSTOMER_REGISTRY_SOURCE, - source_config={"resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC}, - source_revision_config={"image_uri": image_uri}, - secrets=ctx.secrets, - ) - except HostBackendError as err: - if err.status_code == 400 and _LISTENER_REQUIRED_MARKER in err.message: - raise click.ClickException(_HYBRID_LISTENER_GUIDANCE) from None - raise + created, _ = _create_deployment( + ctx.client, + step, + name=name, + source=_CUSTOMER_REGISTRY_SOURCE, + source_config={ + "resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC, + **placement.source_config(), + }, + source_revision_config={"image_uri": image_uri}, + secrets=ctx.secrets, + ) return DeployOutcome( created.id, _image_revision_result(created.resource, "Deployment created") ) @@ -1590,6 +1591,7 @@ def _select_source( image_name: str | None, tag: str | None, remote_build_flag: bool | None, + placement: RequestedPlacement, ) -> DeploymentSource: if push_to is not None: if remote_build_flag is True: @@ -1597,7 +1599,7 @@ def _select_source( reference = _push_reference(push_to, tag) if image is None: _require_local_docker() - return CustomerRegistrySource(reference, prebuilt_image=image) + return CustomerRegistrySource(reference, image, placement) if image and remote_build_flag is True: raise click.UsageError("--image cannot be combined with --remote builds.") use_remote_build, local_build_error = _resolve_build_mode( @@ -1892,6 +1894,21 @@ def _deploy_base_options( "Give the tag here or with --tag (default: latest)." ), ), + click.option( + "--listener-id", + help=( + "Listener that will run the deployment, for workspaces that " + "deploy through a listener in your own cluster. Only used when " + "creating a deployment with --push-to." + ), + ), + click.option( + "--k8s-namespace", + help=( + "Kubernetes namespace the listener deploys into. Only used when " + "creating a deployment with --push-to." + ), + ), click.option( "--config", "-c", @@ -1994,6 +2011,8 @@ def _deploy_cmd( image_name: str | None, image: str | None, push_to: str | None, + listener_id: str | None, + k8s_namespace: str | None, tag: str | None, base_image: str | None, install_command: str | None, @@ -2055,9 +2074,11 @@ def _deploy_cmd( image_name=image_name, tag=tag, remote_build_flag=remote_build_flag, + placement=RequestedPlacement(listener_id, k8s_namespace), ) client = _create_host_backend_client(host_url, api_key, env_vars=env_vars) + endpoints = ControlPlaneEndpoints.from_control_plane_url(client.base_url) try: tracked_packages = find_tracked_packages(config, config_json) or None except Exception as exc: @@ -2067,6 +2088,7 @@ def _deploy_cmd( outcome = source.run( DeployContext( client=client, + endpoints=endpoints, spec=BuildSpec( config=config, config_json=config_json, @@ -2087,7 +2109,7 @@ def _deploy_cmd( dep_status_url = _emit_deployment_status_url( outcome.build_result.updated, outcome.deployment_id, - client.base_url, + endpoints, ) if no_wait: diff --git a/libs/cli/tests/unit_tests/cli/test_deploy_command.py b/libs/cli/tests/unit_tests/cli/test_deploy_command.py index fd9817633..5cff266b1 100644 --- a/libs/cli/tests/unit_tests/cli/test_deploy_command.py +++ b/libs/cli/tests/unit_tests/cli/test_deploy_command.py @@ -17,6 +17,7 @@ from langgraph_cli.host_backend import HostBackendClient from langgraph_cli.image_reference import ImageReference CONTROL_PLANE_URL = "https://control-plane.example.com" +CLOUD_CONTROL_PLANE_URL = "https://api.host.langchain.com" REGISTRY_URL = "https://registry.example.com/team" PUSH_TOKEN = "push-token" PUSHED_IMAGE = "registry.example.com/team/my-app:latest" @@ -24,10 +25,21 @@ PUSHED_DIGEST = "registry.example.com/team/my-app@sha256:abc123" PUSH_REPOSITORY = "registry.example.com/team/agent" EXTERNAL_IMAGE = f"{PUSH_REPOSITORY}:latest" EXTERNAL_DIGEST = f"{PUSH_REPOSITORY}@sha256:abc123" -LISTENER_REQUIRED = ( - "Source configuration error: 'source_config.listener_id' is required for " - "workspace with available listener IDs: ['listener-1']" -) +LISTENER = { + "id": "listener-1", + "compute_id": "prod-cluster", + "compute_config": {"k8s_namespaces": ["agents"]}, +} +OTHER_LISTENER = { + "id": "listener-2", + "compute_id": "other-cluster", + "compute_config": {"k8s_namespaces": ["agents"]}, +} +TWO_NAMESPACE_LISTENER = { + "id": "listener-1", + "compute_id": "prod-cluster", + "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]}, +} CREATED_ID = "dep-created" TRACKED_PACKAGES = ["langgraph:1.0.0"] SIGNED_UPLOAD_URL = "https://storage.example.com/signed" @@ -39,6 +51,7 @@ NOT_A_CLI_DEPLOYMENT = ( "push token is only available for 'internal_docker' source deployments" ) LIST_DEPLOYMENTS = "GET /v2/deployments" +LIST_LISTENERS = "GET /v2/listeners" CREATE_DEPLOYMENT = "POST /v2/deployments" @@ -64,6 +77,7 @@ class ControlPlaneDouble: existing_deployments: list[dict] = field(default_factory=list) push_token_status: int = 200 create_error: str | None = None + listeners: list[dict] = field(default_factory=list) bodies: dict[str, dict] = field(default_factory=dict) def handle(self, request: httpx.Request) -> httpx.Response: @@ -74,6 +88,8 @@ class ControlPlaneDouble: return self._respond(request.method, request.url.path) def _respond(self, method: str, path: str) -> httpx.Response: + if (method, path) == ("GET", "/v2/listeners"): + return httpx.Response(200, json={"resources": self.listeners}) if (method, path) == ("GET", "/v2/deployments"): return httpx.Response(200, json={"resources": self.existing_deployments}) if (method, path) == ("POST", "/v2/deployments"): @@ -199,7 +215,7 @@ class DeployProject: timeline: list[str] uploads: list[tuple[str, str, int]] - def run(self, *args: str) -> Result: + def run(self, *args: str, host_url: str = CONTROL_PLANE_URL) -> Result: return CliRunner().invoke( cli, [ @@ -207,7 +223,7 @@ class DeployProject: "--api-key", "test-key", "--host-url", - CONTROL_PLANE_URL, + host_url, "--name", "my-app", "--no-input", @@ -496,6 +512,7 @@ def test_push_to_builds_pushes_then_creates_an_external_deployment( assert result.exit_code == 0, result.output assert deploy_project.timeline == [ LIST_DEPLOYMENTS, + LIST_LISTENERS, "docker build", "docker push", "docker inspect-digest", @@ -611,18 +628,6 @@ def test_push_to_rejects_a_non_external_deployment_before_any_docker_work( assert deploy_project.docker.verbs() == [] -def test_push_to_explains_the_listener_requirement_of_hybrid_workspaces( - deploy_project: DeployProject, -) -> None: - deploy_project.control_plane.create_error = LISTENER_REQUIRED - - result = deploy_project.run("--push-to", PUSH_REPOSITORY) - - assert result.exit_code != 0 - assert "listener" in result.output - assert "--deployment-id" in result.output - - def test_push_to_with_deployment_id_fetches_the_deployment_once( deploy_project: DeployProject, ) -> None: @@ -652,3 +657,135 @@ def test_invalid_tag_fails_before_any_control_plane_call( assert result.exit_code != 0 assert "Image tag may only contain" in result.output assert deploy_project.timeline == [] + + +def test_push_to_places_a_new_deployment_on_the_only_listener( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + + result = deploy_project.run( + "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL + ) + + assert result.exit_code == 0, result.output + assert deploy_project.timeline == [ + LIST_DEPLOYMENTS, + LIST_LISTENERS, + "docker build", + "docker push", + "docker inspect-digest", + CREATE_DEPLOYMENT, + ] + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == { + "resource_spec": {}, + "listener_id": "listener-1", + "listener_config": {"k8s_namespace": "agents"}, + } + + +def test_push_to_places_a_new_deployment_on_the_chosen_listener( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER, OTHER_LISTENER] + + result = deploy_project.run( + "--push-to", + PUSH_REPOSITORY, + "--listener-id", + "listener-2", + "--k8s-namespace", + "agents", + host_url=CLOUD_CONTROL_PLANE_URL, + ) + + assert result.exit_code == 0, result.output + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == { + "resource_spec": {}, + "listener_id": "listener-2", + "listener_config": {"k8s_namespace": "agents"}, + } + + +@pytest.mark.parametrize( + ("listeners", "args", "message"), + [ + pytest.param( + [LISTENER, OTHER_LISTENER], (), "--listener-id", id="two_listeners" + ), + pytest.param( + [TWO_NAMESPACE_LISTENER], (), "--k8s-namespace", id="two_namespaces" + ), + pytest.param( + [LISTENER], + ("--listener-id", "listener-9"), + "was not found", + id="unknown_listener", + ), + pytest.param( + [LISTENER], + ("--k8s-namespace", "nope"), + "does not serve namespace", + id="unknown_namespace", + ), + ], +) +def test_push_to_refuses_an_unresolved_placement_before_any_docker_work( + deploy_project: DeployProject, listeners, args, message +) -> None: + deploy_project.control_plane.listeners = listeners + + result = deploy_project.run( + "--push-to", PUSH_REPOSITORY, *args, host_url=CLOUD_CONTROL_PLANE_URL + ) + + assert result.exit_code != 0 + assert message in result.output + assert deploy_project.docker.verbs() == [] + assert CREATE_DEPLOYMENT not in deploy_project.timeline + + +def test_self_hosted_control_plane_keeps_its_default_placement( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + + result = deploy_project.run("--push-to", PUSH_REPOSITORY) + + assert result.exit_code == 0, result.output + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == { + "resource_spec": {} + } + + +def test_self_hosted_control_plane_places_when_asked( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + + result = deploy_project.run( + "--push-to", PUSH_REPOSITORY, "--listener-id", "listener-1" + ) + + assert result.exit_code == 0, result.output + assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == { + "resource_spec": {}, + "listener_id": "listener-1", + "listener_config": {"k8s_namespace": "agents"}, + } + + +def test_updating_a_deployment_never_looks_up_listeners( + deploy_project: DeployProject, +) -> None: + deploy_project.control_plane.listeners = [LISTENER] + deploy_project.control_plane.existing_deployments = [ + {"id": "dep-ext", "name": "my-app", "source": "external_docker"} + ] + + result = deploy_project.run( + "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL + ) + + assert result.exit_code == 0, result.output + assert LIST_LISTENERS not in deploy_project.timeline diff --git a/libs/cli/tests/unit_tests/test_deploy_helpers.py b/libs/cli/tests/unit_tests/test_deploy_helpers.py index 8b369a6db..820aa7b8e 100644 --- a/libs/cli/tests/unit_tests/test_deploy_helpers.py +++ b/libs/cli/tests/unit_tests/test_deploy_helpers.py @@ -613,6 +613,7 @@ class TestSelectSource: "image_name": None, "tag": None, "remote_build_flag": None, + "placement": RequestedPlacement(), } REPOSITORY = "registry.example.com/app" @@ -651,6 +652,19 @@ class TestSelectSource: ), id="prebuilt_image_is_retagged_for_push_to_without_docker_checks", ), + pytest.param( + { + "push_to": REPOSITORY, + "placement": RequestedPlacement("listener-1", "agents"), + }, + True, + CustomerRegistrySource( + ImageReference(REPOSITORY, "latest"), + None, + RequestedPlacement("listener-1", "agents"), + ), + id="push_to_carries_the_requested_placement", + ), pytest.param( {"remote_build_flag": True}, True,