UFO³ 设备智能体（Device Agent）开发完全指南：从架构理解到 MobileAgent 实战落地

导读

本指南基于 UFO³ 官方教程系列"Creating Device Agents"整理而成，以 LinuxAgent 为参照实现，系统讲解如何在 UFO³ 中创建一个全新的设备智能体（如 MobileAgent），并将其接入 Galaxy 多设备编排系统。你将掌握设备智能体的三层架构、服务端五件套（Agent Class / Processor / State Manager / Strategy / Prompter）的实现方法、平台专属 MCP Server 的开发范式、设备客户端的 WebSocket 接入流程，以及 third_party.yaml、devices.yaml、提示词模板等配置与部署全链路，最终能够独立完成一个可运行、可测试、可接入 Galaxy 的设备智能体。

---

一、什么是设备智能体（Device Agent）

设备智能体（Device Agent） 是 UFO³ 中一类专门控制并自动化特定计算平台的 AI 智能体。与只扩展某项具体功能的第三方智能体（Third-Party Agent）不同，设备智能体代表的是一个完整的计算平台，具有以下四个核心要素：

要素	说明
执行环境（Execution Environment）	设备特有的操作系统、运行时与 API
控制机制（Control Mechanism）	UI 自动化、CLI 命令或平台 API
通信协议（Communication Protocol）	基于 WebSocket 的客户端-服务端架构
MCP 集成（MCP Integration）	设备专属的 MCP Server 用于执行命令

设备智能体 vs 第三方智能体

维度	设备智能体	第三方智能体
范围	全平台控制（Windows、Linux、移动端）	特定功能（硬件、Web）
架构	客户端-服务端分离	运行在编排器服务端
通信	WebSocket + AIP 协议	直接方法调用
MCP 服务	平台专属 MCP Server	共享 MCP Server
示例	WindowsAgent、LinuxAgent、MobileAgent	HardwareAgent、WebAgent
部署形态	设备上的独立客户端进程	编排器的一部分

何时该创建设备智能体

需要创建设备智能体的场景：

• 控制一个全新平台（移动端、IoT、嵌入式设备）
• 在远程或分布式设备上执行任务
• 需要接入 Galaxy 多设备编排系统
• 出于安全或扩展性考虑需要隔离执行环境

需要创建第三方智能体的场景：

• 在已有平台上扩展新能力
• 添加专用工具或 API
• 与现有智能体并行运行

---

二、前置条件

知识要求

• Python 3.10+：具备中级 Python 编程能力
• 异步编程：理解 async/await 模式
• UFO³ 基础：熟悉 Agent 架构
• MCP 协议：理解 Model Context Protocol
• WebSocket：掌握基本通信知识

推荐阅读

优先级	主题	文档
🥇	Agent 架构总览	Infrastructure/Agents
🥇	LinuxAgent 快速上手	Quick Start: Linux
🥈	服务端-客户端架构	Server Overview、Client Overview
🥈	MCP 集成	MCP Overview
🥉	AIP 协议	AIP Protocol

开发环境准备

git clone https://gitcode.com/GitHub_Trending/uf/UFO
cd UFO

# 安装依赖
pip install -r requirements.txt

# 验证安装
python -c "import ufo; print('UFO³ installed successfully')"

---

三、设备智能体的三层架构

UFO³ 中所有设备智能体遵循统一的三层架构，与仓库 Agent 架构文档 中的描述一致：

1. 状态层（Level-1：FSM）：有限状态机控制智能体生命周期（AgentState、State Machine、State Transitions）
2. 策略层（Level-2：执行逻辑）：由模块化策略组成的处理流水线（ProcessorTemplate → DATA_COLLECTION → LLM_INTERACTION → ACTION_EXECUTION → MEMORY_UPDATE）
3. 命令层（Level-3：系统接口）：通过 MCP 执行的原子级系统操作（CommandDispatcher → MCP Tools → Device Commands）

状态层通过"委托"（delegates to）方式调用策略层，策略层最终通过命令层执行具体动作，形成完整的"感知-推理-执行-记忆"闭环。

服务端-客户端分离架构

设备智能体采用服务端-客户端分离设计，兼顾安全性与可扩展性：

组件	位置	职责	安全模型
Agent Server	编排器	推理、规划、状态管理	不可信（由 LLM 驱动）
Device Client	目标设备	命令执行、资源访问	可信（经过校验的操作）
AIP Protocol	网络层	消息传输、序列化	加密通道

分离带来的收益：

• 安全：隔离 LLM 推理与系统级执行
• 可扩展：单一编排器可管理多台设备
• 灵活：客户端可运行在资源受限设备上（移动端、IoT）
• 安全：客户端在执行所有命令前进行校验

---

四、LinuxAgent：参照实现剖析

为什么选择 LinuxAgent 作为参照

• 架构简单：单层智能体（无 HostAgent 委派）
• 边界清晰：服务端-客户端职责划分干净
• 文档完善：代码与文档配套齐全
• 生产就绪：在真实部署中得到验证
• 复杂度适中：聚焦设备智能体的核心模式

LinuxAgent 组件与文件位置

组件	文件路径	用途
Agent Class	customized_agent.py	LinuxAgent 定义
Processor	customized_agent_processor.py	LinuxAgentProcessor
Strategies	linux_agent_strategy.py	LLM 与动作策略
States	linux_agent_state.py	状态机状态
Prompter	linux_agent_prompter.py	提示词构建
Client	client.py	设备客户端入口
MCP Server	http_servers	命令执行

从仓库源码可以看到，LinuxAgent 通过 @AgentRegistry.register(agent_name="LinuxAgent", third_party=True, processor_cls=LinuxAgentProcessor) 完成注册（见 customized_agent.py），LinuxAgentProcessor 在 customized_agent_processor.py 中定义，状态机相关类（LinuxAgentStatus、LinuxAgentStateManager、ContinueLinuxAgentState 等）均位于 linux_agent_state.py。这套"注册 + 处理器 + 状态"的组织方式就是新设备智能体需要复刻的标准骨架。

一次完整的执行流程

以"列出 /tmp 目录文件"为例，完整的调用链如下：

