本文提供 Mobile Use SDK 的功能介绍、快速入门以及功能说明，便于您在指定项目中快速集成 Mobile Use SDK，无需关注 Agent 实现细节即可对 Mobile Use Agent 通过自然语言指令下发任务并获取任务执行结果。

功能介绍

功能特性

• Agentic Mobile：​支持通过自然语言和程序命令操作云手机，支持复杂规划命令。
• 云手机操作: 完整的云手机控制能力（启动应用、截图、点击以及滑动等）。
• 流式交互: 实时流式响应，支持用户中断和交互反馈。
• 结构化输出: 支持自定义 Pydantic 的 JSON 结构化数据输出。
• MCP 工具集成: 支持 MCP 工具生态扩展，自动管理 MCP 连接池，提高性能和稳定性。

准备工作

开始使用 Mobile Use SDK 之前，您必须准备好以下资源。

• 通过 VeFaaS 部署 Mobile Use Agent 应用
  使用 Mobile Use SDK 之前，您需要部署 Mobile Use Agent 应用（推荐 VeFaaS 一键部署方式）。具体操作参看 Mobile Use Agent 快速部署指南。
• 使用 SDK 所需的产品服务数据
  • 获取您的火山引擎访问密钥 Access Key ID 和 Secret Access Key（AK/SK）。

    注意

    请求云手机服务和对象存储服务的火山引擎 AK/SK 必须使用同一个，否则会造成 Agent 无响应。

  • 参考云手机快速入门步骤一和二完成云手机业务与实例的创建，并记录业务 ID（Product ID）、实例 ID（Pod ID）。

    注意

    • 创建云手机资源期间如遇问题请提交工单。
    • 创建实例时选择 AOSP13 公共镜像。
    • 海外需求需单独与云手机对接人员沟通。
  • 参考对象存储快速入门步骤一和二开通对象存储服务，完成对象存储桶 (Bucket) 创建，并进入存储桶详情页，记录存储桶名称（Bucket）、地域（Region）、地域节点（Endpoint）。
  • 参考火山方舟大模型服务平台快速入门开通模型服务，获取并记录 API Key、Model ID、Base URL。
    请选择视觉理解类模型，当前推荐选择：
    • doubao-1.5-thinking-vision-pro（Mobile Use 体验中心使用的模型，表现效果较为均衡）
    • doubao-seed-1.6-thinking（速度慢、思考时间长，适合深度图像理解的场景）
    • doubao-1.5-ui-tars（操作速度快、不稳定，适合简单直接的任务）

    注意

    由于模型迭代频繁，请在对接前咨询技术支持获取当前最推荐使用的模型名称。

• Python 开发环境
  开发环境要求 Python 版本大于等于 3.11，下载地址请参见 Python Downloads。

快速入门

1. 下载以下示例项目。
   

   

   mobile_use_example.zip

   未知大小

   

2. 下载完成后，在本地解压压缩包，并使用开发工具（例如 TRAE 等）打开项目。
   项目结构说明：

   mobile-use-examples/
   - examples/                           # 示例项目集
      - basic_usage.py                   # SDK 基本使用和设置
      - app_automation.py                # 应用启动和导航
   - .env.example                        # 环境配置模板
   - mcp.json                            # MCP 服务器配置
   - pyproject.toml                      # 项目依赖项
   - README.md                           # 项目自述文档，您可以从中获取到项目介绍以及运行指南
   

3. 打开终端（Terminal），在项目根目录依次运行以下命令，使用 uv 安装依赖。

   uv venv
   source .venv/bin/activate
   uv sync
   

   

4. 运行以下命令，新建 .env 文件。

   cp .env.example .env
   

5. 在项目根目录打开 .env 文件，根据提示依次配置各个参数，并保存。
   参数说明：

   说明

   配置前请确保已参考上文准备工作章节完成了所有的资源创建与配置。

   • ACEP_AK、TOS_AK：火山引擎账号 Access Key ID （AK）。
   • ACEP_SK、TOS_SK：火山引擎账号 Secret Access Key（SK）。
   • ACEP_ACCOUNT_ID：火山引擎账号 ID，登录火山引擎后，在页面右上角账号信息中获取。
     
   • TOS_BUCKET：对象存储桶的名称。
   • TOS_REGION：对象存储桶的地域。例如 cn-beijing
   • TOS_ENDPOINT：对象存储桶的地域节点。
   • ARK_API_KEY：方舟大模型 API Key。
   • ARK_MODEL_ID：方舟大模型 Model ID。
   • ARK_BASE_URL：方舟大模型 Base URL。详细介绍，请参见 Base URL及鉴权。
   • TEST_PRODUCT_ID：云手机业务 ID。
   • TEST_POD_ID：云手机实例 ID。

