Add computer port publish commands

Author: aniketmaurya

src/celesto/computer.py

@@ -15,6 +15,8 @@
 from .sdk.client import Celesto
 
 app = typer.Typer(help="Create, manage, and connect to sandboxed computers.")
+port_app = typer.Typer(help="Publish and unpublish computer ports.")
+app.add_typer(port_app, name="port")
 console = Console()
 
 # Common option types
@@ -73,6 +75,77 @@ def _print_json(data: object) -> None:
     sys.stdout.write(json.dumps(data, indent=2, default=str) + "\n")
 
 
+@port_app.command("publish")
+def publish_port(
+    computer_id: Annotated[str, typer.Argument(help="Computer ID or name")],
+    port: Annotated[int, typer.Option("--port", "-p", help="Port to publish")] = 8000,
+    as_json: JsonOption = False,
+    api_key: ApiKeyOption = None,
+):
+    """Publish a computer port to the internet."""
+    with _get_client(api_key) as client:
+        result = client.computers.publish_port(computer_id, port=port)
+
+    if as_json:
+        _print_json(result)
+        return
+
+    console.print(result.get("url") or "")
+
+
+@port_app.command("list")
+def list_ports(
+    computer_id: Annotated[str, typer.Argument(help="Computer ID or name")],
+    as_json: JsonOption = False,
+    api_key: ApiKeyOption = None,
+):
+    """List published ports for a computer."""
+    with _get_client(api_key) as client:
+        ports = client.computers.list_published_ports(computer_id)
+
+    if as_json:
+        _print_json(ports)
+        return
+
+    if not ports:
+        console.print("[dim]No published ports.[/dim]")
+        return
+
+    table = Table(show_header=True, header_style="bold")
+    table.add_column("Port", justify="right")
+    table.add_column("Status")
+    table.add_column("URL")
+    table.add_column("Created")
+
+    for published_port in ports:
+        table.add_row(
+            str(published_port.get("port", "")),
+            str(published_port.get("status", "")),
+            str(published_port.get("url") or ""),
+            str(published_port.get("created_at") or "")[:19],
+        )
+
+    console.print(table)
+
+
+@port_app.command("unpublish")
+def unpublish_port(
+    computer_id: Annotated[str, typer.Argument(help="Computer ID or name")],
+    port: Annotated[int, typer.Option("--port", "-p", help="Port to unpublish")] = 8000,
+    as_json: JsonOption = False,
+    api_key: ApiKeyOption = None,
+):
+    """Unpublish a computer port."""
+    with _get_client(api_key) as client:
+        result = client.computers.unpublish_port(computer_id, port=port)
+
+    if as_json:
+        _print_json(result)
+        return
+
+    console.print(f"[dim]Port {port} is unpublished.[/dim]")
+
+
 @app.command("create")
 def create_computer(
     cpus: Annotated[

src/celesto/sdk/client.py

@@ -892,6 +892,45 @@ def get(self, computer_id: str) -> dict[str, Any]:
         """
         return self._request("GET", f"/computers/{computer_id}")
 
+    def publish_port(self, computer_id: str, port: int = 8000) -> dict[str, Any]:
+        """Publish a computer port to the internet.
+
+        Args:
+            computer_id: Computer ID or name.
+            port: Port to publish. The MVP supports port 8000.
+
+        Returns:
+            Published port dict with id, computer_id, port, url, status, and created_at.
+        """
+        return self._request(
+            "POST",
+            f"/computers/{computer_id}/published-ports",
+            json_body={"port": port},
+        )
+
+    def list_published_ports(self, computer_id: str) -> list[dict[str, Any]]:
+        """List active published ports for a computer.
+
+        Args:
+            computer_id: Computer ID or name.
+
+        Returns:
+            List of published port dicts.
+        """
+        return self._request("GET", f"/computers/{computer_id}/published-ports")
+
+    def unpublish_port(self, computer_id: str, port: int = 8000) -> dict[str, Any]:
+        """Remove a published computer port.
+
+        Args:
+            computer_id: Computer ID or name.
+            port: Published port to remove. The MVP supports port 8000.
+
+        Returns:
+            Unpublished port dict.
+        """
+        return self._request("DELETE", f"/computers/{computer_id}/published-ports/{port}")
+
     def exec(
         self,
         computer_id: str,

src/celesto/sdk/types.py

@@ -123,6 +123,26 @@ class ComputerConnectionInfo(TypedDict):
     access_url: NotRequired[str]
 
 
+PublishedPortStatus = Literal[
+    "publishing",
+    "published",
+    "unpublishing",
+    "unpublished",
+    "error",
+]
+
+
+class ComputerPublishedPortInfo(TypedDict):
+    """Information about a public computer port route."""
+
+    id: NotRequired[str | None]
+    computer_id: str
+    port: int
+    url: NotRequired[str | None]
+    status: PublishedPortStatus
+    created_at: NotRequired[str | None]
+
+
 class ComputerInfo(TypedDict):
     """Information about a computer."""
 
@@ -136,6 +156,7 @@ class ComputerInfo(TypedDict):
     template_id: str
     template_version: NotRequired[str | None]
     connection: NotRequired[ComputerConnectionInfo | None]
+    published_ports: NotRequired[List[ComputerPublishedPortInfo]]
     last_error: NotRequired[str | None]
     created_at: str
     stopped_at: NotRequired[str | None]
@@ -188,6 +209,8 @@ class ComputerExecResponse(TypedDict):
     # Computer
     "ComputerStatus",
     "ComputerConnectionInfo",
+    "PublishedPortStatus",
+    "ComputerPublishedPortInfo",
     "ComputerInfo",
     "SandboxTemplateInfo",
     "ComputerListResponse",

tests/test_computer_cli.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+import json
+
+from typer.testing import CliRunner
+
+from celesto import computer
+
+
+class _FakeComputers:
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, str, int | None]] = []
+
+    def publish_port(self, computer_id: str, *, port: int = 8000) -> dict:
+        self.calls.append(("publish", computer_id, port))
+        return {
+            "id": "cpp_123",
+            "computer_id": "cmp_123",
+            "port": port,
+            "url": "https://p-test.celesto.ai",
+            "status": "published",
+            "created_at": "2026-06-07T00:00:00Z",
+        }
+
+    def list_published_ports(self, computer_id: str) -> list[dict]:
+        self.calls.append(("list", computer_id, None))
+        return [
+            {
+                "id": "cpp_123",
+                "computer_id": "cmp_123",
+                "port": 8000,
+                "url": "https://p-test.celesto.ai",
+                "status": "published",
+                "created_at": "2026-06-07T00:00:00Z",
+            }
+        ]
+
+    def unpublish_port(self, computer_id: str, *, port: int = 8000) -> dict:
+        self.calls.append(("unpublish", computer_id, port))
+        return {
+            "computer_id": "cmp_123",
+            "port": port,
+            "url": None,
+            "status": "unpublished",
+            "created_at": None,
+        }
+
+
+class _FakeClient:
+    def __init__(self) -> None:
+        self.computers = _FakeComputers()
+
+    def __enter__(self) -> "_FakeClient":
+        return self
+
+    def __exit__(self, *exc: object) -> None:
+        return None
+
+
+def test_computer_port_publish_prints_url(monkeypatch):
+    runner = CliRunner()
+    fake_client = _FakeClient()
+    monkeypatch.setattr(computer, "_get_client", lambda api_key=None: fake_client)
+
+    result = runner.invoke(computer.app, ["port", "publish", "curie"])
+
+    assert result.exit_code == 0
+    assert "https://p-test.celesto.ai" in result.output
+    assert fake_client.computers.calls == [("publish", "curie", 8000)]
+
+
+def test_computer_port_publish_json(monkeypatch):
+    runner = CliRunner()
+    fake_client = _FakeClient()
+    monkeypatch.setattr(computer, "_get_client", lambda api_key=None: fake_client)
+
+    result = runner.invoke(computer.app, ["port", "publish", "cmp_123", "--json"])
+
+    assert result.exit_code == 0
+    payload = json.loads(result.output)
+    assert payload["url"] == "https://p-test.celesto.ai"
+    assert fake_client.computers.calls == [("publish", "cmp_123", 8000)]
+
+
+def test_computer_port_list_json(monkeypatch):
+    runner = CliRunner()
+    fake_client = _FakeClient()
+    monkeypatch.setattr(computer, "_get_client", lambda api_key=None: fake_client)
+
+    result = runner.invoke(computer.app, ["port", "list", "cmp_123", "--json"])
+
+    assert result.exit_code == 0
+    payload = json.loads(result.output)
+    assert payload[0]["port"] == 8000
+    assert fake_client.computers.calls == [("list", "cmp_123", None)]
+
+
+def test_computer_port_unpublish_json(monkeypatch):
+    runner = CliRunner()
+    fake_client = _FakeClient()
+    monkeypatch.setattr(computer, "_get_client", lambda api_key=None: fake_client)
+
+    result = runner.invoke(computer.app, ["port", "unpublish", "cmp_123", "--json"])
+
+    assert result.exit_code == 0
+    payload = json.loads(result.output)
+    assert payload["status"] == "unpublished"
+    assert fake_client.computers.calls == [("unpublish", "cmp_123", 8000)]

tests/test_sdk.py

@@ -106,3 +106,57 @@ def test_computers_list_templates_hits_backend_endpoint():
     assert templates[0]["id"] == "scratch"
     assert session.calls[0]["method"] == "GET"
     assert session.calls[0]["url"] == "https://api.example.test/v1/computers/templates"
+
+
+def test_computers_publish_port_hits_backend_endpoint():
+    session = DummySession(
+        payload={
+            "id": "cpp_123",
+            "computer_id": "cmp_123",
+            "port": 8000,
+            "url": "https://p-test.celesto.ai",
+            "status": "published",
+            "created_at": "2026-06-07T00:00:00Z",
+        }
+    )
+    client = Celesto("test-key", base_url="https://api.example.test/v1")
+    client.session = session
+
+    result = client.computers.publish_port("curie", port=8000)
+
+    assert result["url"] == "https://p-test.celesto.ai"
+    assert session.calls[0]["method"] == "POST"
+    assert session.calls[0]["url"] == "https://api.example.test/v1/computers/curie/published-ports"
+    assert session.calls[0]["json"] == {"port": 8000}
+
+
+def test_computers_list_published_ports_hits_backend_endpoint():
+    session = DummySession(payload=[])
+    client = Celesto("test-key", base_url="https://api.example.test/v1")
+    client.session = session
+
+    result = client.computers.list_published_ports("cmp_123")
+
+    assert result == []
+    assert session.calls[0]["method"] == "GET"
+    assert session.calls[0]["url"] == "https://api.example.test/v1/computers/cmp_123/published-ports"
+
+
+def test_computers_unpublish_port_hits_backend_endpoint():
+    session = DummySession(
+        payload={
+            "computer_id": "cmp_123",
+            "port": 8000,
+            "url": None,
+            "status": "unpublished",
+            "created_at": None,
+        }
+    )
+    client = Celesto("test-key", base_url="https://api.example.test/v1")
+    client.session = session
+
+    result = client.computers.unpublish_port("cmp_123", port=8000)
+
+    assert result["status"] == "unpublished"
+    assert session.calls[0]["method"] == "DELETE"
+    assert session.calls[0]["url"] == "https://api.example.test/v1/computers/cmp_123/published-ports/8000"