Run Minecraft in a Windows sandbox

import asyncio
from cua import Image, QEMURuntime, Sandbox
 
EXTRA_ARGS = [
    # a second, unrestricted user-mode NIC — the default one is restrict=on
    '-netdev', 'user,id=net1,net=10.0.3.0/24,host=10.0.3.2,dns=10.0.3.3',
    '-device', 'virtio-net-pci,netdev=net1,mac=52:55:00:d1:55:02',
    # a CPU the software OpenGL driver can actually use
    '-cpu', 'host',
]
 
async def main():
    sb = await Sandbox.create(
        Image.windows().expose(3000),
        name='mc-win',
        local=True,
        runtime=QEMURuntime(
            mode='bare-metal',
            cpu_count=12,
            memory_mb=16384,
            extra_args=EXTRA_ARGS,
        ),
    )
    mcp_port = sb.exposed_ports[3000]
    print(f'cua-driver MCP on http://127.0.0.1:{mcp_port}/mcp')
    await sb.disconnect()  # the sandbox keeps running
 
asyncio.run(main())

async with Sandbox.connect('mc-win', local=True) as sb:
    check = await sb.shell.run(
        'powershell -Command "(Invoke-WebRequest -UseBasicParsing '
        'https://piston-meta.mojang.com/mc/game/version_manifest.json).StatusCode"'
    )
    print(check.stdout)  # 200

$ErrorActionPreference = 'Stop'
$ProgressPreference = 'SilentlyContinue'
New-Item -ItemType Directory -Force -Path C:\mc | Out-Null
 
# Prism Launcher — signs in with Microsoft device authorization, which needs no browser
Invoke-WebRequest -UseBasicParsing -OutFile C:\mc\prism.zip `
  'https://github.com/PrismLauncher/PrismLauncher/releases/download/11.0.3/PrismLauncher-Windows-MinGW-w64-Portable-11.0.3.zip'
Expand-Archive C:\mc\prism.zip -DestinationPath C:\mc\prismw -Force
 
# 7-Zip, because Mesa ships as .7z
Invoke-WebRequest -UseBasicParsing -OutFile C:\mc\7z.msi 'https://www.7-zip.org/a/7z2408-x64.msi'
Start-Process msiexec.exe -ArgumentList '/i','C:\mc\7z.msi','/qn' -Wait
 
# Mesa3D software OpenGL
Invoke-WebRequest -UseBasicParsing -OutFile C:\mc\mesa.7z `
  'https://github.com/pal1000/mesa-dist-win/releases/download/26.1.6/mesa3d-26.1.6-release-mingw.7z'
& 'C:\Program Files\7-Zip\7z.exe' x C:\mc\mesa.7z -oC:\mc\mesamw -y | Out-Null
 
Start-Process -FilePath C:\mc\prismw\prismlauncher.exe -WorkingDirectory C:\mc\prismw

from pathlib import Path
 
async with Sandbox.connect('mc-win', local=True) as sb:
    await sb.shell.run('if not exist C:\\mc mkdir C:\\mc')
    await sb.files.write_text('C:\\mc\\setup.ps1', Path('setup.ps1').read_text())
    result = await sb.shell.run(
        'powershell -NoProfile -ExecutionPolicy Bypass -File C:\\mc\\setup.ps1',
        timeout=1800,
    )
    print(result.stdout)

async with Sandbox.connect('mc-win', local=True) as sb:
    Path('sandbox.png').write_bytes(await sb.screenshot())   # look at it
    await sb.mouse.click(888, 678)                           # Next

$dirs = Get-ChildItem C:\mc -Recurse -Filter javaw.exe -ErrorAction SilentlyContinue |
  Select-Object -ExpandProperty DirectoryName -Unique
foreach ($d in $dirs) {
  Copy-Item C:\mc\mesamw\x64\opengl32.dll, C:\mc\mesamw\x64\libgallium_wgl.dll $d -Force
  Write-Output "mesa -> $d"
}

async with Sandbox.connect('mc-win', local=True) as sb:
    await sb.files.write_text('C:\\mc\\mesa.ps1', Path('mesa.ps1').read_text())
    result = await sb.shell.run(
        'powershell -NoProfile -ExecutionPolicy Bypass -File C:\\mc\\mesa.ps1', timeout=600,
    )
    print(result.stdout)   # mesa -> C:\mc\prismw\java\java-runtime-gamma\bin

import asyncio, json, os
import litellm
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
 
SYSTEM = """You operate a computer through the provided tools.
 
The desktop is Windows at 1280x800. Work in a loop: look at the screen with
get_desktop_state, decide one action, call one tool, then look again.
 
  * click / type_text / press_key act on a specific application, addressed by
    `pid`. Use list_windows to find the pid, then pass pid with x/y.
  * There is no wait tool. If something is still loading, call
    get_desktop_state again — repeated looks are how you wait.
 
Call exactly one tool per turn. When the task is complete, reply DONE."""
 