1. 用户请求 → LinuxAgent Server 接收请求
2. 状态机 → 激活 ContinueLinuxAgentState
3. 处理器 → 执行 LinuxAgentProcessor 的各阶段策略
4. LLM 交互 → 生成 shell 命令 execute_command("ls -la /tmp")
5. 动作执行 → 通过 AIP 协议将命令发送给客户端
6. MCP 执行 → 客户端调用 Linux MCP Server 的 execute_command 工具，在 Bash 中执行命令
7. 结果处理 → 服务端接收 stdout/stderr/exit_code，更新记忆与黑板（Blackboard）
8. 状态转移 → 进入 FINISH 状态，任务完成

---

五、Part 1：服务端核心组件实现（Core Components）

这是整个教程中内容最完整、最核心的部分，将指导你实现服务端的五个组件。

Step 1：Agent Class

Agent Class 是设备智能体的入口，需要满足：

• 继承自 CustomizedAgent（其父类是 AppAgent）
• 通过 AgentRegistry 注册以实现自动发现
• 初始化 prompter 与默认状态
• 维护黑板（Blackboard）用于多智能体协作

LinuxAgent 的参考实现（核心骨架）：

# ufo/agents/agent/customized_agent.py

from ufo.agents.agent.app_agent import AppAgent
from ufo.agents.agent.basic import AgentRegistry
from ufo.agents.memory.blackboard import Blackboard
from ufo.agents.processors.customized.customized_agent_processor import (
    LinuxAgentProcessor,
)
from ufo.agents.states.linux_agent_state import ContinueLinuxAgentState
from ufo.prompter.customized.linux_agent_prompter import LinuxAgentPrompter


@AgentRegistry.register(
    agent_name="LinuxAgent",      # 唯一标识
    third_party=True,             # 标记为第三方/设备智能体
    processor_cls=LinuxAgentProcessor  # 关联处理器类
)
class LinuxAgent(CustomizedAgent):
    def __init__(self, name: str, main_prompt: str, example_prompt: str) -> None:
        # 非 GUI 智能体：process_name/app_root_name 传 None
        super().__init__(
            name=name,
            main_prompt=main_prompt,
            example_prompt=example_prompt,
            process_name=None,      # Linux 无 Windows 进程概念
            app_root_name=None,     # Linux 无 Windows 应用概念
            is_visual=None,         # CLI 智能体通常为 False
        )
        self._blackboard = Blackboard()
        self.set_state(self.default_state)
        self._context_provision_executed = False

    def get_prompter(self, is_visual, main_prompt, example_prompt) -> LinuxAgentPrompter:
        return LinuxAgentPrompter(main_prompt, example_prompt)

    @property
    def default_state(self) -> ContinueLinuxAgentState:
        return ContinueLinuxAgentState()

    @property
    def blackboard(self) -> Blackboard:
        return self._blackboard

创建 MobileAgent 时，需要针对移动平台做差异化：

维度	LinuxAgent	MobileAgent
is_visual	None（无截图）	True（需要 UI 截图）
平台追踪	不需要	self._platform 存储 "android"/"ios"
处理器	LinuxAgentProcessor	MobileAgentProcessor
Prompter	LinuxAgentPrompter	MobileAgentPrompter
默认状态	ContinueLinuxAgentState	ContinueMobileAgentState

Agent Class 最佳实践：始终先调用 super().__init__()；为多智能体协作初始化黑板；使用截图则设置 is_visual=True；用有意义的日志信息辅助调试；将平台特有元数据保存为属性；初始化逻辑保持精简（委托给处理器）。

Step 2：Processor

Processor 通过模块化策略编排执行流水线，负责：

• 管理四个阶段的策略执行
• 配置中间件（日志、错误处理、指标）
• 校验策略依赖
• 终结处理上下文

四个处理阶段（ProcessingPhase）：

DATA_COLLECTION(数据采集：截图、UI 树) → LLM_INTERACTION(提示词→LLM→响应)
→ ACTION_EXECUTION(执行命令) → MEMORY_UPDATE(更新上下文)

LinuxAgentProcessor 的参考实现：

# ufo/agents/processors/customized/customized_agent_processor.py

class LinuxAgentProcessor(CustomizedProcessor):
    def _setup_strategies(self) -> None:
        # LinuxAgent 不依赖截图，因此没有 DATA_COLLECTION 阶段
        self.strategies[ProcessingPhase.LLM_INTERACTION] = (
            LinuxLLMInteractionStrategy(fail_fast=True)   # LLM 失败应中止
        )
        self.strategies[ProcessingPhase.ACTION_EXECUTION] = (
            LinuxActionExecutionStrategy(fail_fast=False) # 动作失败继续
        )
        self.strategies[ProcessingPhase.MEMORY_UPDATE] = (
            AppMemoryUpdateStrategy(fail_fast=False)      # 记忆失败不阻断
        )

    def _setup_middleware(self) -> None:
        self.middleware_chain = [LinuxLoggingMiddleware()]

    def _finalize_processing_context(self, processing_context) -> None:
        super()._finalize_processing_context(processing_context)
        try:
            result = processing_context.get_local("result")
            if result:
                self.global_context.set(ContextNames.ROUND_RESULT, result)
        except Exception as e:
            self.logger.warning(f"Failed to update ContextNames from results: {e}")

MobileAgentProcessor 则需要在 DATA_COLLECTION 阶段组合多个采集策略：

# ufo/agents/processors/customized/customized_agent_processor.py

class MobileAgentProcessor(CustomizedProcessor):
    def _setup_strategies(self) -> None:
        self.strategies[ProcessingPhase.DATA_COLLECTION] = ComposedStrategy(
            strategies=[
                MobileScreenshotCaptureStrategy(fail_fast=True),   # 截图失败必须停
                MobileAppsCollectionStrategy(fail_fast=False),    # 应用列表失败可继续
                MobileControlsCollectionStrategy(fail_fast=False),# 控件列表失败可继续
            ],
            name="MobileDataCollectionStrategy",
            fail_fast=True,
        )
        self.strategies[ProcessingPhase.LLM_INTERACTION] = (
            MobileLLMInteractionStrategy(fail_fast=True)
        )
        self.strategies[ProcessingPhase.ACTION_EXECUTION] = (
            MobileActionExecutionStrategy(fail_fast=False)
        )
        self.strategies[ProcessingPhase.MEMORY_UPDATE] = (
            AppMemoryUpdateStrategy(fail_fast=False)
        )

各阶段配置指引：

