rho-coding-agent 2.10.0

A fast Rust agent harness with a small footprint and opinionated defaults
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

#!/usr/bin/env python3
"""Stdio-only fake Cua Driver. It never observes or controls a desktop."""
import json
import sys
import os
import socket

assert sys.argv[1:] == ["mcp"]

if os.path.exists("blocked-connect.sock"):
    signal = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    signal.connect("blocked-connect.sock")
    signal.recv(1)
    sys.exit(0)


def send(message):
    print(json.dumps(message), flush=True)


for line in sys.stdin:
    message = json.loads(line)
    if "id" not in message:
        continue
    method = message.get("method")
    if method == "initialize":
        result = {
            "protocolVersion": message["params"]["protocolVersion"],
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "fake-cua", "version": "1"},
            "instructions": "Observe before acting.",
        }
    elif method == "tools/list":
        result = {"tools": [
            {"name": name, "description": name, "inputSchema": {"type": "object"}}
            for name in ["click", "get_window_state", "launch_app", "set_config", "start_recording", "future_tool"]
        ]}
    elif method == "tools/call":
        params = message["params"]
        assert params["name"] in ["click", "get_window_state", "launch_app"]
        arguments = params.get("arguments") or {}
        assert "session" not in arguments
        if arguments.get("fail"):
            result = {"content": [
                {"type": "text", "text": "no action taken"},
            ], "isError": True}
        elif params["name"] == "click" or arguments.get("hang"):
            token = params.get("_meta", {}).get("progressToken")
            if token is not None:
                send({"jsonrpc": "2.0", "method": "notifications/progress", "params": {
                    "progressToken": token, "progress": 1, "message": "action received"
                }})
            continue
        else:
            result = {"content": [
                {"type": "text", "text": "observed"},
                {"type": "image", "mimeType": "image/png", "data": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+jRZkAAAAASUVORK5CYII="},
            ], "isError": False}
    else:
        result = {}
    send({"jsonrpc": "2.0", "id": message["id"], "result": result})