LangChain agents

pip install mandala-computer langchain langchain-anthropic

import base64
import sys

import mandala_computer
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.tools import tool
from mandala_computer import Client


def attempt(action):
    """Run one computer call and describe the outcome for the model."""
    try:
        result = action()
    except (mandala_computer.MandalaError, ValueError) as e:
        return f"Failed: {type(e).__name__}: {e}"
    return "OK" if result is None else str(result)


def mandala_tools(computer):
    width, height = computer.screen

    @tool
    def screenshot() -> list[dict]:
        """Take a screenshot of the desktop. Take one before acting and after each
        few actions. Coordinates in the other tools are pixels in this image."""
        png = computer.screenshot(fresh=True)
        return [{"type": "image", "base64": base64.b64encode(png).decode(), "mime_type": "image/png"}]

    @tool
    def click(x: int, y: int, button: str = "left", count: int = 1, hold: str = "") -> str:
        """Click at pixel (x, y). button is left, right or middle. count is 1, 2
        (double click) or 3 (triple click). hold is keys to hold, like "ctrl" or "shift"."""
        mods = [k for k in hold.split("+") if k]
        if button == "right":
            return attempt(lambda: computer.right_click(x, y, *mods))
        if button == "middle":
            return attempt(lambda: computer.middle_click(x, y, *mods))
        method = {1: computer.click, 2: computer.double_click, 3: computer.triple_click}.get(count)
        if method is None:
            return "Failed: count must be 1, 2 or 3."
        return attempt(lambda: method(x, y, *mods))

    @tool
    def type_text(text: str) -> str:
        """Type text at the keyboard focus. Click the field first."""
        def type_all():
            for i in range(0, len(text), 400):  # at most 400 characters per call
                computer.type(text[i : i + 400])
        return attempt(type_all)

    @tool
    def press_keys(keys: str) -> str:
        """Press a key or a chord: "Return", "Escape", "ctrl+l", "ctrl+shift+t", "Page_Down"."""
        return attempt(lambda: computer.key(*[k for k in keys.split("+") if k]))

    @tool
    def scroll(x: int, y: int, direction: str = "down", amount: int = 3) -> str:
        """Scroll the mouse wheel at pixel (x, y). direction is up, down, left or
        right. amount is the number of wheel notches."""
        return attempt(lambda: computer.scroll(x, y, direction=direction, amount=amount))

    @tool
    def drag(from_x: int, from_y: int, to_x: int, to_y: int) -> str:
        """Press at (from_x, from_y), move to (to_x, to_y) and release."""
        return attempt(lambda: computer.drag(to_x, to_y, from_x=from_x, from_y=from_y))

    @tool
    def open_url(url: str) -> str:
        """Open a URL in the desktop's web browser."""
        def go():
            res = computer.open(url)
            return None if res.ok else f"Failed: exit code {res.exit_code}: {res.stderr_text[-500:]}"
        return attempt(go)

    @tool
    def run_command(command: str) -> str:
        """Run a shell command as root and return its output. There is no display:
        use open_url or the desktop to start programs with a window."""
        def execute():
            res = computer.exec(command, timeout=60)
            return f"exit code {res.exit_code}\n{res.stdout_text[-4000:]}\n{res.stderr_text[-2000:]}"
        return attempt(execute)

    return [screenshot, click, type_text, press_keys, scroll, drag, open_url, run_command]


@wrap_model_call
def one_action_at_a_time(request, handler):
    """One screen, one pointer: ask for actions one at a time, never in parallel."""
    return handler(request.override(model_settings={**request.model_settings, "parallel_tool_calls": False}))


def run(computer, task):
    agent = create_agent(
        model="anthropic:claude-opus-5",  # reads ANTHROPIC_API_KEY
        tools=mandala_tools(computer),
        middleware=[one_action_at_a_time],
        system_prompt=(
            "You operate a Linux desktop (Xfce) through tools. "
            f"The screen is {computer.screen[0]}x{computer.screen[1]} pixels. "
            "Take a screenshot before acting, and check the result after a few actions."
        ),
    )
    result = agent.invoke({"messages": [{"role": "user", "content": task}]}, {"recursion_limit": 80})
    return result["messages"][-1].text


if __name__ == "__main__":
    task = " ".join(sys.argv[1:]) or "Open a terminal and show today's date."
    client = Client()  # reads MANDALA_API_KEY
    with client.computers.ephemeral(template="base", name="langchain-agent") as computer:
        computer.wait_for_guest()
        print(run(computer, task))