6. Agent一句话截真机屏：截图Tool这样写

前面几节把读文档、生成用例这条链路打通了。自动化测试里还有一块绕不开：对着真机截屏。这一节就写截图 Tool——Agent 说一句话，就能把当前屏幕落到本地。

整条链路可以先记个轮廓：先在 .env 里配好 HDC_PATH，再在 json 里写上设备 sn；配置就绪后解析 hdc、检查设备——没设备就失败返回，有设备就截图并拉回本地；文件为空也算失败，成功则保存到 screenshots。Agent 一句话截屏，跑的就是这条链路。下面按关键段落过一遍。

先找 hdc 在哪。_get_hdc_cmd 优先看 HDC_PATH（文件或目录），找不到再退回系统 PATH 里的 hdc：

def _get_hdc_cmd() -> list:
    ...
    return ['hdc']

核心截图函数 _do_take_screenshot 一进来，先拼好 hdc 命令；有 device_id 就加上 -t：

hdc_cmd = _get_hdc_cmd()
hdc_prefix = hdc_cmd + (['-t', device_id] if device_id else [])

本地保存路径也要处理好。传进来的若是目录，就自动建目录并用时间戳拼文件名；若是文件路径，只保证父目录存在：

if save_file.is_dir() or (save_file.suffix == '' and not '.' in save_file.name):
    ...
    save_file = save_file / fname
else:
    save_file.parent.mkdir(parents=True, exist_ok=True)

设备端截图时，临时路径固定。先试新命令 snapshot_display，再试旧命令 screenshot；两个都失败就返回 None：

remote_path = "/data/local/tmp/tmp_screenshot.jpeg"
...
for cmd in [
    hdc_prefix + ['shell', 'snapshot_display', '-f', remote_path],
    hdc_prefix + ['shell', 'screenshot', remote_path],
]:
    ...
if not screenshot_ok:
    return None

截完后用 file recv 拉回电脑。本地文件存在且非空才返回路径，异常统一返回 None：

subprocess.run(
    hdc_prefix + ['file', 'recv', remote_path, str(save_file)],
    ...
)
if save_file.exists() and save_file.stat().st_size > 0:
    return str(save_file)
return None
except Exception:

工具侧，TakeScreenshotInput 声明参数 schema：设备 ID 可选，保存路径必填。TakeScreenshotTool 名字叫 take_screenshot，description 里写清干什么、参数怎么用，Agent 才知道何时调用。

_run 里先做路径规范化——相对路径一律接到项目根目录，已是绝对路径则原样使用：

base_dir = os.path.dirname(os.path.abspath(__file__))
if save_path.startswith('/'):
    save_path = os.path.normpath(os.path.join(base_dir, save_path.lstrip('/')))
elif not os.path.isabs(save_path):
    save_path = os.path.normpath(os.path.join(base_dir, save_path))

接着做前置检查：跑一遍 hdc list targets。工具不可用，或没连上设备，直接返回中文错误：

check = subprocess.run(
    hdc_prefix + ['list', 'targets'],
    ...
)
if check.returncode != 0:
    return f"错误：hdc工具不可用 - ..."
if not output or 'No device' in output or '无设备' in output:
    return "错误：未检测到连接的HarmonyOS设备"

检查通过后调用底层截图。成功报路径和大小；失败、超时、hdc 找不到，则分别返回对应提示：

result_path = _do_take_screenshot(save_path, device_id)
if result_path:
    ...
    return f"截图成功，保存至：{result_path} ({file_size} bytes)"
else:
    return "错误：截图失败"
except subprocess.TimeoutExpired:
    return "错误：截图超时(>15秒)"
...

熟悉截图 Tool 之后，改一下触发工具的提示词，表达清楚就可以了：

def main():
    agent = create_test_agent()
    result = agent.invoke({
        "messages": [{"role": "user", "content": "使用 take_screenshot 工具对当前设备屏幕截一张图，保存到 testcases/screenshots/app_screenshot.png 。"}]
    })
    print(result["messages"][-1].content[0]["text"])

完整test_tools.py的代码如下：

import os
import subprocess
from typing import Optional, Type
from pathlib import Path
from datetime import datetime
from pydantic import BaseModel, Field
from langchain_core.tools import BaseTool
import yaml
"""
B站、抖音：@老陈说编程
""" 
class GenerateTestCasesInput(BaseModel):
    test_cases: list = Field(description="测试用例列表，每个用例含 case_id, case_name, description, steps")
    output_path: str = Field(description="YAML文件输出路径")
