MobileExecutor Server

result = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::capture_screenshot",
        tool_name="capture_screenshot",
        parameters={}
    )
])

# result[0].data = "data:image/png;base64,iVBORw0KGgo..."

{
    "success": bool,
    "ui_tree": str,      # XML content
    "format": "xml",
    # OR
    "error": str         # Error message if failed
}

result = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_ui_tree",
        tool_name="get_ui_tree",
        parameters={}
    )
])

# Parse XML to find elements
import xml.etree.ElementTree as ET
tree = ET.fromstring(result[0].data["ui_tree"])

{
    "success": bool,
    "device_info": {
        "model": str,              # Device model
        "android_version": str,    # Android version (e.g., "13")
        "sdk_version": str,        # SDK version (e.g., "33")
        "screen_size": str,        # Screen resolution (e.g., "Physical size: 1080x2400")
        "screen_density": str,     # Screen density (e.g., "Physical density: 440")
        "battery_level": str,      # Battery percentage
        "battery_status": str      # Charging status
    },
    "from_cache": bool,  # True if returned from cache
    # OR
    "error": str         # Error message if failed
}

result = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_device_info",
        tool_name="get_device_info",
        parameters={}
    )
])

device = result[0].data["device_info"]
print(f"Device: {device['model']}")
print(f"Android: {device['android_version']}")
print(f"Battery: {device['battery_level']}%")

[
    TargetInfo(
        kind=TargetKind.THIRD_PARTY_AGENT,
        id="1",                    # Sequential ID
        name="com.example.app",    # Package name (displayed)
        type="com.example.app"     # Package name (stored)
    ),
    ...
]

# Get all user-installed apps
result = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_mobile_app_target_info",
        tool_name="get_mobile_app_target_info",
        parameters={"include_system_apps": False}
    )
])

apps = result[0].data
for app in apps:
    print(f"ID: {app.id}, Package: {app.name}")

# Filter by package name
result = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_mobile_app_target_info",
        tool_name="get_mobile_app_target_info",
        parameters={"filter": "com.android", "include_system_apps": True}
    )
])

[
    TargetInfo(
        kind=TargetKind.CONTROL,
        id="1",                    # Sequential ID
        name="Button Name",        # Control text or content-desc
        type="Button",             # Control class (short name)
        rect=[x1, y1, x2, y2]     # Bounding box [left, top, right, bottom]
    ),
    ...
]

result = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_app_window_controls_target_info",
        tool_name="get_app_window_controls_target_info",
        parameters={}
    )
])

controls = result[0].data
for ctrl in controls:
    print(f"ID: {ctrl.id}, Name: {ctrl.name}, Type: {ctrl.type}")
    print(f"  Rect: {ctrl.rect}")

{
    "success": bool,
    "action": str,       # "tap(x, y)"
    "output": str,       # Command output
    "error": str         # Error message if failed
}

# Tap at specific coordinates
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::tap",
        tool_name="tap",
        parameters={"x": 500, "y": 1200}
    )
])

{
    "success": bool,
    "action": str,       # "swipe(x1,y1)->(x2,y2) in Nms"
    "output": str,
    "error": str
}

# Swipe up (scroll down content)
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::swipe",
        tool_name="swipe",
        parameters={
            "start_x": 500,
            "start_y": 1500,
            "end_x": 500,
            "end_y": 500,
            "duration": 300
        }
    )
])

# Swipe left (next page)
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::swipe",
        tool_name="swipe",
        parameters={
            "start_x": 800,
            "start_y": 1000,
            "end_x": 200,
            "end_y": 1000,
            "duration": 200
        }
    )
])

{
    "success": bool,
    "action": str,               # Full action description
    "message": str,              # Step-by-step messages
    "control_info": {
        "id": str,
        "name": str,
        "type": str
    },
    # OR
    "error": str                 # Error message
}

# 1. Get controls first
controls = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_app_window_controls_target_info",
        tool_name="get_app_window_controls_target_info",
        parameters={}
    )
])

# 2. Find search input field
search_field = next(c for c in controls[0].data if "Search" in c.name)

# 3. Type text
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::type_text",
        tool_name="type_text",
        parameters={
            "text": "hello world",
            "control_id": search_field.id,
            "control_name": search_field.name,
            "clear_current_text": True
        }
    )
])

