CUA Usage Examples

agentloop_local_host.py

import asyncio
import logging
import os
import signal
import traceback

from cua import Sandbox, Image, Localhost
from cua_agent import ComputerAgent
from dotenv import load_dotenv

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def handle_sigint(sig, frame):
    print("\n\nExecution interrupted by user. Exiting gracefully...")
    exit(0)

async def fill_application():
    try:
        async with Localhost.connect() as host:

            agent = ComputerAgent(
                model="openai/gpt-5.4",
                tools=[host],
                only_n_most_recent_images=3,
                verbosity=logging.INFO,
                trajectory_dir="trajectories",
                use_prompt_caching=True,
                max_trajectory_budget=5.0,
            )

            tasks = [
                "Visit https://www.overleaf.com/latex/templates/jakes-resume/syzfjbzwjncs.pdf and download the pdf.",
                "Visit https://form.jotform.com/252881246782264 and fill the form from the information in the pdf."
            ]

            history = []

            for i, task in enumerate(tasks, 1):
                print(f"\n[Task {i}/{len(tasks)}] {task}")

                # Add user message to history
                history.append({"role": "user", "content": task})

                # Run agent with conversation history
                async for result in agent.run(history, stream=False):
                    history += result.get("output", [])

                    # Print output for debugging
                    for item in result.get("output", []):
                        if item.get("type") == "message":
                            content = item.get("content", [])
                            for content_part in content:
                                if content_part.get("text"):
                                    logger.info(f"Agent: {content_part.get('text')}")
                        elif item.get("type") == "computer_call":
                            action = item.get("action", {})
                            action_type = action.get("type", "")
                            logger.debug(f"Computer Action: {action_type}")

                print(f"✅ Task {i}/{len(tasks)} completed")

            print("\n🎉 All tasks completed successfully!")

    except Exception as e:
        logger.error(f"Error in fill_application: {e}")
        traceback.print_exc()
        raise

def main():
    try:
        load_dotenv()

        if "OPENAI_API_KEY" not in os.environ:
            raise RuntimeError(
                "Please set the OPENAI_API_KEY environment variable.\n"
                "You can add it to a .env file in the project root."
            )

        signal.signal(signal.SIGINT, handle_sigint)

        asyncio.run(fill_application())

    except Exception as e:
        logger.error(f"Error running automation: {e}")
        traceback.print_exc()

if __name__ == "__main__":
    main()

connect_local_host.py

# ! pip install cua

import asyncio
from cua import Localhost

async def main():
    async with Localhost.connect() as host:
        await host.mouse.click(100, 200)
        await host.keyboard.type("Hello, World!")

        screenshot = await host.screenshot()
        print("Screenshot captured successfully")
        
        # screenshot is a bytes object, you can save it to a file if needed
        with open("localhost_screenshot.png", "wb") as f:
            f.write(screenshot)
            

if __name__ == "__main__":
    asyncio.run(main())