class GenerateTestCasesTool(BaseTool):
    name: str = "generate_test_cases"
    description: str = "将测试用例列表保存为YAML文件。用例格式：每个用例包含case_id(用例ID)、case_name(用例名称)、description(描述)、steps(步骤列表)。每个step包含action(操作类型:click/input/clear/assert_exists/assert_not_exists/assert_text/assert_toast/wait/screenshot/swipe/back)和params(参数字典，含selector元素ID)"
    args_schema: Type[BaseModel] = GenerateTestCasesInput
 
    def _run(self, test_cases: list, output_path: str) -> str:
        try:
            base_dir = os.path.dirname(os.path.abspath(__file__))
            if output_path.startswith('/'):
                output_path = os.path.normpath(os.path.join(base_dir, output_path.lstrip('/')))
            elif not os.path.isabs(output_path):
                output_path = os.path.normpath(os.path.join(base_dir, output_path))
 
            for i, case in enumerate(test_cases):
                if 'case_id' not in case or 'steps' not in case:
                    return f"错误：第{i + 1}个用例缺少case_id或steps"
            output_file = Path(output_path)
            output_file.parent.mkdir(parents=True, exist_ok=True)
            with open(output_file, 'w', encoding='utf-8') as f:
                yaml.dump({'test_cases': test_cases}, f, allow_unicode=True, default_flow_style=False)
            return f"成功保存{len(test_cases)}个用例至：{output_path}"
        except Exception as e:
            return f"保存失败：{str(e)}"
 
"""
B站、抖音：@老陈说编程
""" 
 
def _get_hdc_cmd() -> list:
    """
    获取 hdc 命令路径
    hdc（HarmonyOS Device Connector）是 HarmonyOS 的命令行工具，用于与设备通信。
    本函数按以下优先级查找 hdc：
    1. 优先使用环境变量 HDC_PATH 指定的路径（可以是可执行文件或目录）
    2. 如果未配置环境变量或找不到，直接使用 'hdc'（依赖系统 PATH）
    Returns:
        list: hdc 命令列表，例如 ['hdc'] 或 ['C:\\path\\to\\hdc.exe']
    """
    # 从环境变量获取 HDC_PATH 配置
    hdc_path = os.environ.get("HDC_PATH", "")
    if hdc_path:
        p = Path(hdc_path)
        # 如果 HDC_PATH 直接指向一个文件，返回该文件路径
        if p.is_file():
            return [str(p)]
        # 如果 HDC_PATH 是一个目录，在目录下查找 hdc.exe（Windows）或 hdc
        if p.is_dir():
            for name in ['hdc.exe', 'hdc']:
                exe = p / name
                if exe.exists():
                    return [str(exe)]
    # 默认使用系统 PATH 中的 hdc 命令
    return ['hdc']
 