阶段	是否必需	使用场景	示例策略
DATA_COLLECTION	可选	智能体需要观察（截图、传感器数据）	CustomizedScreenshotCaptureStrategy
LLM_INTERACTION	必需	所有智能体都需要 LLM 推理	LinuxLLMInteractionStrategy、MobileLLMInteractionStrategy
ACTION_EXECUTION	必需	所有智能体都需要命令执行	LinuxActionExecutionStrategy、MobileActionExecutionStrategy
MEMORY_UPDATE	推荐	追踪智能体历史与上下文	AppMemoryUpdateStrategy

常见误区：不要跳过 _setup_strategies()（处理器将无法执行）；不要对所有策略都用 fail_fast=True（智能体会变得脆弱）；不要忘记调用 super()._finalize_processing_context()（否则上下文无法传播）。正确的做法是：LLM_INTERACTION 用 fail_fast=True 确保拿到有效响应，ACTION_EXECUTION 用 fail_fast=False 以支持重试逻辑，并添加自定义中间件辅助调试与日志。

Step 3：State Manager（状态机）

State Manager 实现控制智能体生命周期的有限状态机（FSM），定义了状态、转移规则与状态处理器。

LinuxAgent 的状态图：

[*] → CONTINUE → CONTINUE（处理中）
       CONTINUE → FINISH（任务完成）
       CONTINUE → FAIL（发生错误）
       FAIL → FINISH（终态）
       FINISH → [*]

状态基类与三个核心状态类的关键实现（见 linux_agent_state.py）：

class LinuxAgentStatus(Enum):
    FINISH = "FINISH"
    CONTINUE = "CONTINUE"
    FAIL = "FAIL"


class LinuxAgentStateManager(AgentStateManager):
    _state_mapping: Dict[str, Type[LinuxAgentState]] = {}

    @property
    def none_state(self) -> AgentState:
        return NoneLinuxAgentState()


class LinuxAgentState(AgentState):
    def next_agent(self, agent) -> "LinuxAgent":
        return agent  # 设备智能体通常不委派，返回自身

    def next_state(self, agent) -> LinuxAgentState:
        status = agent.status
        return LinuxAgentStateManager().get_state(status)


@LinuxAgentStateManager.register
class ContinueLinuxAgentState(LinuxAgentState):
    async def handle(self, agent, context=None) -> None:
        await agent.process(context)  # 执行处理器策略

    @classmethod
    def name(cls) -> str:
        return LinuxAgentStatus.CONTINUE.value


@LinuxAgentStateManager.register
class FinishLinuxAgentState(LinuxAgentState):
    def next_state(self, agent) -> LinuxAgentState:
        return FinishLinuxAgentState()  # 终态，保持不动

    def is_subtask_end(self) -> bool:
        return True

    def is_round_end(self) -> bool:
        return True

    @classmethod
    def name(cls) -> str:
        return LinuxAgentStatus.FINISH.value


@LinuxAgentStateManager.register
class FailLinuxAgentState(LinuxAgentState):
    def next_state(self, agent) -> LinuxAgentState:
        return FinishLinuxAgentState()  # 失败后优雅收尾

    def is_round_end(self) -> bool:
        return True

    @classmethod
    def name(cls) -> str:
        return LinuxAgentStatus.FAIL.value

MobileAgent 的状态集在 LinuxAgent 基础上增加了一个 WAITING 状态，用于等待应用加载或动画完成（休眠 2 秒后切回 CONTINUE），这体现了移动 UI 自动化的特殊需求。

状态设计指引：

状态	使用时机	必需方法	是否为终态
CONTINUE	正常执行	handle() 调用 agent.process()	否
FINISH	任务完成	is_subtask_end() → True	是
FAIL	发生错误	next_state() → FINISH	是
WAITING	异步等待	handle() 中 await asyncio.sleep()	否
NONE	默认/初始	next_state() → FINISH	是

状态设计最佳实践：始终用 @StateManager.register 注册状态；name() 必须与状态枚举值一致；在 CONTINUE.handle() 中调用 agent.process()；终态将 is_round_end() 设为 True；FAIL 转移到 FINISH 实现优雅终止；不要创建过多状态（保持简单）；不要直接调用处理器（应通过 agent.process()）。

Step 4：Processing Strategies

策略（Strategy） 是执行流水线中实现特定阶段的模块化单元，每个策略：

• 在所属阶段内独立执行
• 用 @depends_on 装饰器声明依赖
• 用 @provides 装饰器声明产出
• 返回带成功/失败状态的 ProcessingResult

LLM 交互策略（以 LinuxAgent 为例）：

# ufo/agents/processors/strategies/linux_agent_strategy.py

@depends_on("request")
@provides(
    "parsed_response",   # LLM 解析后的响应
    "response_text",     # LLM 原始响应文本
    "llm_cost",          # LLM API 成本
    "prompt_message",    # 发送给 LLM 的提示词
    "action",            # 待执行的动作
    "thought",           # LLM 推理
    "comment",           # LLM 注释
)
class LinuxLLMInteractionStrategy(AppLLMInteractionStrategy):
    def __init__(self, fail_fast: bool = True) -> None:
        super().__init__(fail_fast=fail_fast)

    async def execute(self, agent, context) -> ProcessingResult:
        try:
            # 1. 从上下文提取请求与历史计划
            request = context.get("request")
            plan = self._get_prev_plan(agent)

            # 2. 构建完整提示词（含黑板上下文、历史成功动作）
            prompt_message = agent.message_constructor(
                dynamic_examples=[],
                dynamic_knowledge="",
                plan=plan,
                request=request,
                blackboard_prompt=(
                    agent.blackboard.blackboard_to_prompt()
                    if not agent.blackboard.is_empty() else []
                ),
                last_success_actions=self._get_last_success_actions(agent),
            )

            # 3. 调用 LLM
            response_text, llm_cost = await self._get_llm_response(agent, prompt_message)

            # 4. 解析并校验响应
            parsed_response = self._parse_app_response(agent, response_text)

            # 5. 提取结构化数据并返回
            structured_data = parsed_response.model_dump()
            return ProcessingResult(
                success=True,
                data={
                    "parsed_response": parsed_response,
                    "response_text": response_text,
                    "llm_cost": llm_cost,
                    "prompt_message": prompt_message,
                    **structured_data,
                },
                phase=ProcessingPhase.LLM_INTERACTION,
            )
        except Exception as e:
            self.logger.error(f"Linux LLM interaction failed: {str(e)}")
            return self.handle_error(e, ProcessingPhase.LLM_INTERACTION, context)

