Cua Docs

Connect to a sandbox service

Reach a sandbox service through a Fleet service request, a local startup mapping, or a supported tunnel transport.

Choose the connection method for your sandbox's transport. In cua-sandbox 0.7.0, local VM HTTPTransport and pool-backed FleetTransport do not implement sb.tunnel.forward(). Use the supported paths below; see Sandbox runtime support for the versioned transport contract.

Request a named Fleet service#

For a hosted Fleet guest, configure Fleet pool credentials and use a pool whose guest starts the desired service. Within the claimed sandbox's context, send an authenticated request by service name:

response = await sb.services.request('port-3000', method='GET', path='/healthz')
response.raise_for_status()
print(response.text)

An image's .expose(3000) declaration creates the port-3000 service in the default pool template. It does not start an application on that port or create a localhost listener. See Prepare and reference a Fleet image for the complete image and claim workflow. External clients need the Fleet service URL and authentication; an endpoint URL is not anonymous access.

Expose a local QEMU service at startup#

For a local sandbox, declare the port on the image before starting a bare-metal QEMU VM, then read the host mapping from sb.exposed_ports. This example requires QEMU, a compatible host, cua==0.1.6, and httpx:

import asyncio
import httpx
from cua import Image, QEMURuntime, Sandbox
 
async def main():
    image = Image.linux().expose(8080)
    async with Sandbox.ephemeral(
        image, local=True, runtime=QEMURuntime(mode='bare-metal')
    ) as sb:
        await sb.shell.run(
            'nohup python3 -m http.server 8080 --bind 0.0.0.0 '
            '>/tmp/docs-http-server.log 2>&1 </dev/null &'
        )
        host_port = sb.exposed_ports[8080]
        url = f'http://127.0.0.1:{host_port}/'
        async with httpx.AsyncClient(timeout=5.0) as client:
            for attempt in range(30):
                try:
                    response = await client.get(url)
                    response.raise_for_status()
                    print(f'Service is ready at {url}')
                    break
                except httpx.HTTPError:
                    if attempt == 29:
                        raise
                    await asyncio.sleep(1)
 
asyncio.run(main())

The service must listen on a guest-reachable address, not only guest localhost. The mapping lasts for the VM's lifetime; exiting the ephemeral context stops the VM and removes the mapping. Other local runtimes may expose ports differently.

For multiple ports, chain declarations such as Image.linux().expose(8080).expose(9222) and read each guest port's entry from sb.exposed_ports after startup.

Forward through SSH or ADB#

This section applies only to an existing sandbox connected through SSHTransport or ADBTransport. The tunnel helper creates a localhost forwarding endpoint. Start the guest service first, then keep its tunnel open for as long as the client needs it:

# sb must use a transport that implements forwarding, such as SSH or ADB.
async with sb.tunnel.forward(8080) as tunnel:
    print(tunnel.url)
    # Connect your local client before leaving this context.

For several ports, sb.tunnel.forward(8080, 9222) returns a dictionary keyed by guest port. When using tunnel = await sb.tunnel.forward(8080) without a context manager, call await tunnel.close() after the client finishes.

Android abstract sockets#

ADB also accepts a guest abstract socket name, such as the DevTools socket of a running Android Chrome instance:

# sb must be connected to the Android guest through ADBTransport.
async with sb.tunnel.forward('chrome_devtools_remote') as tunnel:
    print(tunnel.url)

A socket name or integer port does not enable forwarding on an unsupported transport. See Sandbox SDK interfaces for the tunnel handle's fields and cleanup behavior.