Create, control, and delete a Windows Cua Fleet with cua-fleet (no Terraform)

cua-fleet-windows-all-code.md

Create, Control, and Delete a Windows Cua Fleet with cua-fleet

This is an all-Python alternative to Terraform. It creates a Windows Cua Fleet pool through the PyPI cua-fleet SDK, waits for a sandbox claim, initializes the Windows computer-server MCP endpoint, and deletes the claim and pool when it exits.

It mirrors Cua's Windows Fleet configuration:

• KubeVirt runtime
• UEFI firmware (efi)
• cua-server-windows:latest container-disk image
• 4 vCPU / 4 GiB memory
• computer-server exposed on TCP port 8000
• TCP readiness probe on port 8000
• warm-pool autoscaling: min 0, initial 1, max 5

This script deletes the Fleet pool and its associated namespace in finally. Set CUA_KEEP_POOL=1 while debugging if you want to keep the pool after it finishes. Claims are always released.

Requirements

• Python 3.10+ and uv
• A run.cua.ai OAuth user key with permission to create pools and claims
• A unique lowercase DNS-label value for CUA_POOL (maximum 63 characters)

Use a user key (ukey-...), not a pool key (key-...). Pool creation and claims need an identity that can manage Kubernetes resources.

Configuration

export CUA_CLIENT_ID='ukey-...'
export CUA_CLIENT_SECRET='...'
export CUA_BASE_URL='https://run.cua.ai'
export CUA_TOKEN_URL='https://auth.cua.ai/realms/cyclops-cs/protocol/openid-connect/token'
export CUA_POOL='windows-fleet-example'

# Optional: retain the pool after the script completes or fails.
# export CUA_KEEP_POOL=1

Script

Save this as windows_fleet_live_control.py:

#!/usr/bin/env python3
import asyncio
import json
import os
import time
import urllib.error
import urllib.request

from cyclops_sdk import (
    CreateClaimRequest,
    CreatePoolRequest,
    CyclopsClient,
    CyclopsConfiguration,
    CyclopsCredentials,
    Firmware,
    HttpClient,
    HttpHeader,
    HttpRequest,
    HttpResponse,
    PoolSpec,
    PoolTemplate,
    PreservedJson,
    RuntimeKind,
    SandboxService,
    ServiceProtocol,
    WarmPoolAutoscaling,
)

WINDOWS_IMAGE = (
    "296062593712.dkr.ecr.us-west-2.amazonaws.com/cua-server-windows:latest"
)


class UrlLibHttpClient(HttpClient):
    async def execute(self, request: HttpRequest) -> HttpResponse:
        return await asyncio.to_thread(self._execute, request)

    def _execute(self, request: HttpRequest) -> HttpResponse:
        native = urllib.request.Request(
            request.url,
            data=request.body,
            method=request.method,
            headers={header.name: header.value for header in request.headers},
        )
        try:
            with urllib.request.urlopen(native, timeout=60) as response:
                return HttpResponse(
                    status=response.status,
                    headers=[
                        HttpHeader(name=name, value=value)
                        for name, value in response.headers.items()
                    ],
                    body=response.read(),
                )
        except urllib.error.HTTPError as error:
            return HttpResponse(
                status=error.code,
                headers=[
                    HttpHeader(name=name, value=value)
                    for name, value in error.headers.items()
                ],
                body=error.read(),
            )


def windows_pool_spec() -> PoolSpec:
    probes = PreservedJson.from_json(
        json.dumps(
            {
                "readinessProbe": {
                    "tcpSocket": {"port": 8000},
                    "initialDelaySeconds": 60,
                    "periodSeconds": 5,
                    "timeoutSeconds": 3,
                    "failureThreshold": 120,
                }
            }
        )
    )
    return PoolSpec(
        replicas=0,
        template=PoolTemplate(
            runtime=RuntimeKind.KUBEVIRT,
            runtime_class_name=None,
            node_selector=None,
            tolerations=None,
            command=None,
            container_disk_image=WINDOWS_IMAGE,
            image_pull_secret="ecr-credentials",
            cpu_cores=4,
            memory="4Gi",
            firmware=Firmware.EFI,
            probes=probes,
            oidc=None,
        ),
        autoscaling=WarmPoolAutoscaling(
            min_pool_size=0,
            initial_pool_size=1,
            max_pool_size=5,
        ),
        services=[
            SandboxService(
                name="computer-server",
                target_port=8000,
                protocol=ServiceProtocol.TCP,
            )
        ],
    )