动作执行策略（调用 Command Dispatcher 将动作发给 MCP Server）：

@depends_on("parsed_response", "command_dispatcher")
@provides("execution_result", "action_info", "control_log", "status")
class LinuxActionExecutionStrategy(AppActionExecutionStrategy):
    def __init__(self, fail_fast: bool = False) -> None:
        super().__init__(fail_fast=fail_fast)

    async def execute(self, agent, context) -> ProcessingResult:
        try:
            parsed_response = context.get_local("parsed_response")
            command_dispatcher = context.global_context.command_dispatcher
            if not parsed_response:
                return ProcessingResult(success=True, data={"message": "No response for action execution"},
                                        phase=ProcessingPhase.ACTION_EXECUTION)

            execution_results = await self._execute_app_action(
                command_dispatcher, parsed_response.action
            )
            actions = self._create_action_info(parsed_response.action, execution_results)
            action_info = ListActionCommandInfo(actions)
            action_info.color_print()
            control_log = action_info.get_target_info()
            status = (parsed_response.action.status
                      if isinstance(parsed_response.action, ActionCommandInfo)
                      else action_info.status)
            return ProcessingResult(
                success=True,
                data={"execution_result": execution_results, "action_info": action_info,
                      "control_log": control_log, "status": status},
                phase=ProcessingPhase.ACTION_EXECUTION,
            )
        except Exception as e:
            self.logger.error(f"Linux action execution failed: {str(e)}")
            return self.handle_error(e, ProcessingPhase.ACTION_EXECUTION, context)

MobileAgent 的 MobileLLMInteractionStrategy 与 LinuxAgent 的差别在于：依赖 request、screenshot、ui_tree 三个上下文键，提示词构建时额外注入截图与 UI 树；MobileActionExecutionStrategy 则执行 tap、swipe、type 等移动专属动作。同时可定义 MobileLoggingMiddleware（继承自 AppAgentLoggingMiddleware）输出移动任务的起始日志。

策略实现检查清单：

• [ ] 用 @depends_on() 声明依赖
• [ ] 用 @provides() 声明产出
• [ ] 返回带成功状态的 ProcessingResult
• [ ] 优雅处理异常（记录日志、返回错误结果）
• [ ] 遵循 fail_fast 配置
• [ ] 使用 self.logger 调试
• [ ] 在 except 块中调用 self.handle_error()

Step 5：Prompter

Prompter 负责构建发给 LLM 的提示词，功能包括：

• 从 YAML 文件加载提示词模板
• 构建 system 与 user 消息
• 插入动态上下文（请求、计划、示例）
• 格式化 API/工具描述

LinuxAgentPrompter 参考实现（见 linux_agent_prompter.py）：

class LinuxAgentPrompter(AppAgentPrompter):
    def __init__(self, prompt_template: str, example_prompt_template: str):
        super().__init__(None, prompt_template, example_prompt_template)
        self.api_prompt_template = None

    def system_prompt_construction(self, additional_examples: List[str] = []) -> str:
        apis = self.api_prompt_helper(verbose=1)          # 格式化 API 描述
        examples = self.examples_prompt_helper(additional_examples=additional_examples)
        return self.prompt_template["system"].format(apis=apis, examples=examples)

    def user_prompt_construction(
        self, prev_plan, user_request, retrieved_docs="", last_success_actions=[]
    ) -> str:
        return self.prompt_template["user"].format(
            prev_plan=json.dumps(prev_plan),
            user_request=user_request,
            retrieved_docs=retrieved_docs,
            last_success_actions=json.dumps(last_success_actions),
        )

    def user_content_construction(self, prev_plan, user_request, retrieved_docs="", last_success_actions=[]):
        return [{
            "type": "text",
            "text": self.user_prompt_construction(prev_plan, user_request, retrieved_docs, last_success_actions),
        }]

MobileAgentPrompter 的关键差异在于支持多模态：user_content_construction() 在文本之外追加 image_url 类型的截图内容（base64 编码），从而让视觉大模型"看见"手机屏幕：

# 多模态用户内容：文本 + 截图
user_content = [{"type": "text", "text": self.user_prompt_construction(...)}]
if screenshot:
    user_content.append({
        "type": "image_url",
        "image_url": {"url": f"data:image/png;base64,{screenshot}"},
    })

Prompter 最佳实践：继承 AppAgentPrompter 复用标准结构；实现 system_prompt_construction() 与 user_prompt_construction()；多模态场景用 user_content_construction()；用 examples_prompt_helper() / api_prompt_helper() 格式化示例与 API；不要把提示词硬编码在代码里，一律使用 YAML 模板。

服务端组件测试

单元测试（Agent Class）：

import pytest
from ufo.agents.agent.customized_agent import MobileAgent
from ufo.agents.processors.customized.customized_agent_processor import MobileAgentProcessor

class TestMobileAgent:
    @pytest.fixture
    def agent(self):
        return MobileAgent(
            name="test_mobile_agent",
            main_prompt="ufo/prompts/third_party/mobile_agent.yaml",
            example_prompt="ufo/prompts/third_party/mobile_agent_example.yaml",
            platform="android",
        )

    def test_agent_initialization(self, agent):
        assert agent.name == "test_mobile_agent"
        assert agent.platform == "android"
        assert agent.prompter is not None
        assert agent.blackboard is not None

    def test_default_state(self, agent):
        from ufo.agents.states.mobile_agent_state import ContinueMobileAgentState
        assert isinstance(agent.default_state, ContinueMobileAgentState)

集成测试（完整流水线）：创建 Context 并设置 request，然后实例化处理器并 await processor.process()，断言 parsed_response 已写入本地上下文。

---

六、Part 2：平台专属 MCP Server 开发

MCP Server 在设备智能体架构中的位置

MCP Server 是在目标设备上（或附近）运行的服务，将平台专属工具以可调用函数的形式暴露给 LLM 智能体：

• 与 Agent Server 分开运行（安全隔离）
• 工具是暴露给 LLM 的原子操作
• Command Dispatcher 负责把 LLM 动作翻译成 MCP 工具调用
• 结果沿原路径回传

设计原则