{
    "success": bool,
    "message": str,
    "package_name": str,    # Actual package launched
    "output": str,          # ADB monkey output
    "error": str,
    "warning": str,         # Optional: name resolution warning
    "app_info": {           # Optional: if id provided
        "id": str,
        "name": str,
        "package": str
    }
}

# Launch by package name
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::launch_app",
        tool_name="launch_app",
        parameters={"package_name": "com.android.settings"}
    )
])

# Launch by app ID (from cache)
apps = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_mobile_app_target_info",
        tool_name="get_mobile_app_target_info",
        parameters={}
    )
])

settings_app = next(a for a in apps[0].data if "settings" in a.name.lower())

result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::launch_app",
        tool_name="launch_app",
        parameters={
            "package_name": settings_app.type,  # Package from cache
            "id": settings_app.id
        }
    )
])

# Launch by app name (auto-resolves package)
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::launch_app",
        tool_name="launch_app",
        parameters={"package_name": "Settings"}  # Resolves to com.android.settings
    )
])

{
    "success": bool,
    "action": str,       # "press_key(KEYCODE_X)"
    "output": str,
    "error": str
}

# Press back button
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::press_key",
        tool_name="press_key",
        parameters={"key_code": "KEYCODE_BACK"}
    )
])

# Press home button
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::press_key",
        tool_name="press_key",
        parameters={"key_code": "KEYCODE_HOME"}
    )
])

# Press enter
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::press_key",
        tool_name="press_key",
        parameters={"key_code": "KEYCODE_ENTER"}
    )
])

{
    "success": bool,
    "action": str,               # Full action description
    "message": str,              # Success message with coordinates
    "control_info": {
        "id": str,
        "name": str,
        "type": str,
        "rect": [int, int, int, int]
    },
    "warning": str,              # Optional: name mismatch warning
    # OR
    "error": str                 # Error message
}

# 1. Get controls
controls = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_app_window_controls_target_info",
        tool_name="get_app_window_controls_target_info",
        parameters={}
    )
])

# 2. Find OK button
ok_button = next(c for c in controls[0].data if c.name == "OK")

# 3. Click it
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::click_control",
        tool_name="click_control",
        parameters={
            "control_id": ok_button.id,
            "control_name": ok_button.name
        }
    )
])

{
    "success": bool,
    "action": str,       # "wait(Ns)"
    "message": str,      # "Waited for N seconds"
    # OR
    "error": str         # Error if invalid seconds
}

# Wait 1 second
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::wait",
        tool_name="wait",
        parameters={"seconds": 1.0}
    )
])

# Wait 500ms
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::wait",
        tool_name="wait",
        parameters={"seconds": 0.5}
    )
])

# Wait 2.5 seconds
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::wait",
        tool_name="wait",
        parameters={"seconds": 2.5}
    )
])

{
    "success": bool,
    "message": str,      # Confirmation message
    # OR
    "error": str         # Invalid cache_type
}

# Invalidate all caches
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::invalidate_cache",
        tool_name="invalidate_cache",
        parameters={"cache_type": "all"}
    )
])

# Invalidate only controls cache
result = await computer.run_actions([
    MCPToolCall(
        tool_key="action::invalidate_cache",
        tool_name="invalidate_cache",
        parameters={"cache_type": "controls"}
    )
])

# Windows agent controlling Android device
MobileAgent:
    default:
        data_collection:
            - { namespace: MobileDataCollector, type: http, host: "localhost", port: 8020, path: "/mcp", auth: "${UFO_MCP_API_KEY}" }
        action:
            - { namespace: MobileActionExecutor, type: http, host: "localhost", port: 8021, path: "/mcp", auth: "${UFO_MCP_API_KEY}" }

# Remote Android device
MobileAgent:
    default:
        data_collection:
            - { namespace: MobileDataCollector, type: http, host: "192.168.1.150", port: 8020, path: "/mcp", auth: "${UFO_MCP_API_KEY}" }
        action:
            - { namespace: MobileActionExecutor, type: http, host: "192.168.1.150", port: 8021, path: "/mcp", auth: "${UFO_MCP_API_KEY}" }

