企业官网建设流程全解析

UFO² 数据收集服务器详解：MCP 框架驱动的只读观测层与 UICollector 实战指南

【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO

导读

本文深入解析 UFO² 框架中的数据收集服务器（Data Collection Servers）—— 一类由框架自动调用、为 LLM 构建观察上下文的只读 MCP 服务器。你将理解它与可被 LLM 主动选择的 Action Servers 的本质区别、工具键与配置体系，掌握内置 UICollector 的 8 个观测工具及其源码实现原理，并学会通过config/ufo/mcp.yaml为 HostAgent / AppAgent 正确配置观测层，最终在"观察—推理—执行—验证"的 Agent 循环中落地一套稳健、高性能的 UI 状态采集方案。

上图展示了 UFO² 中 MCP 服务器层与 AppAgent 的协作关系：Shared GUI MCP Server 承载共享的 GUI 观测与自动化能力，App-Level API MCP Servers 为不同应用（Edge、Outlook、PowerPoint 等）提供专属工具，而数据收集服务器正是这一架构中负责"读取系统状态"的只读部分。

一、什么是数据收集服务器：框架驱动的"眼睛"

Data Collection Servers提供一组只读工具，用于观察并检索系统状态，而不修改任何状态。它们是 Agent 在采取行动之前理解当前环境的关键前提。

与可被 LLM 主动选择的 Action Servers 不同，数据收集服务器具有三个核心定位：

• Framework-Driven（框架驱动）：由 UFO² 框架自动调用，用于采集截图、UI 控件、系统信息等上下文；
• Observation Purpose（观察用途）：采集结果被拼装进 LLM 的观察提示词（observation prompt），作为决策依据；
• Not in Tool List（不在工具列表中）：这些工具不会作为可选动作呈现给 LLM，Agent 无法"选择"它们。

只有 Action Servers（动作服务器）才是 LLM 可选中的。关于动作服务器的完整说明，可参考 Action Servers。

数据收集在 Agent 执行流程中的位置如下：

数据收集服务器的五大特性

这一设计遵循了 MCP 架构中的关注点分离原则（详见 MCP Overview）：Agent 决定"做什么"（高层规划），MCP 服务器实现"怎么做"（底层执行），Computer 负责二者之间的路由。

二、Tool Type 标识符与工具键格式

所有数据收集工具使用统一的工具类型标识符：

tool_type = "data_collection"

工具键（tool_key）遵循如下格式：

tool_key = "data_collection::{tool_name}" # 示例： "data_collection::take_screenshot" "data_collection::get_window_list" "data_collection::get_control_info"

在源码层，工具键被Computer（ufo/client/computer.py）用于将来自 Agent 的Command路由到具体的 MCP 服务器工具上：_data_collection_namespaces = "data_collection"、_action_namespaces = "action"。这意味着框架在构建观察上下文时，只会在data_collection命名空间下寻找并执行工具，而action命名空间的工具则仅作为 LLM 可选项暴露在提示词中。

三、内置数据收集服务器：UICollector 完全解析

3.1 服务器信息总览

UICollector是 UFO² 内置的数据收集 MCP 服务器，负责收集 UI 元素信息与截图，为 LLM 决策构建观察上下文。

该服务器共提供8 个工具，涵盖截图、窗口列表、控件信息与 UI 树等观测能力。完整工具文档参见 UICollector Full Documentation。

3.2 八个观测工具详解

① get_desktop_app_info —— 枚举桌面应用窗口

获取桌面上所有可见应用窗口的列表（窗口名、类型、标识符），是 UI 自动化工作流中典型的第一步。

• 参数：

• 返回：List[Dict[str, Any]]，每个窗口信息字典包含：