原则	说明	示例
原子工具	每个工具只做一件事	execute_command 而非 execute_and_parse_command
类型安全	用 Pydantic Field 做类型注解	Annotated[str, Field(description="...")]
错误韧性	优雅处理所有异常	try/except 返回结构化错误
安全优先	校验与净化所有输入	拦截危险命令、校验路径
平台无关	抽象平台差异	Linux 用 subprocess、Android 用 ADB
异步执行	用 asyncio 保证非阻塞	async def、await subprocess

LinuxAgent MCP Server 分析

参考实现位于 ufo/client/mcp/http_servers/linux_mcp_server.py，核心是 create_bash_mcp_server()：

def create_bash_mcp_server(host: str = "", port: int = 8010) -> None:
    mcp = FastMCP(
        "Linux Bash MCP Server",
        instructions="MCP server for executing shell commands on Linux.",
        stateless_http=False,  # 跨请求保持状态
        json_response=True,    # 返回 JSON 响应
        host=host,
        port=port,
    )

    @mcp.tool()
    async def execute_command(
        command: Annotated[str, Field(description="Shell command to execute...")],
        timeout: Annotated[int, Field(description="Max execution time in seconds...")] = 30,
        cwd: Annotated[Optional[str], Field(description="Working directory...")] = None,
    ) -> Annotated[Dict[str, Any], Field(description="Result dict with 'success', 'exit_code', 'stdout', 'stderr'")]:
        # 安全：拦截危险命令
        dangerous = ["rm -rf /", ":(){ :|:& };:", "mkfs", "dd if=/dev/zero", "shutdown", "reboot"]
        if any(d in command.lower() for d in dangerous):
            return {"success": False, "error": "Blocked dangerous command."}

        try:
            proc = await asyncio.create_subprocess_shell(
                command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=cwd,
            )
            try:
                stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
            except asyncio.TimeoutError:
                proc.kill(); await proc.wait()
                return {"success": False, "error": f"Timeout after {timeout}s."}
            return {
                "success": proc.returncode == 0,
                "exit_code": proc.returncode,
                "stdout": stdout.decode("utf-8", errors="replace"),
                "stderr": stderr.decode("utf-8", errors="replace"),
            }
        except Exception as e:
            return {"success": False, "error": str(e)}

    mcp.run(transport="streamable-http")

四个关键设计模式：

1. 类型安全的工具定义：Annotated[T, Field(description=...)] 让 LLM 理解参数类型与含义，Pydantic 提供运行时校验，并自动生成 API 文档。
2. 安全优先的校验：先黑名单拦截危险模式，再白名单化安全操作，同时校验路径防止目录穿越——不要只依赖净化。
3. 异步执行 + 超时：asyncio.wait_for 强制超时，超时后 proc.kill() 清理进程；服务保持响应，支持并发工具调用。
4. 结构化错误处理：永远返回 dict 而非向 LLM 抛异常；包含 success 布尔字段、详细错误信息、保留 stdout/stderr 便于调试。

创建 MobileAgent MCP Server

完整文件建议置于 ufo/client/mcp/http_servers/mobile_mcp_server.py，提供 6 个移动自动化工具：

工具	参数	平台实现
tap_screen	x、y、duration_ms(默认100)	Android: adb shell input tap；iOS: xcrun simctl io booted tap
swipe	start_x/y、end_x/y、duration_ms(默认300)	Android: adb shell input swipe；iOS: 模拟（真机需 XCUITest）
type_text	text、clear_first(默认False)	Android: adb shell input text（转义特殊字符）；iOS: xcrun simctl io booted text
capture_screenshot	save_path	Android: adb exec-out screencap -p；iOS: xcrun simctl io booted screenshot
get_ui_tree	format(xml/json, 默认xml)	Android: uiautomator dump；iOS: 需 XCUITest/Appium
launch_app	package_name	Android: adb shell monkey -p <pkg> ... 1；iOS: xcrun simctl launch booted <bundle>

服务入口提供 --port（默认 8020）、--host（默认 localhost）、--platform（android/ios，默认 android）三个 CLI 参数。

工具定义最佳实践

• 命名要具描述性：do_thing → tap_screen，get → get_ui_tree
• 返回格式统一：{"success": bool, "message": str, "error": str, "data": Any}，不要混用返回类型（return True 或抛异常都是反模式）
• 编写完整 docstring：说明平台支持、参数含义、返回值与示例（如 >>> await swipe(100, 500, 100, 100) # Swipe up）

输入校验与超时策略

坐标校验、边界检查示例：

@mcp.tool()
async def tap_screen(x: int, y: int) -> Dict[str, Any]:
    if x < 0 or y < 0:
        return {"success": False, "error": f"Invalid coordinates: ({x}, {y}). Must be non-negative."}
    max_x, max_y = 1080, 1920  # 示例分辨率
    if x > max_x or y > max_y:
        return {"success": False, "error": f"Coordinates out of bounds. Screen: {max_x}x{max_y}"}
    try:
        result = subprocess.run([...], timeout=5)
        return {"success": result.returncode == 0}
    except subprocess.TimeoutExpired:
        return {"success": False, "error": "Tap command timeout"}
    except Exception as e:
        return {"success": False, "error": f"Unexpected error: {str(e)}"}

应用包名校验（防止注入非法包名）：

import re
android_pattern = r'^[a-z][a-z0-9_]*(\.[a-z][a-z0-9_]*)+$'   # com.example.app
ios_pattern     = r'^[a-zA-Z][a-zA-Z0-9_]*(\.[a-zA-Z][a-zA-Z0-9_]*)+$'  # com.example.App

三种超时策略：① 命令级 subprocess.run(..., timeout=5)；② 异步超时 + proc.kill() 清理；③ 指数退避重试（await asyncio.sleep(2 ** attempt)）。

MCP Server 测试

• 单元测试：@patch('subprocess.run') mock 执行结果，断言成功/失败路径；测试非法坐标（x=-10）与超时（subprocess.TimeoutExpired）。
• 集成测试：用 HTTPMCPServer 启动真实服务，requests.get(f"{server}/health") 断言 200，POST /execute 调用工具。
• 手动测试：

# 启动 MCP Server
python -m ufo.client.mcp.http_servers.mobile_mcp_server \
  --host localhost --port 8020 --platform android

