UFO 客户端 Computer Manager 与 Computer 架构解析多实例 MCP 工具编排与隔离执行【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO本篇技术指南围绕 UFO 客户端核心执行层——**Computer Manager计算机管理器与 Computer计算机实例**展开深入剖析二者如何通过 MCPModel Context Protocol服务器实现数据采集与行为执行两类工具的命名空间隔离、按 Agent/进程/根应用上下文的路由分发以及带超时保护的线程池隔离执行机制。读完本文你将掌握ComputerManager、Computer、CommandRouter三个核心类的完整职责划分、初始化流程、配置写法与生命周期管理方法并能在 UFO 客户端项目中独立完成多实例工具编排与故障排查。一、整体定位Computer 层是什么在 UFO 客户端架构中ComputerManager是高层编排器负责创建与管理多个Computer实例每个Computer实例代表一个隔离的执行命名空间拥有自己独立的 MCP 服务器集合与工具注册表tool registry。这种分层设计实现了上下文相关的工具路由不同进程、不同 Agent 拥有各自专属的工具集互不干扰数据采集与行为执行的精细管控只读观察类工具与有状态变更类工具被分隔到不同命名空间从机制上防止误用。从客户端完整调用链看详见 UFO Client 与 MCP Integration服务器通过 WebSocket 下发 AIP 命令UFOClient委托给CommandRouterCommandRouter解析目标Computer实例后调用 MCP 工具并返回Result列表最终回传给服务器——Computer正是这一链条中的MCP 工具管理器。1.1 ComputerManager 职责能力说明实现多 Computer 管理创建并管理多个 Computer 实例按进程、按 Agent 的命名空间命名空间隔离为不同上下文隔离工具命名空间每个 Computer 独立的 MCP 服务器命令路由将命令路由到合适的 Computer 实例CommandRouter依据 agent/process/root 解析MCP 服务器配置配置数据采集与行为执行服务器配置驱动的服务器初始化生命周期管理初始化、重置、销毁 Computer异步初始化、级联重置1.2 Computer实例职责能力说明实现工具注册表维护可用 MCP 工具清单_tools_registry字典工具执行带超时保护地执行 MCP 工具调用线程池隔离最多 10 个 worker服务器管理管理数据采集与行为执行两类 MCP 服务器独立命名空间元工具提供内建工具如list_toolsmeta_tool装饰的方法异步初始化异步初始化 MCP 服务器async_init()1.3 架构关系1.4 两种命名空间命名空间类型用途示例数据采集Data Collection信息收集、非侵入式查询截图、UI 元素检测、应用状态行为执行Action执行操作、有状态变更GUI 自动化、文件操作、应用控制数据采集工具被设计为只读的信息获取行为执行工具则拥有完整的控制权以实施状态变更。二者在Computer中分别对应_data_collection_servers与_action_servers两个字典并在aip/messages.py的Command.tool_type中以字面量data_collection/action约束。二、Computer 实例的内部结构2.1 内部结构图2.2 关键属性属性类型用途_namestrComputer 名称标识符_process_namestr关联进程如notepad.exe_data_collection_serversDict[str, BaseMCPServer]命名空间 → MCP 服务器映射数据采集_action_serversDict[str, BaseMCPServer]命名空间 → MCP 服务器映射行为执行_tools_registryDict[str, MCPToolCall]工具键 → 工具信息映射_meta_toolsDict[str, Callable]内建元工具_executorThreadPoolExecutor工具执行线程池10 workers_tool_timeoutint工具执行超时6000 秒100 分钟注意工具执行超时被设置为 6000 秒100 分钟足以支撑长时间运行的操作同时避免无限期挂起。该值与 computer.py 中的self._tool_timeout 6000一致源码注释误写为 5 minutes实际 6000 秒 100 分钟。上述实现位于 computer.py构造函数会创建ThreadPoolExecutor(max_workers10, thread_name_prefixmcp_tool_)并通过遍历dir(self)自动收集所有带_meta_tool_name标记的方法注册为元工具。三、初始化流程3.1 Computer Manager 初始化from ufo.client.computer import ComputerManager from ufo.client.mcp.mcp_server_manager import MCPServerManager from config.config_loader import get_ufo_config # 1. 获取 UFO 配置 ufo_config get_ufo_config() # 2. 初始化 MCP 服务器管理器 mcp_server_manager MCPServerManager() # 3. 创建 Computer Manager computer_manager ComputerManager( ufo_config.to_dict(), mcp_server_manager )3.2 Computer 实例初始化computer Computer( namedefault_agent, process_nameexplorer.exe, mcp_server_managermcp_server_manager, data_collection_servers_config[...], action_servers_config[...] ) # 异步初始化必需 await computer.async_init()3.3 初始化时序3.4 配置示例data_collection_servers: - namespace: screenshot_collector type: local module: ufo.client.mcp.local_servers.screenshot_server reset: false - namespace: ui_collector type: local module: ufo.client.mcp.local_servers.ui_server reset: false action_servers: - namespace: gui_automator type: local module: ufo.client.mcp.local_servers.automation_server reset: false从源码看初始化细节computer.pyasync_init()依次调用_init_data_collection_servers()与_init_action_servers()两者都遍历配置列表取出namespace缺省时分别回退为default_data_collection/default_action与reset标记再交给mcp_server_manager.create_or_get_server(config, resetreset, process_name...)创建或复用服务器随后通过asyncio.gather并行执行register_mcp_servers()分别以tool_typedata_collection与tool_typeaction注册两类工具。3.5 MCP 服务器三种类型MCP 服务器管理器 支持三种部署类型_server_type_mapping类型实现类说明localLocalMCPServer进程内内存服务器从MCPRegistry获取 FastMCP 实例httpHTTPMCPServer远程 HTTP 服务器构造http://{host}:{port}{path}URL可携带auth鉴权stdioStdioMCPServer通过commandstart_argsenvcwd启动子进程传输create_or_get_server()依据namespace判重若resetTrue且已存在则先reset()再复用不存在则create_mcp_server()新建。而ComputerManager.get_or_create()在创建Computer时会先从配置键mcp中取agent_name对应的条目再取root_name子条目若root_name未在配置中定义会打日志并回退到default。四、命令路由CommandRouterCommandRouter根据 agent/process/root 上下文解析出应处理命令的 Computer 实例。路由签名async def execute( self, agent_name: str, process_name: str, root_name: str, commands: List[Command] ) - List[Result]路由逻辑源码级路由细节computer.pyexecute()先调用computer_manager.get_or_create(agent_name, process_name, root_name)实例键格式为f{agent_name}::{process_name}::{root_name or default}同一键的 Computer 会被缓存复用对无tool_name的命令直接返回Result(statusSUCCESS, resultNo action taken.)支持early_exitTrue默认短路语义一旦某条命令失败后续命令不再执行返回ResultStatus.SKIPPED每个命令通过computer.command2tool(command)转换为MCPToolCall后执行成功返回SUCCESS失败记录FAILURE并携带错误文本每条命令之间await asyncio.sleep(0.1)避免瞬间洪峰冲击服务器。command2tool的类型自动推断若Command未显式给出tool_typecommand2tool()会先在data_collection命名空间查找工具名再在action命名空间查找并发出Warning提示两个命名空间都找不到则抛出ValueError。五、工具执行线程池隔离与超时保护MCP 工具在隔离线程中执行目的是防止类似time.sleep的阻塞操作卡住主事件循环进而导致 WebSocket 断连。5.1 执行时序5.2 线程池配置参数值用途max_workers10最大并发工具执行数thread_name_prefixmcp_tool_线程命名便于调试超时6000 秒100 分钟单工具执行超时5.3 核心实现def _call_tool_in_thread(): Execute MCP tool call in an isolated thread with its own event loop. This prevents blocking operations in MCP tools from blocking the main event loop. # Create a new event loop for this thread loop asyncio.new_event_loop() asyncio.set_event_loop(loop) try: async def _do_call(): async with Client(server) as client: return await client.call_tool( nametool_name, argumentsparams, raise_on_errorFalse ) return loop.run_until_complete(_do_call()) finally: loop.close() # Execute in thread pool with timeout protection result await asyncio.wait_for( loop.run_in_executor(self._executor, _call_tool_in_thread), timeoutself._tool_timeout )故障语义computer.pyasyncio.TimeoutError返回is_errorTrue的CallToolResult错误信息为Tool {tool_name} execution timed out after {self._tool_timeout}s其他异常返回is_errorTrue的CallToolResult附带exc_info完整堆栈日志。六、工具注册表Tool Registry6.1 工具键格式tool_type::tool_name 示例 - action::click - action::type_text - data_collection::screenshot - data_collection::get_ui_elements键由 make_tool_key 静态方法生成f{tool_type}::{tool_name}。注册时若键已存在register_one_mcp_server会跳过并打印警告_register_tool则直接抛出ValueError防止重复。6.2 注册流程async def register_one_mcp_server( self, namespace: str, tool_type: str, mcp_server: BaseMCPServer ) - None: async with Client(mcp_server.server) as client: tools await client.list_tools() for tool in tools: tool_key self.make_tool_key(tool_type, tool.name) self._register_tool( tool_keytool_key, tool_nametool.name, titletool.title, namespacenamespace, tool_typetool_type, descriptiontool.description, input_schematool.inputSchema, output_schematool.outputSchema, mcp_servermcp_server )源码实现computer.py在注册 MCP 工具之后还会把该 Computer 的所有元工具按tool_type::meta_tool_name键逐一登记进注册表使其同样可被 LLM 发现与调用。MCPToolCall结构定义在 aip/messages.py其tool_info属性可序列化为不含服务器引用的MCPToolInfo用于对外展示。6.3 MCPToolCall 字段字段类型描述tool_keystr唯一键如action::clicktool_namestr工具名如clicktitlestr显示标题namespacestr服务器命名空间tool_typestraction或data_collectiondescriptionstr工具描述input_schemaDict输入参数 JSON Schemaoutput_schemaDict输出 Schemamcp_serverBaseMCPServer服务器引用6.4 动态增删服务器除初始化外Computer还提供运行时管理接口add_server(namespace, mcp_server, tool_type)把新服务器加入对应命名空间字典并立即注册其工具tool_type必须是data_collection或action否则抛ValueErrordelete_server(namespace, tool_type)按命名空间可叠加类型过滤从注册表移除工具并同步清理服务器字典。七、元工具Meta Tools元工具是带meta_tool装饰器的内建方法提供 Computer 级操作。装饰器实现见 computer.py通过给函数打上_meta_tool_name属性构造函数随后遍历dir(self)自动完成收集。示例list_tools 元工具Computer.meta_tool(list_tools) async def list_tools( self, tool_type: Optional[str] None, namespace: Optional[str] None, remove_meta: bool True ) - CallToolResult: Get available tools of a specific type. tools [] for tool in self._tools_registry.values(): if ((tool_type is None or tool.tool_type tool_type) and (namespace is None or tool.namespace namespace) and (not remove_meta or tool.tool_name not in self._meta_tools)): tools.append(tool.tool_info.model_dump()) return CallToolResult( content[TextContent(typetext, textjson.dumps(tools))] )元工具注册# In __init__: for attr in dir(self): method getattr(self, attr) if callable(method) and hasattr(method, _meta_tool_name): name getattr(method, _meta_tool_name) self._meta_tools[name] method元工具在_run_action()中被特殊处理命中_meta_tools的调用直接本地执行无需服务器若返回 awaitable 则await之。这一机制让list_tools等内建能力对 LLM 呈现为普通工具但执行路径完全不同。八、生命周期管理8.1 重置# Computer Manager 重置级联清空所有 Computer 实例 computer_manager.reset() # Computer 实例重置 computer.reset()组件重置行为Computer Manager清空所有 Computer 实例self.computers.clear()Computer清空工具注册表、重置 MCP 服务器MCP Servers重置服务器状态级联重置链路ufo_client.pyUFOClient.reset()会依次清空会话级状态session_id、agent_name、process_name、root_name、调用computer_manager.reset()以及mcp_server_manager.reset()后者清空全局_servers_mapping实现一次调用、全链路复位适用于会话切换或测试后的重新初始化场景。九、完整集成示例以下组合展示了从配置加载到命令执行的完整链路可直接在 client.py 的入口逻辑中看到对应装配方式from ufo.client.computer import ComputerManager, CommandRouter from ufo.client.mcp.mcp_server_manager import MCPServerManager from config.config_loader import get_ufo_config from aip.messages import Command # 1. 加载 UFO 配置含 mcp 键下的 Agent 服务器配置 ufo_config get_ufo_config() # 2. 初始化 MCP 服务器管理器 mcp_server_manager MCPServerManager() # 3. 创建 Computer Manager按 agent/process/root 管理实例 computer_manager ComputerManager(ufo_config.to_dict(), mcp_server_manager) # 4. 创建命令路由器 command_router CommandRouter(computer_manager) # 5. 执行一次动作Router 自动创建/复用 Computer 并执行 command Command( tool_nameclick, tool_typeaction, parameters{ control_text: Save, control_type: Button } ) results await command_router.execute( agent_nameHostAgent, process_namenotepad.exe, root_namedefault, commands[command] )真实配置参考config/ufo/mcp.yaml仓库中HostAgent、AppAgent含WINWORD.EXE/EXCEL.EXE/POWERPNT.EXE/explorer.exe等根应用、ConstellationAgent、HardwareAgent、LinuxAgent、MobileAgent各自定义了data_collection与action服务器列表。以MobileAgent为例其服务器为http类型并携带auth: ${UFO_MCP_API_KEY}环境变量鉴权HardwareAgent指向localhost:8006/mcp各办公软件根应用还挂载了独立的 COM 执行器如WordCOMExecutor、ExcelCOMExecutor并以reset: true标记在切换 Computer 时重置服务器状态。集成测试佐证仓库的 test_mobile_mcp_server.py 展示了如何用ComputerManagerCommandRouter装配 MobileAgent 的 data collection / action 服务器并对真实设备执行端到端验证前置条件包括 ADB 连接与UFO_MCP_API_KEY可作为二次开发与回归测试的模板。十、最佳实践10.1 监控工具执行耗时import time start time.time() result await computer._run_action(tool_call) duration time.time() - start if duration 300: # 5 minutes logger.warning(fSlow tool: {tool_call.tool_name} took {duration}s)10.2 优雅处理超时100 分钟的超时虽然宽裕但并非无限。设计工具时应让其在合理时间内完成并依赖CommandRouter的early_exit短路与CallToolResult.is_error标记做失败传播与降级处理。10.3 善用命名空间隔离# 将数据采集与行为执行分开 data_tools await computer.list_tools(tool_typedata_collection) action_tools await computer.list_tools(tool_typeaction)在提示词构建阶段仅向模型暴露与当前步骤相关的命名空间工具既能减少上下文噪音又能从机制上约束模型不越权执行状态变更操作。十一、相关文档导航Device Info Provider - 系统画像与设备信息采集MCP Integration - MCP 服务器细节与客户端集成UFO Client - 执行编排与会话管理Quick Start - 客户端快速上手Configuration Overview - UFO 系统配置说明核心源码索引Computer / ComputerManager / CommandRouter、MCP 服务器管理器、MCP 注册表、命令与结果消息定义、客户端装配入口、UFO 客户端会话层。【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