6. （可选）在项目根目录打开 mcp.json 文件，根据实际需要配置其他 MCP 服务。

   说明

   Mobile Use SDK 已内置 Mobile Use MCP，如果没有其他 MCP 服务可配置，则无需操作本步骤。

7. 返回终端内，在项目根目录运行 basic_usage.py 示例代码。

   uv run examples/basic_usage.py
   

   效果说明：

   • 代码中提供的自然语言示例为 当前屏幕有哪些内容。
     
   • 终端返回结果展示了 Agent 获取到的云手机屏幕内容。
     

8. 在项目根目录运行 app_automation.py 示例代码。

   uv run examples/app_automation.py
   

   测试效果说明：

   • 代码中通过自然语言命令 Agent 启动特定应用程序并探索。
     
   • 终端返回结果展示了 Agent 获取到的云手机应用程序信息。
     

在指定项目中集成 Mobile Use SDK

本章节介绍如何将 Mobile Use SDK 集成在项目中。

步骤一：安装 Mobile Use SDK

通过 PyPI 将 SDK 集成至项目。

pip install ve-mobile-use

步骤二：配置项目的 .env 文件

在项目根目录的 .env 文件内，配置 SDK 所必须的环境变量。

# Mobile Use SDK Examples Environment Configuration

# Volcengine Cloud Phone (ACEP) Configuration
ACEP_AK=your_volcengine_access_key
ACEP_SK=your_volcengine_secret_key
ACEP_ACCOUNT_ID=your_volcengine_account_id


# TOS (Object Storage) Configuration
TOS_AK=your_tos_access_key
TOS_SK=your_tos_secret_key
TOS_BUCKET=your_tos_bucket
TOS_REGION=your_tos_region
TOS_ENDPOINT=your_tos_endpoint

# AI Model Configuration (Doubao/ARK)
ARK_API_KEY=your_ark_api_key
ARK_MODEL_ID=your_model_id
ARK_BASE_URL=your_ark_base_url

# Cloud Phone Instance Configuration
TEST_PRODUCT_ID=your_product_id
TEST_POD_ID=your_pod_id

如果您是通过函数服务 veFaaS 一键部署的 Mobile Use Agent 应用，以上变量值可通过以下方式快速获取：

1. 登录函数服务控制台。
2. 在左侧应用 > 我的应用中，找到已部署的应用。
3. 在应用详情页签，找到 Agent 对应的函数 ID。
   
4. 单击 ID 进入函数配置页签，在环境变量中获取对应的数值。
   

步骤三：配置并连接云手机

from mobile_use_sdk.mobile import CloudPhone
from mobile_use_sdk.config import TosInfo
from dotenv import load_dotenv 
import os

# 使用环境变量演示
load_dotenv()

async def main():

    # 云手机管理 Client 
    phone_manager = PhoneManager(
        ak=os.getenv("ACEP_AK"),
        sk=os.getenv("ACEP_SK"),
        account_id=os.getenv("ACEP_ACCOUNT_ID"),
        tos_info=TosInfo(
            ak=os.getenv("TOS_AK"),
            sk=os.getenv("TOS_SK"),
            bucket=os.getenv("TOS_BUCKET"),
            region=os.getenv("TOS_REGION"),
            endpoint=os.getenv("TOS_ENDPOINT"),
        ),
    )
    
    # 连接某个云手机，可以使用这个方法连续创建
    cloud_phone = await phone_manager.create_cloud_phone(
        pod_id=os.getenv("TEST_POD_ID"),
        product_id=os.getenv("TEST_PRODUCT_ID"),
    )    

    # 测试 tos 和 云手机操作
    screen_result = await cloud_phone.screenshot() # 获取手机截图
    print(screen_result)
    # { screenshot: "tos_url", screenshot_dimensions: (720,1440) }
    
    # 关闭云手机连接
    await cloud_phone.aclose()

步骤四：使用 Agent

• 标准使用方式

  from mobile_use_sdk.mobile import CloudPhone
  from mobile_use_sdk.config import TosInfo
  from mobile_use_sdk.agent import MobileUseAgent
  from dotenv import load_dotenv
  
  load_dotenv()
  
  async def main():
  
      # 云手机管理 Client 
      phone_manager = PhoneManager(
          ak=os.getenv("ACEP_AK"),
          sk=os.getenv("ACEP_SK"),
          account_id=os.getenv("ACEP_ACCOUNT_ID"),
          tos_info=TosInfo(
              ak=os.getenv("TOS_AK"),
              sk=os.getenv("TOS_SK"),
              bucket=os.getenv("TOS_BUCKET"),
              region=os.getenv("TOS_REGION"),
              endpoint=os.getenv("TOS_ENDPOINT"),
          ),
      )
  
      
      # 连接某个云手机，可以使用这个方法连续创建
      cloud_phone = await phone_manager.create_cloud_phone(
          pod_id=os.getenv("TEST_POD_ID"),
          product_id=os.getenv("TEST_PRODUCT_ID"),
      )    
  
      # 初始化 Agent
      agent = await MobileUseAgent.init_with_mobile(
          mobile=cloud_phone,
      )
      
      # 测试一个 case 
      await cloud_phone.press_home()
     
      # 执行简单任务
      result = await agent.run("执行xxx任务")
      
      # 任务结果
      print(result)
      
      await cloud_phone.aclose()
      
  if __name__ == '__main__':
      asyncio.run(main())
  

