This is page 11 of 16. Use http://codebase.md/trycua/cua?page={x} to view the full context.
# Directory Structure
```
├── .all-contributorsrc
├── .cursorignore
├── .devcontainer
│ ├── devcontainer.json
│ ├── post-install.sh
│ └── README.md
├── .dockerignore
├── .gitattributes
├── .github
│ ├── FUNDING.yml
│ ├── scripts
│ │ ├── get_pyproject_version.py
│ │ └── tests
│ │ ├── __init__.py
│ │ ├── README.md
│ │ └── test_get_pyproject_version.py
│ └── workflows
│ ├── ci-lume.yml
│ ├── docker-publish-kasm.yml
│ ├── docker-publish-xfce.yml
│ ├── docker-reusable-publish.yml
│ ├── npm-publish-computer.yml
│ ├── npm-publish-core.yml
│ ├── publish-lume.yml
│ ├── pypi-publish-agent.yml
│ ├── pypi-publish-computer-server.yml
│ ├── pypi-publish-computer.yml
│ ├── pypi-publish-core.yml
│ ├── pypi-publish-mcp-server.yml
│ ├── pypi-publish-pylume.yml
│ ├── pypi-publish-som.yml
│ ├── pypi-reusable-publish.yml
│ └── test-validation-script.yml
├── .gitignore
├── .vscode
│ ├── docs.code-workspace
│ ├── launch.json
│ ├── libs-ts.code-workspace
│ ├── lume.code-workspace
│ ├── lumier.code-workspace
│ ├── py.code-workspace
│ └── settings.json
├── blog
│ ├── app-use.md
│ ├── assets
│ │ ├── composite-agents.png
│ │ ├── docker-ubuntu-support.png
│ │ ├── hack-booth.png
│ │ ├── hack-closing-ceremony.jpg
│ │ ├── hack-cua-ollama-hud.jpeg
│ │ ├── hack-leaderboard.png
│ │ ├── hack-the-north.png
│ │ ├── hack-winners.jpeg
│ │ ├── hack-workshop.jpeg
│ │ ├── hud-agent-evals.png
│ │ └── trajectory-viewer.jpeg
│ ├── bringing-computer-use-to-the-web.md
│ ├── build-your-own-operator-on-macos-1.md
│ ├── build-your-own-operator-on-macos-2.md
│ ├── composite-agents.md
│ ├── cua-hackathon.md
│ ├── hack-the-north.md
│ ├── hud-agent-evals.md
│ ├── human-in-the-loop.md
│ ├── introducing-cua-cloud-containers.md
│ ├── lume-to-containerization.md
│ ├── sandboxed-python-execution.md
│ ├── training-computer-use-models-trajectories-1.md
│ ├── trajectory-viewer.md
│ ├── ubuntu-docker-support.md
│ └── windows-sandbox.md
├── CONTRIBUTING.md
├── Development.md
├── Dockerfile
├── docs
│ ├── .gitignore
│ ├── .prettierrc
│ ├── content
│ │ └── docs
│ │ ├── agent-sdk
│ │ │ ├── agent-loops.mdx
│ │ │ ├── benchmarks
│ │ │ │ ├── index.mdx
│ │ │ │ ├── interactive.mdx
│ │ │ │ ├── introduction.mdx
│ │ │ │ ├── meta.json
│ │ │ │ ├── osworld-verified.mdx
│ │ │ │ ├── screenspot-pro.mdx
│ │ │ │ └── screenspot-v2.mdx
│ │ │ ├── callbacks
│ │ │ │ ├── agent-lifecycle.mdx
│ │ │ │ ├── cost-saving.mdx
│ │ │ │ ├── index.mdx
│ │ │ │ ├── logging.mdx
│ │ │ │ ├── meta.json
│ │ │ │ ├── pii-anonymization.mdx
│ │ │ │ └── trajectories.mdx
│ │ │ ├── chat-history.mdx
│ │ │ ├── custom-computer-handlers.mdx
│ │ │ ├── custom-tools.mdx
│ │ │ ├── customizing-computeragent.mdx
│ │ │ ├── integrations
│ │ │ │ ├── hud.mdx
│ │ │ │ └── meta.json
│ │ │ ├── message-format.mdx
│ │ │ ├── meta.json
│ │ │ ├── migration-guide.mdx
│ │ │ ├── prompt-caching.mdx
│ │ │ ├── supported-agents
│ │ │ │ ├── composed-agents.mdx
│ │ │ │ ├── computer-use-agents.mdx
│ │ │ │ ├── grounding-models.mdx
│ │ │ │ ├── human-in-the-loop.mdx
│ │ │ │ └── meta.json
│ │ │ ├── supported-model-providers
│ │ │ │ ├── index.mdx
│ │ │ │ └── local-models.mdx
│ │ │ └── usage-tracking.mdx
│ │ ├── computer-sdk
│ │ │ ├── cloud-vm-management.mdx
│ │ │ ├── commands.mdx
│ │ │ ├── computer-ui.mdx
│ │ │ ├── computers.mdx
│ │ │ ├── meta.json
│ │ │ └── sandboxed-python.mdx
│ │ ├── index.mdx
│ │ ├── libraries
│ │ │ ├── agent
│ │ │ │ └── index.mdx
│ │ │ ├── computer
│ │ │ │ └── index.mdx
│ │ │ ├── computer-server
│ │ │ │ ├── Commands.mdx
│ │ │ │ ├── index.mdx
│ │ │ │ ├── REST-API.mdx
│ │ │ │ └── WebSocket-API.mdx
│ │ │ ├── core
│ │ │ │ └── index.mdx
│ │ │ ├── lume
│ │ │ │ ├── cli-reference.mdx
│ │ │ │ ├── faq.md
│ │ │ │ ├── http-api.mdx
│ │ │ │ ├── index.mdx
│ │ │ │ ├── installation.mdx
│ │ │ │ ├── meta.json
│ │ │ │ └── prebuilt-images.mdx
│ │ │ ├── lumier
│ │ │ │ ├── building-lumier.mdx
│ │ │ │ ├── docker-compose.mdx
│ │ │ │ ├── docker.mdx
│ │ │ │ ├── index.mdx
│ │ │ │ ├── installation.mdx
│ │ │ │ └── meta.json
│ │ │ ├── mcp-server
│ │ │ │ ├── client-integrations.mdx
│ │ │ │ ├── configuration.mdx
│ │ │ │ ├── index.mdx
│ │ │ │ ├── installation.mdx
│ │ │ │ ├── llm-integrations.mdx
│ │ │ │ ├── meta.json
│ │ │ │ ├── tools.mdx
│ │ │ │ └── usage.mdx
│ │ │ └── som
│ │ │ ├── configuration.mdx
│ │ │ └── index.mdx
│ │ ├── meta.json
│ │ ├── quickstart-cli.mdx
│ │ ├── quickstart-devs.mdx
│ │ └── telemetry.mdx
│ ├── next.config.mjs
│ ├── package-lock.json
│ ├── package.json
│ ├── pnpm-lock.yaml
│ ├── postcss.config.mjs
│ ├── public
│ │ └── img
│ │ ├── agent_gradio_ui.png
│ │ ├── agent.png
│ │ ├── cli.png
│ │ ├── computer.png
│ │ ├── som_box_threshold.png
│ │ └── som_iou_threshold.png
│ ├── README.md
│ ├── source.config.ts
│ ├── src
│ │ ├── app
│ │ │ ├── (home)
│ │ │ │ ├── [[...slug]]
│ │ │ │ │ └── page.tsx
│ │ │ │ └── layout.tsx
│ │ │ ├── api
│ │ │ │ └── search
│ │ │ │ └── route.ts
│ │ │ ├── favicon.ico
│ │ │ ├── global.css
│ │ │ ├── layout.config.tsx
│ │ │ ├── layout.tsx
│ │ │ ├── llms.mdx
│ │ │ │ └── [[...slug]]
│ │ │ │ └── route.ts
│ │ │ └── llms.txt
│ │ │ └── route.ts
│ │ ├── assets
│ │ │ ├── discord-black.svg
│ │ │ ├── discord-white.svg
│ │ │ ├── logo-black.svg
│ │ │ └── logo-white.svg
│ │ ├── components
│ │ │ ├── iou.tsx
│ │ │ └── mermaid.tsx
│ │ ├── lib
│ │ │ ├── llms.ts
│ │ │ └── source.ts
│ │ └── mdx-components.tsx
│ └── tsconfig.json
├── examples
│ ├── agent_examples.py
│ ├── agent_ui_examples.py
│ ├── cloud_api_examples.py
│ ├── computer_examples_windows.py
│ ├── computer_examples.py
│ ├── computer_ui_examples.py
│ ├── computer-example-ts
│ │ ├── .env.example
│ │ ├── .gitignore
│ │ ├── .prettierrc
│ │ ├── package-lock.json
│ │ ├── package.json
│ │ ├── pnpm-lock.yaml
│ │ ├── README.md
│ │ ├── src
│ │ │ ├── helpers.ts
│ │ │ └── index.ts
│ │ └── tsconfig.json
│ ├── docker_examples.py
│ ├── evals
│ │ ├── hud_eval_examples.py
│ │ └── wikipedia_most_linked.txt
│ ├── pylume_examples.py
│ ├── sandboxed_functions_examples.py
│ ├── som_examples.py
│ ├── utils.py
│ └── winsandbox_example.py
├── img
│ ├── agent_gradio_ui.png
│ ├── agent.png
│ ├── cli.png
│ ├── computer.png
│ ├── logo_black.png
│ └── logo_white.png
├── libs
│ ├── kasm
│ │ ├── Dockerfile
│ │ ├── LICENSE
│ │ ├── README.md
│ │ └── src
│ │ └── ubuntu
│ │ └── install
│ │ └── firefox
│ │ ├── custom_startup.sh
│ │ ├── firefox.desktop
│ │ └── install_firefox.sh
│ ├── lume
│ │ ├── .cursorignore
│ │ ├── CONTRIBUTING.md
│ │ ├── Development.md
│ │ ├── img
│ │ │ └── cli.png
│ │ ├── Package.resolved
│ │ ├── Package.swift
│ │ ├── README.md
│ │ ├── resources
│ │ │ └── lume.entitlements
│ │ ├── scripts
│ │ │ ├── build
│ │ │ │ ├── build-debug.sh
│ │ │ │ ├── build-release-notarized.sh
│ │ │ │ └── build-release.sh
│ │ │ └── install.sh
│ │ ├── src
│ │ │ ├── Commands
│ │ │ │ ├── Clone.swift
│ │ │ │ ├── Config.swift
│ │ │ │ ├── Create.swift
│ │ │ │ ├── Delete.swift
│ │ │ │ ├── Get.swift
│ │ │ │ ├── Images.swift
│ │ │ │ ├── IPSW.swift
│ │ │ │ ├── List.swift
│ │ │ │ ├── Logs.swift
│ │ │ │ ├── Options
│ │ │ │ │ └── FormatOption.swift
│ │ │ │ ├── Prune.swift
│ │ │ │ ├── Pull.swift
│ │ │ │ ├── Push.swift
│ │ │ │ ├── Run.swift
│ │ │ │ ├── Serve.swift
│ │ │ │ ├── Set.swift
│ │ │ │ └── Stop.swift
│ │ │ ├── ContainerRegistry
│ │ │ │ ├── ImageContainerRegistry.swift
│ │ │ │ ├── ImageList.swift
│ │ │ │ └── ImagesPrinter.swift
│ │ │ ├── Errors
│ │ │ │ └── Errors.swift
│ │ │ ├── FileSystem
│ │ │ │ ├── Home.swift
│ │ │ │ ├── Settings.swift
│ │ │ │ ├── VMConfig.swift
│ │ │ │ ├── VMDirectory.swift
│ │ │ │ └── VMLocation.swift
│ │ │ ├── LumeController.swift
│ │ │ ├── Main.swift
│ │ │ ├── Server
│ │ │ │ ├── Handlers.swift
│ │ │ │ ├── HTTP.swift
│ │ │ │ ├── Requests.swift
│ │ │ │ ├── Responses.swift
│ │ │ │ └── Server.swift
│ │ │ ├── Utils
│ │ │ │ ├── CommandRegistry.swift
│ │ │ │ ├── CommandUtils.swift
│ │ │ │ ├── Logger.swift
│ │ │ │ ├── NetworkUtils.swift
│ │ │ │ ├── Path.swift
│ │ │ │ ├── ProcessRunner.swift
│ │ │ │ ├── ProgressLogger.swift
│ │ │ │ ├── String.swift
│ │ │ │ └── Utils.swift
│ │ │ ├── Virtualization
│ │ │ │ ├── DarwinImageLoader.swift
│ │ │ │ ├── DHCPLeaseParser.swift
│ │ │ │ ├── ImageLoaderFactory.swift
│ │ │ │ └── VMVirtualizationService.swift
│ │ │ ├── VM
│ │ │ │ ├── DarwinVM.swift
│ │ │ │ ├── LinuxVM.swift
│ │ │ │ ├── VM.swift
│ │ │ │ ├── VMDetails.swift
│ │ │ │ ├── VMDetailsPrinter.swift
│ │ │ │ ├── VMDisplayResolution.swift
│ │ │ │ └── VMFactory.swift
│ │ │ └── VNC
│ │ │ ├── PassphraseGenerator.swift
│ │ │ └── VNCService.swift
│ │ └── tests
│ │ ├── Mocks
│ │ │ ├── MockVM.swift
│ │ │ ├── MockVMVirtualizationService.swift
│ │ │ └── MockVNCService.swift
│ │ ├── VM
│ │ │ └── VMDetailsPrinterTests.swift
│ │ ├── VMTests.swift
│ │ ├── VMVirtualizationServiceTests.swift
│ │ └── VNCServiceTests.swift
│ ├── lumier
│ │ ├── .dockerignore
│ │ ├── Dockerfile
│ │ ├── README.md
│ │ └── src
│ │ ├── bin
│ │ │ └── entry.sh
│ │ ├── config
│ │ │ └── constants.sh
│ │ ├── hooks
│ │ │ └── on-logon.sh
│ │ └── lib
│ │ ├── utils.sh
│ │ └── vm.sh
│ ├── python
│ │ ├── agent
│ │ │ ├── .bumpversion.cfg
│ │ │ ├── agent
│ │ │ │ ├── __init__.py
│ │ │ │ ├── __main__.py
│ │ │ │ ├── adapters
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── huggingfacelocal_adapter.py
│ │ │ │ │ ├── human_adapter.py
│ │ │ │ │ ├── mlxvlm_adapter.py
│ │ │ │ │ └── models
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── generic.py
│ │ │ │ │ ├── internvl.py
│ │ │ │ │ ├── opencua.py
│ │ │ │ │ └── qwen2_5_vl.py
│ │ │ │ ├── agent.py
│ │ │ │ ├── callbacks
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── base.py
│ │ │ │ │ ├── budget_manager.py
│ │ │ │ │ ├── image_retention.py
│ │ │ │ │ ├── logging.py
│ │ │ │ │ ├── operator_validator.py
│ │ │ │ │ ├── pii_anonymization.py
│ │ │ │ │ ├── prompt_instructions.py
│ │ │ │ │ ├── telemetry.py
│ │ │ │ │ └── trajectory_saver.py
│ │ │ │ ├── cli.py
│ │ │ │ ├── computers
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── base.py
│ │ │ │ │ ├── cua.py
│ │ │ │ │ └── custom.py
│ │ │ │ ├── decorators.py
│ │ │ │ ├── human_tool
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── __main__.py
│ │ │ │ │ ├── server.py
│ │ │ │ │ └── ui.py
│ │ │ │ ├── integrations
│ │ │ │ │ └── hud
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── agent.py
│ │ │ │ │ └── proxy.py
│ │ │ │ ├── loops
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── anthropic.py
│ │ │ │ │ ├── base.py
│ │ │ │ │ ├── composed_grounded.py
│ │ │ │ │ ├── gemini.py
│ │ │ │ │ ├── glm45v.py
│ │ │ │ │ ├── gta1.py
│ │ │ │ │ ├── holo.py
│ │ │ │ │ ├── internvl.py
│ │ │ │ │ ├── model_types.csv
│ │ │ │ │ ├── moondream3.py
│ │ │ │ │ ├── omniparser.py
│ │ │ │ │ ├── openai.py
│ │ │ │ │ ├── opencua.py
│ │ │ │ │ └── uitars.py
│ │ │ │ ├── proxy
│ │ │ │ │ ├── examples.py
│ │ │ │ │ └── handlers.py
│ │ │ │ ├── responses.py
│ │ │ │ ├── types.py
│ │ │ │ └── ui
│ │ │ │ ├── __init__.py
│ │ │ │ ├── __main__.py
│ │ │ │ └── gradio
│ │ │ │ ├── __init__.py
│ │ │ │ ├── app.py
│ │ │ │ └── ui_components.py
│ │ │ ├── benchmarks
│ │ │ │ ├── .gitignore
│ │ │ │ ├── contrib.md
│ │ │ │ ├── interactive.py
│ │ │ │ ├── models
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── base.py
│ │ │ │ │ └── gta1.py
│ │ │ │ ├── README.md
│ │ │ │ ├── ss-pro.py
│ │ │ │ ├── ss-v2.py
│ │ │ │ └── utils.py
│ │ │ ├── example.py
│ │ │ ├── pyproject.toml
│ │ │ └── README.md
│ │ ├── computer
│ │ │ ├── .bumpversion.cfg
│ │ │ ├── computer
│ │ │ │ ├── __init__.py
│ │ │ │ ├── computer.py
│ │ │ │ ├── diorama_computer.py
│ │ │ │ ├── helpers.py
│ │ │ │ ├── interface
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── base.py
│ │ │ │ │ ├── factory.py
│ │ │ │ │ ├── generic.py
│ │ │ │ │ ├── linux.py
│ │ │ │ │ ├── macos.py
│ │ │ │ │ ├── models.py
│ │ │ │ │ └── windows.py
│ │ │ │ ├── logger.py
│ │ │ │ ├── models.py
│ │ │ │ ├── providers
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── base.py
│ │ │ │ │ ├── cloud
│ │ │ │ │ │ ├── __init__.py
│ │ │ │ │ │ └── provider.py
│ │ │ │ │ ├── docker
│ │ │ │ │ │ ├── __init__.py
│ │ │ │ │ │ └── provider.py
│ │ │ │ │ ├── factory.py
│ │ │ │ │ ├── lume
│ │ │ │ │ │ ├── __init__.py
│ │ │ │ │ │ └── provider.py
│ │ │ │ │ ├── lume_api.py
│ │ │ │ │ ├── lumier
│ │ │ │ │ │ ├── __init__.py
│ │ │ │ │ │ └── provider.py
│ │ │ │ │ ├── types.py
│ │ │ │ │ └── winsandbox
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── provider.py
│ │ │ │ │ └── setup_script.ps1
│ │ │ │ ├── ui
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── __main__.py
│ │ │ │ │ └── gradio
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ └── app.py
│ │ │ │ └── utils.py
│ │ │ ├── poetry.toml
│ │ │ ├── pyproject.toml
│ │ │ └── README.md
│ │ ├── computer-server
│ │ │ ├── .bumpversion.cfg
│ │ │ ├── computer_server
│ │ │ │ ├── __init__.py
│ │ │ │ ├── __main__.py
│ │ │ │ ├── cli.py
│ │ │ │ ├── diorama
│ │ │ │ │ ├── __init__.py
│ │ │ │ │ ├── base.py
│ │ │ │ │ ├── diorama_computer.py
│ │ │ │ │ ├── diorama.py
│ │ │ │ │ ├── draw.py
│ │ │ │ │ ├── macos.py
│ │ │ │ │ └── safezone.py
│ │ │ │ ├── handlers
│ │ │ │ │ ├── base.py
│ │ │ │ │ ├── factory.py
│ │ │ │ │ ├── generic.py
│ │ │ │ │ ├── linux.py
│ │ │ │ │ ├── macos.py
│ │ │ │ │ └── windows.py
│ │ │ │ ├── main.py
│ │ │ │ ├── server.py
│ │ │ │ └── watchdog.py
│ │ │ ├── examples
│ │ │ │ ├── __init__.py
│ │ │ │ └── usage_example.py
│ │ │ ├── pyproject.toml
│ │ │ ├── README.md
│ │ │ ├── run_server.py
│ │ │ └── test_connection.py
│ │ ├── core
│ │ │ ├── .bumpversion.cfg
│ │ │ ├── core
│ │ │ │ ├── __init__.py
│ │ │ │ └── telemetry
│ │ │ │ ├── __init__.py
│ │ │ │ └── posthog.py
│ │ │ ├── poetry.toml
│ │ │ ├── pyproject.toml
│ │ │ └── README.md
│ │ ├── mcp-server
│ │ │ ├── .bumpversion.cfg
│ │ │ ├── CONCURRENT_SESSIONS.md
│ │ │ ├── mcp_server
│ │ │ │ ├── __init__.py
│ │ │ │ ├── __main__.py
│ │ │ │ ├── server.py
│ │ │ │ └── session_manager.py
│ │ │ ├── pdm.lock
│ │ │ ├── pyproject.toml
│ │ │ ├── README.md
│ │ │ └── scripts
│ │ │ ├── install_mcp_server.sh
│ │ │ └── start_mcp_server.sh
│ │ ├── pylume
│ │ │ ├── __init__.py
│ │ │ ├── .bumpversion.cfg
│ │ │ ├── pylume
│ │ │ │ ├── __init__.py
│ │ │ │ ├── client.py
│ │ │ │ ├── exceptions.py
│ │ │ │ ├── lume
│ │ │ │ ├── models.py
│ │ │ │ ├── pylume.py
│ │ │ │ └── server.py
│ │ │ ├── pyproject.toml
│ │ │ └── README.md
│ │ └── som
│ │ ├── .bumpversion.cfg
│ │ ├── LICENSE
│ │ ├── poetry.toml
│ │ ├── pyproject.toml
│ │ ├── README.md
│ │ ├── som
│ │ │ ├── __init__.py
│ │ │ ├── detect.py
│ │ │ ├── detection.py
│ │ │ ├── models.py
│ │ │ ├── ocr.py
│ │ │ ├── util
│ │ │ │ └── utils.py
│ │ │ └── visualization.py
│ │ └── tests
│ │ └── test_omniparser.py
│ ├── typescript
│ │ ├── .gitignore
│ │ ├── .nvmrc
│ │ ├── agent
│ │ │ ├── examples
│ │ │ │ ├── playground-example.html
│ │ │ │ └── README.md
│ │ │ ├── package.json
│ │ │ ├── README.md
│ │ │ ├── src
│ │ │ │ ├── client.ts
│ │ │ │ ├── index.ts
│ │ │ │ └── types.ts
│ │ │ ├── tests
│ │ │ │ └── client.test.ts
│ │ │ ├── tsconfig.json
│ │ │ ├── tsdown.config.ts
│ │ │ └── vitest.config.ts
│ │ ├── biome.json
│ │ ├── computer
│ │ │ ├── .editorconfig
│ │ │ ├── .gitattributes
│ │ │ ├── .gitignore
│ │ │ ├── LICENSE
│ │ │ ├── package.json
│ │ │ ├── README.md
│ │ │ ├── src
│ │ │ │ ├── computer
│ │ │ │ │ ├── index.ts
│ │ │ │ │ ├── providers
│ │ │ │ │ │ ├── base.ts
│ │ │ │ │ │ ├── cloud.ts
│ │ │ │ │ │ └── index.ts
│ │ │ │ │ └── types.ts
│ │ │ │ ├── index.ts
│ │ │ │ ├── interface
│ │ │ │ │ ├── base.ts
│ │ │ │ │ ├── factory.ts
│ │ │ │ │ ├── index.ts
│ │ │ │ │ ├── linux.ts
│ │ │ │ │ ├── macos.ts
│ │ │ │ │ └── windows.ts
│ │ │ │ └── types.ts
│ │ │ ├── tests
│ │ │ │ ├── computer
│ │ │ │ │ └── cloud.test.ts
│ │ │ │ ├── interface
│ │ │ │ │ ├── factory.test.ts
│ │ │ │ │ ├── index.test.ts
│ │ │ │ │ ├── linux.test.ts
│ │ │ │ │ ├── macos.test.ts
│ │ │ │ │ └── windows.test.ts
│ │ │ │ └── setup.ts
│ │ │ ├── tsconfig.json
│ │ │ ├── tsdown.config.ts
│ │ │ └── vitest.config.ts
│ │ ├── core
│ │ │ ├── .editorconfig
│ │ │ ├── .gitattributes
│ │ │ ├── .gitignore
│ │ │ ├── LICENSE
│ │ │ ├── package.json
│ │ │ ├── README.md
│ │ │ ├── src
│ │ │ │ ├── index.ts
│ │ │ │ └── telemetry
│ │ │ │ ├── clients
│ │ │ │ │ ├── index.ts
│ │ │ │ │ └── posthog.ts
│ │ │ │ └── index.ts
│ │ │ ├── tests
│ │ │ │ └── telemetry.test.ts
│ │ │ ├── tsconfig.json
│ │ │ ├── tsdown.config.ts
│ │ │ └── vitest.config.ts
│ │ ├── package.json
│ │ ├── pnpm-lock.yaml
│ │ ├── pnpm-workspace.yaml
│ │ └── README.md
│ └── xfce
│ ├── .dockerignore
│ ├── .gitignore
│ ├── Dockerfile
│ ├── README.md
│ └── src
│ ├── scripts
│ │ ├── resize-display.sh
│ │ ├── start-computer-server.sh
│ │ ├── start-novnc.sh
│ │ ├── start-vnc.sh
│ │ └── xstartup.sh
│ ├── supervisor
│ │ └── supervisord.conf
│ └── xfce-config
│ ├── helpers.rc
│ ├── xfce4-power-manager.xml
│ └── xfce4-session.xml
├── LICENSE.md
├── Makefile
├── notebooks
│ ├── agent_nb.ipynb
│ ├── blog
│ │ ├── build-your-own-operator-on-macos-1.ipynb
│ │ └── build-your-own-operator-on-macos-2.ipynb
│ ├── composite_agents_docker_nb.ipynb
│ ├── computer_nb.ipynb
│ ├── computer_server_nb.ipynb
│ ├── customizing_computeragent.ipynb
│ ├── eval_osworld.ipynb
│ ├── ollama_nb.ipynb
│ ├── pylume_nb.ipynb
│ ├── README.md
│ ├── sota_hackathon_cloud.ipynb
│ └── sota_hackathon.ipynb
├── pdm.lock
├── pyproject.toml
├── pyrightconfig.json
├── README.md
├── samples
│ └── community
│ ├── global-online
│ │ └── README.md
│ └── hack-the-north
│ └── README.md
├── scripts
│ ├── build-uv.sh
│ ├── build.ps1
│ ├── build.sh
│ ├── cleanup.sh
│ ├── playground-docker.sh
│ ├── playground.sh
│ └── run-docker-dev.sh
└── tests
├── pytest.ini
├── shell_cmd.py
├── test_files.py
├── test_mcp_server_session_management.py
├── test_mcp_server_streaming.py
├── test_shell_bash.py
├── test_telemetry.py
├── test_venv.py
└── test_watchdog.py
```
# Files
--------------------------------------------------------------------------------
/libs/python/computer/computer/providers/winsandbox/provider.py:
--------------------------------------------------------------------------------
```python
"""Windows Sandbox VM provider implementation using pywinsandbox."""
import os
import asyncio
import logging
import time
from typing import Dict, Any, Optional, List
from pathlib import Path
from ..base import BaseVMProvider, VMProviderType
# Setup logging
logger = logging.getLogger(__name__)
try:
import winsandbox
HAS_WINSANDBOX = True
except ImportError:
HAS_WINSANDBOX = False
class WinSandboxProvider(BaseVMProvider):
"""Windows Sandbox VM provider implementation using pywinsandbox.
This provider uses Windows Sandbox to create isolated Windows environments.
Storage is always ephemeral with Windows Sandbox.
"""
def __init__(
self,
port: int = 7777,
host: str = "localhost",
storage: Optional[str] = None,
verbose: bool = False,
ephemeral: bool = True, # Windows Sandbox is always ephemeral
memory_mb: int = 4096,
networking: bool = True,
**kwargs
):
"""Initialize the Windows Sandbox provider.
Args:
port: Port for the computer server (default: 7777)
host: Host to use for connections (default: localhost)
storage: Storage path (ignored - Windows Sandbox is always ephemeral)
verbose: Enable verbose logging
ephemeral: Always True for Windows Sandbox
memory_mb: Memory allocation in MB (default: 4096)
networking: Enable networking in sandbox (default: True)
"""
if not HAS_WINSANDBOX:
raise ImportError(
"pywinsandbox is required for WinSandboxProvider. "
"Please install it with 'pip install pywinsandbox'"
)
self.host = host
self.port = port
self.verbose = verbose
self.memory_mb = memory_mb
self.networking = networking
# Windows Sandbox is always ephemeral
if not ephemeral:
logger.warning("Windows Sandbox storage is always ephemeral. Ignoring ephemeral=False.")
self.ephemeral = True
# Storage is always ephemeral for Windows Sandbox
if storage and storage != "ephemeral":
logger.warning("Windows Sandbox does not support persistent storage. Using ephemeral storage.")
self.storage = "ephemeral"
self.logger = logging.getLogger(__name__)
# Track active sandboxes
self._active_sandboxes: Dict[str, Any] = {}
@property
def provider_type(self) -> VMProviderType:
"""Get the provider type."""
return VMProviderType.WINSANDBOX
async def __aenter__(self):
"""Enter async context manager."""
# Verify Windows Sandbox is available
if not HAS_WINSANDBOX:
raise ImportError("pywinsandbox is not available")
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Exit async context manager."""
# Clean up any active sandboxes
for name, sandbox in self._active_sandboxes.items():
try:
sandbox.shutdown()
self.logger.info(f"Terminated sandbox: {name}")
except Exception as e:
self.logger.error(f"Error terminating sandbox {name}: {e}")
self._active_sandboxes.clear()
async def get_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any]:
"""Get VM information by name.
Args:
name: Name of the VM to get information for
storage: Ignored for Windows Sandbox (always ephemeral)
Returns:
Dictionary with VM information including status, IP address, etc.
"""
if name not in self._active_sandboxes:
return {
"name": name,
"status": "stopped",
"ip_address": None,
"storage": "ephemeral"
}
sandbox = self._active_sandboxes[name]
# Check if sandbox is still running
try:
# Try to ping the sandbox to see if it's responsive
try:
sandbox.rpyc.modules.os.getcwd()
sandbox_responsive = True
except Exception:
sandbox_responsive = False
if not sandbox_responsive:
return {
"name": name,
"status": "starting",
"ip_address": None,
"storage": "ephemeral",
"memory_mb": self.memory_mb,
"networking": self.networking
}
# Check for computer server address file
server_address_file = r"C:\Users\WDAGUtilityAccount\Desktop\shared_windows_sandbox_dir\server_address"
try:
# Check if the server address file exists
file_exists = sandbox.rpyc.modules.os.path.exists(server_address_file)
if file_exists:
# Read the server address file
with sandbox.rpyc.builtin.open(server_address_file, 'r') as f:
server_address = f.read().strip()
if server_address and ':' in server_address:
# Parse IP:port from the file
ip_address, port = server_address.split(':', 1)
# Verify the server is actually responding
try:
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(3)
result = sock.connect_ex((ip_address, int(port)))
sock.close()
if result == 0:
# Server is responding
status = "running"
self.logger.debug(f"Computer server found at {ip_address}:{port}")
else:
# Server file exists but not responding
status = "starting"
ip_address = None
except Exception as e:
self.logger.debug(f"Error checking server connectivity: {e}")
status = "starting"
ip_address = None
else:
# File exists but doesn't contain valid address
status = "starting"
ip_address = None
else:
# Server address file doesn't exist yet
status = "starting"
ip_address = None
except Exception as e:
self.logger.debug(f"Error checking server address file: {e}")
status = "starting"
ip_address = None
except Exception as e:
self.logger.error(f"Error checking sandbox status: {e}")
status = "error"
ip_address = None
return {
"name": name,
"status": status,
"ip_address": ip_address,
"storage": "ephemeral",
"memory_mb": self.memory_mb,
"networking": self.networking
}
async def list_vms(self) -> List[Dict[str, Any]]:
"""List all available VMs."""
vms = []
for name in self._active_sandboxes.keys():
vm_info = await self.get_vm(name)
vms.append(vm_info)
return vms
async def run_vm(self, image: str, name: str, run_opts: Dict[str, Any], storage: Optional[str] = None) -> Dict[str, Any]:
"""Run a VM with the given options.
Args:
image: Image name (ignored for Windows Sandbox - always uses host Windows)
name: Name of the VM to run
run_opts: Dictionary of run options (memory, cpu, etc.)
storage: Ignored for Windows Sandbox (always ephemeral)
Returns:
Dictionary with VM run status and information
"""
if name in self._active_sandboxes:
return {
"success": False,
"error": f"Sandbox {name} is already running"
}
try:
# Extract options from run_opts
memory_mb = run_opts.get("memory_mb", self.memory_mb)
if isinstance(memory_mb, str):
# Convert memory string like "4GB" to MB
if memory_mb.upper().endswith("GB"):
memory_mb = int(float(memory_mb[:-2]) * 1024)
elif memory_mb.upper().endswith("MB"):
memory_mb = int(memory_mb[:-2])
else:
memory_mb = self.memory_mb
networking = run_opts.get("networking", self.networking)
# Create folder mappers; always map a persistent venv directory on host for caching packages
folder_mappers = []
# Ensure host side persistent venv directory exists (Path.home()/wsb_venv)
host_wsb_env = Path.home() / ".cua" / "wsb_cache"
try:
host_wsb_env.mkdir(parents=True, exist_ok=True)
except Exception:
# If cannot create, continue without persistent mapping
host_wsb_env = None
shared_directories = run_opts.get("shared_directories", [])
for shared_dir in shared_directories:
if isinstance(shared_dir, dict):
host_path = shared_dir.get("hostPath", "")
elif isinstance(shared_dir, str):
host_path = shared_dir
else:
continue
if host_path and os.path.exists(host_path):
folder_mappers.append(winsandbox.FolderMapper(host_path))
# Add mapping for the persistent venv directory (read/write) so it appears in Sandbox Desktop
if host_wsb_env is not None and host_wsb_env.exists():
try:
folder_mappers.append(
winsandbox.FolderMapper(str(host_wsb_env), read_only=False)
)
except Exception as e:
self.logger.warning(f"Failed to map host winsandbox_venv: {e}")
self.logger.info(f"Creating Windows Sandbox: {name}")
self.logger.info(f"Memory: {memory_mb}MB, Networking: {networking}")
if folder_mappers:
self.logger.info(f"Shared directories: {len(folder_mappers)}")
# Create the sandbox without logon script
try:
# Try with memory_mb parameter (newer pywinsandbox version)
sandbox = winsandbox.new_sandbox(
memory_mb=str(memory_mb),
networking=networking,
folder_mappers=folder_mappers
)
except TypeError as e:
if "memory_mb" in str(e):
# Fallback for older pywinsandbox version that doesn't support memory_mb
self.logger.warning(
f"Your pywinsandbox version doesn't support memory_mb parameter. "
f"Using default memory settings. To use custom memory settings, "
f"please update pywinsandbox: pip install -U git+https://github.com/karkason/pywinsandbox.git"
)
sandbox = winsandbox.new_sandbox(
networking=networking,
folder_mappers=folder_mappers
)
else:
# Re-raise if it's a different TypeError
raise
# Store the sandbox
self._active_sandboxes[name] = sandbox
self.logger.info(f"Windows Sandbox {name} created successfully")
venv_exists = (host_wsb_env / "venv" / "Lib" / "site-packages" / "computer_server").exists() if host_wsb_env else False
# Setup the computer server in the sandbox
await self._setup_computer_server(sandbox, name, wait_for_venv=(not venv_exists))
return {
"success": True,
"name": name,
"status": "starting",
"memory_mb": memory_mb,
"networking": networking,
"storage": "ephemeral"
}
except Exception as e:
self.logger.error(f"Failed to create Windows Sandbox {name}: {e}")
# stack trace
import traceback
self.logger.error(f"Stack trace: {traceback.format_exc()}")
return {
"success": False,
"error": f"Failed to create sandbox: {str(e)}"
}
async def stop_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any]:
"""Stop a running VM.
Args:
name: Name of the VM to stop
storage: Ignored for Windows Sandbox
Returns:
Dictionary with stop status and information
"""
if name not in self._active_sandboxes:
return {
"success": False,
"error": f"Sandbox {name} is not running"
}
try:
sandbox = self._active_sandboxes[name]
# Terminate the sandbox
sandbox.shutdown()
# Remove from active sandboxes
del self._active_sandboxes[name]
self.logger.info(f"Windows Sandbox {name} stopped successfully")
return {
"success": True,
"name": name,
"status": "stopped"
}
except Exception as e:
self.logger.error(f"Failed to stop Windows Sandbox {name}: {e}")
return {
"success": False,
"error": f"Failed to stop sandbox: {str(e)}"
}
async def update_vm(self, name: str, update_opts: Dict[str, Any], storage: Optional[str] = None) -> Dict[str, Any]:
"""Update VM configuration.
Note: Windows Sandbox does not support runtime configuration updates.
The sandbox must be stopped and restarted with new configuration.
Args:
name: Name of the VM to update
update_opts: Dictionary of update options
storage: Ignored for Windows Sandbox
Returns:
Dictionary with update status and information
"""
return {
"success": False,
"error": "Windows Sandbox does not support runtime configuration updates. "
"Please stop and restart the sandbox with new configuration."
}
async def restart_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any]:
raise NotImplementedError("WinSandboxProvider does not support restarting VMs.")
async def get_ip(self, name: str, storage: Optional[str] = None, retry_delay: int = 2) -> str:
"""Get the IP address of a VM, waiting indefinitely until it's available.
Args:
name: Name of the VM to get the IP for
storage: Ignored for Windows Sandbox
retry_delay: Delay between retries in seconds (default: 2)
Returns:
IP address of the VM when it becomes available
"""
total_attempts = 0
# Loop indefinitely until we get a valid IP
while True:
total_attempts += 1
# Log retry message but not on first attempt
if total_attempts > 1:
self.logger.info(f"Waiting for Windows Sandbox {name} IP address (attempt {total_attempts})...")
try:
# Get VM information
vm_info = await self.get_vm(name, storage=storage)
# Check if we got a valid IP
ip = vm_info.get("ip_address", None)
if ip and ip != "unknown" and not ip.startswith("0.0.0.0"):
self.logger.info(f"Got valid Windows Sandbox IP address: {ip}")
return ip
# Check the VM status
status = vm_info.get("status", "unknown")
# If VM is not running yet, log and wait
if status != "running":
self.logger.info(f"Windows Sandbox is not running yet (status: {status}). Waiting...")
# If VM is running but no IP yet, wait and retry
else:
self.logger.info("Windows Sandbox is running but no valid IP address yet. Waiting...")
except Exception as e:
self.logger.warning(f"Error getting Windows Sandbox {name} IP: {e}, continuing to wait...")
# Wait before next retry
await asyncio.sleep(retry_delay)
# Add progress log every 10 attempts
if total_attempts % 10 == 0:
self.logger.info(f"Still waiting for Windows Sandbox {name} IP after {total_attempts} attempts...")
async def _setup_computer_server(self, sandbox, name: str, visible: bool = False, wait_for_venv: bool = True):
"""Setup the computer server in the Windows Sandbox using RPyC.
Args:
sandbox: The Windows Sandbox instance
name: Name of the sandbox
visible: Whether the opened process should be visible (default: False)
"""
try:
self.logger.info(f"Setting up computer server in sandbox {name}...")
# Read the PowerShell setup script
script_path = os.path.join(os.path.dirname(__file__), "setup_script.ps1")
with open(script_path, 'r', encoding='utf-8') as f:
setup_script_content = f.read()
# Write the setup script to the sandbox using RPyC
script_dest_path = r"C:\Users\WDAGUtilityAccount\setup_cua.ps1"
self.logger.info(f"Writing setup script to {script_dest_path}")
with sandbox.rpyc.builtin.open(script_dest_path, 'w') as f:
f.write(setup_script_content)
# Execute the PowerShell script in the background
self.logger.info("Executing setup script in sandbox...")
# Use subprocess to run PowerShell script
import subprocess
powershell_cmd = [
"powershell.exe",
"-ExecutionPolicy", "Bypass",
"-NoExit", # Keep window open after script completes
"-File", script_dest_path
]
# Set creation flags based on visibility preference
if visible:
# CREATE_NEW_CONSOLE - creates a new console window (visible)
creation_flags = 0x00000010
else:
creation_flags = 0x08000000 # CREATE_NO_WINDOW
# Start the process using RPyC
process = sandbox.rpyc.modules.subprocess.Popen(
powershell_cmd,
creationflags=creation_flags,
shell=False
)
if wait_for_venv:
print("Waiting for venv to be created for the first time setup of Windows Sandbox...")
print("This may take a minute...")
await asyncio.sleep(120)
ip = await self.get_ip(name)
self.logger.info(f"Sandbox IP: {ip}")
self.logger.info(f"Setup script started in background in sandbox {name} with PID: {process.pid}")
except Exception as e:
self.logger.error(f"Failed to setup computer server in sandbox {name}: {e}")
import traceback
self.logger.error(f"Stack trace: {traceback.format_exc()}")
```
--------------------------------------------------------------------------------
/libs/python/computer/computer/providers/lume/provider.py:
--------------------------------------------------------------------------------
```python
"""Lume VM provider implementation using curl commands.
This provider uses direct curl commands to interact with the Lume API,
removing the dependency on the pylume Python package.
"""
import os
import re
import asyncio
import json
import logging
import subprocess
import urllib.parse
from typing import Dict, Any, Optional, List, Tuple
from ..base import BaseVMProvider, VMProviderType
from ...logger import Logger, LogLevel
from ..lume_api import (
lume_api_get,
lume_api_run,
lume_api_stop,
lume_api_update,
lume_api_pull,
HAS_CURL,
parse_memory
)
# Setup logging
logger = logging.getLogger(__name__)
class LumeProvider(BaseVMProvider):
"""Lume VM provider implementation using direct curl commands.
This provider uses curl to interact with the Lume API server,
removing the dependency on the pylume Python package.
"""
def __init__(
self,
port: int = 7777,
host: str = "localhost",
storage: Optional[str] = None,
verbose: bool = False,
ephemeral: bool = False,
):
"""Initialize the Lume provider.
Args:
port: Port for the Lume API server (default: 7777)
host: Host to use for API connections (default: localhost)
storage: Path to store VM data
verbose: Enable verbose logging
"""
if not HAS_CURL:
raise ImportError(
"curl is required for LumeProvider. "
"Please ensure it is installed and in your PATH."
)
self.host = host
self.port = port # Default port for Lume API
self.storage = storage
self.verbose = verbose
self.ephemeral = ephemeral # If True, VMs will be deleted after stopping
# Base API URL for Lume API calls
self.api_base_url = f"http://{self.host}:{self.port}"
self.logger = logging.getLogger(__name__)
@property
def provider_type(self) -> VMProviderType:
"""Get the provider type."""
return VMProviderType.LUME
async def __aenter__(self):
"""Enter async context manager."""
# No initialization needed, just return self
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Exit async context manager."""
# No cleanup needed
pass
def _lume_api_get(self, vm_name: str = "", storage: Optional[str] = None, debug: bool = False) -> Dict[str, Any]:
"""Get VM information using shared lume_api function.
Args:
vm_name: Optional name of the VM to get info for.
If empty, lists all VMs.
storage: Optional storage path override. If provided, this will be used instead of self.storage
debug: Whether to show debug output
Returns:
Dictionary with VM status information parsed from JSON response
"""
# Use the shared implementation from lume_api module
return lume_api_get(
vm_name=vm_name,
host=self.host,
port=self.port,
storage=storage if storage is not None else self.storage,
debug=debug,
verbose=self.verbose
)
def _lume_api_run(self, vm_name: str, run_opts: Dict[str, Any], debug: bool = False) -> Dict[str, Any]:
"""Run a VM using shared lume_api function.
Args:
vm_name: Name of the VM to run
run_opts: Dictionary of run options
debug: Whether to show debug output
Returns:
Dictionary with API response or error information
"""
# Use the shared implementation from lume_api module
return lume_api_run(
vm_name=vm_name,
host=self.host,
port=self.port,
run_opts=run_opts,
storage=self.storage,
debug=debug,
verbose=self.verbose
)
def _lume_api_stop(self, vm_name: str, debug: bool = False) -> Dict[str, Any]:
"""Stop a VM using shared lume_api function.
Args:
vm_name: Name of the VM to stop
debug: Whether to show debug output
Returns:
Dictionary with API response or error information
"""
# Use the shared implementation from lume_api module
return lume_api_stop(
vm_name=vm_name,
host=self.host,
port=self.port,
storage=self.storage,
debug=debug,
verbose=self.verbose
)
def _lume_api_update(self, vm_name: str, update_opts: Dict[str, Any], debug: bool = False) -> Dict[str, Any]:
"""Update VM configuration using shared lume_api function.
Args:
vm_name: Name of the VM to update
update_opts: Dictionary of update options
debug: Whether to show debug output
Returns:
Dictionary with API response or error information
"""
# Use the shared implementation from lume_api module
return lume_api_update(
vm_name=vm_name,
host=self.host,
port=self.port,
update_opts=update_opts,
storage=self.storage,
debug=debug,
verbose=self.verbose
)
async def get_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any]:
"""Get VM information by name.
Args:
name: Name of the VM to get information for
storage: Optional storage path override. If provided, this will be used
instead of the provider's default storage path.
Returns:
Dictionary with VM information including status, IP address, etc.
Note:
If storage is not provided, the provider's default storage path will be used.
The storage parameter allows overriding the storage location for this specific call.
"""
if not HAS_CURL:
logger.error("curl is not available. Cannot get VM status.")
return {
"name": name,
"status": "unavailable",
"error": "curl is not available"
}
# First try to get detailed VM info from the API
try:
# Query the Lume API for VM status using the provider's storage_path
vm_info = self._lume_api_get(
vm_name=name,
storage=storage if storage is not None else self.storage,
debug=self.verbose
)
# Check for API errors
if "error" in vm_info:
logger.debug(f"API request error: {vm_info['error']}")
# If we got an error from the API, report the VM as not ready yet
return {
"name": name,
"status": "starting", # VM is still starting - do not attempt to connect yet
"api_status": "error",
"error": vm_info["error"]
}
# Process the VM status information
vm_status = vm_info.get("status", "unknown")
# Check if VM is stopped or not running - don't wait for IP in this case
if vm_status == "stopped":
logger.info(f"VM {name} is in '{vm_status}' state - not waiting for IP address")
# Return the status as-is without waiting for an IP
result = {
"name": name,
"status": vm_status,
**vm_info # Include all original fields from the API response
}
return result
# Handle field name differences between APIs
# Some APIs use camelCase, others use snake_case
if "vncUrl" in vm_info:
vnc_url = vm_info["vncUrl"]
elif "vnc_url" in vm_info:
vnc_url = vm_info["vnc_url"]
else:
vnc_url = ""
if "ipAddress" in vm_info:
ip_address = vm_info["ipAddress"]
elif "ip_address" in vm_info:
ip_address = vm_info["ip_address"]
else:
# If no IP address is provided and VM is supposed to be running,
# report it as still starting
ip_address = None
logger.info(f"VM {name} is in '{vm_status}' state but no IP address found - reporting as still starting")
logger.info(f"VM {name} status: {vm_status}")
# Return the complete status information
result = {
"name": name,
"status": vm_status if vm_status else "running",
"ip_address": ip_address,
"vnc_url": vnc_url,
"api_status": "ok"
}
# Include all original fields from the API response
if isinstance(vm_info, dict):
for key, value in vm_info.items():
if key not in result: # Don't override our carefully processed fields
result[key] = value
return result
except Exception as e:
logger.error(f"Failed to get VM status: {e}")
# Return a fallback status that indicates the VM is not ready yet
return {
"name": name,
"status": "initializing", # VM is still initializing
"error": f"Failed to get VM status: {str(e)}"
}
async def list_vms(self) -> List[Dict[str, Any]]:
"""List all available VMs."""
result = self._lume_api_get(debug=self.verbose)
# Extract the VMs list from the response
if "vms" in result and isinstance(result["vms"], list):
return result["vms"]
elif "error" in result:
logger.error(f"Error listing VMs: {result['error']}")
return []
else:
return []
async def run_vm(self, image: str, name: str, run_opts: Dict[str, Any], storage: Optional[str] = None) -> Dict[str, Any]:
"""Run a VM with the given options.
If the VM does not exist in the storage location, this will attempt to pull it
from the Lume registry first.
Args:
image: Image name to use when pulling the VM if it doesn't exist
name: Name of the VM to run
run_opts: Dictionary of run options (memory, cpu, etc.)
storage: Optional storage path override. If provided, this will be used
instead of the provider's default storage path.
Returns:
Dictionary with VM run status and information
"""
# First check if VM exists by trying to get its info
vm_info = await self.get_vm(name, storage=storage)
if "error" in vm_info:
# VM doesn't exist, try to pull it
self.logger.info(f"VM {name} not found, attempting to pull image {image} from registry...")
# Call pull_vm with the image parameter
pull_result = await self.pull_vm(
name=name,
image=image,
storage=storage
)
# Check if pull was successful
if "error" in pull_result:
self.logger.error(f"Failed to pull VM image: {pull_result['error']}")
return pull_result # Return the error from pull
self.logger.info(f"Successfully pulled VM image {image} as {name}")
# Now run the VM with the given options
self.logger.info(f"Running VM {name} with options: {run_opts}")
from ..lume_api import lume_api_run
return lume_api_run(
vm_name=name,
host=self.host,
port=self.port,
run_opts=run_opts,
storage=storage if storage is not None else self.storage,
debug=self.verbose,
verbose=self.verbose
)
async def stop_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any]:
"""Stop a running VM.
If this provider was initialized with ephemeral=True, the VM will also
be deleted after it is stopped.
Args:
name: Name of the VM to stop
storage: Optional storage path override
Returns:
Dictionary with stop status and information
"""
# Stop the VM first
stop_result = self._lume_api_stop(name, debug=self.verbose)
# Log ephemeral status for debugging
self.logger.info(f"Ephemeral mode status: {self.ephemeral}")
# If ephemeral mode is enabled, delete the VM after stopping
if self.ephemeral and (stop_result.get("success", False) or "error" not in stop_result):
self.logger.info(f"Ephemeral mode enabled - deleting VM {name} after stopping")
try:
delete_result = await self.delete_vm(name, storage=storage)
# Return combined result
return {
**stop_result, # Include all stop result info
"deleted": True,
"delete_result": delete_result
}
except Exception as e:
self.logger.error(f"Failed to delete ephemeral VM {name}: {e}")
# Include the error but still return stop result
return {
**stop_result,
"deleted": False,
"delete_error": str(e)
}
# Just return the stop result if not ephemeral
return stop_result
async def pull_vm(
self,
name: str,
image: str,
storage: Optional[str] = None,
registry: str = "ghcr.io",
organization: str = "trycua",
pull_opts: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Pull a VM image from the registry.
Args:
name: Name for the VM after pulling
image: The image name to pull (e.g. 'macos-sequoia-cua:latest')
storage: Optional storage path to use
registry: Registry to pull from (default: ghcr.io)
organization: Organization in registry (default: trycua)
pull_opts: Additional options for pulling the VM (optional)
Returns:
Dictionary with information about the pulled VM
Raises:
RuntimeError: If pull operation fails or image is not provided
"""
# Validate image parameter
if not image:
raise ValueError("Image parameter is required for pull_vm")
self.logger.info(f"Pulling VM image '{image}' as '{name}'")
self.logger.info("You can check the pull progress using: lume logs -f")
# Set default pull_opts if not provided
if pull_opts is None:
pull_opts = {}
# Log information about the operation
self.logger.debug(f"Pull storage location: {storage or 'default'}")
try:
# Call the lume_api_pull function from lume_api.py
from ..lume_api import lume_api_pull
result = lume_api_pull(
image=image,
name=name,
host=self.host,
port=self.port,
storage=storage if storage is not None else self.storage,
registry=registry,
organization=organization,
debug=self.verbose,
verbose=self.verbose
)
# Check for errors in the result
if "error" in result:
self.logger.error(f"Failed to pull VM image: {result['error']}")
return result
self.logger.info(f"Successfully pulled VM image '{image}' as '{name}'")
return result
except Exception as e:
self.logger.error(f"Failed to pull VM image '{image}': {e}")
return {"error": f"Failed to pull VM: {str(e)}"}
async def delete_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any]:
"""Delete a VM permanently.
Args:
name: Name of the VM to delete
storage: Optional storage path override
Returns:
Dictionary with delete status and information
"""
self.logger.info(f"Deleting VM {name}...")
try:
# Call the lume_api_delete function we created
from ..lume_api import lume_api_delete
result = lume_api_delete(
vm_name=name,
host=self.host,
port=self.port,
storage=storage if storage is not None else self.storage,
debug=self.verbose,
verbose=self.verbose
)
# Check for errors in the result
if "error" in result:
self.logger.error(f"Failed to delete VM: {result['error']}")
return result
self.logger.info(f"Successfully deleted VM '{name}'")
return result
except Exception as e:
self.logger.error(f"Failed to delete VM '{name}': {e}")
return {"error": f"Failed to delete VM: {str(e)}"}
async def update_vm(self, name: str, update_opts: Dict[str, Any], storage: Optional[str] = None) -> Dict[str, Any]:
"""Update VM configuration."""
return self._lume_api_update(name, update_opts, debug=self.verbose)
async def restart_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any]:
raise NotImplementedError("LumeProvider does not support restarting VMs.")
async def get_ip(self, name: str, storage: Optional[str] = None, retry_delay: int = 2) -> str:
"""Get the IP address of a VM, waiting indefinitely until it's available.
Args:
name: Name of the VM to get the IP for
storage: Optional storage path override
retry_delay: Delay between retries in seconds (default: 2)
Returns:
IP address of the VM when it becomes available
"""
# Track total attempts for logging purposes
total_attempts = 0
# Loop indefinitely until we get a valid IP
while True:
total_attempts += 1
# Log retry message but not on first attempt
if total_attempts > 1:
self.logger.info(f"Waiting for VM {name} IP address (attempt {total_attempts})...")
try:
# Get VM information
vm_info = await self.get_vm(name, storage=storage)
# Check if we got a valid IP
ip = vm_info.get("ip_address", None)
if ip and ip != "unknown" and not ip.startswith("0.0.0.0"):
self.logger.info(f"Got valid VM IP address: {ip}")
return ip
# Check the VM status
status = vm_info.get("status", "unknown")
# If VM is not running yet, log and wait
if status != "running":
self.logger.info(f"VM is not running yet (status: {status}). Waiting...")
# If VM is running but no IP yet, wait and retry
else:
self.logger.info("VM is running but no valid IP address yet. Waiting...")
except Exception as e:
self.logger.warning(f"Error getting VM {name} IP: {e}, continuing to wait...")
# Wait before next retry
await asyncio.sleep(retry_delay)
# Add progress log every 10 attempts
if total_attempts % 10 == 0:
self.logger.info(f"Still waiting for VM {name} IP after {total_attempts} attempts...")
```
--------------------------------------------------------------------------------
/libs/python/computer/computer/providers/docker/provider.py:
--------------------------------------------------------------------------------
```python
"""
Docker VM provider implementation.
This provider uses Docker containers running the CUA Ubuntu image to create
Linux VMs with computer-server. It handles VM lifecycle operations through Docker
commands and container management.
"""
import logging
import json
import asyncio
from typing import Dict, List, Optional, Any
import subprocess
import time
import re
from ..base import BaseVMProvider, VMProviderType
# Setup logging
logger = logging.getLogger(__name__)
# Check if Docker is available
try:
subprocess.run(["docker", "--version"], capture_output=True, check=True)
HAS_DOCKER = True
except (subprocess.SubprocessError, FileNotFoundError):
HAS_DOCKER = False
class DockerProvider(BaseVMProvider):
"""
Docker VM Provider implementation using Docker containers.
This provider uses Docker to run containers with the CUA Ubuntu image
that includes computer-server for remote computer use.
"""
def __init__(
self,
port: Optional[int] = 8000,
host: str = "localhost",
storage: Optional[str] = None,
shared_path: Optional[str] = None,
image: str = "trycua/cua-ubuntu:latest",
verbose: bool = False,
ephemeral: bool = False,
vnc_port: Optional[int] = 6901,
):
"""Initialize the Docker VM Provider.
Args:
port: Currently unused (VM provider port)
host: Hostname for the API server (default: localhost)
storage: Path for persistent VM storage
shared_path: Path for shared folder between host and container
image: Docker image to use (default: "trycua/cua-ubuntu:latest")
Supported images:
- "trycua/cua-ubuntu:latest" (Kasm-based)
- "trycua/cua-docker-xfce:latest" (vanilla XFCE)
verbose: Enable verbose logging
ephemeral: Use ephemeral (temporary) storage
vnc_port: Port for VNC interface (default: 6901)
"""
self.host = host
self.api_port = 8000
self.vnc_port = vnc_port
self.ephemeral = ephemeral
# Handle ephemeral storage (temporary directory)
if ephemeral:
self.storage = "ephemeral"
else:
self.storage = storage
self.shared_path = shared_path
self.image = image
self.verbose = verbose
self._container_id = None
self._running_containers = {} # Track running containers by name
# Detect image type and configure user directory accordingly
self._detect_image_config()
def _detect_image_config(self):
"""Detect image type and configure paths accordingly."""
# Detect if this is a docker-xfce image or Kasm image
if "docker-xfce" in self.image.lower() or "xfce" in self.image.lower():
self._home_dir = "/home/cua"
self._image_type = "docker-xfce"
logger.info(f"Detected docker-xfce image: using {self._home_dir}")
else:
# Default to Kasm configuration
self._home_dir = "/home/kasm-user"
self._image_type = "kasm"
logger.info(f"Detected Kasm image: using {self._home_dir}")
@property
def provider_type(self) -> VMProviderType:
"""Return the provider type."""
return VMProviderType.DOCKER
def _parse_memory(self, memory_str: str) -> str:
"""Parse memory string to Docker format.
Examples:
"8GB" -> "8g"
"1024MB" -> "1024m"
"512" -> "512m"
"""
if isinstance(memory_str, int):
return f"{memory_str}m"
if isinstance(memory_str, str):
# Extract number and unit
match = re.match(r"(\d+)([A-Za-z]*)", memory_str)
if match:
value, unit = match.groups()
unit = unit.upper()
if unit == "GB" or unit == "G":
return f"{value}g"
elif unit == "MB" or unit == "M" or unit == "":
return f"{value}m"
# Default fallback
logger.warning(f"Could not parse memory string '{memory_str}', using 4g default")
return "4g" # Default to 4GB
async def get_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any]:
"""Get VM information by name.
Args:
name: Name of the VM to get information for
storage: Optional storage path override. If provided, this will be used
instead of the provider's default storage path.
Returns:
Dictionary with VM information including status, IP address, etc.
"""
try:
# Check if container exists and get its status
cmd = ["docker", "inspect", name]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
# Container doesn't exist
return {
"name": name,
"status": "not_found",
"ip_address": None,
"ports": {},
"image": self.image,
"provider": "docker"
}
# Parse container info
container_info = json.loads(result.stdout)[0]
state = container_info["State"]
network_settings = container_info["NetworkSettings"]
# Determine status
if state["Running"]:
status = "running"
elif state["Paused"]:
status = "paused"
else:
status = "stopped"
# Get IP address
ip_address = network_settings.get("IPAddress", "")
if not ip_address and "Networks" in network_settings:
# Try to get IP from bridge network
for network_name, network_info in network_settings["Networks"].items():
if network_info.get("IPAddress"):
ip_address = network_info["IPAddress"]
break
# Get port mappings
ports = {}
if "Ports" in network_settings and network_settings["Ports"]:
# network_settings["Ports"] is a dict like:
# {'6901/tcp': [{'HostIp': '0.0.0.0', 'HostPort': '6901'}, ...], ...}
for container_port, port_mappings in network_settings["Ports"].items():
if port_mappings: # Check if there are any port mappings
# Take the first mapping (usually the IPv4 one)
for mapping in port_mappings:
if mapping.get("HostPort"):
ports[container_port] = mapping["HostPort"]
break # Use the first valid mapping
return {
"name": name,
"status": status,
"ip_address": ip_address or "127.0.0.1", # Use localhost if no IP
"ports": ports,
"image": container_info["Config"]["Image"],
"provider": "docker",
"container_id": container_info["Id"][:12], # Short ID
"created": container_info["Created"],
"started": state.get("StartedAt", ""),
}
except Exception as e:
logger.error(f"Error getting VM info for {name}: {e}")
import traceback
traceback.print_exc()
return {
"name": name,
"status": "error",
"error": str(e),
"provider": "docker"
}
async def list_vms(self) -> List[Dict[str, Any]]:
"""List all Docker containers managed by this provider."""
try:
# List all containers (running and stopped) with the CUA image
cmd = ["docker", "ps", "-a", "--filter", f"ancestor={self.image}", "--format", "json"]
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
containers = []
if result.stdout.strip():
for line in result.stdout.strip().split('\n'):
if line.strip():
container_data = json.loads(line)
vm_info = await self.get_vm(container_data["Names"])
containers.append(vm_info)
return containers
except subprocess.CalledProcessError as e:
logger.error(f"Error listing containers: {e.stderr}")
return []
except Exception as e:
logger.error(f"Error listing VMs: {e}")
import traceback
traceback.print_exc()
return []
async def run_vm(self, image: str, name: str, run_opts: Dict[str, Any], storage: Optional[str] = None) -> Dict[str, Any]:
"""Run a VM with the given options.
Args:
image: Name/tag of the Docker image to use
name: Name of the container to run
run_opts: Options for running the VM, including:
- memory: Memory limit (e.g., "4GB", "2048MB")
- cpu: CPU limit (e.g., 2 for 2 cores)
- vnc_port: Specific port for VNC interface
- api_port: Specific port for computer-server API
Returns:
Dictionary with VM status information
"""
try:
# Check if container already exists
existing_vm = await self.get_vm(name, storage)
if existing_vm["status"] == "running":
logger.info(f"Container {name} is already running")
return existing_vm
elif existing_vm["status"] in ["stopped", "paused"]:
# Start existing container
logger.info(f"Starting existing container {name}")
start_cmd = ["docker", "start", name]
result = subprocess.run(start_cmd, capture_output=True, text=True, check=True)
# Wait for container to be ready
await self._wait_for_container_ready(name)
return await self.get_vm(name, storage)
# Use provided image or default
docker_image = image if image != "default" else self.image
# Build docker run command
cmd = ["docker", "run", "-d", "--name", name]
# Add memory limit if specified
if "memory" in run_opts:
memory_limit = self._parse_memory(run_opts["memory"])
cmd.extend(["--memory", memory_limit])
# Add CPU limit if specified
if "cpu" in run_opts:
cpu_count = str(run_opts["cpu"])
cmd.extend(["--cpus", cpu_count])
# Add port mappings
vnc_port = run_opts.get("vnc_port", self.vnc_port)
api_port = run_opts.get("api_port", self.api_port)
if vnc_port:
cmd.extend(["-p", f"{vnc_port}:6901"]) # VNC port
if api_port:
cmd.extend(["-p", f"{api_port}:8000"]) # computer-server API port
# Add volume mounts if storage is specified
storage_path = storage or self.storage
if storage_path and storage_path != "ephemeral":
# Mount storage directory using detected home directory
cmd.extend(["-v", f"{storage_path}:{self._home_dir}/storage"])
# Add shared path if specified
if self.shared_path:
# Mount shared directory using detected home directory
cmd.extend(["-v", f"{self.shared_path}:{self._home_dir}/shared"])
# Add environment variables
cmd.extend(["-e", "VNC_PW=password"]) # Set VNC password
cmd.extend(["-e", "VNCOPTIONS=-disableBasicAuth"]) # Disable VNC basic auth
# Add the image
cmd.append(docker_image)
logger.info(f"Running Docker container with command: {' '.join(cmd)}")
# Run the container
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
container_id = result.stdout.strip()
logger.info(f"Container {name} started with ID: {container_id[:12]}")
# Store container info
self._container_id = container_id
self._running_containers[name] = container_id
# Wait for container to be ready
await self._wait_for_container_ready(name)
# Return VM info
vm_info = await self.get_vm(name, storage)
vm_info["container_id"] = container_id[:12]
return vm_info
except subprocess.CalledProcessError as e:
error_msg = f"Failed to run container {name}: {e.stderr}"
logger.error(error_msg)
return {
"name": name,
"status": "error",
"error": error_msg,
"provider": "docker"
}
except Exception as e:
error_msg = f"Error running VM {name}: {e}"
logger.error(error_msg)
return {
"name": name,
"status": "error",
"error": error_msg,
"provider": "docker"
}
async def _wait_for_container_ready(self, container_name: str, timeout: int = 60) -> bool:
"""Wait for the Docker container to be fully ready.
Args:
container_name: Name of the Docker container to check
timeout: Maximum time to wait in seconds (default: 60 seconds)
Returns:
True if the container is running and ready
"""
logger.info(f"Waiting for container {container_name} to be ready...")
start_time = time.time()
while time.time() - start_time < timeout:
try:
# Check if container is running
vm_info = await self.get_vm(container_name)
if vm_info["status"] == "running":
logger.info(f"Container {container_name} is running")
# Additional check: try to connect to computer-server API
# This is optional - we'll just wait a bit more for services to start
await asyncio.sleep(5)
return True
except Exception as e:
logger.debug(f"Container {container_name} not ready yet: {e}")
await asyncio.sleep(2)
logger.warning(f"Container {container_name} did not become ready within {timeout} seconds")
return False
async def stop_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any]:
"""Stop a running VM by stopping the Docker container."""
try:
logger.info(f"Stopping container {name}")
# Stop the container
cmd = ["docker", "stop", name]
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
# Remove from running containers tracking
if name in self._running_containers:
del self._running_containers[name]
logger.info(f"Container {name} stopped successfully")
return {
"name": name,
"status": "stopped",
"message": "Container stopped successfully",
"provider": "docker"
}
except subprocess.CalledProcessError as e:
error_msg = f"Failed to stop container {name}: {e.stderr}"
logger.error(error_msg)
return {
"name": name,
"status": "error",
"error": error_msg,
"provider": "docker"
}
except Exception as e:
error_msg = f"Error stopping VM {name}: {e}"
logger.error(error_msg)
return {
"name": name,
"status": "error",
"error": error_msg,
"provider": "docker"
}
async def restart_vm(self, name: str, storage: Optional[str] = None) -> Dict[str, Any]:
raise NotImplementedError("DockerProvider does not support restarting VMs.")
async def update_vm(self, name: str, update_opts: Dict[str, Any], storage: Optional[str] = None) -> Dict[str, Any]:
"""Update VM configuration.
Note: Docker containers cannot be updated while running.
This method will return an error suggesting to recreate the container.
"""
return {
"name": name,
"status": "error",
"error": "Docker containers cannot be updated while running. Please stop and recreate the container with new options.",
"provider": "docker"
}
async def get_ip(self, name: str, storage: Optional[str] = None, retry_delay: int = 2) -> str:
"""Get the IP address of a VM, waiting indefinitely until it's available.
Args:
name: Name of the VM to get the IP for
storage: Optional storage path override
retry_delay: Delay between retries in seconds (default: 2)
Returns:
IP address of the VM when it becomes available
"""
logger.info(f"Getting IP address for container {name}")
total_attempts = 0
while True:
total_attempts += 1
try:
vm_info = await self.get_vm(name, storage)
if vm_info["status"] == "error":
raise Exception(f"VM is in error state: {vm_info.get('error', 'Unknown error')}")
# TODO: for now, return localhost
# it seems the docker container is not accessible from the host
# on WSL2, unless you port forward? not sure
if True:
logger.warning("Overriding container IP with localhost")
return "localhost"
# Check if we got a valid IP
ip = vm_info.get("ip_address", None)
if ip and ip != "unknown" and not ip.startswith("0.0.0.0"):
logger.info(f"Got valid container IP address: {ip}")
return ip
# For Docker containers, we can also use localhost if ports are mapped
if vm_info["status"] == "running" and vm_info.get("ports"):
logger.info(f"Container is running with port mappings, using localhost")
return "127.0.0.1"
# Check the container status
status = vm_info.get("status", "unknown")
if status == "stopped":
logger.info(f"Container status is {status}, but still waiting for it to start")
elif status != "running":
logger.info(f"Container is not running yet (status: {status}). Waiting...")
else:
logger.info("Container is running but no valid IP address yet. Waiting...")
except Exception as e:
logger.warning(f"Error getting container {name} IP: {e}, continuing to wait...")
# Wait before next retry
await asyncio.sleep(retry_delay)
# Add progress log every 10 attempts
if total_attempts % 10 == 0:
logger.info(f"Still waiting for container {name} IP after {total_attempts} attempts...")
async def __aenter__(self):
"""Async context manager entry."""
logger.debug("Entering DockerProvider context")
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context manager exit.
This method handles cleanup of running containers if needed.
"""
logger.debug(f"Exiting DockerProvider context, handling exceptions: {exc_type}")
try:
# Optionally stop running containers on context exit
# For now, we'll leave containers running as they might be needed
# Users can manually stop them if needed
pass
except Exception as e:
logger.error(f"Error during DockerProvider cleanup: {e}")
if exc_type is None:
raise
return False
```
--------------------------------------------------------------------------------
/blog/build-your-own-operator-on-macos-1.md:
--------------------------------------------------------------------------------
```markdown
# Build Your Own Operator on macOS - Part 1
*Published on March 31, 2025 by Francesco Bonacci*
In this first blogpost, we'll learn how to build our own Computer-Use Operator using OpenAI's `computer-use-preview` model. But first, let's understand what some common terms mean:
- A **Virtual Machine (VM)** is like a computer within your computer - a safe, isolated environment where the AI can work without affecting your main system.
- **computer-use-preview** is OpenAI's specialized language model trained to understand and interact with computer interfaces through screenshots.
- A **Computer-Use Agent** is an AI agent that can control a computer just like a human would - clicking buttons, typing text, and interacting with applications.
Our Operator will run in an isolated macOS VM, by making use of our [cua-computer](https://github.com/trycua/cua/tree/main/libs/computer) package and [lume virtualization CLI](https://github.com/trycua/cua/tree/main/libs/lume).
Check out what it looks like to use your own Operator from a Gradio app:
<div align="center">
<video src="https://github.com/user-attachments/assets/a2cf69ad-2ab2-4eb9-8e1a-45606dd7eec6" width="600" controls></video>
</div>
## What You'll Learn
By the end of this tutorial, you'll be able to:
- Set up a macOS virtual machine for AI automation
- Connect OpenAI's computer-use model to your VM
- Create a basic loop for the AI to interact with your VM
- Handle different types of computer actions (clicking, typing, etc.)
- Implement safety checks and error handling
**Prerequisites:**
- macOS Sonoma (14.0) or later
- 8GB RAM minimum (16GB recommended)
- OpenAI API access (Tier 3+)
- Basic Python knowledge
- Familiarity with terminal commands
**Estimated Time:** 45-60 minutes
## Introduction to Computer-Use Agents
Last March OpenAI released a fine-tuned version of GPT-4o, namely [CUA](https://openai.com/index/computer-using-agent/), introducing pixel-level vision capabilities with advanced reasoning through reinforcement learning. This fine-tuning enables the computer-use model to interpret screenshots and interact with graphical user interfaces on a pixel-level such as buttons, menus, and text fields - mimicking human interactions on a computer screen. It scores a remarkable 38.1% success rate on [OSWorld](https://os-world.github.io) - a benchmark for Computer-Use agents on Linux and Windows. This is the 2nd available model after Anthropic's [Claude 3.5 Sonnet](https://www.anthropic.com/news/3-5-models-and-computer-use) to support computer-use capabilities natively with no external models (e.g. accessory [SoM (Set-of-Mark)](https://arxiv.org/abs/2310.11441) and OCR runs).
Professor Ethan Mollick provides an excellent explanation of computer-use agents in this article: [When you give a Claude a mouse](https://www.oneusefulthing.org/p/when-you-give-a-claude-a-mouse).
### ChatGPT Operator
OpenAI's computer-use model powers [ChatGPT Operator](https://openai.com/index/introducing-operator), a Chromium-based interface exclusively available to ChatGPT Pro subscribers. Users leverage this functionality to automate web-based tasks such as online shopping, expense report submission, and booking reservations by interacting with websites in a human-like manner.
## Benefits of Custom Operators
### Why Build Your Own?
While OpenAI's Operator uses a controlled Chromium VM instance, there are scenarios where you may want to use your own VM with full desktop capabilities. Here are some examples:
- Automating native macOS apps like Finder, Xcode
- Managing files, changing settings, and running terminal commands
- Testing desktop software and applications
- Creating workflows that combine web and desktop tasks
- Automating media editing in apps like Final Cut Pro and Blender
This gives you more control and flexibility to automate tasks beyond just web browsing, with full access to interact with native applications and system-level operations. Additionally, running your own VM locally provides better privacy for sensitive user files and delivers superior performance by leveraging your own hardware instead of renting expensive Cloud VMs.
## Access Requirements
### Model Availability
As we speak, the **computer-use-preview** model has limited availability:
- Only accessible to OpenAI tier 3+ users
- Additional application process may be required even for eligible users
- Cannot be used in the OpenAI Playground
- Outside of ChatGPT Operator, usage is restricted to the new **Responses API**
## Understanding the OpenAI API
### Responses API Overview
Let's start with the basics. In our case, we'll use OpenAI's Responses API to communicate with their computer-use model.
Think of it like this:
1. We send the model a screenshot of our VM and tell it what we want it to do
2. The model looks at the screenshot and decides what actions to take
3. It sends back instructions (like "click here" or "type this")
4. We execute those instructions in our VM
The [Responses API](https://platform.openai.com/docs/guides/responses) is OpenAI's newest way to interact with their AI models. It comes with several built-in tools:
- **Web search**: Let the AI search the internet
- **File search**: Help the AI find documents
- **Computer use**: Allow the AI to control a computer (what we'll be using)
As we speak, the computer-use model is only available through the Responses API.
### Responses API Examples
Let's look at some simple examples. We'll start with the traditional way of using OpenAI's API with Chat Completions, then show the new Responses API primitive.
Chat Completions:
```python
# The old way required managing conversation history manually
messages = [{"role": "user", "content": "Hello"}]
response = client.chat.completions.create(
model="gpt-4",
messages=messages # We had to track all messages ourselves
)
messages.append(response.choices[0].message) # Manual message tracking
```
Responses API:
```python
# Example 1: Simple web search
# The API handles all the complexity for us
response = client.responses.create(
model="gpt-4",
input=[{
"role": "user",
"content": "What's the latest news about AI?"
}],
tools=[{
"type": "web_search", # Tell the API to use web search
"search_query": "latest AI news"
}]
)
# Example 2: File search
# Looking for specific documents becomes easy
response = client.responses.create(
model="gpt-4",
input=[{
"role": "user",
"content": "Find documents about project X"
}],
tools=[{
"type": "file_search",
"query": "project X",
"file_types": ["pdf", "docx"] # Specify which file types to look for
}]
)
```
### Computer-Use Model Setup
For our operator, we'll use the computer-use model. Here's how we set it up:
```python
# Set up the computer-use model to control our VM
response = client.responses.create(
model="computer-use-preview", # Special model for computer control
tools=[{
"type": "computer_use_preview",
"display_width": 1024, # Size of our VM screen
"display_height": 768,
"environment": "mac" # Tell it we're using macOS.
}],
input=[
{
"role": "user",
"content": [
# What we want the AI to do
{"type": "input_text", "text": "Open Safari and go to google.com"},
# Current screenshot of our VM
{"type": "input_image", "image_url": f"data:image/png;base64,{screenshot_base64}"}
]
}
],
truncation="auto" # Let OpenAI handle message length
)
```
### Understanding the Response
When we send a request, the API sends back a response that looks like this:
```json
"output": [
{
"type": "reasoning", # The AI explains what it's thinking
"id": "rs_67cc...",
"summary": [
{
"type": "summary_text",
"text": "Clicking on the browser address bar."
}
]
},
{
"type": "computer_call", # The actual action to perform
"id": "cu_67cc...",
"call_id": "call_zw3...",
"action": {
"type": "click", # What kind of action (click, type, etc.)
"button": "left", # Which mouse button to use
"x": 156, # Where to click (coordinates)
"y": 50
},
"pending_safety_checks": [], # Any safety warnings to consider
"status": "completed" # Whether the action was successful
}
]
```
Each response contains:
1. **Reasoning**: The AI's explanation of what it's doing
2. **Action**: The specific computer action to perform
3. **Safety Checks**: Any potential risks to review
4. **Status**: Whether everything worked as planned
## CUA-Computer Interface
### Architecture Overview
Let's break down the main components of our system and how they work together:
1. **The Virtual Machine (VM)**
- Think of this as a safe playground for our AI
- It's a complete macOS system running inside your computer
- Anything the AI does stays inside this VM, keeping your main system safe
- We use `lume` to create and manage this VM
2. **The Computer Interface (CUI)**
- This is how we control the VM
- It can move the mouse, type text, and take screenshots
- Works like a remote control for the VM
- Built using our `cua-computer` package
3. **The OpenAI Model**
- This is the brain of our operator
- It looks at screenshots of the VM
- Decides what actions to take
- Sends back instructions like "click here" or "type this"
Here's how they all work together:
```mermaid
sequenceDiagram
participant User as You
participant CUI as Computer Interface
participant VM as Virtual Machine
participant AI as OpenAI API
Note over User,AI: The Main Loop
User->>CUI: Start the operator
CUI->>VM: Create macOS sandbox
activate VM
VM-->>CUI: VM is ready
loop Action Loop
Note over CUI,AI: Each iteration
CUI->>VM: Take a screenshot
VM-->>CUI: Return current screen
CUI->>AI: Send screenshot + instructions
AI-->>CUI: Return next action
Note over CUI,VM: Execute the action
alt Mouse Click
CUI->>VM: Move and click mouse
else Type Text
CUI->>VM: Type characters
else Scroll Screen
CUI->>VM: Scroll window
else Press Keys
CUI->>VM: Press keyboard keys
else Wait
CUI->>VM: Pause for a moment
end
end
VM-->>CUI: Task finished
deactivate VM
CUI-->>User: All done!
```
The diagram above shows how information flows through our system:
1. You start the operator
2. The Computer Interface creates a virtual macOS
3. Then it enters a loop:
- Take a picture of the VM screen
- Send it to OpenAI with instructions
- Get back an action to perform
- Execute that action in the VM
- Repeat until the task is done
This design keeps everything organized and safe. The AI can only interact with the VM through our controlled interface, and the VM keeps the AI's actions isolated from your main system.
---
## Implementation Guide
### Prerequisites
1. **Lume CLI Setup**
For installing the standalone lume binary, run the following command from a terminal, or download the [latest pkg](https://github.com/trycua/cua/releases/latest/download/lume.pkg.tar.gz).
```bash
sudo /bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/trycua/cua/main/libs/lume/scripts/install.sh)"
```
**Important Storage Notes:**
- Initial download requires 80GB of free space
- After first run, space usage reduces to ~30GB due to macOS's sparse file system
- VMs are stored in `~/.lume`
- Cached images are stored in `~/.lume/cache`
You can check your downloaded VM images anytime:
```bash
lume ls
```
Example output:
| name | os | cpu | memory | disk | display | status | ip | vnc |
|--------------------------|---------|-------|---------|----------------|-----------|-----------|----------------|---------------------------------------------------|
| macos-sequoia-cua:latest | macOS | 12 | 16.00G | 64.5GB/80.0GB | 1024x768 | running | 192.168.64.78 | vnc://:[email protected]:56085 |
After checking your available images, you can run the VM to ensure everything is working correctly:
```bash
lume run macos-sequoia-cua:latest
```
2. **Python Environment Setup**
**Note**: The `cua-computer` package requires Python 3.10 or later. We recommend creating a dedicated Python environment:
**Using venv:**
```bash
python -m venv cua-env
source cua-env/bin/activate
```
**Using conda:**
```bash
conda create -n cua-env python=3.10
conda activate cua-env
```
Then install the required packages:
```bash
pip install openai
pip install cua-computer
```
Ensure you have an OpenAI API key (set as an environment variable or in your OpenAI configuration).
### Building the Operator
#### Importing Required Modules
With the prerequisites installed and configured, we're ready to build our first operator.
The following example uses asynchronous Python (async/await). You can run it either in a VS Code Notebook or as a standalone Python script.
```python
import asyncio
import base64
import openai
from computer import Computer
```
#### Mapping API Actions to CUA Methods
The following helper function converts a `computer_call` action from the OpenAI Responses API into corresponding commands on the CUI interface. For example, if the API instructs a `click` action, we move the cursor and perform a left click on the lume VM Sandbox. We will use the computer interface to execute the actions.
```python
async def execute_action(computer, action):
action_type = action.type
if action_type == "click":
x = action.x
y = action.y
button = action.button
print(f"Executing click at ({x}, {y}) with button '{button}'")
await computer.interface.move_cursor(x, y)
if button == "right":
await computer.interface.right_click()
else:
await computer.interface.left_click()
elif action_type == "type":
text = action.text
print(f"Typing text: {text}")
await computer.interface.type_text(text)
elif action_type == "scroll":
x = action.x
y = action.y
scroll_x = action.scroll_x
scroll_y = action.scroll_y
print(f"Scrolling at ({x}, {y}) with offsets (scroll_x={scroll_x}, scroll_y={scroll_y})")
await computer.interface.move_cursor(x, y)
await computer.interface.scroll(scroll_y) # Using vertical scroll only
elif action_type == "keypress":
keys = action.keys
for key in keys:
print(f"Pressing key: {key}")
# Map common key names to CUA equivalents
if key.lower() == "enter":
await computer.interface.press_key("return")
elif key.lower() == "space":
await computer.interface.press_key("space")
else:
await computer.interface.press_key(key)
elif action_type == "wait":
wait_time = action.time
print(f"Waiting for {wait_time} seconds")
await asyncio.sleep(wait_time)
elif action_type == "screenshot":
print("Taking screenshot")
# This is handled automatically in the main loop, but we can take an extra one if requested
screenshot = await computer.interface.screenshot()
return screenshot
else:
print(f"Unrecognized action: {action_type}")
```
#### Implementing the Computer-Use Loop
This section defines a loop that:
1. Initializes the cua-computer instance (connecting to a macOS sandbox).
2. Captures a screenshot of the current state.
3. Sends the screenshot (with a user prompt) to the OpenAI Responses API using the `computer-use-preview` model.
4. Processes the returned `computer_call` action and executes it using our helper function.
5. Captures an updated screenshot after the action (this example runs one iteration, but you can wrap it in a loop).
For a full loop, you would repeat these steps until no further actions are returned.
```python
async def cua_openai_loop():
# Initialize the lume computer instance (macOS sandbox)
async with Computer(
display="1024x768",
memory="4GB",
cpu="2",
os_type="macos"
) as computer:
await computer.run() # Start the lume VM
# Capture the initial screenshot
screenshot = await computer.interface.screenshot()
screenshot_base64 = base64.b64encode(screenshot).decode('utf-8')
# Initial request to start the loop
response = openai.responses.create(
model="computer-use-preview",
tools=[{
"type": "computer_use_preview",
"display_width": 1024,
"display_height": 768,
"environment": "mac"
}],
input=[
{
"role": "user",
"content": [
{"type": "input_text", "text": "Open Safari, download and install Cursor."},
{"type": "input_image", "image_url": f"data:image/png;base64,{screenshot_base64}"}
]
}
],
truncation="auto"
)
# Continue the loop until no more computer_call actions
while True:
# Check for computer_call actions
computer_calls = [item for item in response.output if item and item.type == "computer_call"]
if not computer_calls:
print("No more computer calls. Loop complete.")
break
# Get the first computer call
call = computer_calls[0]
last_call_id = call.call_id
action = call.action
print("Received action from OpenAI Responses API:", action)
# Handle any pending safety checks
if call.pending_safety_checks:
print("Safety checks pending:", call.pending_safety_checks)
# In a real implementation, you would want to get user confirmation here
acknowledged_checks = call.pending_safety_checks
else:
acknowledged_checks = []
# Execute the action
await execute_action(computer, action)
await asyncio.sleep(1) # Allow time for changes to take effect
# Capture new screenshot after action
new_screenshot = await computer.interface.screenshot()
new_screenshot_base64 = base64.b64encode(new_screenshot).decode('utf-8')
# Send the screenshot back as computer_call_output
response = openai.responses.create(
model="computer-use-preview",
tools=[{
"type": "computer_use_preview",
"display_width": 1024,
"display_height": 768,
"environment": "mac"
}],
input=[{
"type": "computer_call_output",
"call_id": last_call_id,
"acknowledged_safety_checks": acknowledged_checks,
"output": {
"type": "input_image",
"image_url": f"data:image/png;base64,{new_screenshot_base64}"
}
}],
truncation="auto"
)
# End the session
await computer.stop()
# Run the loop
if __name__ == "__main__":
asyncio.run(cua_openai_loop())
```
You can find the full code in our [notebook](https://github.com/trycua/cua/blob/main/notebooks/blog/build-your-own-operator-on-macos-1.ipynb).
#### Request Handling Differences
The first request to the OpenAI Responses API is special in that it includes the initial screenshot and prompt. Subsequent requests are handled differently, using the `computer_call_output` type to provide feedback on the executed action.
##### Initial Request Format
- We use `role: "user"` with `content` that contains both `input_text` (the prompt) and `input_image` (the screenshot)
##### Subsequent Request Format
- We use `type: "computer_call_output"` instead of the user role
- We include the `call_id` to link the output to the specific previous action that was executed
- We provide any `acknowledged_safety_checks` that were approved
- We include the new screenshot in the `output` field
This structured approach allows the API to maintain context and continuity throughout the interaction session.
**Note**: For multi-turn conversations, you should include the `previous_response_id` in your initial requests when starting a new conversation with prior context. However, when using `computer_call_output` for action feedback, you don't need to explicitly manage the conversation history - OpenAI's API automatically tracks the context using the `call_id`. The `previous_response_id` is primarily important when the user provides additional instructions or when starting a new request that should continue from a previous session.
## Conclusion
### Summary
This blogpost demonstrates a single iteration of a OpenAI Computer-Use loop where:
- A macOS sandbox is controlled using the CUA interface.
- A screenshot and prompt are sent to the OpenAI Responses API.
- The returned action (e.g. a click or type command) is executed via the CUI interface.
In a production setting, you would wrap the action-response cycle in a loop, handling multiple actions and safety checks as needed.
### Next Steps
In the next blogpost, we'll introduce our Agent framework which abstracts away all these tedious implementation steps. This framework provides a higher-level API that handles the interaction loop between OpenAI's computer-use model and the macOS sandbox, allowing you to focus on building sophisticated applications rather than managing the low-level details we've explored here. Can't wait? Check out the [cua-agent](https://github.com/trycua/cua/tree/main/libs/agent) package!
### Resources
- [OpenAI Computer-Use docs](https://platform.openai.com/docs/guides/tools-computer-use)
- [cua-computer](https://github.com/trycua/cua/tree/main/libs/computer)
- [lume](https://github.com/trycua/cua/tree/main/libs/lume)
```
--------------------------------------------------------------------------------
/libs/python/computer-server/computer_server/diorama/diorama.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""Diorama: A virtual desktop manager for macOS"""
import os
import asyncio
import logging
import sys
import io
from typing import Union
from PIL import Image, ImageDraw
from computer_server.diorama.draw import capture_all_apps, AppActivationContext, get_frontmost_and_active_app, get_all_windows, get_running_apps
from computer_server.diorama.diorama_computer import DioramaComputer
from computer_server.handlers.macos import *
# simple, nicely formatted logging
logger = logging.getLogger(__name__)
automation_handler = MacOSAutomationHandler()
class Diorama:
"""Virtual desktop manager that provides automation capabilities for macOS applications.
Manages application windows and provides an interface for taking screenshots,
mouse interactions, keyboard input, and coordinate transformations between
screenshot space and screen space.
"""
_scheduler_queue = None
_scheduler_task = None
_loop = None
_scheduler_started = False
@classmethod
def create_from_apps(cls, *args) -> DioramaComputer:
"""Create a DioramaComputer instance from a list of application names.
Args:
*args: Variable number of application names to include in the desktop
Returns:
DioramaComputer: A computer interface for the specified applications
"""
cls._ensure_scheduler()
return cls(args).computer
# Dictionary to store cursor positions for each unique app_list hash
_cursor_positions = {}
def __init__(self, app_list):
"""Initialize a Diorama instance for the specified applications.
Args:
app_list: List of application names to manage
"""
self.app_list = app_list
self.interface = self.Interface(self)
self.computer = DioramaComputer(self)
self.focus_context = None
# Create a hash for this app_list to use as a key
self.app_list_hash = hash(tuple(sorted(app_list)))
# Initialize cursor position for this app_list if it doesn't exist
if self.app_list_hash not in Diorama._cursor_positions:
Diorama._cursor_positions[self.app_list_hash] = (0, 0)
@classmethod
def _ensure_scheduler(cls):
"""Ensure the async scheduler loop is running.
Creates and starts the scheduler task if it hasn't been started yet.
"""
if not cls._scheduler_started:
logger.info("Starting Diorama scheduler loop…")
cls._scheduler_queue = asyncio.Queue()
cls._loop = asyncio.get_event_loop()
cls._scheduler_task = cls._loop.create_task(cls._scheduler_loop())
cls._scheduler_started = True
@classmethod
async def _scheduler_loop(cls):
"""Main scheduler loop that processes automation commands.
Continuously processes commands from the scheduler queue, handling
screenshots, mouse actions, keyboard input, and scrolling operations.
"""
while True:
cmd = await cls._scheduler_queue.get()
action = cmd.get("action")
args = cmd.get("arguments", {})
future = cmd.get("future")
logger.info(f"Processing command: {action} | args={args}")
app_whitelist = args.get("app_list", [])
all_windows = get_all_windows()
running_apps = get_running_apps()
frontmost_app, active_app_to_use, active_app_pid = get_frontmost_and_active_app(all_windows, running_apps, app_whitelist)
focus_context = AppActivationContext(active_app_pid, active_app_to_use, logger)
with focus_context:
try:
if action == "screenshot":
logger.info(f"Taking screenshot for apps: {app_whitelist}")
result, img = capture_all_apps(
app_whitelist=app_whitelist,
save_to_disk=False,
take_focus=False
)
logger.info("Screenshot complete.")
if future:
future.set_result((result, img))
# Mouse actions
elif action in ["left_click", "right_click", "double_click", "move_cursor", "drag_to"]:
x = args.get("x")
y = args.get("y")
duration = args.get("duration", 0.5)
if action == "left_click":
await automation_handler.left_click(x, y)
elif action == "right_click":
await automation_handler.right_click(x, y)
elif action == "double_click":
await automation_handler.double_click(x, y)
elif action == "move_cursor":
await automation_handler.move_cursor(x, y)
elif action == "drag_to":
await automation_handler.drag_to(x, y, duration=duration)
if future:
future.set_result(None)
elif action in ["scroll_up", "scroll_down"]:
x = args.get("x")
y = args.get("y")
if x is not None and y is not None:
await automation_handler.move_cursor(x, y)
clicks = args.get("clicks", 1)
if action == "scroll_up":
await automation_handler.scroll_up(clicks)
else:
await automation_handler.scroll_down(clicks)
if future:
future.set_result(None)
# Keyboard actions
elif action == "type_text":
text = args.get("text")
await automation_handler.type_text(text)
if future:
future.set_result(None)
elif action == "press_key":
key = args.get("key")
await automation_handler.press_key(key)
if future:
future.set_result(None)
elif action == "hotkey":
keys = args.get("keys", [])
await automation_handler.hotkey(keys)
if future:
future.set_result(None)
elif action == "get_cursor_position":
pos = await automation_handler.get_cursor_position()
if future:
future.set_result(pos)
else:
logger.warning(f"Unknown action: {action}")
if future:
future.set_exception(ValueError(f"Unknown action: {action}"))
except Exception as e:
logger.error(f"Exception during {action}: {e}", exc_info=True)
if future:
future.set_exception(e)
class Interface():
"""Interface for interacting with the virtual desktop.
Provides methods for taking screenshots, mouse interactions, keyboard input,
and coordinate transformations between screenshot and screen coordinates.
"""
def __init__(self, diorama):
"""Initialize the interface with a reference to the parent Diorama instance.
Args:
diorama: The parent Diorama instance
"""
self._diorama = diorama
self._scene_hitboxes = []
self._scene_size = None
async def _send_cmd(self, action, arguments=None):
"""Send a command to the scheduler queue.
Args:
action (str): The action to perform
arguments (dict, optional): Arguments for the action
Returns:
The result of the command execution
"""
Diorama._ensure_scheduler()
loop = asyncio.get_event_loop()
future = loop.create_future()
logger.info(f"Enqueuing {action} command for apps: {self._diorama.app_list}")
await Diorama._scheduler_queue.put({
"action": action,
"arguments": {"app_list": self._diorama.app_list, **(arguments or {})},
"future": future
})
try:
return await future
except asyncio.CancelledError:
logger.warning(f"Command was cancelled: {action}")
return None
async def screenshot(self, as_bytes: bool = True) -> Union[str, Image.Image]:
"""Take a screenshot of the managed applications.
Args:
as_bytes (bool): If True, return base64-encoded bytes; if False, return PIL Image
Returns:
Union[str, Image.Image]: Base64-encoded PNG bytes or PIL Image object
"""
import base64
result, img = await self._send_cmd("screenshot")
self._scene_hitboxes = result.get("hitboxes", [])
self._scene_size = img.size
if as_bytes:
# PIL Image to bytes, then base64 encode for JSON
import io
img_byte_arr = io.BytesIO()
img.save(img_byte_arr, format="PNG")
img_bytes = img_byte_arr.getvalue()
img_b64 = base64.b64encode(img_bytes).decode("ascii")
return img_b64
else:
return img
async def left_click(self, x, y):
"""Perform a left mouse click at the specified coordinates.
Args:
x (int): X coordinate in screenshot space (or None to use last position)
y (int): Y coordinate in screenshot space (or None to use last position)
"""
# Get last cursor position for this app_list hash
app_list_hash = hash(tuple(sorted(self._diorama.app_list)))
last_pos = Diorama._cursor_positions.get(app_list_hash, (0, 0))
x, y = x or last_pos[0], y or last_pos[1]
# Update cursor position for this app_list hash
Diorama._cursor_positions[app_list_hash] = (x, y)
sx, sy = await self.to_screen_coordinates(x, y)
await self._send_cmd("left_click", {"x": sx, "y": sy})
async def right_click(self, x, y):
"""Perform a right mouse click at the specified coordinates.
Args:
x (int): X coordinate in screenshot space (or None to use last position)
y (int): Y coordinate in screenshot space (or None to use last position)
"""
# Get last cursor position for this app_list hash
app_list_hash = hash(tuple(sorted(self._diorama.app_list)))
last_pos = Diorama._cursor_positions.get(app_list_hash, (0, 0))
x, y = x or last_pos[0], y or last_pos[1]
# Update cursor position for this app_list hash
Diorama._cursor_positions[app_list_hash] = (x, y)
sx, sy = await self.to_screen_coordinates(x, y)
await self._send_cmd("right_click", {"x": sx, "y": sy})
async def double_click(self, x, y):
"""Perform a double mouse click at the specified coordinates.
Args:
x (int): X coordinate in screenshot space (or None to use last position)
y (int): Y coordinate in screenshot space (or None to use last position)
"""
# Get last cursor position for this app_list hash
app_list_hash = hash(tuple(sorted(self._diorama.app_list)))
last_pos = Diorama._cursor_positions.get(app_list_hash, (0, 0))
x, y = x or last_pos[0], y or last_pos[1]
# Update cursor position for this app_list hash
Diorama._cursor_positions[app_list_hash] = (x, y)
sx, sy = await self.to_screen_coordinates(x, y)
await self._send_cmd("double_click", {"x": sx, "y": sy})
async def move_cursor(self, x, y):
"""Move the mouse cursor to the specified coordinates.
Args:
x (int): X coordinate in screenshot space (or None to use last position)
y (int): Y coordinate in screenshot space (or None to use last position)
"""
# Get last cursor position for this app_list hash
app_list_hash = hash(tuple(sorted(self._diorama.app_list)))
last_pos = Diorama._cursor_positions.get(app_list_hash, (0, 0))
x, y = x or last_pos[0], y or last_pos[1]
# Update cursor position for this app_list hash
Diorama._cursor_positions[app_list_hash] = (x, y)
sx, sy = await self.to_screen_coordinates(x, y)
await self._send_cmd("move_cursor", {"x": sx, "y": sy})
async def drag_to(self, x, y, duration=0.5):
"""Drag the mouse from current position to the specified coordinates.
Args:
x (int): X coordinate in screenshot space (or None to use last position)
y (int): Y coordinate in screenshot space (or None to use last position)
duration (float): Duration of the drag operation in seconds
"""
# Get last cursor position for this app_list hash
app_list_hash = hash(tuple(sorted(self._diorama.app_list)))
last_pos = Diorama._cursor_positions.get(app_list_hash, (0, 0))
x, y = x or last_pos[0], y or last_pos[1]
# Update cursor position for this app_list hash
Diorama._cursor_positions[app_list_hash] = (x, y)
sx, sy = await self.to_screen_coordinates(x, y)
await self._send_cmd("drag_to", {"x": sx, "y": sy, "duration": duration})
async def get_cursor_position(self):
"""Get the current cursor position in screen coordinates.
Returns:
tuple: (x, y) coordinates of the cursor in screen space
"""
return await self._send_cmd("get_cursor_position")
async def type_text(self, text):
"""Type the specified text using the keyboard.
Args:
text (str): The text to type
"""
await self._send_cmd("type_text", {"text": text})
async def press_key(self, key):
"""Press a single key on the keyboard.
Args:
key (str): The key to press
"""
await self._send_cmd("press_key", {"key": key})
async def hotkey(self, keys):
"""Press a combination of keys simultaneously.
Args:
keys (list): List of keys to press together
"""
await self._send_cmd("hotkey", {"keys": list(keys)})
async def scroll_up(self, clicks: int = 1):
"""Scroll up at the current cursor position.
Args:
clicks (int): Number of scroll clicks to perform
"""
# Get last cursor position for this app_list hash
app_list_hash = hash(tuple(sorted(self._diorama.app_list)))
last_pos = Diorama._cursor_positions.get(app_list_hash, (0, 0))
x, y = last_pos[0], last_pos[1]
await self._send_cmd("scroll_up", {"clicks": clicks, "x": x, "y": y})
async def scroll_down(self, clicks: int = 1):
"""Scroll down at the current cursor position.
Args:
clicks (int): Number of scroll clicks to perform
"""
# Get last cursor position for this app_list hash
app_list_hash = hash(tuple(sorted(self._diorama.app_list)))
last_pos = Diorama._cursor_positions.get(app_list_hash, (0, 0))
x, y = last_pos[0], last_pos[1]
await self._send_cmd("scroll_down", {"clicks": clicks, "x": x, "y": y})
async def get_screen_size(self) -> dict[str, int]:
"""Get the size of the screenshot area.
Returns:
dict[str, int]: Dictionary with 'width' and 'height' keys
"""
if not self._scene_size:
await self.screenshot()
return { "width": self._scene_size[0], "height": self._scene_size[1] }
async def to_screen_coordinates(self, x: float, y: float) -> tuple[float, float]:
"""Convert screenshot coordinates to screen coordinates.
Args:
x: X absolute coordinate in screenshot space
y: Y absolute coordinate in screenshot space
Returns:
tuple[float, float]: (x, y) absolute coordinates in screen space
"""
if not self._scene_hitboxes:
await self.screenshot() # get hitboxes
# Try all hitboxes
for h in self._scene_hitboxes[::-1]:
rect_from = h.get("hitbox")
rect_to = h.get("target")
if not rect_from or len(rect_from) != 4:
continue
# check if (x, y) is inside rect_from
x0, y0, x1, y1 = rect_from
if x0 <= x <= x1 and y0 <= y <= y1:
logger.info(f"Found hitbox: {h}")
# remap (x, y) to rect_to
tx0, ty0, tx1, ty1 = rect_to
# calculate offset from x0, y0
offset_x = x - x0
offset_y = y - y0
# remap offset to rect_to
tx = tx0 + offset_x
ty = ty0 + offset_y
return tx, ty
return x, y
async def to_screenshot_coordinates(self, x: float, y: float) -> tuple[float, float]:
"""Convert screen coordinates to screenshot coordinates.
Args:
x: X absolute coordinate in screen space
y: Y absolute coordinate in screen space
Returns:
tuple[float, float]: (x, y) absolute coordinates in screenshot space
"""
if not self._scene_hitboxes:
await self.screenshot() # get hitboxes
# Try all hitboxes
for h in self._scene_hitboxes[::-1]:
rect_from = h.get("target")
rect_to = h.get("hitbox")
if not rect_from or len(rect_from) != 4:
continue
# check if (x, y) is inside rect_from
x0, y0, x1, y1 = rect_from
if x0 <= x <= x1 and y0 <= y <= y1:
# remap (x, y) to rect_to
tx0, ty0, tx1, ty1 = rect_to
# calculate offset from x0, y0
offset_x = x - x0
offset_y = y - y0
# remap offset to rect_to
tx = tx0 + offset_x
ty = ty0 + offset_y
return tx, ty
return x, y
import pyautogui
import time
async def main():
"""Main function demonstrating Diorama usage with multiple desktops and mouse tracking."""
desktop1 = Diorama.create_from_apps(["Discord", "Notes"])
desktop2 = Diorama.create_from_apps(["Terminal"])
img1 = await desktop1.interface.screenshot(as_bytes=False)
img2 = await desktop2.interface.screenshot(as_bytes=False)
img1.save("app_screenshots/desktop1.png")
img2.save("app_screenshots/desktop2.png")
# Initialize Diorama desktop
desktop3 = Diorama.create_from_apps("Safari")
screen_size = await desktop3.interface.get_screen_size()
print(screen_size)
# Take initial screenshot
img = await desktop3.interface.screenshot(as_bytes=False)
img.save("app_screenshots/desktop3.png")
# Prepare hitboxes and draw on the single screenshot
hitboxes = desktop3.interface._scene_hitboxes[::-1]
base_img = img.copy()
draw = ImageDraw.Draw(base_img)
for h in hitboxes:
rect = h.get("hitbox")
if not rect or len(rect) != 4:
continue
draw.rectangle(rect, outline="red", width=2)
# Track and draw mouse position in real time (single screenshot size)
last_mouse_pos = None
print("Tracking mouse... Press Ctrl+C to stop.")
try:
while True:
mouse_x, mouse_y = pyautogui.position()
if last_mouse_pos != (mouse_x, mouse_y):
last_mouse_pos = (mouse_x, mouse_y)
# Map to screenshot coordinates
sx, sy = await desktop3.interface.to_screenshot_coordinates(mouse_x, mouse_y)
# Draw on a copy of the screenshot
frame = base_img.copy()
frame_draw = ImageDraw.Draw(frame)
frame_draw.ellipse((sx-5, sy-5, sx+5, sy+5), fill="blue", outline="blue")
# Save the frame
frame.save("app_screenshots/desktop3_mouse.png")
print(f"Mouse at screen ({mouse_x}, {mouse_y}) -> screenshot ({sx:.1f}, {sy:.1f})")
time.sleep(0.05) # Throttle updates to ~20 FPS
except KeyboardInterrupt:
print("Stopped tracking.")
draw.text((rect[0], rect[1]), str(idx), fill="red")
canvas.save("app_screenshots/desktop3_hitboxes.png")
# move mouse in a square spiral around the screen
import math
import random
step = 20 # pixels per move
dot_radius = 10
width = screen_size["width"]
height = screen_size["height"]
x, y = 0, 10
while x < width and y < height:
await desktop3.interface.move_cursor(x, y)
img = await desktop3.interface.screenshot(as_bytes=False)
draw = ImageDraw.Draw(img)
draw.ellipse((x-dot_radius, y-dot_radius, x+dot_radius, y+dot_radius), fill="red")
img.save("current.png")
await asyncio.sleep(0.03)
x += step
y = math.sin(x / width * math.pi * 2) * 50 + 25
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/libs/lume/src/Server/Server.swift:
--------------------------------------------------------------------------------
```swift
import Darwin
import Foundation
import Network
// MARK: - Error Types
enum PortError: Error, LocalizedError {
case alreadyInUse(port: UInt16)
var errorDescription: String? {
switch self {
case .alreadyInUse(let port):
return "Port \(port) is already in use by another process"
}
}
}
// MARK: - Server Class
@MainActor
final class Server {
// MARK: - Route Type
private struct Route {
let method: String
let path: String
let handler: (HTTPRequest) async throws -> HTTPResponse
func matches(_ request: HTTPRequest) -> Bool {
if method != request.method { return false }
// Handle path parameters
let routeParts = path.split(separator: "/")
let requestParts = request.path.split(separator: "/")
if routeParts.count != requestParts.count { return false }
for (routePart, requestPart) in zip(routeParts, requestParts) {
if routePart.hasPrefix(":") { continue } // Path parameter
if routePart != requestPart { return false }
}
return true
}
func extractParams(_ request: HTTPRequest) -> [String: String] {
var params: [String: String] = [:]
let routeParts = path.split(separator: "/")
// Split request path to remove query parameters
let requestPathOnly = request.path.split(separator: "?", maxSplits: 1)[0]
let requestParts = requestPathOnly.split(separator: "/")
for (routePart, requestPart) in zip(routeParts, requestParts) {
if routePart.hasPrefix(":") {
let paramName = String(routePart.dropFirst())
params[paramName] = String(requestPart)
}
}
return params
}
}
// MARK: - Properties
private let port: NWEndpoint.Port
private let controller: LumeController
private var isRunning = false
private var listener: NWListener?
private var routes: [Route]
// MARK: - Initialization
init(port: UInt16 = 7777) {
self.port = NWEndpoint.Port(rawValue: port)!
self.controller = LumeController()
self.routes = []
// Define API routes after self is fully initialized
self.setupRoutes()
}
// MARK: - Route Setup
private func setupRoutes() {
routes = [
Route(
method: "GET", path: "/lume/vms",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
// Extract storage from query params if present
let storage = self.extractQueryParam(request: request, name: "storage")
return try await self.handleListVMs(storage: storage)
}),
Route(
method: "GET", path: "/lume/vms/:name",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
let params = Route(
method: "GET", path: "/lume/vms/:name",
handler: { _ in
HTTPResponse(statusCode: .ok, body: "")
}
).extractParams(request)
guard let name = params["name"] else {
return HTTPResponse(statusCode: .badRequest, body: "Missing VM name")
}
// Extract storage from query params if present
let storage = self.extractQueryParam(request: request, name: "storage")
return try await self.handleGetVM(name: name, storage: storage)
}),
Route(
method: "DELETE", path: "/lume/vms/:name",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
let params = Route(
method: "DELETE", path: "/lume/vms/:name",
handler: { _ in
HTTPResponse(statusCode: .ok, body: "")
}
).extractParams(request)
guard let name = params["name"] else {
return HTTPResponse(statusCode: .badRequest, body: "Missing VM name")
}
// Extract storage from query params if present
let storage = self.extractQueryParam(request: request, name: "storage")
return try await self.handleDeleteVM(name: name, storage: storage)
}),
Route(
method: "POST", path: "/lume/vms",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
return try await self.handleCreateVM(request.body)
}),
Route(
method: "POST", path: "/lume/vms/clone",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
return try await self.handleCloneVM(request.body)
}),
Route(
method: "PATCH", path: "/lume/vms/:name",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
let params = Route(
method: "PATCH", path: "/lume/vms/:name",
handler: { _ in
HTTPResponse(statusCode: .ok, body: "")
}
).extractParams(request)
guard let name = params["name"] else {
return HTTPResponse(statusCode: .badRequest, body: "Missing VM name")
}
return try await self.handleSetVM(name: name, body: request.body)
}),
Route(
method: "POST", path: "/lume/vms/:name/run",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
let params = Route(
method: "POST", path: "/lume/vms/:name/run",
handler: { _ in
HTTPResponse(statusCode: .ok, body: "")
}
).extractParams(request)
guard let name = params["name"] else {
return HTTPResponse(statusCode: .badRequest, body: "Missing VM name")
}
return try await self.handleRunVM(name: name, body: request.body)
}),
Route(
method: "POST", path: "/lume/vms/:name/stop",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
let params = Route(
method: "POST", path: "/lume/vms/:name/stop",
handler: { _ in
HTTPResponse(statusCode: .ok, body: "")
}
).extractParams(request)
guard let name = params["name"] else {
return HTTPResponse(statusCode: .badRequest, body: "Missing VM name")
}
Logger.info("Processing stop VM request", metadata: ["method": request.method, "path": request.path])
// Extract storage from the request body
var storage: String? = nil
if let bodyData = request.body, !bodyData.isEmpty {
do {
if let json = try JSONSerialization.jsonObject(with: bodyData) as? [String: Any],
let bodyStorage = json["storage"] as? String {
storage = bodyStorage
Logger.info("Extracted storage from request body", metadata: ["storage": bodyStorage])
}
} catch {
Logger.error("Failed to parse request body JSON", metadata: ["error": error.localizedDescription])
}
}
return try await self.handleStopVM(name: name, storage: storage)
}),
Route(
method: "GET", path: "/lume/ipsw",
handler: { [weak self] _ in
guard let self else { throw HTTPError.internalError }
return try await self.handleIPSW()
}),
Route(
method: "POST", path: "/lume/pull",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
return try await self.handlePull(request.body)
}),
Route(
method: "POST", path: "/lume/prune",
handler: { [weak self] _ in
guard let self else { throw HTTPError.internalError }
return try await self.handlePruneImages()
}),
Route(
method: "GET", path: "/lume/images",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
return try await self.handleGetImages(request)
}),
// New config endpoint
Route(
method: "GET", path: "/lume/config",
handler: { [weak self] _ in
guard let self else { throw HTTPError.internalError }
return try await self.handleGetConfig()
}),
Route(
method: "POST", path: "/lume/config",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
return try await self.handleUpdateConfig(request.body)
}),
Route(
method: "GET", path: "/lume/config/locations",
handler: { [weak self] _ in
guard let self else { throw HTTPError.internalError }
return try await self.handleGetLocations()
}),
Route(
method: "POST", path: "/lume/config/locations",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
return try await self.handleAddLocation(request.body)
}),
Route(
method: "DELETE", path: "/lume/config/locations/:name",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
let params = Route(
method: "DELETE", path: "/lume/config/locations/:name",
handler: { _ in
HTTPResponse(statusCode: .ok, body: "")
}
).extractParams(request)
guard let name = params["name"] else {
return HTTPResponse(statusCode: .badRequest, body: "Missing location name")
}
return try await self.handleRemoveLocation(name)
}),
// Logs retrieval route
Route(
method: "GET", path: "/lume/logs",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
// Extract query parameters
let type = self.extractQueryParam(request: request, name: "type") // "info", "error", or "all"
let linesParam = self.extractQueryParam(request: request, name: "lines")
let lines = linesParam.flatMap { Int($0) } // Convert to Int if present
return try await self.handleGetLogs(type: type, lines: lines)
}),
Route(
method: "POST", path: "/lume/config/locations/default/:name",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
let params = Route(
method: "POST", path: "/lume/config/locations/default/:name",
handler: { _ in
HTTPResponse(statusCode: .ok, body: "")
}
).extractParams(request)
guard let name = params["name"] else {
return HTTPResponse(statusCode: .badRequest, body: "Missing location name")
}
return try await self.handleSetDefaultLocation(name)
}),
Route(
method: "POST", path: "/lume/vms/push",
handler: { [weak self] request in
guard let self else { throw HTTPError.internalError }
return try await self.handlePush(request.body)
}),
]
}
// Helper to extract query parameters from the URL
private func extractQueryParam(request: HTTPRequest, name: String) -> String? {
// Extract only the query part by splitting on '?'
let parts = request.path.split(separator: "?", maxSplits: 1)
guard parts.count > 1 else { return nil } // No query parameters
let queryString = String(parts[1])
// Create a placeholder URL with the query string
if let urlComponents = URLComponents(string: "http://placeholder.com?"+queryString),
let queryItems = urlComponents.queryItems
{
return queryItems.first(where: { $0.name == name })?.value?.removingPercentEncoding
}
return nil
}
// MARK: - Port Utilities
private func isPortAvailable(port: Int) async -> Bool {
// Create a socket
let socketFD = socket(AF_INET, SOCK_STREAM, 0)
if socketFD == -1 {
return false
}
// Set socket options to allow reuse
var value: Int32 = 1
if setsockopt(
socketFD, SOL_SOCKET, SO_REUSEADDR, &value, socklen_t(MemoryLayout<Int32>.size)) == -1
{
close(socketFD)
return false
}
// Set up the address structure
var addr = sockaddr_in()
addr.sin_family = sa_family_t(AF_INET)
addr.sin_port = UInt16(port).bigEndian
addr.sin_addr.s_addr = INADDR_ANY.bigEndian
// Bind to the port
let bindResult = withUnsafePointer(to: &addr) { addrPtr in
addrPtr.withMemoryRebound(to: sockaddr.self, capacity: 1) { addrPtr in
Darwin.bind(socketFD, addrPtr, socklen_t(MemoryLayout<sockaddr_in>.size))
}
}
// Clean up
close(socketFD)
// If bind failed, the port is in use
return bindResult == 0
}
// MARK: - Server Lifecycle
func start() async throws {
// First check if the port is already in use
if !(await isPortAvailable(port: Int(port.rawValue))) {
// Don't log anything here, just throw the error
throw PortError.alreadyInUse(port: port.rawValue)
}
let parameters = NWParameters.tcp
listener = try NWListener(using: parameters, on: port)
// Create an actor to safely manage state transitions
actor StartupState {
var error: Error?
var isComplete = false
func setError(_ error: Error) {
self.error = error
self.isComplete = true
}
func setComplete() {
self.isComplete = true
}
func checkStatus() -> (isComplete: Bool, error: Error?) {
return (isComplete, error)
}
}
let startupState = StartupState()
// Set up a state update handler to detect port binding errors
listener?.stateUpdateHandler = { state in
Task {
switch state {
case .setup:
// Initial state, no action needed
Logger.info("Listener setup", metadata: ["port": "\(self.port.rawValue)"])
break
case .waiting(let error):
// Log the full error details to see what we're getting
Logger.error(
"Listener waiting",
metadata: [
"error": error.localizedDescription,
"debugDescription": error.debugDescription,
"localizedDescription": error.localizedDescription,
"port": "\(self.port.rawValue)",
])
// Check for different port in use error messages
if error.debugDescription.contains("Address already in use")
|| error.localizedDescription.contains("in use")
|| error.localizedDescription.contains("address already in use")
{
Logger.error(
"Port conflict detected", metadata: ["port": "\(self.port.rawValue)"])
await startupState.setError(
PortError.alreadyInUse(port: self.port.rawValue))
} else {
// Wait for a short period to see if the listener recovers
// Some network errors are transient
try? await Task.sleep(nanoseconds: 1_000_000_000) // 1 second
// If we're still waiting after delay, consider it an error
if case .waiting = await self.listener?.state {
await startupState.setError(error)
}
}
case .failed(let error):
// Log the full error details
Logger.error(
"Listener failed",
metadata: [
"error": error.localizedDescription,
"debugDescription": error.debugDescription,
"port": "\(self.port.rawValue)",
])
await startupState.setError(error)
case .ready:
// Listener successfully bound to port
Logger.info("Listener ready", metadata: ["port": "\(self.port.rawValue)"])
await startupState.setComplete()
case .cancelled:
// Listener was cancelled
Logger.info("Listener cancelled", metadata: ["port": "\(self.port.rawValue)"])
break
@unknown default:
Logger.info(
"Unknown listener state",
metadata: ["state": "\(state)", "port": "\(self.port.rawValue)"])
break
}
}
}
listener?.newConnectionHandler = { [weak self] connection in
Task { @MainActor [weak self] in
guard let self else { return }
self.handleConnection(connection)
}
}
listener?.start(queue: .main)
// Wait for either successful startup or an error
var status: (isComplete: Bool, error: Error?) = (false, nil)
repeat {
try await Task.sleep(nanoseconds: 100_000_000) // 100ms
status = await startupState.checkStatus()
} while !status.isComplete
// If there was a startup error, throw it
if let error = status.error {
self.stop()
throw error
}
isRunning = true
Logger.info("Server started", metadata: ["port": "\(port.rawValue)"])
// Keep the server running
while isRunning {
try await Task.sleep(nanoseconds: 1_000_000_000)
}
}
func stop() {
isRunning = false
listener?.cancel()
}
// MARK: - Connection Handling
private func handleConnection(_ connection: NWConnection) {
connection.stateUpdateHandler = { [weak self] state in
switch state {
case .ready:
Task { @MainActor [weak self] in
guard let self else { return }
self.receiveData(connection)
}
case .failed(let error):
Logger.error("Connection failed", metadata: ["error": error.localizedDescription])
connection.cancel()
case .cancelled:
// Connection is already cancelled, no need to cancel again
break
default:
break
}
}
connection.start(queue: .main)
}
private func receiveData(_ connection: NWConnection) {
connection.receive(minimumIncompleteLength: 1, maximumLength: 65536) {
[weak self] content, _, isComplete, error in
if let error = error {
Logger.error("Receive error", metadata: ["error": error.localizedDescription])
connection.cancel()
return
}
guard let data = content, !data.isEmpty else {
if isComplete {
connection.cancel()
}
return
}
Task { @MainActor [weak self] in
guard let self else { return }
do {
let response = try await self.handleRequest(data)
self.send(response, on: connection)
} catch {
let errorResponse = self.errorResponse(error)
self.send(errorResponse, on: connection)
}
}
}
}
private func send(_ response: HTTPResponse, on connection: NWConnection) {
let data = response.serialize()
Logger.info(
"Serialized response", metadata: ["data": String(data: data, encoding: .utf8) ?? ""])
connection.send(
content: data,
completion: .contentProcessed { [weak connection] error in
if let error = error {
Logger.error(
"Failed to send response", metadata: ["error": error.localizedDescription])
} else {
Logger.info("Response sent successfully")
}
if connection?.state != .cancelled {
connection?.cancel()
}
})
}
// MARK: - Request Handling
private func handleRequest(_ data: Data) async throws -> HTTPResponse {
Logger.info(
"Received request data", metadata: ["data": String(data: data, encoding: .utf8) ?? ""])
guard let request = HTTPRequest(data: data) else {
Logger.error("Failed to parse request")
return HTTPResponse(statusCode: .badRequest, body: "Invalid request")
}
Logger.info(
"Parsed request",
metadata: [
"method": request.method,
"path": request.path,
"headers": "\(request.headers)",
"body": String(data: request.body ?? Data(), encoding: .utf8) ?? "",
])
// Find matching route
guard let route = routes.first(where: { $0.matches(request) }) else {
return HTTPResponse(statusCode: .notFound, body: "Not found")
}
// Handle the request
let response = try await route.handler(request)
Logger.info(
"Sending response",
metadata: [
"statusCode": "\(response.statusCode.rawValue)",
"headers": "\(response.headers)",
"body": String(data: response.body ?? Data(), encoding: .utf8) ?? "",
])
return response
}
private func errorResponse(_ error: Error) -> HTTPResponse {
HTTPResponse(
statusCode: .internalServerError,
headers: ["Content-Type": "application/json"],
body: try! JSONEncoder().encode(APIError(message: error.localizedDescription))
)
}
}
```
--------------------------------------------------------------------------------
/libs/typescript/computer/tests/interface/macos.test.ts:
--------------------------------------------------------------------------------
```typescript
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { WebSocket, WebSocketServer } from 'ws';
import { MacOSComputerInterface } from '../../src/interface/macos.ts';
describe('MacOSComputerInterface', () => {
// Define test parameters
const testParams = {
ipAddress: 'localhost',
username: 'testuser',
password: 'testpass',
// apiKey: "test-api-key", No API Key for local testing
vmName: 'test-vm',
};
// WebSocket server mock
let wss: WebSocketServer;
let serverPort: number;
let connectedClients: WebSocket[] = [];
// Track received messages for verification
interface ReceivedMessage {
action: string;
[key: string]: unknown;
}
let receivedMessages: ReceivedMessage[] = [];
// Set up WebSocket server before all tests
beforeEach(async () => {
receivedMessages = [];
connectedClients = [];
// Create WebSocket server on a random available port
wss = new WebSocketServer({ port: 0 });
serverPort = (wss.address() as { port: number }).port;
// Update test params with the actual server address
testParams.ipAddress = `localhost:${serverPort}`;
// Handle WebSocket connections
wss.on('connection', (ws) => {
connectedClients.push(ws);
// Handle incoming messages
ws.on('message', (data) => {
try {
const message = JSON.parse(data.toString());
receivedMessages.push(message);
// Send appropriate responses based on action
switch (message.command) {
case 'screenshot':
ws.send(
JSON.stringify({
image_data: Buffer.from('fake-screenshot-data').toString(
'base64'
),
success: true,
})
);
break;
case 'get_screen_size':
ws.send(
JSON.stringify({
size: { width: 1920, height: 1080 },
success: true,
})
);
break;
case 'get_cursor_position':
ws.send(
JSON.stringify({
position: { x: 100, y: 200 },
success: true,
})
);
break;
case 'copy_to_clipboard':
ws.send(
JSON.stringify({
content: 'clipboard content',
success: true,
})
);
break;
case 'file_exists':
ws.send(
JSON.stringify({
exists: true,
success: true,
})
);
break;
case 'directory_exists':
ws.send(
JSON.stringify({
exists: true,
success: true,
})
);
break;
case 'list_dir':
ws.send(
JSON.stringify({
files: ['file1.txt', 'file2.txt'],
success: true,
})
);
break;
case 'read_text':
ws.send(
JSON.stringify({
content: 'file content',
success: true,
})
);
break;
case 'read_bytes':
ws.send(
JSON.stringify({
content_b64: Buffer.from('binary content').toString('base64'),
success: true,
})
);
break;
case 'run_command':
ws.send(
JSON.stringify({
stdout: 'command output',
stderr: '',
success: true,
})
);
break;
case 'get_accessibility_tree':
ws.send(
JSON.stringify({
role: 'window',
title: 'Test Window',
bounds: { x: 0, y: 0, width: 1920, height: 1080 },
children: [],
success: true,
})
);
break;
case 'to_screen_coordinates':
case 'to_screenshot_coordinates':
ws.send(
JSON.stringify({
coordinates: [message.params?.x || 0, message.params?.y || 0],
success: true,
})
);
break;
default:
// For all other actions, just send success
ws.send(JSON.stringify({ success: true }));
break;
}
} catch (error) {
ws.send(JSON.stringify({ error: (error as Error).message }));
}
});
ws.on('error', (error) => {
console.error('WebSocket error:', error);
});
});
});
// Clean up WebSocket server after each test
afterEach(async () => {
// Close all connected clients
for (const client of connectedClients) {
if (client.readyState === WebSocket.OPEN) {
client.close();
}
}
// Close the server
await new Promise<void>((resolve) => {
wss.close(() => resolve());
});
});
describe('Connection Management', () => {
it('should connect with proper authentication headers', async () => {
const macosInterface = new MacOSComputerInterface(
testParams.ipAddress,
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
await macosInterface.connect();
// Verify the interface is connected
expect(macosInterface.isConnected()).toBe(true);
expect(connectedClients.length).toBe(1);
await macosInterface.disconnect();
});
it('should handle connection without API key', async () => {
// Create a separate server that doesn't check auth
const noAuthWss = new WebSocketServer({ port: 0 });
const noAuthPort = (noAuthWss.address() as { port: number }).port;
noAuthWss.on('connection', (ws) => {
ws.on('message', () => {
ws.send(JSON.stringify({ success: true }));
});
});
const macosInterface = new MacOSComputerInterface(
`localhost:${noAuthPort}`,
testParams.username,
testParams.password,
undefined,
undefined
);
await macosInterface.connect();
expect(macosInterface.isConnected()).toBe(true);
await macosInterface.disconnect();
await new Promise<void>((resolve) => {
noAuthWss.close(() => resolve());
});
});
});
describe('Mouse Actions', () => {
let macosInterface: MacOSComputerInterface;
beforeEach(async () => {
macosInterface = new MacOSComputerInterface(
testParams.ipAddress,
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
await macosInterface.connect();
});
afterEach(async () => {
if (macosInterface) {
await macosInterface.disconnect();
}
});
it('should send mouse_down command', async () => {
await macosInterface.mouseDown(100, 200, 'left');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'mouse_down',
params: {
x: 100,
y: 200,
button: 'left',
},
});
});
it('should send mouse_up command', async () => {
await macosInterface.mouseUp(100, 200, 'right');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'mouse_up',
params: {
x: 100,
y: 200,
button: 'right',
},
});
});
it('should send left_click command', async () => {
await macosInterface.leftClick(150, 250);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'left_click',
params: {
x: 150,
y: 250,
},
});
});
it('should send right_click command', async () => {
await macosInterface.rightClick(200, 300);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'right_click',
params: {
x: 200,
y: 300,
},
});
});
it('should send double_click command', async () => {
await macosInterface.doubleClick(250, 350);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'double_click',
params: {
x: 250,
y: 350,
},
});
});
it('should send move_cursor command', async () => {
await macosInterface.moveCursor(300, 400);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'move_cursor',
params: {
x: 300,
y: 400,
},
});
});
it('should send drag_to command', async () => {
await macosInterface.dragTo(400, 500, 'left', 1.5);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'drag_to',
params: {
x: 400,
y: 500,
button: 'left',
duration: 1.5,
},
});
});
it('should send drag command with path', async () => {
const path: Array<[number, number]> = [
[100, 100],
[200, 200],
[300, 300],
];
await macosInterface.drag(path, 'middle', 2.0);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'drag',
params: {
path: path,
button: 'middle',
duration: 2.0,
},
});
});
});
describe('Keyboard Actions', () => {
let macosInterface: MacOSComputerInterface;
beforeEach(async () => {
macosInterface = new MacOSComputerInterface(
testParams.ipAddress,
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
await macosInterface.connect();
});
afterEach(async () => {
if (macosInterface) {
await macosInterface.disconnect();
}
});
it('should send key_down command', async () => {
await macosInterface.keyDown('a');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'key_down',
params: {
key: 'a',
},
});
});
it('should send key_up command', async () => {
await macosInterface.keyUp('b');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'key_up',
params: {
key: 'b',
},
});
});
it('should send type_text command', async () => {
await macosInterface.typeText('Hello, World!');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'type_text',
params: {
text: 'Hello, World!',
},
});
});
it('should send press_key command', async () => {
await macosInterface.pressKey('enter');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'press_key',
params: {
key: 'enter',
},
});
});
it('should send hotkey command', async () => {
await macosInterface.hotkey('cmd', 'c');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'hotkey',
params: {
keys: ['cmd', 'c'],
},
});
});
});
describe('Scrolling Actions', () => {
let macosInterface: MacOSComputerInterface;
beforeEach(async () => {
macosInterface = new MacOSComputerInterface(
testParams.ipAddress,
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
await macosInterface.connect();
});
afterEach(async () => {
if (macosInterface) {
await macosInterface.disconnect();
}
});
it('should send scroll command', async () => {
await macosInterface.scroll(10, -5);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'scroll',
params: {
x: 10,
y: -5,
},
});
});
it('should send scroll_down command', async () => {
await macosInterface.scrollDown(3);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'scroll_down',
params: {
clicks: 3,
},
});
});
it('should send scroll_up command', async () => {
await macosInterface.scrollUp(2);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'scroll_up',
params: {
clicks: 2,
},
});
});
});
describe('Screen Actions', () => {
let macosInterface: MacOSComputerInterface;
beforeEach(async () => {
macosInterface = new MacOSComputerInterface(
testParams.ipAddress,
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
await macosInterface.connect();
});
afterEach(async () => {
if (macosInterface) {
await macosInterface.disconnect();
}
});
it('should get screenshot', async () => {
const screenshot = await macosInterface.screenshot();
expect(screenshot).toBeInstanceOf(Buffer);
expect(screenshot.toString()).toBe('fake-screenshot-data');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'screenshot',
params: {},
});
});
it('should get screen size', async () => {
const size = await macosInterface.getScreenSize();
expect(size).toEqual({ width: 1920, height: 1080 });
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'get_screen_size',
params: {},
});
});
it('should get cursor position', async () => {
const position = await macosInterface.getCursorPosition();
expect(position).toEqual({ x: 100, y: 200 });
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'get_cursor_position',
params: {},
});
});
});
describe('Clipboard Actions', () => {
let macosInterface: MacOSComputerInterface;
beforeEach(async () => {
macosInterface = new MacOSComputerInterface(
testParams.ipAddress,
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
await macosInterface.connect();
});
afterEach(async () => {
if (macosInterface) {
await macosInterface.disconnect();
}
});
it('should copy to clipboard', async () => {
const text = await macosInterface.copyToClipboard();
expect(text).toBe('clipboard content');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'copy_to_clipboard',
params: {},
});
});
it('should set clipboard', async () => {
await macosInterface.setClipboard('new clipboard text');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'set_clipboard',
params: {
text: 'new clipboard text',
},
});
});
});
describe('File System Actions', () => {
let macosInterface: MacOSComputerInterface;
beforeEach(async () => {
macosInterface = new MacOSComputerInterface(
testParams.ipAddress,
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
await macosInterface.connect();
});
afterEach(async () => {
if (macosInterface) {
await macosInterface.disconnect();
}
});
it('should check file exists', async () => {
const exists = await macosInterface.fileExists('/path/to/file');
expect(exists).toBe(true);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'file_exists',
params: {
path: '/path/to/file',
},
});
});
it('should check directory exists', async () => {
const exists = await macosInterface.directoryExists('/path/to/dir');
expect(exists).toBe(true);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'directory_exists',
params: {
path: '/path/to/dir',
},
});
});
it('should list directory', async () => {
const files = await macosInterface.listDir('/path/to/dir');
expect(files).toEqual(['file1.txt', 'file2.txt']);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'list_dir',
params: {
path: '/path/to/dir',
},
});
});
it('should read text file', async () => {
const content = await macosInterface.readText('/path/to/file.txt');
expect(content).toBe('file content');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'read_text',
params: {
path: '/path/to/file.txt',
},
});
});
it('should write text file', async () => {
await macosInterface.writeText('/path/to/file.txt', 'new content');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'write_text',
params: {
path: '/path/to/file.txt',
content: 'new content',
},
});
});
it('should read binary file', async () => {
const content = await macosInterface.readBytes('/path/to/file.bin');
expect(content).toBeInstanceOf(Buffer);
expect(content.toString()).toBe('binary content');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'read_bytes',
params: {
path: '/path/to/file.bin',
},
});
});
it('should write binary file', async () => {
const buffer = Buffer.from('binary data');
await macosInterface.writeBytes('/path/to/file.bin', buffer);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'write_bytes',
params: {
path: '/path/to/file.bin',
content_b64: buffer.toString('base64'),
},
});
});
it('should delete file', async () => {
await macosInterface.deleteFile('/path/to/file');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'delete_file',
params: {
path: '/path/to/file',
},
});
});
it('should create directory', async () => {
await macosInterface.createDir('/path/to/new/dir');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'create_dir',
params: {
path: '/path/to/new/dir',
},
});
});
it('should delete directory', async () => {
await macosInterface.deleteDir('/path/to/dir');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'delete_dir',
params: {
path: '/path/to/dir',
},
});
});
it('should run command', async () => {
const [stdout, stderr] = await macosInterface.runCommand('ls -la');
expect(stdout).toBe('command output');
expect(stderr).toBe('');
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'run_command',
params: {
command: 'ls -la',
},
});
});
});
describe('Accessibility Actions', () => {
let macosInterface: MacOSComputerInterface;
beforeEach(async () => {
macosInterface = new MacOSComputerInterface(
testParams.ipAddress,
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
await macosInterface.connect();
});
afterEach(async () => {
if (macosInterface) {
await macosInterface.disconnect();
}
});
it('should get accessibility tree', async () => {
const tree = await macosInterface.getAccessibilityTree();
expect(tree).toEqual({
role: 'window',
title: 'Test Window',
bounds: { x: 0, y: 0, width: 1920, height: 1080 },
children: [],
success: true,
});
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'get_accessibility_tree',
params: {},
});
});
it('should convert to screen coordinates', async () => {
const [x, y] = await macosInterface.toScreenCoordinates(100, 200);
expect(x).toBe(100);
expect(y).toBe(200);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'to_screen_coordinates',
params: {
x: 100,
y: 200,
},
});
});
it('should convert to screenshot coordinates', async () => {
const [x, y] = await macosInterface.toScreenshotCoordinates(300, 400);
expect(x).toBe(300);
expect(y).toBe(400);
const lastMessage = receivedMessages[receivedMessages.length - 1];
expect(lastMessage).toEqual({
command: 'to_screenshot_coordinates',
params: {
x: 300,
y: 400,
},
});
});
});
describe('Error Handling', () => {
it('should handle WebSocket connection errors', async () => {
// Use a valid but unreachable IP to avoid DNS errors
const macosInterface = new MacOSComputerInterface(
'localhost:9999',
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
// Connection should fail
await expect(macosInterface.connect()).rejects.toThrow();
});
it('should handle command errors', async () => {
// Create a server that returns errors
const errorWss = new WebSocketServer({ port: 0 });
const errorPort = (errorWss.address() as { port: number }).port;
errorWss.on('connection', (ws) => {
ws.on('message', () => {
ws.send(JSON.stringify({ error: 'Command failed', success: false }));
});
});
const macosInterface = new MacOSComputerInterface(
`localhost:${errorPort}`,
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
await macosInterface.connect();
// Command should throw error
await expect(macosInterface.leftClick(100, 100)).rejects.toThrow(
'Command failed'
);
await macosInterface.disconnect();
await new Promise<void>((resolve) => {
errorWss.close(() => resolve());
});
});
it('should handle disconnection gracefully', async () => {
const macosInterface = new MacOSComputerInterface(
testParams.ipAddress,
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
await macosInterface.connect();
expect(macosInterface.isConnected()).toBe(true);
// Disconnect
macosInterface.disconnect();
expect(macosInterface.isConnected()).toBe(false);
// Should reconnect automatically on next command
await macosInterface.leftClick(100, 100);
expect(macosInterface.isConnected()).toBe(true);
await macosInterface.disconnect();
});
it('should handle force close', async () => {
const macosInterface = new MacOSComputerInterface(
testParams.ipAddress,
testParams.username,
testParams.password,
undefined,
testParams.vmName
);
await macosInterface.connect();
expect(macosInterface.isConnected()).toBe(true);
// Force close
macosInterface.forceClose();
expect(macosInterface.isConnected()).toBe(false);
});
});
});
```
--------------------------------------------------------------------------------
/libs/python/computer-server/computer_server/handlers/windows.py:
--------------------------------------------------------------------------------
```python
"""
Windows implementation of automation and accessibility handlers.
This implementation uses pyautogui for GUI automation and Windows-specific APIs
for accessibility and system operations.
"""
from typing import Dict, Any, List, Tuple, Optional
import logging
import subprocess
import asyncio
import base64
import os
from io import BytesIO
from pynput.mouse import Controller as MouseController
from pynput.keyboard import Controller as KeyboardController
# Configure logger
logger = logging.getLogger(__name__)
# Try to import pyautogui
try:
import pyautogui
pyautogui.FAILSAFE = False
logger.info("pyautogui successfully imported, GUI automation available")
except Exception as e:
logger.error(f"pyautogui import failed: {str(e)}. GUI operations will not work.")
pyautogui = None
# Try to import Windows-specific modules
try:
import win32gui
import win32con
import win32api
logger.info("Windows API modules successfully imported")
WINDOWS_API_AVAILABLE = True
except Exception as e:
logger.error(f"Windows API modules import failed: {str(e)}. Some Windows-specific features will be unavailable.")
WINDOWS_API_AVAILABLE = False
from .base import BaseAccessibilityHandler, BaseAutomationHandler
class WindowsAccessibilityHandler(BaseAccessibilityHandler):
"""Windows implementation of accessibility handler."""
async def get_accessibility_tree(self) -> Dict[str, Any]:
"""Get the accessibility tree of the current window.
Returns:
Dict[str, Any]: A dictionary containing the success status and either
the accessibility tree or an error message.
Structure: {"success": bool, "tree": dict} or
{"success": bool, "error": str}
"""
if not WINDOWS_API_AVAILABLE:
return {"success": False, "error": "Windows API not available"}
try:
# Get the foreground window
hwnd = win32gui.GetForegroundWindow()
if not hwnd:
return {"success": False, "error": "No foreground window found"}
# Get window information
window_text = win32gui.GetWindowText(hwnd)
rect = win32gui.GetWindowRect(hwnd)
tree = {
"role": "Window",
"title": window_text,
"position": {"x": rect[0], "y": rect[1]},
"size": {"width": rect[2] - rect[0], "height": rect[3] - rect[1]},
"children": []
}
# Enumerate child windows
def enum_child_proc(hwnd_child, children_list):
"""Callback function to enumerate child windows and collect their information.
Args:
hwnd_child: Handle to the child window being enumerated.
children_list: List to append child window information to.
Returns:
bool: True to continue enumeration, False to stop.
"""
try:
child_text = win32gui.GetWindowText(hwnd_child)
child_rect = win32gui.GetWindowRect(hwnd_child)
child_class = win32gui.GetClassName(hwnd_child)
child_info = {
"role": child_class,
"title": child_text,
"position": {"x": child_rect[0], "y": child_rect[1]},
"size": {"width": child_rect[2] - child_rect[0], "height": child_rect[3] - child_rect[1]},
"children": []
}
children_list.append(child_info)
except Exception as e:
logger.debug(f"Error getting child window info: {e}")
return True
win32gui.EnumChildWindows(hwnd, enum_child_proc, tree["children"])
return {"success": True, "tree": tree}
except Exception as e:
logger.error(f"Error getting accessibility tree: {e}")
return {"success": False, "error": str(e)}
async def find_element(self, role: Optional[str] = None,
title: Optional[str] = None,
value: Optional[str] = None) -> Dict[str, Any]:
"""Find an element in the accessibility tree by criteria.
Args:
role (Optional[str]): The role or class name of the element to find.
title (Optional[str]): The title or text of the element to find.
value (Optional[str]): The value of the element (not used in Windows implementation).
Returns:
Dict[str, Any]: A dictionary containing the success status and either
the found element or an error message.
Structure: {"success": bool, "element": dict} or
{"success": bool, "error": str}
"""
if not WINDOWS_API_AVAILABLE:
return {"success": False, "error": "Windows API not available"}
try:
# Find window by title if specified
if title:
hwnd = win32gui.FindWindow(None, title)
if hwnd:
rect = win32gui.GetWindowRect(hwnd)
return {
"success": True,
"element": {
"role": "Window",
"title": title,
"position": {"x": rect[0], "y": rect[1]},
"size": {"width": rect[2] - rect[0], "height": rect[3] - rect[1]}
}
}
# Find window by class name if role is specified
if role:
hwnd = win32gui.FindWindow(role, None)
if hwnd:
window_text = win32gui.GetWindowText(hwnd)
rect = win32gui.GetWindowRect(hwnd)
return {
"success": True,
"element": {
"role": role,
"title": window_text,
"position": {"x": rect[0], "y": rect[1]},
"size": {"width": rect[2] - rect[0], "height": rect[3] - rect[1]}
}
}
return {"success": False, "error": "Element not found"}
except Exception as e:
logger.error(f"Error finding element: {e}")
return {"success": False, "error": str(e)}
class WindowsAutomationHandler(BaseAutomationHandler):
"""Windows implementation of automation handler using pyautogui and Windows APIs."""
mouse = MouseController()
keyboard = KeyboardController()
# Mouse Actions
async def mouse_down(self, x: Optional[int] = None, y: Optional[int] = None, button: str = "left") -> Dict[str, Any]:
"""Press and hold a mouse button at the specified coordinates.
Args:
x (Optional[int]): The x-coordinate to move to before pressing. If None, uses current position.
y (Optional[int]): The y-coordinate to move to before pressing. If None, uses current position.
button (str): The mouse button to press ("left", "right", or "middle").
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
if x is not None and y is not None:
pyautogui.moveTo(x, y)
pyautogui.mouseDown(button=button)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def mouse_up(self, x: Optional[int] = None, y: Optional[int] = None, button: str = "left") -> Dict[str, Any]:
"""Release a mouse button at the specified coordinates.
Args:
x (Optional[int]): The x-coordinate to move to before releasing. If None, uses current position.
y (Optional[int]): The y-coordinate to move to before releasing. If None, uses current position.
button (str): The mouse button to release ("left", "right", or "middle").
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
if x is not None and y is not None:
pyautogui.moveTo(x, y)
pyautogui.mouseUp(button=button)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def move_cursor(self, x: int, y: int) -> Dict[str, Any]:
"""Move the mouse cursor to the specified coordinates.
Args:
x (int): The x-coordinate to move to.
y (int): The y-coordinate to move to.
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
pyautogui.moveTo(x, y)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def left_click(self, x: Optional[int] = None, y: Optional[int] = None) -> Dict[str, Any]:
"""Perform a left mouse click at the specified coordinates.
Args:
x (Optional[int]): The x-coordinate to click at. If None, clicks at current position.
y (Optional[int]): The y-coordinate to click at. If None, clicks at current position.
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
if x is not None and y is not None:
pyautogui.moveTo(x, y)
pyautogui.click()
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def right_click(self, x: Optional[int] = None, y: Optional[int] = None) -> Dict[str, Any]:
"""Perform a right mouse click at the specified coordinates.
Args:
x (Optional[int]): The x-coordinate to click at. If None, clicks at current position.
y (Optional[int]): The y-coordinate to click at. If None, clicks at current position.
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
if x is not None and y is not None:
pyautogui.moveTo(x, y)
pyautogui.rightClick()
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def double_click(self, x: Optional[int] = None, y: Optional[int] = None) -> Dict[str, Any]:
"""Perform a double left mouse click at the specified coordinates.
Args:
x (Optional[int]): The x-coordinate to double-click at. If None, clicks at current position.
y (Optional[int]): The y-coordinate to double-click at. If None, clicks at current position.
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
if x is not None and y is not None:
pyautogui.moveTo(x, y)
pyautogui.doubleClick(interval=0.1)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def drag_to(self, x: int, y: int, button: str = "left", duration: float = 0.5) -> Dict[str, Any]:
"""Drag from the current position to the specified coordinates.
Args:
x (int): The x-coordinate to drag to.
y (int): The y-coordinate to drag to.
button (str): The mouse button to use for dragging ("left", "right", or "middle").
duration (float): The time in seconds to take for the drag operation.
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
pyautogui.dragTo(x, y, duration=duration, button=button)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def drag(self, path: List[Tuple[int, int]], button: str = "left", duration: float = 0.5) -> Dict[str, Any]:
"""Drag the mouse through a series of coordinates.
Args:
path (List[Tuple[int, int]]): A list of (x, y) coordinate tuples to drag through.
button (str): The mouse button to use for dragging ("left", "right", or "middle").
duration (float): The total time in seconds for the entire drag operation.
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
if not path:
return {"success": False, "error": "Path is empty"}
# Move to first position
pyautogui.moveTo(*path[0])
# Drag through all positions
for x, y in path[1:]:
pyautogui.dragTo(x, y, duration=duration/len(path), button=button)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
# Keyboard Actions
async def key_down(self, key: str) -> Dict[str, Any]:
"""Press and hold a keyboard key.
Args:
key (str): The key to press down (e.g., 'ctrl', 'shift', 'a').
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
pyautogui.keyDown(key)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def key_up(self, key: str) -> Dict[str, Any]:
"""Release a keyboard key.
Args:
key (str): The key to release (e.g., 'ctrl', 'shift', 'a').
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
pyautogui.keyUp(key)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def type_text(self, text: str) -> Dict[str, Any]:
"""Type the specified text.
Args:
text (str): The text to type.
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
try:
# use pynput for Unicode support
self.keyboard.type(text)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def press_key(self, key: str) -> Dict[str, Any]:
"""Press and release a keyboard key.
Args:
key (str): The key to press (e.g., 'enter', 'space', 'tab').
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
pyautogui.press(key)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def hotkey(self, keys: List[str]) -> Dict[str, Any]:
"""Press a combination of keys simultaneously.
Args:
keys (List[str]): The keys to press together (e.g., ['ctrl', 'c'], ['alt', 'tab']).
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
pyautogui.hotkey(*keys)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
# Scrolling Actions
async def scroll(self, x: int, y: int) -> Dict[str, Any]:
"""Scroll vertically at the current cursor position.
Args:
x (int): Horizontal scroll amount (not used in pyautogui implementation).
y (int): Vertical scroll amount. Positive values scroll up, negative values scroll down.
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
self.mouse.scroll(x, y)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def scroll_down(self, clicks: int = 1) -> Dict[str, Any]:
"""Scroll down by the specified number of clicks.
Args:
clicks (int): The number of scroll clicks to perform downward.
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
pyautogui.scroll(-clicks)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
async def scroll_up(self, clicks: int = 1) -> Dict[str, Any]:
"""Scroll up by the specified number of clicks.
Args:
clicks (int): The number of scroll clicks to perform upward.
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
pyautogui.scroll(clicks)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
# Screen Actions
async def screenshot(self) -> Dict[str, Any]:
"""Capture a screenshot of the entire screen.
Returns:
Dict[str, Any]: A dictionary containing the success status and either
base64-encoded image data or an error message.
Structure: {"success": bool, "image_data": str} or
{"success": bool, "error": str}
"""
if not pyautogui:
return {"success": False, "error": "pyautogui not available"}
try:
from PIL import Image
screenshot = pyautogui.screenshot()
if not isinstance(screenshot, Image.Image):
return {"success": False, "error": "Failed to capture screenshot"}
buffered = BytesIO()
screenshot.save(buffered, format="PNG", optimize=True)
buffered.seek(0)
image_data = base64.b64encode(buffered.getvalue()).decode()
return {"success": True, "image_data": image_data}
except Exception as e:
return {"success": False, "error": f"Screenshot error: {str(e)}"}
async def get_screen_size(self) -> Dict[str, Any]:
"""Get the size of the screen in pixels.
Returns:
Dict[str, Any]: A dictionary containing the success status and either
screen size information or an error message.
Structure: {"success": bool, "size": {"width": int, "height": int}} or
{"success": bool, "error": str}
"""
try:
if pyautogui:
size = pyautogui.size()
return {"success": True, "size": {"width": size.width, "height": size.height}}
elif WINDOWS_API_AVAILABLE:
# Fallback to Windows API
width = win32api.GetSystemMetrics(win32con.SM_CXSCREEN)
height = win32api.GetSystemMetrics(win32con.SM_CYSCREEN)
return {"success": True, "size": {"width": width, "height": height}}
else:
return {"success": False, "error": "No screen size detection method available"}
except Exception as e:
return {"success": False, "error": str(e)}
async def get_cursor_position(self) -> Dict[str, Any]:
"""Get the current position of the mouse cursor.
Returns:
Dict[str, Any]: A dictionary containing the success status and either
cursor position or an error message.
Structure: {"success": bool, "position": {"x": int, "y": int}} or
{"success": bool, "error": str}
"""
try:
if pyautogui:
pos = pyautogui.position()
return {"success": True, "position": {"x": pos.x, "y": pos.y}}
elif WINDOWS_API_AVAILABLE:
# Fallback to Windows API
pos = win32gui.GetCursorPos()
return {"success": True, "position": {"x": pos[0], "y": pos[1]}}
else:
return {"success": False, "error": "No cursor position detection method available"}
except Exception as e:
return {"success": False, "error": str(e)}
# Clipboard Actions
async def copy_to_clipboard(self) -> Dict[str, Any]:
"""Get the current content of the clipboard.
Returns:
Dict[str, Any]: A dictionary containing the success status and either
clipboard content or an error message.
Structure: {"success": bool, "content": str} or
{"success": bool, "error": str}
"""
try:
import pyperclip
content = pyperclip.paste()
return {"success": True, "content": content}
except Exception as e:
return {"success": False, "error": str(e)}
async def set_clipboard(self, text: str) -> Dict[str, Any]:
"""Set the clipboard content to the specified text.
Args:
text (str): The text to copy to the clipboard.
Returns:
Dict[str, Any]: A dictionary with success status and optional error message.
"""
try:
import pyperclip
pyperclip.copy(text)
return {"success": True}
except Exception as e:
return {"success": False, "error": str(e)}
# Command Execution
async def run_command(self, command: str) -> Dict[str, Any]:
"""Execute a shell command asynchronously.
Args:
command (str): The shell command to execute.
Returns:
Dict[str, Any]: A dictionary containing the success status and either
command output or an error message.
Structure: {"success": bool, "stdout": str, "stderr": str, "return_code": int} or
{"success": bool, "error": str}
"""
try:
# Create subprocess
process = await asyncio.create_subprocess_shell(
command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
# Wait for the subprocess to finish
stdout, stderr = await process.communicate()
# Return decoded output
return {
"success": True,
"stdout": stdout.decode() if stdout else "",
"stderr": stderr.decode() if stderr else "",
"return_code": process.returncode
}
except Exception as e:
return {"success": False, "error": str(e)}
```