# curl 调用工具
curl -X POST http://localhost:8020/mcp \
  -H "Content-Type: application/json" \
  -d '{"tool": "tap_screen", "parameters": {"x": 500, "y": 1000}}'

---

七、Part 3：设备客户端设置（Client Setup）

客户端角色

设备客户端运行在目标设备上，负责执行命令并与 Agent Server 通过 WebSocket 通信。核心组件：

组件	职责	关键方法
Entry Point（client.py）	解析参数、初始化服务	main()
UFOClient（ufo_client.py）	执行命令、路由动作	execute_step()、execute_actions()
WebSocketClient（websocket.py）	双向通信	connect_and_listen()、handle_messages()
MCP Server Manager	启停 MCP Server	register_server()、start_server()
Command Router	将命令路由到 MCP 工具	execute()

UFOClient 核心逻辑

# ufo/client/ufo_client.py

class UFOClient:
    def __init__(self, mcp_server_manager, computer_manager,
                 client_id=None, platform=None):
        self.mcp_server_manager = mcp_server_manager
        self.computer_manager = computer_manager
        self.command_router = CommandRouter(computer_manager=computer_manager)
        self.task_lock = asyncio.Lock()  # 线程安全
        self.client_id = client_id or "client_001"
        self.platform = platform

    async def execute_step(self, response: ServerMessage) -> List[Result]:
        # 更新智能体上下文
        self.agent_name = response.agent_name
        self.process_name = response.process_name
        self.root_name = response.root_name
        return await self.execute_actions(response.actions)

    async def execute_actions(self, commands):
        if commands:
            return await self.command_router.execute(
                agent_name=self.agent_name,
                process_name=self.process_name,
                root_name=self.root_name,
                commands=commands,
            )
        return []

UFOClient 通过属性 setter（session_id、agent_name、process_name、root_name）维护会话上下文，并附带类型校验（如 session_id 必须为 str 或 None）。

WebSocket 客户端与 AIP 协议

UFOWebSocketClient 使用 AIP（Agent Interaction Protocol）完成结构化通信，连接时初始化四个协议实例：WebSocketTransport、RegistrationProtocol、HeartbeatProtocol、TaskExecutionProtocol。

async def connect_and_listen(self):
    while True:
        try:
            if self.retry_count >= self.max_retries:
                self.logger.error(f"Max retries ({self.max_retries}) reached")
                break
            async with websockets.connect(
                self.ws_url,
                ping_interval=20, ping_timeout=180, close_timeout=10,
                max_size=100 * 1024 * 1024,  # 100MB 最大消息
            ) as ws:
                self._ws = ws
                self.transport = WebSocketTransport(ws)
                self.registration_protocol = RegistrationProtocol(self.transport)
                self.heartbeat_protocol = HeartbeatProtocol(self.transport)
                self.task_protocol = TaskExecutionProtocol(self.transport)
                await self.register_client()
                self.retry_count = 0
                await self.handle_messages()
        except (websockets.ConnectionClosed, asyncio.TimeoutError) as e:
            self.retry_count += 1
            await self._maybe_retry()   # 指数退避：2 ** retry_count 秒

消息处理循环：handle_messages() 根据 ServerMessageType 分发——TASK_REQUEST 调用 ufo_client.execute_step() 后经 task_protocol.send_result() 回传结果；HEARTBEAT 返回 heartbeat_ack；RESULT_ACK 仅记录日志。

完整通信时序：

1. REGISTER（client_id + system_info + platform + client_version）→ REGISTER_ACK
2. 服务端周期性发送 HEARTBEAT → 客户端回复 HEARTBEAT_ACK
3. 服务端发送 TASK_REQUEST → 客户端 execute_step() → Command Router 调用 MCP Server → 返回 TASK_RESULT
4. 服务端回 RESULT_ACK，进入下一轮循环

注册时通过 DeviceInfoProvider.collect_system_info() 收集设备系统信息（见 device_info_provider.py），并附带 platform 与 client_version 元数据。

MCP Server Manager

MCPServerManager 支持三种服务类型（见 ufo/client/mcp/mcp_server_manager.py）：

类型	说明	配置关键字段
HTTPMCPServer	HTTP 远程服务（设备智能体最常见）	host、port、path（默认 /mcp）
LocalMCPServer	进程内注册表服务	namespace
StdioMCPServer	子进程标准 I/O 服务	command、start_args

平台注册示例：Android → HTTP 8020 /mcp（namespace mobile）；Linux → HTTP 8010 /mcp（namespace linux）；Windows → Local（namespace windows）。

平台检测与客户端入口

# ufo/client/client.py 中的平台自动检测
if args.platform is None:
    detected = platform_module.system().lower()
    args.platform = detected if detected in ["windows", "linux"] else "windows"

client.py 的 CLI 参数：--client-id（默认 client_001）、--ws-server（默认 ws://localhost:5000/ws）、--ws（启用 WebSocket 模式，必需）、--max-retries（默认 5）、--platform（windows|linux|android|ios，默认自动检测）、--log-level（默认 WARNING）。

部署命令示例：

# Android 移动客户端
python -m ufo.client.client \
  --ws \
  --ws-server ws://192.168.1.100:5010/ws \
  --client-id mobile_agent_1 \
  --platform android \
  --log-level INFO

# Linux 客户端
python -m ufo.client.client \
  --ws \
  --ws-server ws://localhost:5001/ws \
  --client-id linux_agent_1 \
  --platform linux \
  --max-retries 10

可选配置文件 config/client_config.yaml 可集中管理 client、websocket、logging、mcp_servers 四段配置。

---

八、Part 4：配置与部署（Configuration & Deployment）

配置文件层级

文件	用途	是否必需
config/ufo/third_party.yaml	向 UFO 注册智能体	✅ 是
config/galaxy/devices.yaml	注册设备实例	✅ 是（Galaxy 场景）
config/galaxy/constellation.yaml	多设备编排	可选
ufo/prompts/third_party/<name>.yaml	主提示词模板	✅ 是
ufo/prompts/third_party/<name>_example.yaml	少样本示例	✅ 是

third_party.yaml：智能体注册

# config/ufo/third_party.yaml

ENABLED_THIRD_PARTY_AGENTS: ["MobileAgent", "LinuxAgent"]