def _do_take_screenshot(save_path: str, device_id: Optional[str] = None, label: str = "") -> Optional[str]:
    """
    统一的截图底层实
    完整流程：
    1. 解析保存路径（支持传目录或完整文件路径）
    2. 在设备端执行截图命令（优先 snapshot_display，兼容旧版 screenshot）
    3. 验证截图是否在设备端成功生成
    4. 将截图从设备拉取到本地
    5. 验证本地文件有效性
    Args:
        save_path: 保存文件路径或目录
            - 如果是目录（如 'testcases/screenshots/'），自动生成带时间戳的文件名
            - 如果是完整路径（如 'testcases/screenshots/before.png'），直接使用该路径
        device_id: 设备ID（可选）
            - 如果有多个设备连接，通过此参数指定目标设备
            - 如果为 None，使用默认设备
        label: 截图标签（仅当 save_path 是目录时使用）
            - 用于自动生成文件名，例如 label='before' → 'before_20260723_160000.png'
    Returns:
        成功: 返回截图文件的绝对路径（字符串）
        失败: 返回 None
    """
    try:
        # 获取 hdc 基础命令
        hdc_cmd = _get_hdc_cmd()
 
        # 构建设备前缀命令：如果指定了 device_id，添加 '-t ' 参数
        # 例如 ['hdc', '-t', '2L1111110223916']
        hdc_prefix = hdc_cmd + (['-t', device_id] if device_id else [])
 
        # ========== 步骤1：解析并准备本地保存路径 ==========
        save_file = Path(save_path)
 
        # 判断 save_path 是目录还是文件路径：
        # - 如果是已存在的目录
        # - 或者路径没有后缀名（没有 '.'），也视为目录
        if save_file.is_dir() or (save_file.suffix == '' and not '.' in save_file.name):
            # 自动创建目录（parents=True 允许递归创建，exist_ok=True 目录已存在不报错）
            save_file.mkdir(parents=True, exist_ok=True)
            # 生成时间戳，格式：年月日_时分秒（例如 20260723_160000）
            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
            # 生成文件名：有 label 时用 label_时间戳.png，否则用 screenshot_时间戳.png
            fname = f"{label}_{timestamp}.png" if label else f"screenshot_{timestamp}.png"
            # 拼接完整文件路径
            save_file = save_file / fname
        else:
            # 如果是文件路径，确保父目录存在
            save_file.parent.mkdir(parents=True, exist_ok=True)
 
        # ========== 步骤2：设备端临时路径 ==========
        # HarmonyOS 设备上的临时截图保存位置（应用沙盒外的公共临时目录）
        remote_path = "/data/local/tmp/tmp_screenshot.jpeg"
 
        # ========== 步骤3：在设备端执行截图命令（兼容新旧版本） ==========
        screenshot_ok = False
        # 尝试两种截图命令：
        # 1. snapshot_display：新版 HarmonyOS 的截图命令（API 9+）
        # 2. screenshot：旧版兼容命令
        for cmd in [
            hdc_prefix + ['shell', 'snapshot_display', '-f', remote_path],
            hdc_prefix + ['shell', 'screenshot', remote_path],
        ]:
            # 执行截图命令，超时时间 15 秒
            r = subprocess.run(
                cmd, capture_output=True, text=True, timeout=15,
                encoding='utf-8', errors='replace'
            )
            # 合并 stdout 和 stderr，转为小写用于错误检测
            combined = (r.stdout + r.stderr).lower()
            # 检查命令是否执行成功：
            # - 返回码为 0
            # - 输出中不包含 'fail' 或 'error' 关键字
            if r.returncode == 0 and 'fail' not in combined and 'error' not in combined:
                # ========== 步骤4：验证设备端截图文件确实生成了 ==========
                # 执行 ls -l 命令检查文件是否存在、大小是否正常
                check = subprocess.run(
                    hdc_prefix + ['shell', 'ls', '-l', remote_path],
                    capture_output=True, text=True, timeout=5,
                    encoding='utf-8', errors='replace'
                )
                # 再次验证：ls 命令成功、没有 "No such file"、文件名出现在输出中
                if check.returncode == 0 and 'No such file' not in check.stdout and remote_path.split('/')[-1] in check.stdout:
                    screenshot_ok = True
                    # 截图成功，跳出循环
                    break
 
        # 如果两种命令都失败了，直接返回 None
        if not screenshot_ok:
            return None
 
        # ========== 步骤5：将截图从设备拉取到本地 ==========
        # 使用 hdc file recv 命令：从设备接收文件到本地
        # 命令格式：hdc file recv <设备端路径> <本地路径>
        subprocess.run(
            hdc_prefix + ['file', 'recv', remote_path, str(save_file)],
            capture_output=True, text=True, timeout=10,
            encoding='utf-8', errors='replace'
        )
 
        # ========== 步骤6：验证本地文件有效性 ==========
        # 检查文件是否存在，并且文件大小大于 0（避免空文件）
        if save_file.exists() and save_file.stat().st_size > 0:
            # 返回绝对路径字符串
            return str(save_file)
        # 文件无效，返回 None
        return None
 
    except Exception:
        # 捕获所有异常（超时、文件IO错误等），统一返回 None
        return None
 
class TakeScreenshotInput(BaseModel):
    """
    TakeScreenshotTool 的参数定义模型
    使用 Pydantic BaseModel 定义工具的输入参数 schema，
    LangChain 会自动根据这个模型生成参数描述和验证。
    """
    # 设备ID，可选参数，默认 None（使用默认设备）
    device_id: Optional[str] = Field(default=None, description="设备ID(可选)")
    # 截图保存路径，必填参数
    save_path: str = Field(description="截图保存路径")
 
