Services, tunnels and MCP

Reach ports inside a sandbox: named services, tunnels, shareable URLs and MCP servers.

Named services#

ServiceHandle#

class ServiceHandle

One named service of a sandbox (sb.service("mcp")).

The same calls work local and in the cloud: requests carry your headers (the cloud gateway's credentials are added for you), url() is usable from this machine and public_url() can be shared.

ServiceHandle.request#

async def request(
    method: str,
    path: str = '/',
    *,
    headers: Headers = None,
    body: bytes | str | None = None,
    json: Any = None,
    timeout: float = 30.0,
) -> Any

One HTTP request to path on the service; returns an httpx.Response. authorization is refused in the cloud, where the gateway credentials own it.

ServiceHandle.url#

async def url() -> str

A URL for the service usable from this machine with no credentials (no trailing slash): the published loopback port locally, a signed URL (1 h, renewed on later calls) in the cloud.

ServiceHandle.public_url#

async def public_url(ttl: float = 3600, *, label: Optional[str] = None) -> 'PublicUrl'

A shareable URL that stops working after ttl seconds (60 s to 24 h). See cua_sandbox.Sandbox.public_url.

Services#

class Services

Request a named service exposed by the active sandbox connection.

Services.request#

async def request(
    name: str,
    *,
    method: str,
    path: str,
    json: Any = None,
    headers: dict[str, str] | None = None,
) -> Any

Send a request to a named service on this sandbox.

Prefer sb.service(name).request(...), which works the same local and in the cloud.

Services.create_signed_url#

async def create_signed_url(
    name: str,
    *,
    expires_in_seconds: int,
    label: str | None = None,
) -> SignedServiceURL

Create a revocable public URL for a named cloud service.

Prefer sb.public_url(name, ttl=...), which also works locally.

Services.list_signed_urls#

async def list_signed_urls() -> list[SignedServiceURL]

List signed service URLs created for this cloud sandbox.

Services.revoke_signed_url#

async def revoke_signed_url(signed_url: SignedServiceURL) -> None

Revoke a previously created signed service URL.

SignedServiceURL#

class SignedServiceURL

A revocable public URL for one sandbox service.

Provider internals (the cloud namespace and claim) live in details.

SignedServiceURL.from_resource#

@classmethod
def from_resource(resource: Any) -> 'SignedServiceURL'

Tunnels#

Tunnel#

class Tunnel

Port-forwarding interface: exposes sandbox ports on the host.

Tunnel.forward#

def forward(*ports: int | str) -> _TunnelContext

Forward one or more sandbox ports (or abstract sockets) to the host.

ports may be:

• int: TCP port inside the sandbox (e.g. 8080)
• str: Android abstract socket name (e.g. "chrome_devtools_remote")

Returns a context manager (or awaitable) that yields:

• a single TunnelInfo when one target is given
• a dict[sandbox_port, TunnelInfo] when multiple targets are given

Example: Chrome DevTools over ADB:

async with sb.tunnel.forward("chrome_devtools_remote") as t:
    # http://localhost:<random>/json lists CDP targets
    print(t.url)

TunnelInfo#

class TunnelInfo

A single forwarded port.

TunnelInfo.close#

async def close() -> None

Close this tunnel (no-op if already closed or inside a context manager).

MCP#

mcp_config#

async def mcp_config(
    service: Callable[[], Awaitable[Any]],
    path: str = '/mcp',
) -> Dict[str, Any]

{"url": ..., "headers": {...}} of the MCP endpoint at path.

connect#

async def connect(config: Dict[str, Any]) -> AsyncIterator[Any]

An initialized official-SDK client for config (url, headers).

With mcp 2.x this is an mcp.Client (protocol negotiated with server/discover, falling back to initialize); with 1.x an initialized mcp.ClientSession.

open_mcp#

async def open_mcp(
    service: Callable[[], Awaitable[Any]],
    path: str = '/mcp',
) -> AsyncIterator[Any]

connect(await mcp_config(service, path)).