THIRD_PARTY_AGENT_CONFIG:
  MobileAgent:
    VISUAL_MODE: True
    AGENT_NAME: "MobileAgent"
    APPAGENT_PROMPT: "ufo/prompts/third_party/mobile_agent.yaml"
    APPAGENT_EXAMPLE_PROMPT: "ufo/prompts/third_party/mobile_agent_example.yaml"
    # API_PROMPT: "ufo/prompts/third_party/mobile_agent_api.yaml"  # 可选
    INTRODUCTION: >
      The MobileAgent controls Android and iOS mobile devices.
      It can perform UI automation, tap/swipe gestures, type text,
      launch apps, and capture screenshots. Use it for mobile
      app testing, automation, and device control tasks.

  LinuxAgent:
    VISUAL_MODE: False
    AGENT_NAME: "LinuxAgent"
    APPAGENT_PROMPT: "ufo/prompts/third_party/linux_agent.yaml"
    APPAGENT_EXAMPLE_PROMPT: "ufo/prompts/third_party/linux_agent_example.yaml"
    INTRODUCTION: >
      The LinuxAgent executes commands on Linux systems.
      It can run bash commands, manage files, inspect processes,
      configure services, and perform system administration tasks.

字段参考：

字段	类型	必需	说明	示例
VISUAL_MODE	boolean	✅	是否启用截图采集	True（移动/GUI）、False（CLI）
AGENT_NAME	string	✅	必须与 @AgentRegistry.register 一致	"MobileAgent"
APPAGENT_PROMPT	string	✅	主提示词模板路径	"ufo/prompts/third_party/mobile_agent.yaml"
APPAGENT_EXAMPLE_PROMPT	string	✅	示例提示词模板路径	"ufo/prompts/third_party/mobile_agent_example.yaml"
API_PROMPT	string	可选	自定义 API 描述	"ufo/prompts/third_party/mobile_agent_api.yaml"
INTRODUCTION	string	✅	给 HostAgent 的智能体描述	多行能力说明

仓库中的 config/ufo/third_party.yaml 真实存在并采用完全相同的 ENABLED_THIRD_PARTY_AGENTS + THIRD_PARTY_AGENT_CONFIG 结构，新增智能体时只需追加一个配置块。

devices.yaml：设备注册

# config/galaxy/devices.yaml

devices:
  - device_id: "mobile_agent_1"
    server_url: "ws://192.168.1.100:5010/ws"
    os: "android"
    capabilities:
      - "ui_automation"
      - "mobile_app_testing"
      - "touch_gestures"
      - "screenshot_capture"
    metadata:
      device_model: "Google Pixel 6"
      android_version: "14"
      screen_resolution: "1080x2400"
      tips: >
        This device runs Android 14 on Google Pixel 6.
        For app testing, use package name format: com.example.app
    auto_connect: true
    max_retries: 5

字段参考：

字段	类型	必需	说明	默认值
device_id	string	✅	设备唯一标识	—
server_url	string	✅	WebSocket 服务地址	—
os	string	✅	android/ios/linux/windows	—
capabilities	list[string]	✅	能力清单（Galaxy 任务路由依据）	—
metadata	dict	可选	自定义元数据（{tips} 可注入提示词）	—
metadata.tips	string	推荐	设备专属指令	—
auto_connect	boolean	可选	启动时自动连接	false
max_retries	integer	可选	连接重试上限	3

建议在 metadata 中包含日志路径、开发路径、日志模式等运维信息（如 LinuxAgent 的 logs_file_path、warning_log_pattern: "WARN"、error_log_pattern: "ERROR|FATAL"），这些信息对任务执行非常实用。

提示词模板创建

主提示词模板 ufo/prompts/third_party/mobile_agent.yaml 的核心结构（精简展示关键占位符与 JSON 契约）：

version: 1.0

system: |-
  You are **MobileAgent**, the UFO framework's intelligent agent for mobile device automation.
  ...
  ## Task Status
  - `CONTINUE` — the request is partially complete; further actions are required.
  - `FINISH` — the request has been successfully fulfilled; no further actions are needed.
  - `FAIL` — the request cannot be completed due to errors...

  ## Response Format
  Always respond **only** with valid JSON that strictly follows the structure below.

  Required JSON keys:
    {{
      "observation": str, "...",
      "thought": str, "...",
      "action": {{
        "function": str, "...",
        "arguments": Dict[str, Any], "...",
        "status": str, "<CONTINUE | FINISH | FAIL>"
      }},
      "plan": List[str], "...",
      "result": str, "..."
    }}

  ## Actions
  - You are able to use the following APIs to interact with the mobile device.
  {{apis}}

  ## Examples
  {{examples}}

user: |-
  <Overall User Request:> {{user_request}}
  <Current Screenshot:> [See attached image]
  <Installed Apps:> {{installed_apps}}
  <Current Screen Controls:> {{current_controls}}
  <The actions you took at the last step and their results:> {{last_success_actions}}
  <Your Next Plan:> {{prev_plan}}
  <Your response:>

示例提示词模板 mobile_agent_example.yaml 通常包含 4 个典型场景的少样本（launch app、tap、swipe、type），每个示例包含完整的 observation / thought / action / plan / result JSON 结构。

提示词模板要点：

• 用 {{variable}}（双花括号）声明模板变量
• 提供带类型注解的清晰 JSON 结构
• 包含平台专属指引（Android vs iOS）
• 示例覆盖成功与失败场景
• 强调"只输出 JSON，禁止 markdown"
• API 描述用 {{apis}} 占位符注入，不要硬编码

逐步部署（10 步）

1. 配置 third_party.yaml（nano config/ufo/third_party.yaml）
2. 注册 devices.yaml 设备（nano config/galaxy/devices.yaml）
3. 创建提示词模板（touch mobile_agent.yaml / mobile_agent_example.yaml）
4. 实现智能体组件（Agent/Processor/States/Strategies/Prompter）
5. 创建 MCP Server（touch ufo/client/mcp/http_servers/mobile_mcp_server.py）
6. 运行测试（pytest tests/unit/test_mobile_agent.py 等）
7. 启动 Agent Server（python -m ufo.server.app --port 5010）
8. 启动 MCP Server（python -m ufo.client.mcp.http_servers.mobile_mcp_server --host localhost --port 8020 --platform android）
9. 启动设备客户端（python -m ufo.client.client --ws --ws-server ws://localhost:5010/ws --client-id mobile_agent_1 --platform android --log-level INFO）
10. 验证连接（客户端日志出现 "✅ Client registered: mobile_agent_1"）