async def complete(**kwargs):
    """Two workarounds for the endpoint used here — yours may need neither.
 
    It is streaming-only (a plain request comes back with empty output), and it
    rejects role=system, so the system prompt travels as the first user turn.
    """
    messages, system = [], []
    for m in kwargs['messages']:
        (system if m.get('role') == 'system' else messages).append(m)
    if system:
        text = '\n\n'.join(m['content'] for m in system)
        messages = [{'role': 'user', 'content': text}] + messages
    kwargs['messages'] = messages
 
    stream = await litellm.acompletion(**kwargs, stream=True)
    chunks = [c async for c in stream]
    return litellm.stream_chunk_builder(chunks, messages=messages)
 
def prune_images(messages, keep=3):
    """Each get_desktop_state returns a full screenshot; keep only the newest."""
    seen = 0
    for msg in reversed(messages):
        if not isinstance(msg.get('content'), list):
            continue
        for part in msg['content']:
            if part.get('type') == 'image_url':
                seen += 1
                if seen > keep:
                    part.clear()
                    part.update({'type': 'text', 'text': '[older screenshot dropped]'})
    return messages
 
async def run(mcp_url, task, model, max_steps=60, headers=None):
    client = Client(StreamableHttpTransport(mcp_url, headers=headers))
    async with client:
        # list_tools() returned 55 tools on the image used here, most of them
        # browser and recording plumbing this task never needs. Hand the model
        # only what the job requires: dozens of schemas is a lot of context to
        # spend, and a shorter menu is a shorter list of ways to go wrong. Worth
        # doing whether or not your driver already filters denied tools out.
        wanted = {
            'get_desktop_state', 'list_windows', 'list_apps', 'click',
            'double_click', 'type_text', 'press_key', 'hotkey', 'launch_app',
            'bring_to_front', 'scroll',
        }
        mcp_tools = [t for t in await client.list_tools() if t.name in wanted]
        tools = [{
            'type': 'function',
            'function': {
                'name': t.name,
                'description': (t.description or '')[:800],
                'parameters': t.inputSchema or {'type': 'object', 'properties': {}},
            },
        } for t in mcp_tools]
 
        messages = [{'role': 'system', 'content': SYSTEM},
                    {'role': 'user', 'content': task}]
 
        for _ in range(max_steps):
            resp = await complete(
                model=model, messages=prune_images(messages), tools=tools,
                tool_choice='auto', temperature=0.0,
            )
            msg = resp.choices[0].message
            messages.append(msg.model_dump())
            if not msg.tool_calls:
                break                      # model said DONE
 
            for call in msg.tool_calls:
                args = json.loads(call.function.arguments or '{}')
                result = await client.call_tool(call.function.name, args,
                                                raise_on_error=False)
                text = ''.join(getattr(b, 'text', '') for b in (result.content or []))
                messages.append({'role': 'tool', 'tool_call_id': call.id,
                                 'name': call.function.name, 'content': text[:1500]})
 
                shot = next((b.data for b in (result.content or [])
                             if getattr(b, 'data', None)), None)
                if shot:
                    messages.append({'role': 'user', 'content': [
                        {'type': 'text', 'text': 'screenshot after that action:'},
                        {'type': 'image_url',
                         'image_url': {'url': f'data:image/png;base64,{shot}'}},
                    ]})

TASK = """
Prism Launcher is open, with a Minecraft instance and a signed-in account.
Select the instance and click Launch. Minecraft uses a software renderer, so the
window takes minutes to appear and repaints slowly — keep calling
get_desktop_state to watch it, and do not restart anything. On the title screen
click Singleplayer, then Create New World, then Create New World again. Stop as
soon as you are inside the world (terrain in first person, hotbar and hearts
visible) and reply DONE. Never press Escape while Minecraft is in the foreground.
"""
 
asyncio.run(run(f'http://127.0.0.1:{mcp_port}/mcp', TASK, 'your-model'))

import httpx
from cua import Image, Sandbox
 
sb = await Sandbox.create(Image.windows().expose(3000), name='mc-fleet', time_to_start=900)
 
tunnel = await sb.tunnel.forward(3000)
# https://run.cua.ai/api/svc/<namespace>/<sandbox>-port-3000/
 
token = httpx.post(
    'https://auth.cua.ai/realms/cyclops-cs/protocol/openid-connect/token',
    data={'grant_type': 'client_credentials',
          'client_id': os.environ['CUA_CLIENT_ID'],
          'client_secret': os.environ['CUA_CLIENT_SECRET']},
).json()['access_token']
 
await run(tunnel.url + 'mcp', TASK, 'your-model',
          headers={'Authorization': f'Bearer {token}'})