# Windows (via Android SDK or standalone)
# Download from: https://developer.android.com/studio/releases/platform-tools

# Linux
sudo apt-get install android-tools-adb

# macOS
brew install android-platform-tools

# Check connected devices
adb devices

# Output:
# List of devices attached
# R5CR20XXXXX    device

# Linux or macOS
export UFO_MCP_API_KEY="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')"

# Windows PowerShell
$env:UFO_MCP_API_KEY = python -c "import secrets; print(secrets.token_urlsafe(32))"

# Start both servers (recommended)
python -m ufo.client.mcp.http_servers.mobile_mcp_server --server both --host 0.0.0.0 --data-port 8020 --action-port 8021

# Output:
# ==================================================
# UFO Mobile MCP Servers (Android)
# Android device control via ADB and Model Context Protocol
# ==================================================
# Using ADB: C:\...\adb.exe
# Found 1 connected device(s)
# ✅ Starting both servers in same process (shared MobileServerState)
#    - Data Collection Server: 0.0.0.0:8020
#    - Action Server: 0.0.0.0:8021
# Both servers share MobileServerState cache. Press Ctrl+C to stop.

# Start only data collection server
python -m ufo.client.mcp.http_servers.mobile_mcp_server --server data --host 0.0.0.0 --data-port 8020

# Start only action server
python -m ufo.client.mcp.http_servers.mobile_mcp_server --server action --host 0.0.0.0 --action-port 8021

python -m ufo.client.mcp.http_servers.mobile_mcp_server --adb-path "C:\custom\path\adb.exe"

# Servers on same machine as client
--host localhost

# Servers accessible from network
--host 0.0.0.0

# ✅ Good: Both servers in same process (shared state)
python -m ufo.client.mcp.http_servers.mobile_mcp_server --server both

# ❌ Bad: Separate processes (no shared state)
python -m ufo.client.mcp.http_servers.mobile_mcp_server --server data &
python -m ufo.client.mcp.http_servers.mobile_mcp_server --server action &

# ✅ Good: Get controls first
controls = await computer.run_data_collection([
    MCPToolCall(tool_key="data_collection::get_app_window_controls_target_info", ...)
])

# Then click/type
await computer.run_actions([
    MCPToolCall(tool_key="action::click_control", parameters={"control_id": "5", ...})
])

# ❌ Bad: Click without getting controls
await computer.run_actions([
    MCPToolCall(tool_key="action::click_control", parameters={"control_id": "5", ...})
])
# Error: Control not found in cache

# ✅ Good: Use click_control (reliable)
await computer.run_actions([
    MCPToolCall(
        tool_key="action::click_control",
        parameters={"control_id": "3", "control_name": "Submit"}
    )
])

# ⚠️ OK: Use tap only when control not available
await computer.run_actions([
    MCPToolCall(
        tool_key="action::tap",
        parameters={"x": 500, "y": 1200}
    )
])

# Check if controls are stale
controls = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_app_window_controls_target_info",
        parameters={"force_refresh": False}  # Use cache if available
    )
])

# For critical operations, force refresh
controls = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_app_window_controls_target_info",
        parameters={"force_refresh": True}  # Always query device
    )
])

# ✅ Good: Wait for UI to settle
await computer.run_actions([
    MCPToolCall(tool_key="action::tap", parameters={"x": 500, "y": 1200})
])
await computer.run_actions([
    MCPToolCall(tool_key="action::wait", parameters={"seconds": 1.0})
])

# Get updated controls
controls = await computer.run_data_collection([
    MCPToolCall(tool_key="data_collection::get_app_window_controls_target_info", ...)
])

# Check device info before operations
device_info = await computer.run_data_collection([
    MCPToolCall(tool_key="data_collection::get_device_info", parameters={})
])

if device_info[0].is_error:
    raise RuntimeError("No Android device connected")

# Launch app
await computer.run_actions([
    MCPToolCall(
        tool_key="action::launch_app",
        tool_name="launch_app",
        parameters={"package_name": "com.example.app"}
    )
])

# Wait for app to load
await computer.run_actions([
    MCPToolCall(tool_key="action::wait", parameters={"seconds": 2.0})
])

