agentframe_hands.py

"""
AgentFrame 任务3: OpenClaw 工具接入 (browser + computer-use)
===========================================================
扩展 Hands 层: 在 exec/read_file 基础上加入 browser 操控

browser CLI 命令 (OpenClaw):
  openclaw browser --browser-profile openclaw <cmd>
  - open <url>: 打开网页
  - snapshot: 页面快照 (AI 可读)
  - click <ref>: 点击元素
  - fill <form>: 填表
  - navigate <url>: 导航
  - screenshot: 截图 (MEDIA:<path>)
  - evaluate <js>: 执行 JS
  - pdf: 存 PDF
"""
import subprocess
import time
import re
from typing import Dict, List, Optional

BROWSER_CMD = "openclaw browser --browser-profile openclaw"


class BrowserHands:
    """OpenClaw 浏览器操控 (Agent 的"眼睛和手")"""

    def __init__(self, timeout=30):
        self.timeout = timeout

    def _run(self, args: str) -> str:
        """执行 openclaw browser 命令"""
        cmd = f"{BROWSER_CMD} {args}"
        try:
            r = subprocess.run(cmd, shell=True, capture_output=True,
                               text=True, timeout=self.timeout)
            # 过滤 gateway 噪音
            out = r.stdout
            for line in out.split('\n'):
                if 'Config warning' in line or 'gateway connect' in line:
                    continue
            return out.strip()
        except subprocess.TimeoutExpired:
            return "⚠️ 浏览器命令超时"
        except Exception as e:
            return f"❌ {e}"

    def open(self, url: str) -> str:
        """打开网页"""
        return self._run(f'open "{url}"')

    def navigate(self, url: str) -> str:
        """导航当前标签页"""
        return self._run(f'navigate "{url}"')

    def snapshot(self) -> str:
        """页面快照 (AI 可读的 DOM 描述)"""
        return self._run("snapshot")

    def click(self, ref: str) -> str:
        """点击元素"""
        return self._run(f'click "{ref}"')

    def fill(self, form_json: str) -> str:
        """填表 (JSON 格式)"""
        return self._run(f'fill \'{form_json}\'')

    def press(self, key: str) -> str:
        """按键"""
        return self._run(f'press "{key}"')

    def screenshot(self) -> str:
        """截图 (返回 MEDIA 路径)"""
        return self._run("screenshot")

    def evaluate(self, js: str) -> str:
        """执行 JS"""
        return self._run(f'evaluate "{js}"')

    def pdf(self, path: str) -> str:
        """存 PDF"""
        return self._run(f'pdf "{path}"')

    def focus(self, target: str) -> str:
        """聚焦标签页"""
        return self._run(f'focus "{target}"')


class ComputerUseHands:
    """电脑操控 (Codex Computer Use / cua-driver)"""
    # 需要安装 cua-driver MCP 后启用
    def __init__(self):
        self.available = False

    def check(self) -> bool:
        """检查是否可用"""
        try:
            r = subprocess.run("cua-driver --version", shell=True,
                               capture_output=True, text=True, timeout=5)
            self.available = r.returncode == 0
        except Exception:
            self.available = False
        return self.available


# ============================================================
# 集成到 AgentFrame Hands
# ============================================================
class AgentFrameHands:
    """完整 Hands: exec + 文件 + 浏览器 + (可选)电脑操控"""

    def __init__(self, workspace="/root/.openclaw/workspace"):
        self.workspace = workspace
        self.browser = BrowserHands()
        self.computer = ComputerUseHands()

    def tool_schemas(self) -> List[Dict]:
        """喂给 DeepSeek 的工具定义"""
        return [
            {
                "type": "function",
                "function": {
                    "name": "exec",
                    "description": "执行 shell 命令",
                    "parameters": {"type": "object",
                                   "properties": {"command": {"type": "string"}},
                                   "required": ["command"]},
                },
            },
            {
                "type": "function",
                "function": {
                    "name": "browser_open",
                    "description": "打开网页 (如 https://example.com)",
                    "parameters": {"type": "object",
                                   "properties": {"url": {"type": "string"}},
                                   "required": ["url"]},
                },
            },
            {
                "type": "function",
                "function": {
                    "name": "browser_snapshot",
                    "description": "获取当前网页内容快照 (AI 可读)",
                    "parameters": {"type": "object", "properties": {}},
                },
            },
            {
                "type": "function",
                "function": {
                    "name": "browser_click",
                    "description": "点击网页元素 (ref 来自 snapshot)",
                    "parameters": {"type": "object",
                                   "properties": {"ref": {"type": "string"}},
                                   "required": ["ref"]},
                },
            },
            {
                "type": "function",
                "function": {
                    "name": "read_file",
                    "description": "读取文件内容",
                    "parameters": {"type": "object",
                                   "properties": {"path": {"type": "string"}},
                                   "required": ["path"]},
                },
            },
        ]

    def execute(self, name: str, args: Dict) -> str:
        """执行工具"""
        if name == "exec":
            r = subprocess.run(args.get("command", ""), shell=True,
                               capture_output=True, text=True, timeout=15,
                               cwd=self.workspace)
            return (r.stdout or r.stderr)[:600]

        elif name == "browser_open":
            return self.browser.open(args.get("url", ""))

        elif name == "browser_snapshot":
            return self.browser.snapshot()

        elif name == "browser_click":
            return self.browser.click(args.get("ref", ""))

        elif name == "read_file":
            try:
                with open(args.get("path", "")) as f:
                    return f.read()[:600]
            except Exception as e:
                return f"错误: {e}"

        return f"未知工具: {name}"


# ============================================================
# 演示
# ============================================================
if __name__ == "__main__":
    print("=" * 60)
    print("AgentFrame 任务3: OpenClaw 工具接入 演示")
    print("=" * 60)

    hands = AgentFrameHands()
    print("\n📋 工具清单:")
    for t in hands.tool_schemas():
        print(f"  🛠 {t['function']['name']}: {t['function']['description']}")

    # 测试浏览器
    print("\n🌐 测试浏览器 (无头模式验证):")
    try:
        r = hands.browser.open("https://example.com")
        print(f"  open: {'✅' if '成功' in r or r else '⚠️ 需 gateway 授权: ' + r[:50]}")
    except Exception as e:
        print(f"  ⚠️ {e}")

    # 测试 exec (无需浏览器授权)
    print("\n💻 测试 exec (本地):")
    r = hands.execute("exec", {"command": "echo 'AgentFrame Hands 就绪!'"})
    print(f"  → {r.strip()}")

    print("\n✅ 任务3完成: OpenClaw 工具集已接入 Hands 层")
    print("   (browser 需 gateway scope 授权后完整可用)")