class TakeScreenshotTool(BaseTool):
    """
    HarmonyOS 截图工具（LangChain Tool 封装）
    这是供 DeepAgent 调用的工具类，Agent 通过调用 take_screenshot 工具来截取设备屏幕。
    工具会：
    1. 自动处理相对路径/绝对路径转换（基于项目根目录）
    2. 检查 hdc 工具可用性
    3. 检查设备连接状态
    4. 调用底层 _do_take_screenshot 执行实际截图
    5. 返回友好的成功/失败信息（包含文件路径和大小）
    """
    # 工具名称（Agent 通过此名称调用工具）
    name: str = "take_screenshot"
    # 工具描述（Agent 会根据此描述理解工具用途和参数）
    description: str = "对HarmonyOS设备当前屏幕截图并保存到本地文件。参数save_path为截图保存的本地路径（建议testcases/screenshots/目录下）。成功返回文件路径和大小，失败返回错误信息。"
    # 参数 schema，指向上面定义的 TakeScreenshotInput
    args_schema: Type[BaseModel] = TakeScreenshotInput
 
    def _run(self, save_path: str, device_id: Optional[str] = None) -> str:
        """
        工具执行入口（LangChain 调用此方法）
        Args:
            save_path: 截图保存路径（可以是相对路径或绝对路径）
            device_id: 设备ID（可选）
        Returns:
            str: 执行结果消息
                - 成功："截图成功，保存至：<绝对路径> (<文件大小> bytes)"
                - 失败："错误：<错误原因>"
        """
        try:
            # ========== 步骤1：路径规范化处理 ==========
            # 获取当前文件所在目录（项目根目录）
            base_dir = os.path.dirname(os.path.abspath(__file__))
 
            # 将传入的路径转换为绝对路径：
            if save_path.startswith('/'):
                # 如果以 '/' 开头，视为相对于项目根目录的路径（去掉开头的 / 后拼接）
                save_path = os.path.normpath(os.path.join(base_dir, save_path.lstrip('/')))
            elif not os.path.isabs(save_path):
                # 如果是相对路径（不以 / 开头，也不是 Windows 绝对路径如 C:\），
                # 基于项目根目录拼接成绝对路径
                save_path = os.path.normpath(os.path.join(base_dir, save_path))
            # 如果已经是绝对路径（Windows 如 E:\...），直接使用
 
            # ========== 步骤2：准备 hdc 命令 ==========
            hdc_cmd = _get_hdc_cmd()
            hdc_prefix = hdc_cmd + (['-t', device_id] if device_id else [])
 
            # ========== 步骤3：前置检查 - hdc 可用性和设备连接 ==========
            # 执行 'hdc list targets' 命令列出已连接设备
            check = subprocess.run(
                hdc_prefix + ['list', 'targets'],
                capture_output=True, text=True, timeout=5,
                encoding='utf-8', errors='replace'
            )
            # 检查 hdc 命令本身是否可用
            if check.returncode != 0:
                return f"错误：hdc工具不可用 - {check.stderr.strip() or check.stdout.strip()}"
            output = check.stdout.strip()
            # 检查是否有设备连接
            # hdc list targets 在无设备时可能输出 "No device"、空行、或 "无设备"
            if not output or 'No device' in output or '无设备' in output:
                return "错误：未检测到连接的HarmonyOS设备"
 
            # ========== 步骤4：执行截图 ==========
            result_path = _do_take_screenshot(save_path, device_id)
            if result_path:
                # 截图成功：获取文件大小，返回成功消息
                save_file = Path(result_path)
                file_size = save_file.stat().st_size
                return f"截图成功，保存至：{result_path} ({file_size} bytes)"
            else:
                # 截图失败
                return "错误：截图失败"
 
        except subprocess.TimeoutExpired:
            # 命令执行超时
            return "错误：截图超时(>15秒)"
        except FileNotFoundError:
            # hdc 命令未找到（系统找不到 hdc.exe）
            return "错误：hdc命令未找到，请确保已安装HarmonyOS SDK并配置HDC_PATH环境变量"
        except Exception as e:
            # 其他未知异常
            return f"截图失败：{str(e)}"
 
def get_all_tools() -> list:
    return [
        GenerateTestCasesTool(),
        TakeScreenshotTool()
    ]

完整run_automation_test.py的代码如下：

import os
from deepagents.backends import LocalShellBackend
from dotenv import load_dotenv
from deepagents import create_deep_agent
from test_tools import get_all_tools
"""
B站、抖音：@老陈说编程
""" 
 
load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("API_KEY")
os.environ["OPENAI_BASE_URL"] = os.getenv("BASE_URL")
model = f"openai:{os.getenv('MODEL')}"
 
def create_test_agent():
    return create_deep_agent(
        model=model,
        system_prompt="HarmonyOS 自动化测试专家。",
        tools=get_all_tools(),
        skills=["./skills/"],
        backend=LocalShellBackend(root_dir=os.getcwd(), virtual_mode=True, inherit_env=True),
        debug=True
    )
 
def main():
    agent = create_test_agent()
    result = agent.invoke({
        "messages": [{"role": "user", "content": "使用take_screentshot工具对当前设备屏幕截一张，"
                                                 "保存到 testcases/screenshots/app_screenshot.png 。"}]
    })
    print(result["messages"][-1].content[0]["text"])
 
if __name__ == '__main__':
    main()

跑起来后，Agent 会调用 take_screenshot，把当前屏幕保存到指定路径。有了这步，后面的启停应用、测前测后截图才有落点。

配套动画视频在：《6. Agent一句话截真机屏：截图Tool这样写》。