pyro-mcp/examples/python_workspace.py

from __future__ import annotations

import tempfile
from pathlib import Path

from pyro_mcp import Pyro


def main() -> None:
    pyro = Pyro()
    with (
        tempfile.TemporaryDirectory(prefix="pyro-workspace-seed-") as seed_dir,
        tempfile.TemporaryDirectory(prefix="pyro-workspace-sync-") as sync_dir,
        tempfile.TemporaryDirectory(prefix="pyro-workspace-export-") as export_dir,
        tempfile.TemporaryDirectory(prefix="pyro-workspace-disk-") as disk_dir,
        tempfile.TemporaryDirectory(prefix="pyro-workspace-secret-") as secret_dir,
    ):
        Path(seed_dir, "note.txt").write_text("hello from seed\n", encoding="utf-8")
        Path(sync_dir, "note.txt").write_text("hello from sync\n", encoding="utf-8")
        secret_file = Path(secret_dir, "token.txt")
        secret_file.write_text("from-file\n", encoding="utf-8")
        created = pyro.create_workspace(
            environment="debian:12",
            seed_path=seed_dir,
            name="repro-fix",
            labels={"issue": "123"},
            network_policy="egress+published-ports",
            secrets=[
                {"name": "API_TOKEN", "value": "expected"},
                {"name": "FILE_TOKEN", "file_path": str(secret_file)},
            ],
        )
        workspace_id = str(created["workspace_id"])
        try:
            listed = pyro.list_workspaces()
            print(f"workspace_count={listed['count']}")
            updated = pyro.update_workspace(
                workspace_id,
                labels={"owner": "codex"},
            )
            print(updated["labels"]["owner"])
            pyro.push_workspace_sync(workspace_id, sync_dir)
            files = pyro.list_workspace_files(workspace_id, path="/workspace", recursive=True)
            print(f"workspace_entries={len(files['entries'])}")
            note = pyro.read_workspace_file(workspace_id, "note.txt")
            print(note["content"], end="")
            written = pyro.write_workspace_file(
                workspace_id,
                "src/app.py",
                text="print('hello from file ops')\n",
            )
            print(f"written_bytes={written['bytes_written']}")
            patched = pyro.apply_workspace_patch(
                workspace_id,
                patch=(
                    "--- a/note.txt\n"
                    "+++ b/note.txt\n"
                    "@@ -1 +1 @@\n"
                    "-hello from sync\n"
                    "+hello from patch\n"
                ),
            )
            print(f"patch_changed={patched['changed']}")
            result = pyro.exec_workspace(workspace_id, command="cat note.txt")
            print(result["stdout"], end="")
            secret_result = pyro.exec_workspace(
                workspace_id,
                command='sh -lc \'printf "%s\\n" "$API_TOKEN"\'',
                secret_env={"API_TOKEN": "API_TOKEN"},
            )
            print(secret_result["stdout"], end="")
            diff_result = pyro.diff_workspace(workspace_id)
            print(f"changed={diff_result['changed']} total={diff_result['summary']['total']}")
            snapshot = pyro.create_snapshot(workspace_id, "checkpoint")
            print(snapshot["snapshot"]["snapshot_name"])
            exported_path = Path(export_dir, "note.txt")
            pyro.export_workspace(workspace_id, "note.txt", output_path=exported_path)
            print(exported_path.read_text(encoding="utf-8"), end="")
            shell = pyro.open_shell(workspace_id, secret_env={"API_TOKEN": "API_TOKEN"})
            shell_id = str(shell["shell_id"])
            pyro.write_shell(
                workspace_id,
                shell_id,
                input='printf "%s\\n" "$API_TOKEN"',
            )
            shell_output = pyro.read_shell(
                workspace_id,
                shell_id,
                cursor=0,
                plain=True,
                wait_for_idle_ms=300,
            )
            print(f"shell_output_len={len(shell_output['output'])}")
            pyro.close_shell(workspace_id, shell_id)
            pyro.start_service(
                workspace_id,
                "web",
                command="touch .web-ready && while true; do sleep 60; done",
                readiness={"type": "file", "path": ".web-ready"},
                secret_env={"API_TOKEN": "API_TOKEN"},
                published_ports=[{"guest_port": 8080}],
            )
            services = pyro.list_services(workspace_id)
            print(f"services={services['count']} running={services['running_count']}")
            service_status = pyro.status_service(workspace_id, "web")
            print(f"service_state={service_status['state']} ready_at={service_status['ready_at']}")
            print(f"published_ports={service_status['published_ports']}")
            service_logs = pyro.logs_service(workspace_id, "web", tail_lines=20)
            print(f"service_stdout_len={len(service_logs['stdout'])}")
            pyro.stop_service(workspace_id, "web")
            stopped = pyro.stop_workspace(workspace_id)
            print(f"stopped_state={stopped['state']}")
            disk_listing = pyro.list_workspace_disk(workspace_id, path="/workspace", recursive=True)
            print(f"disk_entries={len(disk_listing['entries'])}")
            disk_read = pyro.read_workspace_disk(workspace_id, "note.txt")
            print(disk_read["content"], end="")
            disk_image = Path(disk_dir, "workspace.ext4")
            pyro.export_workspace_disk(workspace_id, output_path=disk_image)
            print(f"disk_bytes={disk_image.stat().st_size}")
            started = pyro.start_workspace(workspace_id)
            print(f"started_state={started['state']}")
            reset = pyro.reset_workspace(workspace_id, snapshot="checkpoint")
            print(f"reset_count={reset['reset_count']}")
            print(f"secret_count={len(reset['secrets'])}")
            logs = pyro.logs_workspace(workspace_id)
            print(f"workspace_id={workspace_id} command_count={logs['count']}")
        finally:
            pyro.delete_workspace(workspace_id)


if __name__ == "__main__":
    main()