async def initialize_mcp(client: CyclopsClient, sandbox) -> int:
    body = json.dumps(
        {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-03-26",
                "capabilities": {},
                "clientInfo": {
                    "name": "cua-fleet-windows-all-code",
                    "version": "0.1.0",
                },
            },
        },
        separators=(",", ":"),
    ).encode()
    deadline = time.monotonic() + 300

    while True:
        response = await client.service_request(
            sandbox,
            "computer-server",
            "/mcp",
            HttpRequest(
                method="POST",
                url="https://ignored.invalid/mcp",
                headers=[
                    HttpHeader(
                        name="accept",
                        value="application/json, text/event-stream",
                    ),
                    HttpHeader(name="content-type", value="application/json"),
                ],
                body=body,
            ),
        )
        if 200 <= response.status < 300:
            return response.status
        if response.status in (502, 503, 504) and time.monotonic() < deadline:
            print(f"computer-server is starting: HTTP {response.status}")
            await asyncio.sleep(5)
            continue
        raise RuntimeError(
            f"MCP initialize failed with HTTP {response.status}: {response.body!r}"
        )


async def main() -> None:
    pool_name = os.environ["CUA_POOL"]
    keep_pool = os.environ.get("CUA_KEEP_POOL") == "1"

    client = CyclopsClient.connect(
        CyclopsConfiguration(
            base_url=os.environ["CUA_BASE_URL"],
            token_url=os.environ["CUA_TOKEN_URL"],
            credentials=CyclopsCredentials(
                os.environ["CUA_CLIENT_ID"],
                os.environ["CUA_CLIENT_SECRET"],
            ),
            pool_poll_interval_ms=5000,
            pool_poll_limit=240,
            claim_poll_interval_ms=5000,
            claim_poll_limit=120,
        ),
        UrlLibHttpClient(),
    )

    pool = None
    claim = None
    try:
        pool = await client.create_pool(
            CreatePoolRequest(namespace=pool_name, spec=windows_pool_spec())
        )
        print(f"created pool {pool.metadata.name!r}; waiting for its first Windows VM")

        claim = await client.create_claim(CreateClaimRequest(pool=pool, spec=None))
        sandbox = await client.wait_claim(claim)
        mcp_status = await initialize_mcp(client, sandbox)
        print(
            json.dumps(
                {
                    "namespace": pool_name,
                    "pool": pool.metadata.name,
                    "claim": claim.metadata.name,
                    "sandbox": sandbox.name,
                    "mcp_status": mcp_status,
                },
                sort_keys=True,
            )
        )
    finally:
        if claim is not None:
            await client.delete_claim(claim)
        if pool is not None and not keep_pool:
            await client.delete_pool(pool)
        elif pool is not None:
            print(f"keeping pool {pool.metadata.name!r} because CUA_KEEP_POOL=1")


asyncio.run(main())

Run it with uv

uv run --with cua-fleet==0.0.5 python windows_fleet_live_control.py

A fresh Windows fleet may take several minutes to pull the container-disk image, boot through UEFI, and start computer-server. The SDK waits for the claim to bind, and the script separately retries the MCP request while the in-guest service finishes starting.

Cleanup behavior

The default is safe for one-off experiments:

• The claim is deleted in all cases.
• The pool is deleted in all cases after the claim cleanup succeeds.
• Set CUA_KEEP_POOL=1 to keep the pool and namespace; release any remaining claim before deleting the pool manually with a later run or your Fleet admin tooling.