• 结构化输出示例
  Mobile Use 支持对内容进行结构化提取，示例如下，您可以参考并接入到项目当中。

  from pydantic import BaseModel
  from mobile_use_sdk.agent import MobileUseAgent
  
  class AppInfo(BaseModel):
      app_name: str
      version: str
      size: str
      rating: float
  
  result = await agent.run("执行xxx任务", response_format=AppInfo)
  app_info = AppInfo(**result)
  print(result.app_name, result.version)
  

SDK 功能说明

SDK 导出模块

SDK 有 4 块导出模块，其中：agent、mobile 为核心；cli、config 为辅助模块。

from mobile_use_sdk.mobile import PhoneManager
from mobile_use_sdk.agent import MobileUseAgent
from mobile_use_sdk.config import LLMConfig, AgentConfig, TosInfo

Mobile 使用

云手机创建和管理

1. 创建一个维持签名和鉴权信息的实例

   from mobile_use_sdk.mobile import PhoneManager
   
   phone_manager = PhoneManager(
           ak=os.getenv("ACEP_AK"),
           sk=os.getenv("ACEP_SK"),
           account_id=os.getenv("ACEP_ACCOUNT_ID"),
           tos_info=TosInfo(
               ak=os.getenv("TOS_AK"),
               sk=os.getenv("TOS_SK"),
               bucket=os.getenv("TOS_BUCKET"),
               region=os.getenv("TOS_REGION"),
               endpoint=os.getenv("TOS_ENDPOINT"),
           ),
       )
   

2. 使用 Manager 实例，可以创建多个云手机的实例

   cloud_phone = await phone_manager.create_cloud_phone(
           pod_id=os.getenv("TEST_POD_ID"),
           product_id=os.getenv("TEST_PRODUCT_ID"),
   )
   
   await phone.aclose()
   

Mobile Use 工具

支持多种手机操作，包括：

• tap：单点
• Swipe：单指滑动
• long_press：长按
• press_home：点击 home 键
• press_back：点击 back 键
• screenshot：截屏
• launch_app：打开 app
• close_app：关闭 app
• list_apps：列出所有 app 的 packagename 和 app display 名称
• clear_apps：清除所有已打开的应用

tap

## 函数声明
async def tap(self, point: tuple[int, int]) -> str:

## 使用方式
await phone.tap((x ,y))

swipe

## 函数声明
async def swipe(self, start_point: tuple[int, int], end_point: tuple[int, int],) -> str

## 使用方式
await phone.swipe((x,y), (x,y))

long_press

async def long_press(self, point: tuple[int, int], duration: int = 1) -> str

await phone.long_press((x,y), 1)

press_home

async def press_home(self) -> str
 
await phone.press_home()

press_back

async def press_back(self, count: int = 1) -> str
 
await phone.press_back(1)

screenshot

async def screenshot(self) -> dict:
 
 response = await phone.screenshot()
 
 print(response)
# => 
# {
#    "screenshot": screenshot_url,
#    "screenshot_dimensions": (width, height),
#  }

launch_app

async def launch_app(self, package_name: str) -> str:
 
await phone.launch_app("com.ss.android.ugc.aweme") # 抖音包

close_app

async def close_app(self, package_name: str) -> str:

 await phone.close_app("com.ss.android.ugc.aweme") # 抖音包

list_apps

async def list_apps(self) -> dict:
 
await phone.list_apps()
 
 # [{app_name: "抖音", package_name: "com.ss.android.ugc.aweme"}]

close_apps

async def close_app(self) -> str:

await phone.close_apps()

Agent 使用

初始化

Mobile Use 默认使用豆包视觉理解系列的模型，模型列表参考上文准备工作章节。

from mobile_use_sdk.agent import MobileUseAgent
from mobile_use_sdk.config import LLMConfig


# 使用默认配置
llm_config=LLMConfig(
        model=os.getenv("ARK_MODEL_ID"), # 'doubao-1.5-thinking-vision-pro' , '或您的推理点id'
        api_key=os.getenv("ARK_API_KEY"), # 'ak-xxx'
        base_url=os.getenv("ARK_BASE_URL"), # https://ark.cn-beijing.volces.com/api/v3
)

agent = await MobileUseAgent.init_with_mobile(
    llm_config=llm_config,
    mobile=cloud_phone,
)

