Upload agentframe_hands.py with huggingface_hub

agentframe_hands.py

@@ -0,0 +1,224 @@
+"""
+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 授权后完整可用)")