# Get controls
controls = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_app_window_controls_target_info",
        parameters={}
    )
])

# Find and click button
login_btn = next(c for c in controls[0].data if "Login" in c.name)
await computer.run_actions([
    MCPToolCall(
        tool_key="action::click_control",
        parameters={
            "control_id": login_btn.id,
            "control_name": login_btn.name
        }
    )
])

# Get controls
controls = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_app_window_controls_target_info",
        parameters={}
    )
])

# Type username
username_field = next(c for c in controls[0].data if "username" in c.name.lower())
await computer.run_actions([
    MCPToolCall(
        tool_key="action::type_text",
        tool_name="type_text",
        parameters={
            "text": "john.doe@example.com",
            "control_id": username_field.id,
            "control_name": username_field.name,
            "clear_current_text": True
        }
    )
])

# Get updated controls (after typing)
await computer.run_actions([
    MCPToolCall(tool_key="action::wait", parameters={"seconds": 0.5})
])
controls = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_app_window_controls_target_info",
        parameters={"force_refresh": True}
    )
])

# Type password
password_field = next(c for c in controls[0].data if "password" in c.name.lower())
await computer.run_actions([
    MCPToolCall(
        tool_key="action::type_text",
        parameters={
            "text": "SecureP@ssw0rd",
            "control_id": password_field.id,
            "control_name": password_field.name
        }
    )
])

# Submit
submit_btn = next(c for c in controls[0].data if "Submit" in c.name)
await computer.run_actions([
    MCPToolCall(
        tool_key="action::click_control",
        parameters={
            "control_id": submit_btn.id,
            "control_name": submit_btn.name
        }
    )
])

# Swipe up to scroll down content
await computer.run_actions([
    MCPToolCall(
        tool_key="action::swipe",
        tool_name="swipe",
        parameters={
            "start_x": 500,
            "start_y": 1500,
            "end_x": 500,
            "end_y": 500,
            "duration": 300
        }
    )
])

# Wait for scrolling to complete
await computer.run_actions([
    MCPToolCall(tool_key="action::wait", parameters={"seconds": 0.5})
])

# Get updated controls
controls = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_app_window_controls_target_info",
        parameters={"force_refresh": True}
    )
])

# Get device info
device_info = await computer.run_data_collection([
    MCPToolCall(tool_key="data_collection::get_device_info", parameters={})
])

print(f"Testing on: {device_info[0].data['device_info']['model']}")
print(f"Android: {device_info[0].data['device_info']['android_version']}")

# Take screenshot before test
screenshot_before = await computer.run_data_collection([
    MCPToolCall(tool_key="data_collection::capture_screenshot", parameters={})
])

# Perform test actions
# ...

# Take screenshot after test
screenshot_after = await computer.run_data_collection([
    MCPToolCall(tool_key="data_collection::capture_screenshot", parameters={})
])

# Compare screenshots (external comparison logic)

# Restart ADB server
adb kill-server
adb start-server

# Check device connection
adb devices

# If no devices shown:
# 1. Check USB cable
# 2. Verify USB debugging enabled on device
# 3. Accept "Allow USB debugging" prompt on device

# Check if ports are in use
netstat -an | findstr "8020"
netstat -an | findstr "8021"

# Change ports if needed
python -m ufo.client.mcp.http_servers.mobile_mcp_server --data-port 8030 --action-port 8031

# Force refresh cache
controls = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_app_window_controls_target_info",
        parameters={"force_refresh": True}
    )
])

# Or invalidate cache manually
await computer.run_actions([
    MCPToolCall(
        tool_key="action::invalidate_cache",
        parameters={"cache_type": "controls"}
    )
])

# Ensure control is in cache
controls = await computer.run_data_collection([
    MCPToolCall(
        tool_key="data_collection::get_app_window_controls_target_info",
        parameters={}
    )
])

# Verify control ID and name match
field = next(c for c in controls[0].data if c.id == "5")
print(f"Control name: {field.name}")

# Use exact ID and name
await computer.run_actions([
    MCPToolCall(
        tool_key="action::type_text",
        parameters={
            "text": "test",
            "control_id": field.id,
            "control_name": field.name
        }
    )
])