系统提示词自定义

为了效果的稳定，系统提示词会追加到到当前系统提示词之后，请确保新增系统提示词不要过多对 Agent 的输出产生影响。如果需要新增工具，请使用第三方 MCP 功能集成。

from mobile_use_sdk.agent import MobileUseAgent
from mobile_use_sdk.config import AgentConfig

agent_config = AgentConfig(
        addiontional_system_prompt="请使用英语回答问题", 
        step_interval=0.5, 
        max_steps=25
    )

agent = await MobileUseAgent.init_with_mobile(
    llm_config=llm_config,
    agent_config=agent_config,
    mobile=cloud_phone,
)

第三方 MCP

如果想要集成第三方 API、MCP，可以如下集成，背后 Mobile Use 会帮忙做 McpHub 的链接池管理，封装的 Langgraph 的 McpAdaptor，可以支持现阶段所有的 mcp json 协议。

from mobile_use_sdk.agent import MobileUseAgent, mcp_manager

# 自定义MCP配置
mcp_config = {
    "mcpServers": {
        "a_map": {
            "url": "https://xxx",
            "transport": "sse"
        }
    }
}

agent = await MobileUseAgent.init_with_mobile(
    mobile=cloud_phone,
    mcp_json=mcp_json,
)

# 注意在程序结束，需要对 mcp_manager 进行清除, 背后 agent 使用 mcp 链接池管理
await mcp_manager.cleanup_all_connections()

结构化输出

run用于执行单个移动设备操作任务并返回完整的执行结果。

async def run(
    self,
    user_prompt: str,
    thread_id: Optional[str] = uuid.uuid4(),
    output_format: Optional[Type[BaseModel]] = None,
) -> AgentResponse:

参数说明

thread_id = uuid.uuid4()

response = await agent.run("执行xxx任务", thread_id=thread_id)

# 传递相同的 thread_id 可保留之前对话的记忆
response = await agent.run("执行xxx任务", thread_id=thread_id)

from pydantic import BaseModel

class AppInfo(BaseModel):
    name: str
    package: str
    is_installed: bool

response = await agent.run("执行xxx任务", output_format=AppInfo)

返回值

返回类型为 AgentResponse。
主要属性：

实例方法：

response = await agent.run("执行xxx任务")
total_time = response.total_duration_seconds()
print(f"总执行时间: {total_time}秒")

if response.is_done():
    print("任务执行完成")

error_msg = response.errors()
if error_msg:
    print(f"执行出错: {error_msg}")

tools = response.tool_calls()
print(f"共调用了 {len(tools)} 个工具")

# 获取所有截图
all_screenshots = response.screenshots()

# 获取最后一张截图
last_screenshot = response.screenshots(n_last=1)

messages = response.messages()
print(f"共有 {len(messages)} 条消息")

流式输出

import asyncio
from mobile_use_sdk.cli.stream_output import StreamOutput
from mobile_use_sdk.agent import MobileUseAgent
from dotenv import load_dotenv
import os

from mobile_use_sdk.mobile.cloud_phone import CloudPhone

load_dotenv()

async def run_task_with_user_interaction(agent, user_prompt, stream_output):
    print(f"\n🎯 开始执行任务: {user_prompt}")
    print("=" * 50)
    user_input_for_resume = None
    def handle_user_input(user_input: str):
        nonlocal user_input_for_resume
        user_input_for_resume = user_input
    stream_output.set_user_input_callback(handle_user_input)
    try:
        async for chunk in agent.astream(user_prompt, is_stream=True):
            stream_output.register_task(chunk)
        while user_input_for_resume is not None:
            current_input = user_input_for_resume
            user_input_for_resume = None
            print(f"用户反馈，{current_input}, 请等待...")
            async for chunk in agent.astream(current_input, is_stream=True):
                stream_output.register_task(chunk)

        print(f"\n✅ 任务完成: {user_prompt}")

    except Exception as e:
        print(f"\n❌ 任务执行出错: {e}")

async def agent_run():
    stream_output = StreamOutput()

    async with CloudPhone() as cloud_phone:
        await cloud_phone.initialize(
            pod_id=os.getenv("TEST_POD_ID"),
            product_id=os.getenv("TEST_PRODUCT_ID"),
        )

        try:
            await cloud_phone.clear_apps()
            agent = MobileUseAgent()
            await agent.initialize(
                mobile=cloud_phone,
            )
            tasks = [
                "执行xxx1任务",
                "执行xxx2任务",
                "执行xxx3任务",
            ]

            for task in tasks:
                await run_task_with_user_interaction(agent, task, stream_output)
                await asyncio.sleep(1)
        except Exception as e:
            print(f"❌ 执行出错: {e}")

if __name__ == "__main__":
    asyncio.run(agent_run())