UFO 客户端核心执行层深度解析Computer 与 ComputerManager 的 MCP 工具编排架构与实战指南【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFOComputer 层是 UFO 客户端ufo.client的执行引擎核心负责承载 MCPModel Context Protocol工具的注册、路由与线程隔离执行是多 AgentHostAgent / AppAgent / ConstellationAgent 等在真实设备上完成「感知 → 决策 → 操作」闭环的落地通道。本文将基于仓库文档与 computer.py 源码系统讲解Computer、ComputerManager、CommandRouter三大组件的职责边界、实例化流程、工具命名空间、超时与线程隔离机制、元工具Meta Tools、动态服务器管理以及错误处理与最佳实践读者读完后可独立完成自定义 MCP 工具接入与命令路由编排。架构总览一条从命令到 MCP 工具调用的完整链路Computer 层位于 UFO 客户端内部向上承接 AIPAgent Interaction Protocol协议消息中的Command向下驱动各类 MCP 服务器本地进程、HTTP 服务、stdio 子进程执行具体工具。其核心由三个组件构成Computer单个逻辑计算机的执行上下文管理自己的一组 MCP 服务器与工具提供线程隔离10 工作线程与超时保护6000 秒。ComputerManager基于 Agent 配置创建并缓存多个Computer实例按agent_name::process_name::root_name命名空间路由。CommandRouter把协议层的Command列表路由到正确的Computer实例上执行支持early_exit首错即停。对应源码位于 ufo/client/computer.py其中ComputerL22、ComputerManagerL595、CommandRouterL679三个类定义完整可查。核心职责工具注册从多个 MCP 服务器批量注册工具并以tool_type::tool_name作为全局唯一键做命名空间隔离命令路由将高层命令Command转换为MCPToolCall并分发执行执行管理在线程池中隔离执行工具叠加超时保护避免阻塞操作拖垮事件循环元工具内置自省能力如list_tools让 Agent 能够动态查询自身可用工具。Computer 类单个执行上下文Computer类管理一台「逻辑计算机」每台计算机拥有独立的 MCP 服务器集合与工具注册表。它的核心属性如下表与 computer.py 的实现一一对应属性类型说明_namestr计算机实例唯一标识_process_namestr关联的进程名用于 MCP 服务器进程隔离_data_collection_serversDict[str, BaseMCPServer]数据采集服务器截图、UI 检测等_action_serversDict[str, BaseMCPServer]动作服务器GUI 自动化、文件操作等_tools_registryDict[str, MCPToolCall]全部可用工具注册表键为tool_type::tool_name_meta_toolsDict[str, Callable]内置自省工具_executorThreadPoolExecutor线程池10 个并发 worker线程名前缀为mcp_tool__tool_timeoutint工具执行超时6000 秒100 分钟从源码看线程池在__init__中创建self._executor concurrent.futures.ThreadPoolExecutor( max_workers10, thread_name_prefixmcp_tool_ ) # Tool execution timeout (seconds) self._tool_timeout 6000工具命名空间Tool NamespacesComputer支持两类工具命名空间类级常量定义于源码 L27-L28data_collection信息采集类工具非破坏性操作如截图、UI 检测action状态变更类操作如点击、输入文本。工具键格式统一为tool_type::tool_namedata_collection::screenshot # 截图 data_collection::ui_detection # 检测 UI 元素 action::click # 点击 UI 元素 action::type_text # 输入文本说明由于命名空间不同同一个工具名可以在数据采集与动作两个上下文中并存。例如data_collection::get_file_info与action::get_file_info可以同时注册。make_tool_key()静态方法computer.py即负责拼接这个键。ComputerManager多实例管理与配置解析ComputerManager根据 Agent 配置创建并缓存多个Computer实例避免重复初始化开销。所有实例缓存在self.computers字典中通过get_or_create惰性创建computer.py。实例键格式key f{agent_name}::{process_name}::{root_name}示例host_agent::chrome::default # host_agent 的默认 chrome 计算机 host_agent::vscode::custom_config # host_agent 的自定义 VSCode 计算机需要说明的是process_name与root_name均为可选参数缺省时以default填充源码 L624root_name or default。配置结构配置入口为mcp键ComputerManager._configs_key mcp其结构与仓库真实配置 config/ufo/mcp.yaml 一致mcp: host_agent: default: data_collection: - namespace: screenshot server_type: local module: ufo.client.mcp.local_servers.screenshot reset: false - namespace: ui_detection server_type: local module: ufo.client.mcp.local_servers.ui_detection reset: false action: - namespace: gui_automation server_type: local module: ufo.client.mcp.local_servers.gui_automation reset: false仓库真实配置示例config/ufo/mcp.yaml展示了更完整的用法每个条目是一个 MCP 服务器定义AppAgent: default: data_collection: - namespace: UICollector type: local start_args: [] reset: false action: - namespace: AppUIExecutor type: local start_args: [] reset: false - namespace: CommandLineExecutor type: local start_args: [] reset: false EXCEL.EXE: data_collection: - namespace: UICollector type: local start_args: [] reset: false action: - namespace: AppUIExecutor type: local start_args: [] reset: false - namespace: ExcelCOMExecutor type: local start_args: [] reset: true字段说明字段类型说明namespacestr服务器命名空间也是工具的命名空间前缀typestr服务器类型local/http/stdiostart_argslist启动参数stdio 类型使用host/port/pathstr/intHTTP 类型服务器的地址如 HardwareAgent 的localhost:8006/mcpauthstrHTTP 鉴权信息如${UFO_MCP_API_KEY}支持环境变量未解析的环境变量会被拒绝见 mcp_server_manager.pyresetbool切换到新计算机时是否重置 MCP 服务器进程服务器类型映射定义在 mcp_server_manager.pyhttp → HTTPMCPServer、local → LocalMCPServer、stdio → StdioMCPServer。其中local从MCPRegistry按命名空间取出已注册的FastMCP实例http组装 URL可选鉴权stdio以StdioTransport拉起子进程。配置要求每个 Agent 必须至少包含一个defaultroot 配置若指定的root_name不存在管理器会打印日志并回退到default源码 L634-L641缺失 Agent 或 root 配置会抛出ValueError。CommandRouter命令的编排执行器CommandRouter负责把一组Command路由到正确的Computer实例并依次执行。其执行流程如下核心方法是execute()computer.py调用链为通过computer_manager.get_or_create(agent_name, process_name, root_name)获取/创建目标计算机对每条Command调用computer.command2tool(command)转换为MCPToolCall调用computer.run_actions([tool_call])执行根据CallToolResult.is_error构造ResultSUCCESS/FAILURE失败时记录has_failed若early_exitTrue且此前已有失败则后续命令被跳过并标记为ResultStatus.SKIPPED每条命令执行后await asyncio.sleep(0.1)限速避免压垮服务器。初始化流程Computer 初始化from ufo.client.computer import Computer from ufo.client.mcp.mcp_server_manager import MCPServerManager # 创建 MCP server manager mcp_manager MCPServerManager() # 初始化 computer computer Computer( namemy_computer, process_namemy_process, mcp_server_managermcp_manager, data_collection_servers_config[ { namespace: screenshot, server_type: local, module: ufo.client.mcp.local_servers.screenshot } ], action_servers_config[ { namespace: gui_automation, server_type: local, module: ufo.client.mcp.local_servers.gui_automation } ] ) # 异步初始化必需 await computer.async_init()⚠️ 重要创建Computer实例后必须调用await computer.async_init()。它会构建数据采集与动作两类服务器字典并通过asyncio.gather并行注册全部 MCP 服务器及其工具computer.py。async_init的注册流程值得细看每个 MCP 服务器通过register_one_mcp_server()用Client连接后list_tools()拉取工具清单逐个以tool_type::tool_name为键写入_tools_registry若键已存在则打印 warning 跳过防止覆盖computer.py。ComputerManager 初始化from ufo.client.computer import ComputerManager # 加载配置 with open(config.yaml) as f: configs yaml.safe_load(f) # 创建 manager manager ComputerManager( configsconfigs, mcp_server_managermcp_manager ) # 获取或创建 computer 实例 computer await manager.get_or_create( agent_namehost_agent, process_namechrome, root_namedefault )ComputerManager提供reset()方法清空全部缓存实例computer.py适用于不重启应用的场景下重新初始化。工具执行基础工具执行from aip.messages import MCPToolCall # 创建工具调用 tool_call MCPToolCall( tool_keydata_collection::screenshot, tool_namescreenshot, parameters{region: full_screen} ) # 执行工具 results await computer.run_actions([tool_call]) # 检查结果 if results[0].is_error: print(fError: {results[0].content}) else: print(fSuccess: {results[0].data})run_actions()会按顺序逐个执行而非并发内部逐条调用_run_action()并返回List[CallToolResult]computer.py。命令到工具的转换command2tool()方法把高层Command对象转换为MCPToolCallfrom aip.messages import Command # 创建命令 command Command( tool_namescreenshot, tool_typedata_collection, parameters{region: active_window} ) # 转换为工具调用 tool_call computer.command2tool(command) # 执行 results await computer.run_actions([tool_call])如果命令未指定tool_typecommand2tool()会自动在注册表中探测先在data_collection键下查找命中则作为数据采集工具并发出 Warning 提示否则在action键下查找两处都未命中则抛出ValueErrorcomputer.py。转换时参数会经过copy.deepcopy拷贝后写入tool_info.parameters。批量工具执行# 顺序执行多个工具 tool_calls [ MCPToolCall(tool_keydata_collection::screenshot, tool_namescreenshot), MCPToolCall(tool_keydata_collection::ui_detection, tool_namedetect_ui), MCPToolCall(tool_keyaction::click, tool_nameclick, parameters{x: 100, y: 200}) ] results await computer.run_actions(tool_calls) for i, result in enumerate(results): print(fTool {i}: {Success if not result.is_error else Failed})线程隔离与超时机制为什么需要线程隔离MCP 工具内部可能包含阻塞操作如time.sleep()、同步 I/O若直接在主事件循环中执行会卡死整个进程进而引发 WebSocket 断连。为防止这一点设计了三重保护每个工具调用在独立线程中运行并创建属于自己的事件循环线程池提供10 个并发 worker每个工具调用有6000 秒100 分钟超时。实现细节源码实现位于 computer.pydef _call_tool_in_thread(): 在独立线程及其事件循环中执行 MCP 工具调用。 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() # 在线程池中执行并叠加超时 result await asyncio.wait_for( loop.run_in_executor(self._executor, _call_tool_in_thread), timeoutself._tool_timeout )若工具执行超过 6000 秒asyncio.wait_for抛出TimeoutError_run_action捕获后构造超时错误结果返回而非抛出异常保证调用方始终拿到结构化的CallToolResultCallToolResult( is_errorTrue, content[TextContent(textTool execution timed out after 6000s)] )同理其他异常也会被捕获并包装为is_errorTrue的结果同时以exc_infoTrue记录日志computer.py。Meta Tools内置自省工具元工具是内建的内省工具让 Agent 无需真实调用 MCP 服务器即可查询计算机的能力清单。当前内置的唯一元工具是list_tools。注册方式通过Computer.meta_tool()装饰器注册。装饰器只是把元工具名写入函数属性_meta_tool_name__init__时遍历实例属性自动收集到_meta_tools字典computer.py。注册阶段每个命名空间都会自动挂载一份元工具computer.pyclass Computer: meta_tool(list_tools) async def list_tools( self, tool_type: Optional[str] None, namespace: Optional[str] None, remove_meta: bool True ) - CallToolResult: 列出所有可用工具。实现上按tool_type、namespace过滤注册表remove_metaTrue时排除元工具自身最终返回CallToolResult(datatools, ...)其中data是MCPToolInfo.model_dump()的列表computer.py。使用方式# 列出所有 action 工具 tool_call MCPToolCall( tool_keyaction::list_tools, tool_namelist_tools, parameters{tool_type: action} ) result await computer.run_actions([tool_call]) tools result[0].data # 可用 action 工具列表_run_action在发现目标工具名命中_meta_tools时走特殊分支直接调用元工具函数若是协程则await否则直接返回全程不发起服务器调用computer.py。示例列出screenshot命名空间下的全部工具result await computer.run_actions([ MCPToolCall( tool_keydata_collection::list_tools, tool_namelist_tools, parameters{namespace: screenshot, remove_meta: True} ) ]) # 返回形如[{tool_name: take_screenshot, description: ..., ...}]动态服务器管理Computer支持在运行期动态增删 MCP 服务器非常适合按需扩展工具集。添加服务器from ufo.client.mcp.mcp_server_manager import BaseMCPServer # 创建新的 MCP 服务器 new_server mcp_manager.create_or_get_server( mcp_config{ namespace: custom_tools, server_type: local, module: my_custom_mcp_server }, resetFalse, process_namemy_process ) # 添加到 computer await computer.add_server( namespacecustom_tools, mcp_servernew_server, tool_typeaction )add_server()要求显式指定tool_typedata_collection或action否则抛出ValueError它会写入对应服务器字典并立即调用register_one_mcp_server完成工具注册computer.py。删除服务器# 删除服务器及其全部工具 await computer.delete_server( namespacecustom_tools, tool_typeaction )delete_server()会扫描注册表删除namespace匹配且可选按tool_type过滤的所有工具键再弹出服务器字典中的对应条目computer.py。动态管理的典型场景为特定任务临时挂载专用工具移除服务器以释放内存占用开发期热重载 MCP 服务器。命令路由实战CommandRouter编排跨计算机的命令执行early_exit参数控制失败时的行为。基础用法from ufo.client.computer import CommandRouter from aip.messages import Command, Result # 创建 router router CommandRouter(computer_managermanager) # 执行命令 commands [ Command(tool_namescreenshot, tool_typedata_collection), Command(tool_nameclick, tool_typeaction, parameters{x: 100, y: 200}) ] results await router.execute( agent_namehost_agent, process_namechrome, root_namedefault, commandscommands, early_exitTrue # 首个错误即停止 ) for result in results: print(fStatus: {result.status}) print(fData: {result.data})Result对象定义于 aip/messages.py携带status、error、result、namespace、call_id五个字段其中call_id与Command.call_id一一对应便于协议层对齐结果。错误处理语义# early_exitTrue首个命令失败后立即停止后续执行 results await router.execute( agent_namehost_agent, process_namechrome, root_namedefault, commandscommands, early_exitTrue ) # early_exitFalse即使部分失败也执行完所有命令 results await router.execute( agent_namehost_agent, process_namechrome, root_namedefault, commandscommands, early_exitFalse )⚠️ 警告early_exitTrue时若某条命令失败后续命令不会执行其结果会被置为ResultStatus.SKIPPEDResultStatus定义于 aip/messages.py取值为SUCCESS/FAILURE/SKIPPED/NONE。另外execute()对无tool_name的空命令会直接返回SUCCESSresult 为No action taken.而非报错保证协议消息的健壮性computer.py。工具注册表_tools_registry维护全部可用工具的映射值为MCPToolCall对象Pydantic 模型定义于 aip/messages.py它同时承载工具元数据与运行时参数。工具键格式tool_key f{tool_type}::{tool_name} # 示例 data_collection::screenshot action::click data_collection::list_tools # 元工具访问工具信息# 获取工具信息 tool_info computer._tools_registry.get(action::click) # 工具信息包含 print(tool_info.tool_name) # click print(tool_info.tool_type) # action print(tool_info.namespace) # 如 gui_automation print(tool_info.description) # 工具描述 print(tool_info.input_schema) # 输入参数的 JSON schema print(tool_info.mcp_server) # 指向所属 MCP 服务器注意_tools_registry是私有属性正常业务路径应优先通过list_tools元工具或command2tool间接访问。最佳实践配置层面合理使用命名空间把相关工具归组到有意义的命名空间下关注点分离只读操作用data_collection状态变更用action调优超时长耗时操作如大文件下载按需调大_tool_timeout保留 default root始终提供defaultroot 配置作为兜底避免ValueError。性能优化并行注册服务器async_init()已通过asyncio.gather并行注册数据采集与动作服务器复用 Computer 实例让ComputerManager缓存实例避免重复创建与重复注册的开销控制并发工具数线程池只有 10 个 worker过量并行工具会排队谨慎使用 reset配置resetTrue会重启 MCP 服务器进程只在需要重新初始化服务器状态时使用。常见陷阱⚠️ 重要以下是高频踩坑点——忘记async_init()创建Computer实例后必须调用否则服务器与工具未注册任何工具调用都会报「工具未注册」工具键冲突同一tool_type内的工具名必须唯一重复注册会被跳过并输出 warning超时设置过短部分操作如文件下载天然耗时长需放宽超时元工具内阻塞元工具应保持轻量避免 I/O 操作拖慢内省响应。错误处理工具执行错误try: results await computer.run_actions([tool_call]) if results[0].is_error: error_message results[0].content[0].text print(fTool error: {error_message}) except ValueError as e: print(fTool not registered: {e}) except asyncio.TimeoutError: print(Tool execution timed out) except Exception as e: print(fUnexpected error: {e})需要强调_run_action内部已把超时与一般异常包装为is_errorTrue的CallToolResult因此run_actions本身通常不会抛异常ValueError工具未注册与TimeoutError主要出现在直接调用command2tool/_run_action等更底层接口时。配置错误try: computer await manager.get_or_create( agent_namehost_agent, process_namechrome, root_nameinvalid_root ) except ValueError as e: print(fConfiguration error: {e}) # 回退到 default computer await manager.get_or_create( agent_namehost_agent, process_namechrome, root_namedefault )值得注意的是get_or_create对不存在的root_name会自动回退到default打印 info 日志真正抛ValueError的情况是 Agent 级配置缺失或default也不存在。集成点与 UFOClient 的集成Computer由UFOClient创建并管理客户端把CommandRouter挂在自身接收到服务端下发的Command列表后统一路由执行# UFOClient 内部 self.command_router CommandRouter(computer_manager) # 执行来自服务端的命令 results await self.command_router.execute( agent_nameself.agent_name, process_nameself.process_name, root_nameself.root_name, commandscommand_list )与 MCPServerManager 的集成Computer的服务器生命周期完全委托给MCPServerManager。create_or_get_server遵循「存在即复用不存在即创建」的策略namespace已注册则直接返回既有实例resetTrue时先调用reset()再重建mcp_server_manager.pymcp_server self.mcp_server_manager.create_or_get_server( mcp_configserver_config, resetFalse, process_nameself._process_name )进一步阅读UFO 客户端总览 - 客户端高层架构UFO 客户端 - 命令执行编排ComputerManager - 多计算机实例管理MCP 集成 - MCP 服务器细节AIP 消息定义 - Command 与 Result 消息格式【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考