Claude Computer Use 桌面自动化完全教程
掌握 Claude Computer Use 技术，实现 AI 驱动的桌面自动化操作。
目录
- Computer Use 概述
- 技术原理：视觉理解 + 操作执行
- Claude Computer Use API 接入
- 屏幕截图与视觉分析
- 鼠标键盘操作自动化
- 桌面应用场景
- 网页自动化操作
- 与 Selenium/Playwright 对比
- 安全防护与操作审计
- 企业级 RPA 集成方案
- Computer Use 最佳实践与限制
- 总结
1. Computer Use 概述
1.1 什么是 Computer Use
Claude Computer Use 是 Anthropic 推出的一项革命性功能，让 AI 模型能够像人类一样直接操作计算机——观看屏幕、移动鼠标、点击按钮、输入文字、浏览网页。它不是通过 API 接口与软件交互，而是通过视觉感知 + 操作执行的方式，直接操控桌面环境。
这意味着 Claude 可以：
- 看到屏幕上的所有内容（通过截图）
- 理解界面元素和它们的含义
- 操作鼠标和键盘来完成任务
- 验证操作结果（再次截图确认）
1.2 与传统自动化的本质区别
| 维度 | 传统 RPA | Claude Computer Use |
|---|---|---|
| 工作方式 | 基于固定脚本/规则 | 基于视觉理解 + AI 决策 |
| 适应性 | 界面变化需重写脚本 | 自动适应界面变化 |
| 开发成本 | 需要专业 RPA 开发 | 自然语言描述任务即可 |
| 维护成本 | 高（脚本需持续维护） | 低（AI 自动适应） |
| 适用场景 | 固定流程自动化 | 灵活、非结构化任务 |
| 错误处理 | 预定义异常处理 | AI 理解上下文自主处理 |
1.3 核心优势
- 零脚本开发：用自然语言描述任务，无需编写自动化脚本
- 界面自适应：UI 变化不影响任务执行，AI 会"看懂"新界面
- 跨应用操作：可以在多个应用程序之间无缝切换操作
- 智能错误恢复：遇到意外情况能自主判断和处理
2. 技术原理：视觉理解 + 操作执行
2.1 工作流程
┌──────────────────────────────────────────────────┐
│ Claude Computer Use │
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 截图模块 │───→│ 视觉理解 │───→│ 决策引擎 │ │
│ └─────────┘ └─────────┘ └────┬────┘ │
│ ↑ │ │
│ │ ▼ │
│ ┌────┴────┐ ┌─────────┐ │
│ │ 结果验证 │←──────────────────│ 操作执行 │ │
│ └─────────┘ └─────────┘ │
│ │ │
│ ┌─────┴─────┐ │
│ │ 鼠标/键盘 │ │
│ └───────────┘ │
└──────────────────────────────────────────────────┘
2.2 视觉理解能力
Claude 的视觉理解基于多模态大语言模型，能够：
- 元素识别：识别按钮、输入框、菜单、图标等 UI 元素
- 文字识别（OCR）：读取屏幕上的所有文字内容
- 布局理解：理解页面结构和元素之间的关系
- 状态判断：判断当前页面状态（加载中、已登录、错误等）
- 坐标定位：精确计算需要点击/操作的屏幕坐标
2.3 操作执行机制
Claude Computer Use 通过以下工具（tools）与计算机交互：
# Claude 可调用的三种核心工具
# 1. computer - 计算机操作（鼠标、键盘、截图）
computer_tool = {
"type": "computer_20250124",
"name": "computer",
"display_width_px": 1920,
"display_height_px": 1080,
}
# 2. text_editor - 文本文件编辑
text_editor_tool = {
"type": "text_editor_20250124",
"name": "str_replace_editor",
}
# 3. bash - 命令行执行
bash_tool = {
"type": "bash_20250124",
"name": "bash",
}
3. Claude Computer Use API 接入
3.1 环境准备
# 安装依赖
pip install anthropic pillow
3.2 基础调用示例
import anthropic
import base64
from PIL import Image
import io
client = anthropic.Anthropic()
def take_screenshot() -> str:
"""获取屏幕截图并转为 base64"""
import subprocess
# Linux 使用 scrot 或 gnome-screenshot
subprocess.run(["scrot", "/tmp/screenshot.png"], check=True)
with open("/tmp/screenshot.png", "rb") as f:
return base64.standard_b64encode(f.read()).decode("utf-8")
def computer_use_demo():
"""Computer Use 基础示例"""
screenshot_b64 = take_screenshot()
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
tools=[
{
"type": "computer_20250124",
"name": "computer",
"display_width_px": 1920,
"display_height_px": 1080,
"display_number": 0,
},
{
"type": "text_editor_20250124",
"name": "str_replace_editor",
},
{
"type": "bash_20250124",
"name": "bash",
}
],
messages=[
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": screenshot_b64,
},
},
{
"type": "text",
"text": "请打开系统自带的计算器应用，计算 123 * 456 的结果。"
}
]
}
]
)
return response
# 执行
response = computer_use_demo()
print(response)
3.3 完整的 Agent 循环
Computer Use 的核心是一个交互循环：截图 → 分析 → 操作 → 再截图 → 验证 → 继续...
import anthropic
import base64
import subprocess
import time
class ComputerUseAgent:
"""Computer Use Agent - 完整的交互循环"""
def __init__(self, display_size: tuple = (1920, 1080)):
self.client = anthropic.Anthropic()
self.display_size = display_size
self.conversation_history = []
self.max_iterations = 50
self.tools = [
{
"type": "computer_20250124",
"name": "computer",
"display_width_px": display_size[0],
"display_height_px": display_size[1],
"display_number": 0,
},
{
"type": "text_editor_20250124",
"name": "str_replace_editor",
},
{
"type": "bash_20250124",
"name": "bash",
}
]
def take_screenshot(self) -> str:
"""获取屏幕截图"""
subprocess.run(["scrot", "/tmp/screenshot.png"], check=True)
with open("/tmp/screenshot.png", "rb") as f:
return base64.standard_b64encode(f.read()).decode("utf-8")
def execute_tool(self, tool_use) -> dict:
"""执行工具调用"""
tool_name = tool_use.name
tool_input = tool_use.input
if tool_name == "computer":
return self._execute_computer_action(tool_input)
elif tool_name == "bash":
return self._execute_bash(tool_input)
elif tool_name == "str_replace_editor":
return self._execute_editor(tool_input)
else:
return {"error": f"Unknown tool: {tool_name}"}
def _execute_computer_action(self, action_input: dict) -> dict:
"""执行计算机操作"""
action = action_input.get("action")
if action == "screenshot":
screenshot = self.take_screenshot()
return {
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": screenshot,
}
}
elif action == "mouse_move":
x, y = action_input["coordinate"]
subprocess.run(["xdotool", "mousemove", str(x), str(y)], check=True)
return {"result": f"鼠标移动到 ({x}, {y})"}
elif action == "left_click":
x, y = action_input["coordinate"]
subprocess.run(["xdotool", "mousemove", str(x), str(y), "click", "1"], check=True)
return {"result": f"左键点击 ({x}, {y})"}
elif action == "type":
text = action_input["text"]
subprocess.run(["xdotool", "type", "--clearmodifiers", text], check=True)
return {"result": f"输入文字: {text}"}
elif action == "key":
key = action_input["key"]
subprocess.run(["xdotool", "key", key], check=True)
return {"result": f"按键: {key}"}
elif action == "left_click_drag":
start = action_input["start_coordinate"]
end = action_input["end_coordinate"]
subprocess.run([
"xdotool", "mousemove", str(start[0]), str(start[1]),
"mousedown", "1", "mousemove", str(end[0]), str(end[1]),
"mouseup", "1"
], check=True)
return {"result": f"拖拽 ({start[0]},{start[1]}) → ({end[0]},{end[1]})"}
return {"error": f"未支持的操作: {action}"}
def _execute_bash(self, bash_input: dict) -> dict:
"""执行 Bash 命令"""
command = bash_input.get("command", "")
try:
result = subprocess.run(
command, shell=True, capture_output=True, text=True, timeout=30
)
return {
"stdout": result.stdout,
"stderr": result.stderr,
"returncode": result.returncode
}
except subprocess.TimeoutExpired:
return {"error": "命令执行超时"}
def _execute_editor(self, editor_input: dict) -> dict:
"""执行文本编辑操作"""
# 简化实现，实际需要完整支持 view/create/str_replace 等
command = editor_input.get("command")
path = editor_input.get("path")
if command == "view":
try:
with open(path, "r") as f:
return {"content": f.read()}
except FileNotFoundError:
return {"error": f"文件不存在: {path}"}
return {"error": f"未支持的编辑命令: {command}"}
def run(self, task: str) -> str:
"""执行任务的完整 Agent 循环"""
print(f"开始执行任务: {task}")
# 初始截图
screenshot = self.take_screenshot()
self.conversation_history = [
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": screenshot,
},
},
{"type": "text", "text": task}
]
}
]
for iteration in range(self.max_iterations):
print(f"\n--- 迭代 {iteration + 1} ---")
# 调用 Claude
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
tools=self.tools,
messages=self.conversation_history
)
# 检查是否完成
if response.stop_reason == "end_turn":
# 提取最终文本回复
for block in response.content:
if block.type == "text":
print(f"任务完成: {block.text}")
return block.text
return "任务完成"
# 处理工具调用
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f"调用工具: {block.name} → {block.input}")
result = self.execute_tool(block)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": [result] if isinstance(result, dict) else [result],
})
# 更新对话历史
self.conversation_history.append({"role": "assistant", "content": response.content})
self.conversation_history.append({"role": "user", "content": tool_results})
# 短暂等待，让操作生效
time.sleep(0.5)
return "达到最大迭代次数，任务未完成"
# 使用示例
agent = ComputerUseAgent()
result = agent.run("打开 Firefox 浏览器，访问 google.com，搜索 'Claude Computer Use'")
print(result)
4. 屏幕截图与视觉分析
4.1 截图方案对比
| 平台 | 工具 | 命令 | 特点 |
|---|---|---|---|
| Linux | scrot | scrot /tmp/screenshot.png |
轻量、快速 |
| Linux | gnome-screenshot | gnome-screenshot -f /tmp/screenshot.png |
支持窗口截图 |
| macOS | screencapture | screencapture /tmp/screenshot.png |
系统自带 |
| Windows | PIL/PyScreeze | pyautogui.screenshot() |
跨平台兼容 |
4.2 高效截图实现
import platform
import subprocess
import base64
from PIL import Image
import io
class ScreenCapture:
"""跨平台截图工具"""
def __init__(self):
self.system = platform.system()
def capture(self, save_path: str = "/tmp/screenshot.png") -> str:
"""截图并返回 base64 编码"""
if self.system == "Linux":
self._capture_linux(save_path)
elif self.system == "Darwin":
self._capture_macos(save_path)
elif self.system == "Windows":
self._capture_windows(save_path)
return self._to_base64(save_path)
def capture_region(self, x: int, y: int, width: int, height: int,
save_path: str = "/tmp/screenshot_region.png") -> str:
"""截取指定区域"""
# 先全屏截图
full_path = "/tmp/screenshot_full.png"
self.capture(full_path)
# 裁剪
img = Image.open(full_path)
cropped = img.crop((x, y, x + width, y + height))
cropped.save(save_path)
return self._to_base64(save_path)
def _capture_linux(self, save_path: str):
subprocess.run(["scrot", save_path], check=True)
def _capture_macos(self, save_path: str):
subprocess.run(["screencapture", save_path], check=True)
def _capture_windows(self, save_path: str):
import pyautogui
screenshot = pyautogui.screenshot()
screenshot.save(save_path)
def _to_base64(self, path: str) -> str:
with open(path, "rb") as f:
return base64.standard_b64encode(f.read()).decode("utf-8")
4.3 视觉分析提示词设计
def build_analysis_prompt(task: str, context: str = "") -> str:
"""构建视觉分析提示词"""
prompt = f"""你正在操作一台计算机。当前屏幕截图已提供。
任务：{task}
请分析当前屏幕状态，决定下一步操作。
操作选项：
1. 点击某个元素 → 使用 computer 工具的 left_click 操作，提供精确坐标
2. 输入文字 → 使用 computer 工具的 type 操作
3. 按键 → 使用 computer 工具的 key 操作（如 Enter、Tab 等）
4. 截图查看 → 使用 computer 工具的 screenshot 操作
5. 执行命令 → 使用 bash 工具
注意：
- 坐标必须精确，基于屏幕分辨率
- 每次只执行一个操作
- 操作后需要截图验证结果
- 如果任务已完成，请说明完成情况
"""
if context:
prompt += f"\n上下文信息：{context}"
return prompt
5. 鼠标键盘操作自动化
5.1 鼠标操作详解
class MouseController:
"""鼠标控制器"""
@staticmethod
def move(x: int, y: int):
"""移动鼠标"""
subprocess.run(["xdotool", "mousemove", str(x), str(y)])
@staticmethod
def click(x: int, y: int, button: int = 1):
"""点击（1=左键, 2=中键, 3=右键）"""
subprocess.run(["xdotool", "mousemove", str(x), str(y), "click", str(button)])
@staticmethod
def double_click(x: int, y: int):
"""双击"""
subprocess.run(["xdotool", "mousemove", str(x), str(y), "click", "--repeat", "2", "1"])
@staticmethod
def right_click(x: int, y: int):
"""右键点击"""
subprocess.run(["xdotool", "mousemove", str(x), str(y), "click", "3"])
@staticmethod
def drag(start_x: int, start_y: int, end_x: int, end_y: int):
"""拖拽"""
subprocess.run([
"xdotool", "mousemove", str(start_x), str(start_y),
"mousedown", "1",
"mousemove", str(end_x), str(end_y),
"mouseup", "1"
])
@staticmethod
def scroll(x: int, y: int, direction: str = "up", clicks: int = 3):
"""滚动（direction: up/down）"""
button = "4" if direction == "up" else "5"
subprocess.run(["xdotool", "mousemove", str(x), str(y)])
for _ in range(clicks):
subprocess.run(["xdotool", "click", button])
5.2 键盘操作详解
class KeyboardController:
"""键盘控制器"""
@staticmethod
def type_text(text: str, delay: int = 0):
"""输入文字"""
if delay > 0:
for char in text:
subprocess.run(["xdotool", "type", "--clearmodifiers", char])
time.sleep(delay / 1000.0)
else:
subprocess.run(["xdotool", "type", "--clearmodifiers", text])
@staticmethod
def press_key(key: str):
"""按下按键"""
subprocess.run(["xdotool", "key", key])
@staticmethod
def hotkey(*keys: str):
"""组合键"""
combo = "+".join(keys)
subprocess.run(["xdotool", "key", combo])
@staticmethod
def press_enter():
KeyboardController.press_key("Return")
@staticmethod
def press_tab():
KeyboardController.press_key("Tab")
@staticmethod
def press_escape():
KeyboardController.press_key("Escape")
@staticmethod
def select_all():
KeyboardController.hotkey("ctrl", "a")
@staticmethod
def copy():
KeyboardController.hotkey("ctrl", "c")
@staticmethod
def paste():
KeyboardController.hotkey("ctrl", "v")
@staticmethod
def undo():
KeyboardController.hotkey("ctrl", "z")
# 常用按键映射
KEY_MAP = {
"enter": "Return",
"return": "Return",
"tab": "Tab",
"escape": "Escape",
"esc": "Escape",
"space": "space",
"backspace": "BackSpace",
"delete": "Delete",
"up": "Up",
"down": "Down",
"left": "Left",
"right": "Right",
"home": "Home",
"end": "End",
"pageup": "Page_Up",
"pagedown": "Page_Down",
"f1": "F1", "f2": "F2", "f3": "F3", "f4": "F4",
"f5": "F5", "f6": "F6", "f7": "F7", "f8": "F8",
"f9": "F9", "f10": "F10", "f11": "F11", "f12": "F12",
}
5.3 等待与同步
import time
class WaitHelper:
"""等待辅助工具"""
@staticmethod
def wait(seconds: float):
"""简单等待"""
time.sleep(seconds)
@staticmethod
def wait_for_pixel_color(
x: int, y: int,
target_color: tuple,
timeout: float = 10.0,
tolerance: int = 10
) -> bool:
"""等待像素变为指定颜色"""
import pyautogui
start = time.time()
while time.time() - start < timeout:
pixel = pyautogui.pixel(x, y)
if all(abs(pixel[i] - target_color[i]) <= tolerance for i in range(3)):
return True
time.sleep(0.2)
return False
@staticmethod
def wait_for_stable_screen(interval: float = 0.5, stable_count: int = 3) -> bool:
"""等待屏幕稳定（连续多次截图相同）"""
import hashlib
previous_hash = None
stable = 0
for _ in range(20): # 最多等待 10 秒
screenshot = ScreenCapture().capture()
current_hash = hashlib.md5(screenshot.encode()).hexdigest()
if current_hash == previous_hash:
stable += 1
if stable >= stable_count:
return True
else:
stable = 0
previous_hash = current_hash
time.sleep(interval)
return False
6. 桌面应用场景
6.1 文件管理自动化
def automate_file_management():
"""文件管理自动化示例"""
agent = ComputerUseAgent()
tasks = [
# 任务1：整理下载文件夹
"打开文件管理器，进入下载文件夹，将所有 PDF 文件移动到 '文档' 文件夹，"
"将所有图片文件（jpg/png）移动到 '图片' 文件夹",
# 任务2：批量重命名
"在当前文件夹中，将所有 .txt 文件重命名，添加日期前缀（格式：YYYY-MM-DD_原文件名）",
# 任务3：创建报告目录结构
"在桌面创建一个名为 '2024年度报告' 的文件夹，"
"里面包含 '数据'、'图表'、'文档'、'最终版' 四个子文件夹"
]
for task in tasks:
print(f"\n执行任务: {task}")
result = agent.run(task)
print(f"结果: {result}")
6.2 数据录入自动化
def automate_data_entry(data_source: str, target_app: str):
"""数据录入自动化"""
agent = ComputerUseAgent()
# 读取数据源
import csv
with open(data_source, "r") as f:
reader = csv.DictReader(f)
rows = list(reader)
for i, row in enumerate(rows):
task = f"""
当前需要录入第 {i+1}/{len(rows)} 条数据：
- 姓名: {row['name']}
- 邮箱: {row['email']}
- 电话: {row['phone']}
- 地址: {row['address']}
请在当前打开的表单中依次填入以上信息，然后点击"保存"按钮。
保存后点击"新建"按钮准备录入下一条。
"""
result = agent.run(task)
print(f"第 {i+1} 条录入完成: {result}")
6.3 表单填写自动化
def automate_form_filling():
"""表单填写自动化"""
agent = ComputerUseAgent()
form_data = {
"company": "示例科技有限公司",
"contact": "张三",
"email": "zhangsan@example.com",
"phone": "13800138000",
"address": "北京市海淀区中关村大街1号",
"description": "这是一份自动生成的商业合作提案..."
}
task = f"""
请在当前网页表单中填写以下信息：
1. 公司名称：{form_data['company']}
2. 联系人：{form_data['contact']}
3. 邮箱：{form_data['email']}
4. 电话：{form_data['phone']}
5. 地址：{form_data['address']}
6. 描述：{form_data['description']}
填写完成后，请检查所有字段是否正确，然后点击提交按钮。
"""
result = agent.run(task)
return result
6.4 应用间协作
def cross_app_workflow():
"""跨应用协作工作流"""
agent = ComputerUseAgent()
task = """
请执行以下跨应用工作流：
1. 打开浏览器，访问 weather.com，获取今天北京的天气信息
2. 打开备忘录应用，创建一个新笔记
3. 在笔记中记录今天的天气情况
4. 打开邮件客户端，创建一封新邮件
5. 将天气信息作为邮件内容，发送给 team@example.com
6. 主题为"今日天气播报"
"""
result = agent.run(task)
return result
7. 网页自动化操作
7.1 浏览器操作
def web_automation_example():
"""网页自动化示例"""
agent = ComputerUseAgent()
# 示例：在线购物比价
task = """
请帮我完成以下网页操作：
1. 打开浏览器，访问 jd.com
2. 在搜索框中搜索 "机械键盘"
3. 按价格从低到高排序
4. 记录前 5 个商品的名称和价格
5. 然后访问 taobao.com，搜索同样的商品
6. 比较两个平台的价格，给出推荐
"""
result = agent.run(task)
return result
7.2 网页数据采集
def web_scraping_with_computer_use():
"""使用 Computer Use 进行网页数据采集"""
agent = ComputerUseAgent()
task = """
请在当前网页上：
1. 读取页面标题
2. 找到所有新闻标题和链接
3. 将信息整理为表格格式
4. 保存到桌面的 news_data.txt 文件中
"""
result = agent.run(task)
return result
7.3 表单提交与验证
def form_submission_workflow():
"""表单提交工作流"""
agent = ComputerUseAgent()
task = """
请在当前网页表单中完成以下操作：
1. 填写用户名：admin_test
2. 填写密码：********
3. 填写邮箱：test@example.com
4. 选择用户类型为"企业用户"
5. 勾选"同意服务条款"
6. 上传桌面上的 logo.png 文件作为头像
7. 点击注册按钮
8. 检查是否出现注册成功的提示
9. 如果有错误，截图记录并报告
"""
result = agent.run(task)
return result
8. 与 Selenium/Playwright 对比
8.1 技术对比
| 维度 | Claude Computer Use | Selenium | Playwright |
|---|---|---|---|
| 工作原理 | 视觉理解 + 模拟操作 | WebDriver 协议 | CDP 协议 |
| 元素定位 | 视觉识别（坐标） | CSS/XPath 选择器 | CSS/XPath 选择器 |
| 跨应用 | ✅ 支持桌面任意应用 | ❌ 仅浏览器 | ❌ 仅浏览器 |
| 维护成本 | 低（AI 自适应） | 高（选择器需维护） | 中等 |
| 执行速度 | 较慢（需截图分析） | 快 | 快 |
| 可靠性 | 中等（依赖视觉识别） | 高（精确定位） | 高 |
| 开发难度 | 低（自然语言） | 中等 | 中等 |
| 适用场景 | 复杂 UI、跨应用 | Web 自动化 | Web 自动化 |
8.2 混合方案
class HybridAutomation:
"""混合自动化方案 - 结合 Computer Use 与 Playwright"""
def __init__(self):
self.computer_agent = ComputerUseAgent()
self.playwright = None # Playwright 实例
async def init_playwright(self):
"""初始化 Playwright"""
from playwright.async_api import async_playwright
self.playwright = await async_playwright().start()
self.browser = await self.playwright.chromium.launch(headless=False)
self.page = await self.browser.new_page()
async def smart_automate(self, task: str):
"""智能选择自动化方式"""
# 判断任务类型
if self._is_web_task(task):
# Web 任务优先使用 Playwright（更可靠、更快）
return await self._playwright_execute(task)
elif self._is_desktop_task(task):
# 桌面任务使用 Computer Use
return self.computer_agent.run(task)
else:
# 混合任务：先用 Playwright 打开网页，再用 Computer Use 操作
return await self._hybrid_execute(task)
def _is_web_task(self, task: str) -> bool:
"""判断是否为纯 Web 任务"""
web_keywords = ["网页", "浏览器", "网站", "URL", "http"]
return any(kw in task for kw in web_keywords)
def _is_desktop_task(self, task: str) -> bool:
"""判断是否为纯桌面任务"""
desktop_keywords = ["桌面", "文件", "应用", "窗口", "系统设置"]
return any(kw in task for kw in desktop_keywords)
async def _playwright_execute(self, task: str):
"""使用 Playwright 执行 Web 任务"""
# Playwright 代码执行逻辑
pass
async def _hybrid_execute(self, task: str):
"""混合执行"""
# 先用 Playwright 打开目标网页
await self.page.goto("https://example.com")
# 再用 Computer Use 进行复杂操作
result = self.computer_agent.run(task)
return result
8.3 选择建议
任务是否涉及桌面应用？
├── 是 → Claude Computer Use
└── 否 → 任务是否为 Web 自动化？
├── 是 → 是否需要频繁维护？
│ ├── 是 → Claude Computer Use（自适应）
│ └── 否 → Playwright（性能更好）
└── 否 → 根据具体场景选择
9. 安全防护与操作审计
9.1 安全风险
Computer Use 的安全风险主要来自：
- 误操作风险：AI 可能误点击导致数据丢失
- 权限提升：自动化操作可能绕过人工确认
- 数据泄露：截图可能包含敏感信息
- 恶意利用：被用于自动化攻击
9.2 安全防护层
class SecurityGuard:
"""安全防护层"""
def __init__(self):
self.dangerous_actions = [
"delete", "remove", "format", "shutdown",
"restart", "logout", "drop", "truncate"
]
self.sensitive_patterns = [
"password", "密码", "secret", "密钥",
"token", "credit card", "信用卡"
]
self.allowed_apps = [
"firefox", "chrome", "terminal", "calculator",
"text_editor", "file_manager"
]
def check_action(self, action: str, context: str = "") -> dict:
"""检查操作安全性"""
result = {"safe": True, "warnings": [], "blocked": False}
# 检查危险操作
action_lower = action.lower()
for dangerous in self.dangerous_actions:
if dangerous in action_lower:
result["warnings"].append(f"检测到潜在危险操作: {dangerous}")
# 检查敏感信息
for pattern in self.sensitive_patterns:
if pattern in context.lower():
result["warnings"].append(f"可能涉及敏感信息: {pattern}")
# 高风险操作需要人工确认
if result["warnings"]:
result["safe"] = False
return result
def require_confirmation(self, action: str) -> bool:
"""要求人工确认"""
print(f"\n⚠️ 安全确认")
print(f"即将执行操作: {action}")
response = input("是否继续？(yes/no): ")
return response.lower() in ["yes", "y", "是"]
def sanitize_screenshot(self, screenshot_b64: str) -> str:
"""对截图进行脱敏处理"""
# 在实际应用中，可以使用 OCR + 正则表达式
# 识别并遮盖敏感信息（密码框、信用卡号等）
return screenshot_b64 # 简化示例
9.3 操作审计系统
import json
import datetime
from dataclasses import dataclass, asdict
from typing import List
@dataclass
class AuditRecord:
"""审计记录"""
timestamp: str
action_type: str
action_detail: dict
screenshot_before: str # base64
screenshot_after: str # base64
result: str
risk_level: str # low/medium/high
user_confirmed: bool
class AuditSystem:
"""操作审计系统"""
def __init__(self, log_path: str = "/var/log/computer_use_audit.jsonl"):
self.log_path = log_path
self.records: List[AuditRecord] = []
def log_action(self, record: AuditRecord):
"""记录操作"""
self.records.append(record)
# 写入日志文件
with open(self.log_path, "a") as f:
f.write(json.dumps(asdict(record), ensure_ascii=False) + "\n")
def create_record(
self,
action_type: str,
action_detail: dict,
screenshot_before: str = "",
screenshot_after: str = "",
result: str = "",
risk_level: str = "low",
user_confirmed: bool = False
) -> AuditRecord:
"""创建审计记录"""
return AuditRecord(
timestamp=datetime.datetime.now().isoformat(),
action_type=action_type,
action_detail=action_detail,
screenshot_before=screenshot_before,
screenshot_after=screenshot_after,
result=result,
risk_level=risk_level,
user_confirmed=user_confirmed
)
def get_summary(self, hours: int = 24) -> dict:
"""获取审计摘要"""
cutoff = datetime.datetime.now() - datetime.timedelta(hours=hours)
recent = [
r for r in self.records
if datetime.datetime.fromisoformat(r.timestamp) > cutoff
]
return {
"total_actions": len(recent),
"risk_distribution": {
"low": sum(1 for r in recent if r.risk_level == "low"),
"medium": sum(1 for r in recent if r.risk_level == "medium"),
"high": sum(1 for r in recent if r.risk_level == "high"),
},
"confirmed": sum(1 for r in recent if r.user_confirmed),
"action_types": list(set(r.action_type for r in recent))
}
9.4 沙箱环境
class SandboxEnvironment:
"""沙箱环境 - 限制 Computer Use 的操作范围"""
def __init__(self):
self.restricted_paths = [
"/etc", "/usr", "/var", "/boot",
"~/.ssh", "~/.gnupg", "~/.aws"
]
self.restricted_commands = [
"rm -rf", "sudo", "chmod 777", "mkfs",
"dd if=", "shutdown", "reboot"
]
self.screen_regions = {
"allowed": None, # None 表示全屏允许
"blocked": [] # 需要屏蔽的区域（如密码输入框）
}
def check_path_access(self, path: str) -> bool:
"""检查路径是否允许访问"""
import os
abs_path = os.path.abspath(path)
for restricted in self.restricted_paths:
if abs_path.startswith(os.path.expanduser(restricted)):
return False
return True
def check_command_safety(self, command: str) -> bool:
"""检查命令安全性"""
for restricted in self.restricted_commands:
if restricted in command:
return False
return True
def get_safe_region(self, screenshot: Image.Image) -> Image.Image:
"""遮盖屏幕上的敏感区域"""
# 在实际应用中，使用 OCR 检测密码框等敏感区域并遮盖
return screenshot
10. 企业级 RPA 集成方案
10.1 架构设计
┌───────────────────────────────────────────────────────────┐
│ 企业 RPA 平台 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 任务调度 │ │ 流程管理 │ │ 监控告警 │ │
│ └─────┬────┘ └─────┬────┘ └─────┬────┘ │
│ │ │ │ │
│ ┌─────┴─────────────┴─────────────┴─────┐ │
│ │ Claude Computer Use │ │
│ │ ┌─────────────────────────────────┐ │ │
│ │ │ Agent 执行层 │ │ │
│ │ │ - 截图 → 视觉分析 → 操作执行 │ │ │
│ │ └─────────────────────────────────┘ │ │
│ │ ┌─────────────────────────────────┐ │ │
│ │ │ 安全防护层 │ │ │
│ │ │ - 权限控制 → 操作审计 → 沙箱 │ │ │
│ │ └─────────────────────────────────┘ │ │
│ │ ┌─────────────────────────────────┐ │ │
│ │ │ 集成适配层 │ │ │
│ │ │ - API 网关 → 消息队列 → 缓存 │ │ │
│ │ └─────────────────────────────────┘ │ │
│ └───────────────────────────────────────┘ │
└───────────────────────────────────────────────────────────┘
10.2 任务调度系统
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable
import asyncio
from datetime import datetime
class TaskStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
@dataclass
class RPATask:
id: str
name: str
description: str
priority: int # 1-10
status: TaskStatus = TaskStatus.PENDING
created_at: datetime = None
started_at: datetime = None
completed_at: datetime = None
result: Optional[str] = None
error: Optional[str] = None
retry_count: int = 0
max_retries: int = 3
class TaskScheduler:
"""任务调度器"""
def __init__(self, max_concurrent: int = 3):
self.task_queue = asyncio.PriorityQueue()
self.running_tasks = {}
self.completed_tasks = {}
self.max_concurrent = max_concurrent
self.agent = ComputerUseAgent()
async def submit_task(self, task: RPATask):
"""提交任务"""
task.created_at = datetime.now()
await self.task_queue.put((task.priority, task))
print(f"任务已提交: {task.name} (优先级: {task.priority})")
async def run(self):
"""运行调度器"""
while True:
if len(self.running_tasks) < self.max_concurrent:
try:
priority, task = await asyncio.wait_for(
self.task_queue.get(), timeout=1.0
)
asyncio.create_task(self._execute_task(task))
except asyncio.TimeoutError:
continue
# 清理已完成任务
done = [t for t, s in self.running_tasks.items()
if s.done()]
for t in done:
del self.running_tasks[t]
await asyncio.sleep(0.1)
async def _execute_task(self, task: RPATask):
"""执行任务"""
task.status = TaskStatus.RUNNING
task.started_at = datetime.now()
try:
result = await asyncio.to_thread(
self.agent.run, task.description
)
task.result = result
task.status = TaskStatus.COMPLETED
except Exception as e:
task.error = str(e)
if task.retry_count < task.max_retries:
task.retry_count += 1
task.status = TaskStatus.PENDING
await self.task_queue.put((task.priority, task))
print(f"任务重试: {task.name} (第 {task.retry_count} 次)")
else:
task.status = TaskStatus.FAILED
task.completed_at = datetime.now()
self.completed_tasks[task.id] = task
10.3 流程编排
class WorkflowEngine:
"""流程编排引擎"""
def __init__(self):
self.scheduler = TaskScheduler()
def create_workflow(self, steps: list) -> list:
"""创建工作流"""
tasks = []
for i, step in enumerate(steps):
task = RPATask(
id=f"workflow_step_{i}",
name=step.get("name", f"步骤 {i+1}"),
description=step["description"],
priority=step.get("priority", 5),
max_retries=step.get("max_retries", 3)
)
tasks.append(task)
return tasks
async def execute_workflow(self, workflow: list):
"""顺序执行工作流"""
for task in workflow:
print(f"\n{'='*50}")
print(f"执行: {task.name}")
print(f"{'='*50}")
await self.scheduler.submit_task(task)
# 等待任务完成
while task.status in [TaskStatus.PENDING, TaskStatus.RUNNING]:
await asyncio.sleep(1)
if task.status == TaskStatus.FAILED:
print(f"任务失败: {task.error}")
# 可以选择继续或中断
break
print(f"完成: {task.result[:200] if task.result else '无结果'}...")
# 使用示例
async def main():
engine = WorkflowEngine()
workflow = engine.create_workflow([
{
"name": "打开浏览器",
"description": "打开 Firefox 浏览器，等待加载完成",
"priority": 10
},
{
"name": "访问目标网站",
"description": "在地址栏输入 https://example.com 并回车",
"priority": 9
},
{
"name": "登录",
"description": "找到登录按钮并点击，输入用户名和密码",
"priority": 8
},
{
"name": "导出数据",
"description": "进入数据页面，点击导出按钮，选择 CSV 格式，保存到桌面",
"priority": 7
}
])
await engine.execute_workflow(workflow)
# asyncio.run(main())
10.4 监控与告警
class RPAMonitor:
"""RPA 监控系统"""
def __init__(self):
self.metrics = {
"tasks_submitted": 0,
"tasks_completed": 0,
"tasks_failed": 0,
"avg_execution_time": 0,
"success_rate": 0.0
}
self.execution_times = []
self.alerts = []
def record_task_completion(self, task: RPATask):
"""记录任务完成"""
self.metrics["tasks_completed"] += 1
if task.started_at and task.completed_at:
duration = (task.completed_at - task.started_at).total_seconds()
self.execution_times.append(duration)
self.metrics["avg_execution_time"] = (
sum(self.execution_times[-100:]) / len(self.execution_times[-100:])
)
# 更新成功率
total = self.metrics["tasks_completed"] + self.metrics["tasks_failed"]
if total > 0:
self.metrics["success_rate"] = self.metrics["tasks_completed"] / total
# 检查告警条件
self._check_alerts(task)
def _check_alerts(self, task: RPATask):
"""检查告警条件"""
# 连续失败告警
if task.status == TaskStatus.FAILED:
self.metrics["tasks_failed"] += 1
if self.metrics["tasks_failed"] >= 3:
self._alert("连续任务失败超过 3 次")
# 执行时间过长告警
if task.started_at and task.completed_at:
duration = (task.completed_at - task.started_at).total_seconds()
if duration > 300: # 超过 5 分钟
self._alert(f"任务执行时间过长: {task.name} ({duration:.0f}s)")
def _alert(self, message: str):
"""发送告警"""
alert = {
"timestamp": datetime.now().isoformat(),
"message": message
}
self.alerts.append(alert)
print(f"[ALERT] {message}")
# 集成实际告警系统（Slack、邮件、短信等）
def get_dashboard(self) -> dict:
"""获取监控面板数据"""
return {
**self.metrics,
"recent_alerts": self.alerts[-10:],
"total_tasks": (
self.metrics["tasks_submitted"] +
self.metrics["tasks_completed"] +
self.metrics["tasks_failed"]
)
}
11. Computer Use 最佳实践与限制
11.1 最佳实践
实践一：任务拆分
# ❌ 不推荐：一次性描述复杂任务
bad_task = "帮我完成整个电商订单处理流程，包括登录、查看订单、处理退款、更新库存、发送通知"
# ✅ 推荐：拆分为小步骤
good_tasks = [
"打开浏览器，访问电商后台管理系统",
"使用管理员账号登录系统",
"进入订单管理页面，找到待处理退款的订单",
"点击退款按钮，填写退款原因，确认退款",
"进入库存管理页面，更新相关商品的库存数量",
]
实践二：明确的验证点
task = """
请执行以下操作，并在每一步后验证结果：
1. 打开计算器应用
验证：确认计算器窗口已打开
2. 计算 123 * 456
验证：确认显示结果为 56088
3. 将结果记录到记事本
验证：确认记事本中已保存结果
4. 保存记事本文件为 calculation.txt
验证：确认文件已成功保存
"""
实践三：错误处理指导
task = """
请打开邮箱客户端并发送一封邮件。
如果遇到以下情况，请按对应方式处理：
- 如果邮箱客户端未安装 → 报告"邮箱客户端未安装"并停止
- 如果需要登录 → 使用保存的凭据自动登录
- 如果网络连接失败 → 等待 30 秒后重试，最多重试 3 次
- 如果发送失败 → 截图记录错误信息并报告
"""
实践四：使用结构化指令
task = """
## 任务
在 Excel 中创建月度销售报表
## 步骤
1. 打开桌面上的 "sales_data.xlsx" 文件
2. 在 A1 单元格输入标题 "月度销售报表"
3. 在第 3 行创建表头：日期 | 产品 | 数量 | 金额
4. 从第 4 行开始填入数据
5. 在最后一行添加合计公式
6. 保存文件
## 约束
- 使用微软雅黑字体
- 标题字号 16，表头字号 12
- 金额列使用货币格式
- 合计行使用黄色背景
## 验证
完成后截图确认报表格式正确
"""
11.2 当前限制
| 限制 | 说明 | 影响 |
|---|---|---|
| 执行速度 | 每次操作需要截图-分析-执行循环 | 不适合高频实时操作 |
| 准确性 | 视觉识别可能存在误差 | 复杂 UI 可能误操作 |
| 成本 | 每次操作消耗 API Token | 大规模任务成本较高 |
| 延迟 | 网络延迟 + 模型推理时间 | 单次操作 2-5 秒 |
| 复杂 UI | 动态渲染、Canvas 等可能识别困难 | 特定应用兼容性问题 |
| 多显示器 | 多屏幕场景支持有限 | 需要额外配置 |
| 无障碍 | 依赖视觉，无法操作无界面程序 | 命令行任务需用 bash |
11.3 常见问题排查
# 问题1：点击位置不准确
# 解决：检查屏幕分辨率设置是否匹配
agent = ComputerUseAgent(display_size=(1920, 1080)) # 确保与实际分辨率一致
# 问题2：操作后界面未更新
# 解决：增加等待时间
task = """
点击按钮后，等待 2 秒让页面加载完成，然后再截图确认结果。
"""
# 问题3：文字输入乱码
# 解决：使用 xdotool 的 --clearmodifiers 参数
# 或使用剪贴板粘贴方式输入
# 问题4：弹窗干扰
# 解决：在任务描述中加入弹窗处理逻辑
task = """
执行任务过程中，如果出现任何弹窗或对话框：
- 如果是更新提示 → 点击"稍后"或关闭
- 如果是权限请求 → 点击"允许"
- 如果是错误提示 → 截图并报告
"""
12. 总结
12.1 核心要点回顾
- Computer Use 的本质：通过视觉理解 + 操作执行实现真正的桌面自动化
- 三大核心工具：computer（鼠标键盘）、text_editor（文件编辑）、bash（命令行）
- Agent 循环：截图 → 分析 → 操作 → 验证的闭环
- 安全第一：必须建立完善的权限控制、操作审计和沙箱机制
- 混合方案：与传统自动化工具（Selenium/Playwright）结合使用效果最佳
12.2 适用场景推荐
| 场景 | 推荐度 | 说明 |
|---|---|---|
| 非结构化数据录入 | ⭐⭐⭐⭐⭐ | AI 理解能力强 |
| 跨应用工作流 | ⭐⭐⭐⭐⭐ | 不受应用限制 |
| 网页表单填写 | ⭐⭐⭐⭐ | 适合复杂表单 |
| 文件管理 | ⭐⭐⭐⭐ | 简单高效的替代方案 |
| 高频批量操作 | ⭐⭐ | 速度和成本限制 |
| 精确像素级操作 | ⭐⭐ | 视觉识别精度有限 |
12.3 快速开始清单
- 安装 Python 依赖：
pip install anthropic pillow - 配置 Anthropic API Key
- 安装系统工具：
xdotool（Linux）或pyautogui（跨平台） - 运行第一个基础示例
- 实现截图 → 分析 → 操作的完整循环
- 添加安全防护层
- 实现操作审计日志
- 根据业务需求扩展功能
提示：本教程基于 Claude Computer Use 的公开文档和实践经验编写。由于技术持续迭代，具体 API 参数和功能请以 Anthropic 官方文档 为准。Computer Use 处于 Beta 阶段，生产环境使用请充分评估风险。