{ "id": str, # 唯一窗口标识（如 "1"、"2"、"3"） "name": str, # 窗口标题/文本 "type": str, # 控件类型（如 "Window"、"Pane"） "kind": str # 目标类型："window" }

在源码实现中（ufo/client/mcp/local_servers/ui_mcp_server.py），当refresh_app_windows=True时会调用ui_state.control_inspector.get_desktop_app_dict(remove_empty=remove_empty)实时刷新窗口字典，否则复用缓存在ui_state.last_app_windows中的结果——这正是下文"缓存结果"最佳实践的直接实现。

② get_desktop_app_target_info —— 结构化窗口目标信息

与get_desktop_app_info类似，但返回TargetInfo对象而非普通字典，为框架内部使用提供更结构化的窗口表示。属性包括id、name、type、kind=TargetKind.WINDOW。

③ get_app_window_info —— 获取当前选中窗口的详细信息

检索当前激活/选中窗口的指定字段。必须先通过select_application_window（HostUIExecutor 的动作工具）选中窗口，再调用本工具。

• 参数：field_list: List[str]（必填），常见可用字段：

• 返回：Dict[str, Any]，字段名到值的映射。若未选中窗口，源码会抛出ToolError("No window is selected， please select a window first.")（ui_mcp_server.py）。

④ get_app_window_controls_info —— 枚举窗口内全部 UI 控件

扫描当前选中窗口，返回所有可交互控件（按钮、文本框等）的信息，是理解"该窗口可执行哪些操作"的关键工具。

• 参数：field_list: List[str]（必填），常见字段：label（控件标识）、control_text（控件文本）、control_type（Button/Edit 等）、control_rect、is_enabled、is_visible。
• 返回：List[Dict[str, Any]]。

⑤ get_app_window_controls_target_info —— 结构化控件目标信息

与get_app_window_controls_info类似，但返回TargetInfo对象（kind=TargetKind.CONTROL，含source: "uia"），供框架内部消费。

⑥ capture_window_screenshot —— 捕获当前选中窗口截图

截取当前活动窗口，返回base64 编码的 PNG 图像数据，是支撑 LLM 视觉能力的关键工具。

• 参数：无。
• 返回：str（base64 PNG 字符串）。
• 错误处理：截图失败时返回错误字符串：

"Error: No window selected" "Error capturing screenshot: {error_details}"

⑦ capture_desktop_screenshot —— 捕获桌面/主屏幕截图

截取整个桌面环境，可选所有显示器或仅主屏。

• 参数：all_screens: bool（可选，默认True）——True截取所有屏幕，False仅主屏。
• 返回：str（base64 PNG 图像数据）。

⑧ get_ui_tree —— 获取完整 UI 树结构

检索窗口中全部 UI 元素的层级结构树，深入洞察窗口布局与控件关系。

• 参数：无。
• 返回：Dict[str, Any]，嵌套字典表示的控件层级：

{ "control_type": "Window", "name": "Calculator", "children": [ {"control_type": "Pane", "name": "Display", "children": [...]}, {"control_type": "Button", "name": "1"} ] }

• 错误处理：失败时返回错误字典{"error": "No window selected"}或{"error": "Error getting UI tree: {details}"}。

3.3 源码实现原理：单例状态 + 工厂注册

从源码结构看，UICollector 的底层实现包含两个关键设计（ufo/client/mcp/local_servers/ui_mcp_server.py）：

1. 单例 UI 状态（UIServerState）：数据服务器与动作服务器共享同一个单例状态对象，包括photographer（截图门面，对应PhotographerFacade）、control_inspector（控件检查门面，对应ControlInspectorFacade）、selected_app_window（当前选中窗口，由 HostUIExecutor 设置）、last_app_windows（桌面窗口缓存）以及control_dict（控件 ID 到控件对象的映射）。这保证了"HostUIExecutor 选中窗口 → UICollector 读取该窗口"的跨服务器协调。窗口/控件还会被转换为WindowInfo/ControlInfo结构化对象（定义见 aip/messages.py）。

2. 工厂注册（MCPRegistry）：数据服务器通过装饰器注册进注册表：

@MCPRegistry.register_factory_decorator("UICollector") def create_data_mcp_server(*args, **kwargs) -> FastMCP: # 获取单例 UI 状态 ui_state = UIServerState() data_mcp = FastMCP("UFO UI Data MCP Server") # ... 注册 8 个 data_mcp.tool() 工具

MCPRegistry（ufo/client/mcp/mcp_registry.py）支持实例注册与工厂延迟初始化两种模式：register_factory注册工厂函数，get()时若实例不存在则通过工厂创建，实现了懒加载。配置文件中type: local的服务器正是从该注册表获取实例并在 Agent 进程内运行的。

3.4 快速上手示例

直接使用aip.messages.Command构造数据收集命令（aip是 UFO² 的 Agent Interaction Protocol 实现，参见 AIP Messages）：

from aip.messages import Command # 截取活动窗口截图 screenshot_cmd = Command( tool_name="take_screenshot", tool_type="data_collection", parameters={ "region": "active_window", "save_path": "screenshots/current.png" } ) # 获取所有窗口列表 windows_cmd = Command( tool_name="get_window_list", tool_type="data_collection", parameters={} )

更贴近当前仓库的实际调用方式是通过computer.run_actions([...])提交MCPToolCall，例如调用 UICollector 的capture_desktop_screenshot：

result = await computer.run_actions([ MCPToolCall( tool_key="data_collection::capture_desktop_screenshot", tool_name="capture_desktop_screenshot", parameters={"all_screens": True} ) ]) # 返回 base64 字符串："iVBORw0KGgoAAAANSUhEUgAA..."

四、配置详解：让观测层接入 Agent

数据收集服务器统一在config/ufo/mcp.yaml中配置，采用分层 YAML 结构：AgentName → SubType → tool_type（data_collection / action）→ Server List。完整字段说明可参考 MCP Configuration Guide。

4.1 三种典型配置

基础配置（HostAgent 默认）：

HostAgent: default: data_collection: - namespace: UICollector type: local start_args: [] reset: false

多服务器配置（同一观测类型挂载多个服务器）：

HostAgent: default: data_collection: - namespace: UICollector type: local reset: false

按应用定制配置（AppAgent 子类型覆盖）：

AppAgent: WINWORD.EXE: data_collection: - namespace: UICollector type: local reset: false # 在文档间切换时不重置 EXCEL.EXE: data_collection: - namespace: UICollector type: local reset: true # 在表格间切换时重置

4.2 配置字段速查

对于type: http的远程服务器，还需配置host（主机名或 IP）、port（端口）、path（MCP 端点路径，如/mcp）；type: stdio则需配置command、env、cwd等子进程字段。数据收集服务器在仓库默认配置中均使用type: local（进程内运行，无 IPC 开销），而远程场景（如 HardwareCollector、MobileDataCollector）可切换为 HTTP 部署——远程部署细节见 Remote Servers。

4.3 仓库默认配置全貌

仓库根目录下的 config/ufo/mcp.yaml 展示了全部 Agent 的实际数据收集配置：

HostAgent: default: data_collection: - namespace: UICollector type: local start_args: [] reset: false # 切换到新计算机时是否重置 MCP 服务器 AppAgent: default: data_collection: - namespace: UICollector type: local start_args: [] reset: false # WINWORD.EXE / EXCEL.EXE / POWERPNT.EXE / explorer.exe 各自挂载 UICollector HardwareAgent: default: data_collection: - namespace: HardwareCollector type: http host: "localhost" port: 8006 path: "/mcp" reset: false MobileAgent: default: data_collection: - namespace: MobileDataCollector type: http host: "localhost" port: 8020 path: "/mcp" auth: "${UFO_MCP_API_KEY}" # 密钥通过环境变量注入 reset: false

可见，Windows 桌面侧的 UI 观测统一复用UICollector（local），而跨平台硬件观测（HardwareCollector）、Android 设备观测（MobileDataCollector）则采用 HTTP 远程模式，并且支持用${UFO_MCP_API_KEY}这类环境变量管理认证凭据，避免把密钥硬编码进配置文件。

五、最佳实践

1. 行动前先观察（Call Before Action）：永远在执行动作之前采集数据，做出有依据的决策。源码中processing_context.py将DATA_COLLECTION列为 Agent 处理管线的独立阶段（ufo/agents/processors/context/processing_context.py），与LLM_INTERACTION、ACTION_EXECUTION等阶段并列，从流程层面强制了"先观察后决策"。

2. 缓存结果（Cache Results）：当状态未变化时可缓存采集结果以提升性能。get_desktop_app_info的refresh_app_windows=False参数正是为此设计——源码中该分支会直接复用ui_state.last_app_windows，跳过重新枚举（ui_mcp_server.py）。

3. 优雅处理失败（Handle Failures Gracefully）：窗口关闭或控件消失会导致采集失败，务必实现错误处理。典型模式是先检查返回内容中是否含"error"：

window_info = await computer.run_actions([ MCPToolCall(tool_key="data_collection::get_app_window_info", ...) ]) if "error" in window_info[0].content[0].text: # 未选中窗口，先执行 select_application_window...

4. 最小化截图调用（Minimize Screenshot Calls）：截图是昂贵操作——拍一张图分析多次，而非反复拍摄。每次截图都伴随 base64 编码与图像数据传输，调用次数应控制到最低。

5. 使用合适的区域（Use Appropriate Regions）：选择包含所需信息的最小区域（如 active window 而非 full screen），缩小截图与扫描范围。

6. 按需取字段（Selective Field Retrieval）：field_list只请求当前真正需要的字段。请求过多字段（如一次性要 8 个字段）会拖慢控件信息处理，违背性能原则。

六、常见用例

• UI 元素检测（UI Element Detection）：通过get_desktop_app_info+get_app_window_controls_info发现窗口与控件，为自动化定位目标。
• 屏幕监控（Screen Monitoring）：周期性capture_desktop_screenshot/capture_window_screenshot，驱动事件型自动化（如界面变更检测）。
• 系统健康检查（System Health Check）：在执行重型任务前通过观测工具检查系统资源状态。

七、错误处理

数据收集工具常见错误及应对策略：

在框架层面，所有 MCP 工具在Computer的线程池中执行（ThreadPoolExecutor(max_workers=10)），并受 6000 秒工具超时保护（ufo/client/computer.py），超时的工具会被取消并返回超时错误，从而避免阻塞主事件循环与 WebSocket 连接。

八、性能考量

• 截图优化：善用region/all_screens参数，只捕获所需区域；
• 并行数据采集：相互独立的采集可并行执行（Computer的线程池支持并发工具执行）；
• 缓存：状态未变化时复用缓存（refresh_app_windows=False、last_app_windows）。

九、与 Agent 的集成：Observe → Reason → Act → Verify 循环

数据收集服务器通常用于 Agent 执行的观察阶段。仓库中 AppAgent 的处理策略（ufo/agents/processors/strategies/app_agent_processing_strategy.py）会在构建 LLM 观察提示词时自动构造tool_type="data_collection"的Command，依次调用capture_window_screenshot、get_app_window_info、get_ui_tree等工具，这正是文档所强调的"框架自动调用、LLM 不选择"的落地形态。

Agent 的标准执行循环如下：

# Agent 执行循环（Observe → Reason → Act → Verify） while not task_complete: # 1. Observe: 收集当前状态 screenshot = await data_collection_server.take_screenshot() # 2. Reason: Agent 基于观察结果决定下一步动作 next_action = agent.plan(screenshot) # 3. Act: 执行动作（Action Server，LLM 可选） result = await action_server.execute(next_action) # 4. Verify: 再次观察，验证动作效果 new_screenshot = await data_collection_server.take_screenshot()

完整的"观察 → 行动 → 验证"模式可参考 Action Servers 中的 Integration with Data Collection 章节，以及 Computer（工具执行层）、HostAgent 概览 与 AppAgent 概览。

十、相关文档与关键要点

• UICollector Full Documentation —— 全部工具参数与示例
• Action Servers —— 可改变状态、由 LLM 选择的执行工具
• Configuration Guide —— MCP 分层配置完整参考
• Local Servers —— 内置本地 MCP 服务器清单
• Remote Servers —— HTTP/Stdio 远程部署
• MCP Overview —— MCP 高层架构
• Computer —— MCP 工具执行层
• 核心源码：ufo/client/mcp/local_servers/ui_mcp_server.py、ufo/client/mcp/mcp_registry.py、ufo/client/computer.py
• 默认配置：config/ufo/mcp.yaml

核心要点回顾：

• 数据收集服务器是只读、可安全重试的观测层；
• 始终先观察后行动，为决策提供依据；
• 状态未变化时缓存结果以提升性能；
• 用重试与回退逻辑优雅处理错误；
• 用合适的区域与并行采集换取性能；
• 完整细节请见 UICollector 文档。

【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO