feat(cli): place new self-hosted deployments on a listener

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 <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:

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

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,