feat(cli): read the workspace listeners from the control plane

libs/cli/langgraph_cli/deploy.py

@@ -155,6 +155,29 @@ class ByName:
 DeploymentSelector = ById | ByName
 
 
+@dataclass(frozen=True, slots=True)
+class Listener:
+    id: str
+    compute_id: str
+    namespaces: tuple[str, ...]
+
+    @classmethod
+    def from_resource(cls, resource: Mapping[str, object]) -> "Listener":
+        compute_config = resource.get("compute_config")
+        namespaces = (
+            compute_config.get("k8s_namespaces")
+            if isinstance(compute_config, Mapping)
+            else None
+        )
+        return cls(
+            str(resource.get("id", "")),
+            str(resource.get("compute_id", "")),
+            tuple(str(namespace) for namespace in namespaces)
+            if isinstance(namespaces, list)
+            else (),
+        )
+
+
 @dataclass(frozen=True, slots=True)
 class ExistingDeployment:
     id: str

libs/cli/langgraph_cli/host_backend.py

@@ -18,6 +18,7 @@ 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"]
 
 
@@ -190,6 +191,11 @@ class HostBackendClient:
             )
         )
 
+    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]:
         return self._request("GET", f"/v2/deployments/{deployment_id}")
 

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -15,6 +15,7 @@ import langgraph_cli.deploy as deploy_mod
 from langgraph_cli.deploy import (
     CustomerRegistrySource,
     DockerBuildCommand,
+    Listener,
     ManagedRegistrySource,
     RemoteBuildSource,
     _call_host_backend_with_optional_tenant,
@@ -892,3 +893,37 @@ class TestResolvePushedImageDigest:
         frame_locals = captured["coro"].cr_frame.f_locals
         assert "--config" not in frame_locals["args"]
         captured["coro"].close()
+
+
+class TestListener:
+    @pytest.mark.parametrize(
+        ("resource", "expected"),
+        [
+            pytest.param(
+                {
+                    "id": "listener-1",
+                    "compute_id": "prod-cluster",
+                    "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
+                },
+                Listener("listener-1", "prod-cluster", ("agents", "agents-staging")),
+                id="reads_id_cluster_and_namespaces",
+            ),
+            pytest.param(
+                {"id": "listener-1", "compute_id": "c", "compute_config": {}},
+                Listener("listener-1", "c", ()),
+                id="missing_namespaces",
+            ),
+            pytest.param(
+                {"id": "listener-1", "compute_id": "c", "compute_config": None},
+                Listener("listener-1", "c", ()),
+                id="null_compute_config",
+            ),
+            pytest.param(
+                {"id": "listener-1"},
+                Listener("listener-1", "", ()),
+                id="only_an_id",
+            ),
+        ],
+    )
+    def test_from_resource_reads_the_control_plane_shape(self, resource, expected):
+        assert Listener.from_resource(resource) == expected

libs/cli/tests/unit_tests/test_host_backend.py

@@ -573,3 +573,18 @@ def test_list_endpoints_return_resource_objects(payload, expected):
     )
 
     assert c.list_deployments() == expected
+
+
+def test_list_listeners_asks_for_a_full_page():
+    seen: dict = {}
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        seen["url"] = str(req.url)
+        return httpx.Response(200, json={"resources": [{"id": "listener-1"}]})
+
+    c = HostBackendClient(
+        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
+    )
+
+    assert c.list_listeners() == [{"id": "listener-1"}]
+    assert seen["url"] == "https://api.example.com/v2/listeners?limit=100"