预期日志：Agent Server 输出 Registered agents: MobileAgent, LinuxAgent 与 WebSocket endpoint: ws://localhost:5010/ws；MCP Server 输出 Registered tools: tap_screen, swipe, type_text, ...；客户端输出 ✅ Client registered: mobile_agent_1。

可选验证命令：

curl -X POST http://localhost:5010/api/v1/task \
  -H "Content-Type: application/json" \
  -d '{"request": "Tap at coordinates (500, 1000)", "device_id": "mobile_agent_1"}'

Galaxy 多设备集成

constellation.yaml 配置多设备编排：

constellation:
  constellation_id: "mobile_test_constellation"
  heartbeat_interval: 30
  task_timeout: 300
  max_task_retries: 3
  retry_delay: 5
  load_balancing_strategy: "round_robin"   # round_robin | least_loaded | capability_based
  device_selection_strategy: "capability_match"  # capability_match | explicit | random

routing_rules:
  - task_type: "mobile_app_testing"
    preferred_devices: ["mobile_agent_1", "mobile_agent_2"]
    required_capabilities: ["ui_automation"]
  - task_type: "server_management"
    preferred_devices: ["linux_agent_1", "linux_agent_2"]
    required_capabilities: ["bash_commands"]

多设备启动示例：先启动 Galaxy 编排器（python -m galaxy --constellation-id mobile_test_constellation --config config/galaxy/constellation.yaml），再并行启动多个设备客户端，最后通过 python -m galaxy.client.submit_task --constellation ... --request ... --devices mobile_agent_1,mobile_agent_2 提交跨设备任务。

常见配置模式

1. 开发 vs 生产：开发环境用 auto_connect: false 便于手动调试，生产环境 auto_connect: true + 更高 max_retries（如 10）保证可靠性。
2. 多平台支持：同一个 MobileAgent 配置可对应 Android 与 iOS 两个设备注册条目（os 与 capabilities 区分）。
3. 设备池管理：同一类型多台设备（android_pool_1、android_pool_2）通过 metadata.pool、metadata.device_index 标记，供负载均衡使用。

---

九、Part 5：测试与调试

注意：该章节在官方文档中标注为"开发中"，以下为其中的临时快速指南。

基础测试模板：

import pytest
from ufo.agents.agent.customized_agent import MobileAgent

def test_agent_initialization():
    agent = MobileAgent(
        name="test_agent",
        main_prompt="ufo/prompts/third_party/mobile_agent.yaml",
        example_prompt="ufo/prompts/third_party/mobile_agent_example.yaml",
        platform="android",
    )
    assert agent.name == "test_agent"
    assert agent.platform == "android"

常见问题速查：

问题	解决方案
智能体未注册	检查 @AgentRegistry.register() 装饰器
MCP Server 无响应	确认 MCP Server 正在正确端口运行
WebSocket 连接失败	检查服务 URL 与网络连通性

更完整的测试策略可参考仓库 tests/ 目录中的设备与集成测试，以及 Part 1 中给出的单元/集成测试模板。

---

十、Part 6：完整示例与学习路径

完整 MobileAgent 实战教程（Android/iOS 平台细节、UI Automator 集成）在官方文档中标注为"开发中"，暂以 LinuxAgent 作为完整参照。规划内容涵盖：

• 平台专属设置（Android：ADB、UI Automator、无障碍服务；iOS：XCTest、Accessibility API、Instrument 工具）
• 完整实现（Agent 类、处理器与策略、状态机、带移动工具的 MCP Server、移动 UI 的 Prompter）
• 高级特性（多设备协同、应用专属自动化、错误恢复、性能优化）

---

十一、学习路径与快速导航

推荐学习路径（来自教程索引 index.md）：

阶段	内容	预计耗时	难度
Part 0：总览	理解设备智能体与架构	15 分钟	⭐
Part 1：核心组件	构建服务端组件	45 分钟	⭐⭐⭐
Part 2：MCP Server	创建平台专属 MCP Server	30 分钟	⭐⭐
Part 3：客户端设置	配置设备客户端	20 分钟	⭐⭐
Part 4：配置与部署	配置并部署智能体	25 分钟	⭐⭐
Part 5：测试与调试	测试与调试实现	30 分钟	⭐⭐⭐
Part 6：完整示例	从零实现 MobileAgent	60 分钟	⭐⭐⭐⭐

经验开发者快速起步清单：

1. 定义 @AgentRegistry.register(agent_name="MobileAgent", third_party=True, processor_cls=MobileAgentProcessor) 修饰的 Agent Class
2. 在 Processor 的 _setup_strategies() 中按四阶段组合策略（移动端需 DATA_COLLECTION）
3. 用 FastMCP 创建平台专属 MCP Server（tap_screen、swipe、type_text 等工具）
4. 在 third_party.yaml 中注册智能体
5. 在 devices.yaml 中注册设备
6. 三个终端分别启动 Agent Server、MCP Server、Device Client

扩展阅读：

• Agent 架构总览 — 三层架构深入
• Linux 快速上手 — LinuxAgent 部署指南
• 服务端概览 — 服务端编排
• 客户端概览 — 客户端执行
• MCP 概览 — Model Context Protocol
• AIP 协议 — Agent Interaction Protocol
• 创建第三方智能体教程 — 关联教程
• Galaxy 概览 — 多设备编排

---

十二、总结

通过本指南，你可以掌握 UFO³ 设备智能体的完整开发链路：

• 设备智能体控制整个平台（Windows、Linux、移动端），采用服务端-客户端分离架构将推理与执行隔离
• 三层设计（状态层 / 策略层 / 命令层）提供了模块化、可扩展的框架
• LinuxAgent 是理想的参照实现，其组件分布在 customized_agent.py、customized_agent_processor.py、linux_agent_strategy.py、linux_agent_state.py 中
• 6 部分教程覆盖设备智能体创建的全部环节
• MCP 集成实现了平台专属命令执行
• Galaxy 集成支持多设备编排

无论你是要接入移动端、IoT 还是嵌入式平台，遵循"以 LinuxAgent 为骨架、按四阶段策略编排、以 MCP 暴露平台能力"的路线，即可快速打造属于自己的设备智能体。