From e0c05dc72a12278f8ffcd56bc084750efc5c137e Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Wed, 11 Feb 2026 03:05:27 -0700 Subject: [PATCH] =?UTF-8?q?Add=20connectivity=20and=20settings=20mixins=20?= =?UTF-8?q?(50=20=E2=86=92=2065=20tools)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New mixins: - connectivity.py: adb_connect, adb_disconnect, adb_tcpip, adb_pair, device_properties (batch getprop) - settings.py: settings_get/put, wifi/bluetooth/airplane toggles, screen_brightness, screen_timeout, notification_list, clipboard_get, media_control Also fixes clipboard_set false-positive on devices where cmd clipboard returns exit 0 but has no implementation. --- CLAUDE.md | 4 +- README.md | 38 +- pyproject.toml | 2 +- src/mixins/__init__.py | 4 + src/mixins/connectivity.py | 276 +++++++++++++++ src/mixins/input.py | 20 +- src/mixins/settings.py | 701 +++++++++++++++++++++++++++++++++++++ src/server.py | 40 ++- uv.lock | 2 +- 9 files changed, 1078 insertions(+), 9 deletions(-) create mode 100644 src/mixins/connectivity.py create mode 100644 src/mixins/settings.py diff --git a/CLAUDE.md b/CLAUDE.md index db8c18b..d4963a3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -34,18 +34,20 @@ docker compose up --build ## Architecture -Modular MCPMixin architecture with 50 tools across 6 domain mixins: +Modular MCPMixin architecture with 65 tools across 8 domain mixins: - **`src/server.py`**: FastMCP app and `ADBServer` class (thin orchestrator inheriting all mixins) - **`src/config.py`**: Persistent singleton config (`~/.config/adb-mcp/config.json`) - **`src/models.py`**: Pydantic models (`DeviceInfo`, `CommandResult`, `ScreenshotResult`) - **`src/mixins/base.py`**: Core ADB execution with `run_adb()`, `run_shell()`, and injection-safe `run_shell_args()` using `shlex.quote()` - **`src/mixins/devices.py`**: Device discovery, info, logcat, reboot +- **`src/mixins/connectivity.py`**: TCP/IP connect/disconnect, wireless pairing, device properties - **`src/mixins/input.py`**: Tap, swipe, scroll, keys, text, clipboard, shell command - **`src/mixins/apps.py`**: Launch, close, install, intents, broadcasts - **`src/mixins/screenshot.py`**: Screen capture, recording, display settings - **`src/mixins/ui.py`**: Accessibility tree dump, element search, text polling - **`src/mixins/files.py`**: Push, pull, list, delete, exists +- **`src/mixins/settings.py`**: System settings, radio toggles, brightness, notifications, clipboard read, media control ### Key Patterns diff --git a/README.md b/README.md index da3e38b..02835d2 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ A [Model Context Protocol](https://modelcontextprotocol.io/) server that gives AI assistants direct control over Android devices through ADB. Point any MCP-compatible client at a phone plugged into USB, and it can take screenshots, tap buttons, launch apps, inspect UI elements, transfer files, and run shell commands — all through structured, type-safe tool calls. -Built on [FastMCP](https://gofastmcp.com/) with a modular mixin architecture. 50 tools across 6 domains. Tested on real hardware. +Built on [FastMCP](https://gofastmcp.com/) with a modular mixin architecture. 65 tools across 8 domains. Tested on real hardware. ## Quick Start @@ -57,6 +57,10 @@ claude mcp add mcadb -- uv run --directory /path/to/mcp-adb mcadb | | `devices_use` | Set active device for multi-device setups | | | `devices_current` | Show which device is selected | | | `device_info` | Battery, WiFi, storage, Android version, model | +| **Connectivity** | `adb_connect` | Connect to device over TCP/IP | +| | `adb_disconnect` | Disconnect a network device | +| | `adb_pair` | Wireless debugging pairing (Android 11+) | +| | `device_properties` | Batch getprop (model, SoC, versions, serial, ABI) | | **Input** | `input_tap` | Tap at screen coordinates | | | `input_swipe` | Swipe between two points | | | `input_scroll_down` | Scroll down (auto-detects screen size) | @@ -80,6 +84,10 @@ claude mcp add mcadb -- uv run --directory /path/to/mcp-adb mcadb | | `wait_for_text` | Poll until text appears on screen | | | `wait_for_text_gone` | Poll until text disappears | | | `tap_text` | Find an element by text and tap it | +| **Settings** | `settings_get` | Read any system/global/secure setting | +| | `notification_list` | List recent notifications (title, text, package) | +| | `clipboard_get` | Read clipboard contents | +| | `media_control` | Play/pause/next/previous/stop/volume | | **Config** | `config_status` | Show current settings | | | `config_set_developer_mode` | Toggle developer tools | | | `config_set_screenshot_dir` | Set where screenshots are saved | @@ -109,6 +117,13 @@ Enable with `config_set_developer_mode(true)` to unlock power-user tools. Destru | | `file_list` | List directory contents | | | `file_delete` | Delete file (with confirmation) | | | `file_exists` | Check if file exists | +| **Connectivity** | `adb_tcpip` | Switch USB device to TCP/IP mode | +| **Settings** | `settings_put` | Write system/global/secure setting | +| | `wifi_toggle` | Enable/disable WiFi | +| | `bluetooth_toggle` | Enable/disable Bluetooth | +| | `airplane_mode_toggle` | Toggle airplane mode (with confirmation) | +| | `screen_brightness` | Set brightness 0-255 | +| | `screen_timeout` | Set screen timeout duration | ### Resources @@ -146,6 +161,13 @@ Enable with `config_set_developer_mode(true)` to unlock power-user tools. Destru 4. logcat_capture(filter_spec="MyApp:D *:S") ``` +**Connect to a WiFi device:** +``` +1. adb_connect("10.20.0.25") → Connect to network device +2. devices_list() → Verify it appears +3. screenshot() → Capture from network device +``` + **Multi-device workflow:** ``` 1. devices_list() → See all connected devices @@ -154,9 +176,17 @@ Enable with `config_set_developer_mode(true)` to unlock power-user tools. Destru 4. screenshot() → Capture from selected device ``` +**Read settings and control media:** +``` +1. settings_get("global", "wifi_on") → Check WiFi state +2. notification_list() → See recent notifications +3. media_control("pause") → Pause media playback +4. clipboard_get() → Read clipboard contents +``` + ## Architecture -The server uses FastMCP's [MCPMixin](https://gofastmcp.com/) pattern to organize 50 tools into focused, single-responsibility modules: +The server uses FastMCP's [MCPMixin](https://gofastmcp.com/) pattern to organize 65 tools into focused, single-responsibility modules: ``` src/ @@ -166,14 +196,16 @@ src/ mixins/ base.py ← ADB command execution, injection-safe shell quoting devices.py ← Device discovery, info, logcat, reboot + connectivity.py ← TCP/IP connect/disconnect, wireless pairing, properties input.py ← Tap, swipe, scroll, keys, text, clipboard, shell apps.py ← Launch, close, install, intents, broadcasts screenshot.py ← Capture, recording, display settings ui.py ← Accessibility tree, element search, text polling files.py ← Push, pull, list, delete, exists + settings.py ← System settings, radios, brightness, notifications, media ``` -`ADBServer` inherits all six mixins. Each mixin calls `run_shell_args()` (injection-safe) or `run_adb()` on the base class. The base handles device targeting, subprocess execution, and timeouts. +`ADBServer` inherits all eight mixins. Each mixin calls `run_shell_args()` (injection-safe) or `run_adb()` on the base class. The base handles device targeting, subprocess execution, and timeouts. ## Security Model diff --git a/pyproject.toml b/pyproject.toml index 6a2708c..ed90c23 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "mcadb" -version = "0.3.1" +version = "0.4.0" description = "Android ADB MCP Server for device automation via Model Context Protocol" authors = [ {name = "Ryan Malloy", email = "ryan@supported.systems"} diff --git a/src/mixins/__init__.py b/src/mixins/__init__.py index 3fb2992..5ac2eaf 100644 --- a/src/mixins/__init__.py +++ b/src/mixins/__init__.py @@ -2,18 +2,22 @@ from .apps import AppsMixin from .base import ADBBaseMixin +from .connectivity import ConnectivityMixin from .devices import DevicesMixin from .files import FilesMixin from .input import InputMixin from .screenshot import ScreenshotMixin +from .settings import SettingsMixin from .ui import UIMixin __all__ = [ "ADBBaseMixin", "DevicesMixin", + "ConnectivityMixin", "InputMixin", "AppsMixin", "ScreenshotMixin", "UIMixin", "FilesMixin", + "SettingsMixin", ] diff --git a/src/mixins/connectivity.py b/src/mixins/connectivity.py new file mode 100644 index 0000000..40fa912 --- /dev/null +++ b/src/mixins/connectivity.py @@ -0,0 +1,276 @@ +"""Connectivity mixin for Android ADB MCP Server. + +Provides tools for managing ADB network connections and device properties. +""" + +import re +from typing import Any + +from fastmcp import Context +from fastmcp.contrib.mcp_mixin import mcp_tool + +from ..config import is_developer_mode +from .base import ADBBaseMixin + + +class ConnectivityMixin(ADBBaseMixin): + """Mixin for ADB connectivity management. + + Provides tools for: + - Connecting/disconnecting devices over TCP/IP + - Switching USB devices to network mode + - Wireless debugging pairing (Android 11+) + - Batch device property queries + """ + + @mcp_tool() + async def adb_connect( + self, + host: str, + port: int = 5555, + ) -> dict[str, Any]: + """Connect to a device over TCP/IP. + + Establishes an ADB connection to a device on the network. + The device must already have ADB over TCP/IP enabled + (via adb_tcpip or developer options). + + Args: + host: Device IP address or hostname + port: ADB port (default 5555) + + Returns: + Connection result with device address + """ + target = f"{host}:{port}" + result = await self.run_adb(["connect", target]) + + # ADB connect returns 0 even on failure — check stdout + connected = result.success and "connected" in result.stdout.lower() + already = "already connected" in result.stdout.lower() + + return { + "success": connected, + "already_connected": already, + "address": target, + "output": result.stdout, + "error": result.stderr if not connected else None, + } + + @mcp_tool() + async def adb_disconnect( + self, + host: str, + port: int = 5555, + ) -> dict[str, Any]: + """Disconnect a network-connected device. + + Drops the ADB TCP/IP connection to the specified device. + + Args: + host: Device IP address or hostname + port: ADB port (default 5555) + + Returns: + Disconnection result + """ + target = f"{host}:{port}" + result = await self.run_adb(["disconnect", target]) + + disconnected = result.success and "disconnected" in result.stdout.lower() + + return { + "success": disconnected, + "address": target, + "output": result.stdout, + "error": result.stderr if not disconnected else None, + } + + @mcp_tool( + tags={"developer"}, + annotations={"requires": "developer_mode"}, + ) + async def adb_tcpip( + self, + ctx: Context, + port: int = 5555, + device_id: str | None = None, + ) -> dict[str, Any]: + """Switch a USB-connected device to TCP/IP mode. + + [DEVELOPER MODE] Restarts ADB on the device in TCP/IP mode, + then fetches the device's IP address so you can immediately + call adb_connect() to reconnect over the network. + + The device must be connected via USB. After switching, the + USB ADB connection will be dropped and you'll need to use + adb_connect(ip, port) to reconnect. + + Args: + ctx: MCP context for logging + port: TCP port for ADB to listen on (default 5555) + device_id: Target USB device (required if multiple connected) + + Returns: + Result with device IP address for subsequent adb_connect + """ + if not is_developer_mode(): + return { + "success": False, + "error": "Developer mode required", + } + + # Reject if device_id looks like a network device (IP:port format) + target = device_id or self.get_current_device() + if target and re.match(r"\d+\.\d+\.\d+\.\d+:\d+", target): + return { + "success": False, + "error": ( + f"Device '{target}' is already a network device. " + "adb_tcpip only works on USB-connected devices." + ), + } + + # Get device IP before switching (wlan0) + ip_result = await self.run_shell_args( + ["ip", "addr", "show", "wlan0"], device_id + ) + device_ip = None + if ip_result.success: + match = re.search(r"inet (\d+\.\d+\.\d+\.\d+)/", ip_result.stdout) + if match: + device_ip = match.group(1) + + if not device_ip: + return { + "success": False, + "error": ( + "Could not determine device IP address. " + "Ensure the device is connected to WiFi." + ), + } + + await ctx.info(f"Switching device to TCP/IP mode on port {port}...") + + # Switch to TCP/IP mode — this is an ADB client command, not shell + result = await self.run_adb(["tcpip", str(port)], device_id) + + if not result.success: + return { + "success": False, + "error": result.stderr or result.stdout, + } + + await ctx.info( + f"Device switched to TCP/IP on port {port}. " + f"Connect with: adb_connect('{device_ip}', {port})" + ) + + return { + "success": True, + "port": port, + "device_ip": device_ip, + "connect_address": f"{device_ip}:{port}", + "message": ( + f"Device now listening on {device_ip}:{port}. " + "USB connection will drop. Use adb_connect() to reconnect." + ), + } + + @mcp_tool() + async def adb_pair( + self, + host: str, + port: int, + pairing_code: str, + ) -> dict[str, Any]: + """Pair with a device for wireless debugging (Android 11+). + + Pairs with a device using the wireless debugging pairing code + shown in Developer Options > Wireless Debugging > Pair device. + + After pairing, use adb_connect() with the connection address + (different from the pairing address) to establish the session. + + Args: + host: Device IP address from pairing dialog + port: Pairing port from pairing dialog + pairing_code: 6-digit pairing code from device screen + + Returns: + Pairing result + """ + target = f"{host}:{port}" + result = await self.run_adb(["pair", target, pairing_code]) + + paired = result.success and "successfully paired" in result.stdout.lower() + + return { + "success": paired, + "address": target, + "output": result.stdout, + "error": result.stderr if not paired else None, + } + + @mcp_tool() + async def device_properties( + self, + device_id: str | None = None, + ) -> dict[str, Any]: + """Get detailed device properties via getprop. + + Fetches a comprehensive batch of system properties including + hardware info, software versions, and device identifiers. + + Args: + device_id: Target device + + Returns: + Dictionary of device properties grouped by category + """ + props_to_fetch = { + "identity": [ + ("model", "ro.product.model"), + ("manufacturer", "ro.product.manufacturer"), + ("brand", "ro.product.brand"), + ("device", "ro.product.device"), + ("product", "ro.product.name"), + ("serial", "ro.serialno"), + ], + "software": [ + ("android_version", "ro.build.version.release"), + ("sdk_version", "ro.build.version.sdk"), + ("build_id", "ro.build.display.id"), + ("security_patch", "ro.build.version.security_patch"), + ("build_type", "ro.build.type"), + ("build_fingerprint", "ro.build.fingerprint"), + ], + "hardware": [ + ("chipset", "ro.board.platform"), + ("cpu_abi", "ro.product.cpu.abi"), + ("hardware", "ro.hardware"), + ], + "system": [ + ("timezone", "persist.sys.timezone"), + ("language", "persist.sys.language"), + ("locale", "ro.product.locale"), + ], + } + + result: dict[str, Any] = {"success": True} + + for category, prop_list in props_to_fetch.items(): + category_data: dict[str, str] = {} + for friendly_name, prop_key in prop_list: + value = await self.get_device_property(prop_key, device_id) + if value: + category_data[friendly_name] = value + if category_data: + result[category] = category_data + + # Check if we got anything at all + if len(result) == 1: # only "success" key + result["success"] = False + result["error"] = "No properties returned. Is the device connected?" + + return result diff --git a/src/mixins/input.py b/src/mixins/input.py index e4b55c8..b56a7b4 100644 --- a/src/mixins/input.py +++ b/src/mixins/input.py @@ -474,10 +474,13 @@ class InputMixin(ADBBaseMixin): Success status """ # Try cmd clipboard set (Android 12+, injection-safe via args) + # Some devices return exit 0 but "No shell command implementation" + # in stderr — treat that as failure so the fallback triggers result = await self.run_shell_args(["cmd", "clipboard", "set", text], device_id) + cmd_worked = result.success and "no shell command" not in result.stderr.lower() # Fallback: try am broadcast (Clipper app or similar) - if not result.success: + if not cmd_worked: result = await self.run_shell_args( [ "am", @@ -490,6 +493,21 @@ class InputMixin(ADBBaseMixin): ], device_id, ) + # Broadcast returns exit 0 even with no receiver. + # Check for "result=-1" (receiver acknowledged) vs + # "result=0" (default, likely no receiver) + if result.success and "result=0" in result.stdout: + result = result.model_copy( + update={ + "success": False, + "stderr": ( + "Device does not support cmd clipboard " + "and no broadcast receiver found. Use " + "input_text() for simple text or install " + "a clipboard helper app." + ), + } + ) preview = text[:100] + "..." if len(text) > 100 else text response: dict[str, Any] = { diff --git a/src/mixins/settings.py b/src/mixins/settings.py new file mode 100644 index 0000000..9c8a265 --- /dev/null +++ b/src/mixins/settings.py @@ -0,0 +1,701 @@ +"""Settings mixin for Android ADB MCP Server. + +Provides tools for reading/writing system settings, toggling radios, +display configuration, notification access, clipboard, and media control. +""" + +import re +from typing import Any + +from fastmcp import Context +from fastmcp.contrib.mcp_mixin import mcp_tool + +from ..config import is_developer_mode +from .base import ADBBaseMixin + +_VALID_NAMESPACES = {"system", "global", "secure"} +_SETTING_KEY_PATTERN = re.compile(r"^[a-zA-Z0-9_.]+$") + +_MEDIA_KEYCODES: dict[str, str] = { + "play": "KEYCODE_MEDIA_PLAY", + "pause": "KEYCODE_MEDIA_PAUSE", + "play_pause": "KEYCODE_MEDIA_PLAY_PAUSE", + "next": "KEYCODE_MEDIA_NEXT", + "previous": "KEYCODE_MEDIA_PREVIOUS", + "stop": "KEYCODE_MEDIA_STOP", + "mute": "KEYCODE_VOLUME_MUTE", + "volume_up": "KEYCODE_VOLUME_UP", + "volume_down": "KEYCODE_VOLUME_DOWN", +} + + +class SettingsMixin(ADBBaseMixin): + """Mixin for Android system settings and controls. + + Provides tools for: + - Reading/writing system, global, and secure settings + - WiFi, Bluetooth, and airplane mode toggles + - Display brightness and timeout + - Notification listing + - Clipboard reading + - Media playback control + """ + + @mcp_tool() + async def settings_get( + self, + namespace: str, + key: str, + device_id: str | None = None, + ) -> dict[str, Any]: + """Read an Android system setting. + + Reads a value from the device's settings database. + Settings are organized into three namespaces: + - system: User-facing settings (brightness, font size, ringtone) + - global: Device-wide settings (wifi_on, airplane_mode_on, adb_enabled) + - secure: Security/privacy settings (location_mode, accessibility) + + Args: + namespace: Settings namespace — "system", "global", or "secure" + key: Setting key (e.g., "screen_brightness", "wifi_on") + device_id: Target device + + Returns: + The setting value + """ + if namespace not in _VALID_NAMESPACES: + return { + "success": False, + "error": ( + f"Invalid namespace '{namespace}'. " + f"Must be one of: {', '.join(sorted(_VALID_NAMESPACES))}" + ), + } + + if not _SETTING_KEY_PATTERN.match(key): + return { + "success": False, + "error": ( + f"Invalid key '{key}'. Keys must contain " + "only letters, digits, underscores, and dots." + ), + } + + result = await self.run_shell_args( + ["settings", "get", namespace, key], device_id + ) + + if not result.success: + return { + "success": False, + "error": result.stderr, + } + + value = result.stdout.strip() + is_null = value == "null" + + return { + "success": True, + "namespace": namespace, + "key": key, + "value": None if is_null else value, + "exists": not is_null, + } + + @mcp_tool( + tags={"developer"}, + annotations={"requires": "developer_mode"}, + ) + async def settings_put( + self, + ctx: Context, + namespace: str, + key: str, + value: str, + device_id: str | None = None, + ) -> dict[str, Any]: + """Write an Android system setting. + + [DEVELOPER MODE] Writes a value to the device's settings database. + Writing to the "secure" namespace requires additional confirmation + as it can affect security-sensitive behavior. + + Args: + ctx: MCP context for elicitation/logging + namespace: Settings namespace — "system", "global", or "secure" + key: Setting key + value: Value to set + device_id: Target device + + Returns: + Result with read-back verification + """ + if not is_developer_mode(): + return { + "success": False, + "error": "Developer mode required", + } + + if namespace not in _VALID_NAMESPACES: + return { + "success": False, + "error": ( + f"Invalid namespace '{namespace}'. " + f"Must be one of: {', '.join(sorted(_VALID_NAMESPACES))}" + ), + } + + if not _SETTING_KEY_PATTERN.match(key): + return { + "success": False, + "error": ( + f"Invalid key '{key}'. Keys must contain " + "only letters, digits, underscores, and dots." + ), + } + + # Extra confirmation for secure namespace + if namespace == "secure": + confirmation = await ctx.elicit( + f"Writing to the 'secure' namespace can affect security-sensitive " + f"device behavior.\n\n" + f"Setting: {namespace}/{key} = {value}\n\n" + f"Proceed?", + ["Yes, write setting", "Cancel"], + ) + if confirmation.action != "accept" or confirmation.content == "Cancel": + return { + "success": False, + "cancelled": True, + "message": "Settings write cancelled by user", + } + + result = await self.run_shell_args( + ["settings", "put", namespace, key, value], device_id + ) + + if not result.success: + return { + "success": False, + "error": result.stderr, + } + + # Read back to verify + verify = await self.run_shell_args( + ["settings", "get", namespace, key], device_id + ) + readback = verify.stdout.strip() if verify.success else None + + await ctx.info(f"Set {namespace}/{key} = {value}") + + return { + "success": True, + "namespace": namespace, + "key": key, + "value": value, + "readback": readback, + "verified": readback == value, + } + + @mcp_tool( + tags={"developer"}, + annotations={"requires": "developer_mode"}, + ) + async def wifi_toggle( + self, + enabled: bool, + device_id: str | None = None, + ) -> dict[str, Any]: + """Toggle WiFi on or off. + + [DEVELOPER MODE] Enables or disables WiFi using the svc command. + Verifies the state change after toggling. + + Args: + enabled: True to enable WiFi, False to disable + device_id: Target device + + Returns: + Result with verified WiFi state + """ + if not is_developer_mode(): + return { + "success": False, + "error": "Developer mode required", + } + + action = "enable" if enabled else "disable" + result = await self.run_shell_args(["svc", "wifi", action], device_id) + + if not result.success: + return { + "success": False, + "error": result.stderr, + } + + # Verify state change + verify = await self.run_shell_args( + ["settings", "get", "global", "wifi_on"], device_id + ) + current = verify.stdout.strip() if verify.success else "unknown" + + return { + "success": True, + "action": action, + "wifi_on": current, + "verified": current == ("1" if enabled else "0"), + } + + @mcp_tool( + tags={"developer"}, + annotations={"requires": "developer_mode"}, + ) + async def bluetooth_toggle( + self, + enabled: bool, + device_id: str | None = None, + ) -> dict[str, Any]: + """Toggle Bluetooth on or off. + + [DEVELOPER MODE] Enables or disables Bluetooth using the svc command. + + Args: + enabled: True to enable Bluetooth, False to disable + device_id: Target device + + Returns: + Result with action taken + """ + if not is_developer_mode(): + return { + "success": False, + "error": "Developer mode required", + } + + action = "enable" if enabled else "disable" + result = await self.run_shell_args(["svc", "bluetooth", action], device_id) + + return { + "success": result.success, + "action": action, + "error": result.stderr if not result.success else None, + } + + @mcp_tool( + tags={"developer"}, + annotations={"requires": "developer_mode"}, + ) + async def airplane_mode_toggle( + self, + ctx: Context, + enabled: bool, + device_id: str | None = None, + ) -> dict[str, Any]: + """Toggle airplane mode on or off. + + [DEVELOPER MODE] Enables or disables airplane mode. + Requires confirmation because enabling airplane mode on a + network-connected device will sever the ADB connection. + + Args: + ctx: MCP context for elicitation/logging + enabled: True to enable airplane mode, False to disable + device_id: Target device + + Returns: + Result with airplane mode state + """ + if not is_developer_mode(): + return { + "success": False, + "error": "Developer mode required", + } + + # Warn about network disconnection risk + if enabled: + target = device_id or self.get_current_device() + is_network = target and re.match(r"\d+\.\d+\.\d+\.\d+:\d+", target) + + warning = "Enable airplane mode?" + if is_network: + warning = ( + "WARNING: Enabling airplane mode on a network-connected " + f"device ({target}) will sever the ADB connection. " + "You will not be able to reconnect until airplane mode " + "is manually disabled on the device.\n\nProceed?" + ) + + confirmation = await ctx.elicit( + warning, + ["Yes, enable airplane mode", "Cancel"], + ) + if confirmation.action != "accept" or confirmation.content == "Cancel": + return { + "success": False, + "cancelled": True, + "message": "Airplane mode toggle cancelled by user", + } + + # Set the setting + value = "1" if enabled else "0" + put_result = await self.run_shell_args( + ["settings", "put", "global", "airplane_mode_on", value], device_id + ) + + if not put_result.success: + return { + "success": False, + "error": put_result.stderr, + } + + # Broadcast the change so the system acts on it + await self.run_shell_args( + [ + "am", + "broadcast", + "-a", + "android.intent.action.AIRPLANE_MODE", + "--ez", + "state", + str(enabled).lower(), + ], + device_id, + ) + + action = "enabled" if enabled else "disabled" + await ctx.info(f"Airplane mode {action}") + + return { + "success": True, + "airplane_mode": enabled, + "action": action, + } + + @mcp_tool( + tags={"developer"}, + annotations={"requires": "developer_mode"}, + ) + async def screen_brightness( + self, + level: int, + device_id: str | None = None, + ) -> dict[str, Any]: + """Set screen brightness level. + + [DEVELOPER MODE] Sets the screen brightness to a specific level. + Automatically disables auto-brightness first. + + Args: + level: Brightness level 0-255 (0=minimum, 255=maximum) + device_id: Target device + + Returns: + Result with brightness level set + """ + if not is_developer_mode(): + return { + "success": False, + "error": "Developer mode required", + } + + if not 0 <= level <= 255: + return { + "success": False, + "error": f"Brightness level must be 0-255, got {level}", + } + + # Disable auto-brightness first + await self.run_shell_args( + ["settings", "put", "system", "screen_brightness_mode", "0"], + device_id, + ) + + # Set brightness + result = await self.run_shell_args( + ["settings", "put", "system", "screen_brightness", str(level)], + device_id, + ) + + return { + "success": result.success, + "brightness": level, + "auto_brightness": False, + "error": result.stderr if not result.success else None, + } + + @mcp_tool( + tags={"developer"}, + annotations={"requires": "developer_mode"}, + ) + async def screen_timeout( + self, + seconds: int, + device_id: str | None = None, + ) -> dict[str, Any]: + """Set screen timeout duration. + + [DEVELOPER MODE] Sets how long the screen stays on before + going to sleep. Capped at 1800 seconds (30 minutes). + + Args: + seconds: Timeout in seconds (max 1800) + device_id: Target device + + Returns: + Result with timeout value set + """ + if not is_developer_mode(): + return { + "success": False, + "error": "Developer mode required", + } + + if seconds < 1 or seconds > 1800: + return { + "success": False, + "error": f"Timeout must be 1-1800 seconds, got {seconds}", + } + + # Android stores timeout in milliseconds + ms = seconds * 1000 + result = await self.run_shell_args( + ["settings", "put", "system", "screen_off_timeout", str(ms)], + device_id, + ) + + return { + "success": result.success, + "timeout_seconds": seconds, + "timeout_ms": ms, + "error": result.stderr if not result.success else None, + } + + @mcp_tool() + async def notification_list( + self, + limit: int = 50, + device_id: str | None = None, + ) -> dict[str, Any]: + """List recent notifications. + + Retrieves notifications from the notification shade. + Parses title, text, package, and post time from dumpsys output. + + Args: + limit: Maximum number of notifications to return (default 50) + device_id: Target device + + Returns: + List of notifications with title, text, package, and time + """ + result = await self.run_shell_args( + ["dumpsys", "notification", "--noredact"], device_id + ) + + if not result.success: + return { + "success": False, + "error": result.stderr, + } + + notifications: list[dict[str, str | None]] = [] + current: dict[str, str | None] = {} + + for line in result.stdout.split("\n"): + stripped = line.strip() + + # Look for notification record boundaries + if "NotificationRecord" in stripped: + if current: + notifications.append(current) + if len(notifications) >= limit: + break + current = {} + # Extract package from NotificationRecord line + pkg_match = re.search(r"pkg=(\S+)", stripped) + if pkg_match: + current["package"] = pkg_match.group(1) + + # Extract fields from extras or notification content + if "android.title=" in stripped: + title_match = re.search(r"android\.title=(.+?)(?:\s*$)", stripped) + if title_match: + current["title"] = title_match.group(1).strip() + + if "android.text=" in stripped: + text_match = re.search(r"android\.text=(.+?)(?:\s*$)", stripped) + if text_match: + current["text"] = text_match.group(1).strip() + + if "postTime=" in stripped: + time_match = re.search(r"postTime=(\d+)", stripped) + if time_match: + current["post_time"] = time_match.group(1) + + # Don't forget the last one + if current and len(notifications) < limit: + notifications.append(current) + + return { + "success": True, + "notifications": notifications, + "count": len(notifications), + } + + @mcp_tool() + async def clipboard_get( + self, + device_id: str | None = None, + ) -> dict[str, Any]: + """Read the device clipboard contents. + + Retrieves the current text from the device clipboard. + Uses service call to read the primary clip via Binder IPC, + which works across Android versions without requiring + a specific shell command implementation. + + Args: + device_id: Target device + + Returns: + Clipboard text content + """ + # getPrimaryClip via Binder (transaction 4 on Android 12+) + result = await self.run_shell_args( + [ + "service", + "call", + "clipboard", + "4", + "s16", + "com.android.shell", + ], + device_id, + ) + + if result.success and "Parcel(" in result.stdout: + text = self._parse_clipboard_parcel(result.stdout) + if text is not None: + return { + "success": True, + "text": text, + "method": "service_call", + } + + return { + "success": False, + "error": ( + "Could not read clipboard. The device may have " + "an empty clipboard or use an unsupported format." + ), + } + + @staticmethod + def _parse_clipboard_parcel(raw: str) -> str | None: + """Parse clipboard text from service call parcel output. + + The Parcel contains a ClipData object with: + - Status word, non-null marker + - ClipDescription (label, MIME types in UTF-16LE) + - Extras, timestamps, flags + - Item count, then for each item: CharSequence as + a type marker + length-prefixed UTF-8 text + + We scan for length-prefixed UTF-8 text blocks after + the MIME type section and return the longest valid one. + """ + # Extract hex words from parcel dump + hex_words = re.findall(r"\b([0-9a-f]{8})\b", raw) + if not hex_words: + return None + + # Convert to bytes (little-endian 32-bit words) + data = b"".join(int(w, 16).to_bytes(4, "little") for w in hex_words) + + # Check status (first 4 bytes should be 0 for success) + if len(data) < 8: + return None + status = int.from_bytes(data[0:4], "little") + if status != 0: + return None + + # Find "text/plain" in UTF-16LE to locate the MIME section + mime_marker = "text/plain".encode("utf-16-le") + mime_pos = data.find(mime_marker) + if mime_pos == -1: + return None + + # Scan from after MIME section for length-prefixed UTF-8 + # text blocks. Return the longest valid one found. + search_start = mime_pos + len(mime_marker) + best_text: str | None = None + best_len = 0 + + pos = search_start + while pos < len(data) - 8: + text_len = int.from_bytes(data[pos : pos + 4], "little") + if 2 < text_len < 100000 and pos + 4 + text_len <= len(data): + text_bytes = data[pos + 4 : pos + 4 + text_len] + try: + text = text_bytes.decode("utf-8") + # Verify mostly printable (not binary garbage) + printable = sum(1 for c in text if c.isprintable() or c in "\n\r\t") + ratio = printable / max(len(text), 1) + if ratio > 0.8 and text_len > best_len: + best_text = text + best_len = text_len + except UnicodeDecodeError: + pass + pos += 4 + + return best_text + + @mcp_tool() + async def media_control( + self, + action: str, + device_id: str | None = None, + ) -> dict[str, Any]: + """Control media playback. + + Sends media key events to control the active media player. + + Available actions: + - play: Start playback + - pause: Pause playback + - play_pause: Toggle play/pause + - next: Skip to next track + - previous: Go to previous track + - stop: Stop playback + - mute: Toggle mute + - volume_up: Increase volume + - volume_down: Decrease volume + + Args: + action: Media action (play, pause, next, previous, stop, etc.) + device_id: Target device + + Returns: + Success status with action performed + """ + action_lower = action.lower().strip() + keycode = _MEDIA_KEYCODES.get(action_lower) + + if not keycode: + return { + "success": False, + "error": ( + f"Unknown action '{action}'. " + f"Available: {', '.join(sorted(_MEDIA_KEYCODES))}" + ), + } + + result = await self.run_shell_args(["input", "keyevent", keycode], device_id) + + return { + "success": result.success, + "action": action_lower, + "keycode": keycode, + "error": result.stderr if not result.success else None, + } diff --git a/src/server.py b/src/server.py index 211d479..5e1a8b7 100644 --- a/src/server.py +++ b/src/server.py @@ -21,26 +21,37 @@ from fastmcp.contrib.mcp_mixin import mcp_resource, mcp_tool from .config import get_config from .mixins import ( AppsMixin, + ConnectivityMixin, DevicesMixin, FilesMixin, InputMixin, ScreenshotMixin, + SettingsMixin, UIMixin, ) class ADBServer( - DevicesMixin, InputMixin, AppsMixin, ScreenshotMixin, UIMixin, FilesMixin + DevicesMixin, + ConnectivityMixin, + InputMixin, + AppsMixin, + ScreenshotMixin, + UIMixin, + FilesMixin, + SettingsMixin, ): """Android ADB MCP Server combining all functionality. Inherits from mixins: - DevicesMixin: Device listing, selection, info, logcat + - ConnectivityMixin: TCP/IP connect/disconnect, pairing, device properties - InputMixin: Tap, swipe, keys, text input, clipboard - AppsMixin: App launching, URL opening, package management, intents - ScreenshotMixin: Screen capture, recording, display control - UIMixin: UI hierarchy inspection, element finding, text waiting - FilesMixin: File push/pull between host and device + - SettingsMixin: System settings, radios, brightness, notifications, media Developer mode enables additional tools for power users. """ @@ -131,6 +142,12 @@ class ADBServer( "devices_current - Get current device info", "device_info - Battery, wifi, storage, system info", ], + "connectivity": [ + "adb_connect - Connect to device over TCP/IP", + "adb_disconnect - Disconnect network device", + "adb_pair - Wireless debugging pairing (Android 11+)", + "device_properties - Batch getprop (model, SoC, versions)", + ], "input": [ "input_tap - Tap at coordinates", "input_swipe - Swipe between points", @@ -158,6 +175,12 @@ class ADBServer( "wait_for_text_gone - Wait for text to disappear", "tap_text - Find element by text and tap it", ], + "settings": [ + "settings_get - Read any system/global/secure setting", + "notification_list - List recent notifications", + "clipboard_get - Read clipboard contents", + "media_control - Play/pause/next/previous/stop/volume", + ], "config": [ "config_status - Show current settings", "config_set_developer_mode - Toggle developer tools", @@ -180,6 +203,13 @@ class ADBServer( "logcat_capture / logcat_clear - Android logs", "file_push / file_pull - Transfer files", "file_list / file_delete / file_exists - File operations", + "adb_tcpip - Switch USB device to TCP/IP mode", + "settings_put - Write system/global/secure setting", + "wifi_toggle - Enable/disable WiFi", + "bluetooth_toggle - Enable/disable Bluetooth", + "airplane_mode_toggle - Toggle airplane mode", + "screen_brightness - Set brightness 0-255", + "screen_timeout - Set screen timeout", ] return { @@ -202,12 +232,18 @@ mcp = FastMCP( Use devices_list() first to see connected devices. If multiple devices are connected, use devices_use(device_id) to select one. +For network devices, use adb_connect(host, port) to establish a connection. +Use adb_pair() for wireless debugging on Android 11+. + Common workflows: 1. Take screenshot: screenshot() 2. Tap on screen: input_tap(x, y) 3. Launch app: app_launch("com.android.chrome") 4. Open URL: app_open_url("https://example.com") 5. Navigate back: input_back() +6. Connect WiFi device: adb_connect("10.20.0.25") +7. Read a setting: settings_get("global", "wifi_on") +8. Control media: media_control("pause") Enable developer mode for advanced tools: config_set_developer_mode(True) @@ -226,7 +262,7 @@ def main(): package_version = version("mcadb") except Exception: - package_version = "0.3.1" + package_version = "0.4.0" print(f"📱 mcadb v{package_version}", flush=True) diff --git a/uv.lock b/uv.lock index 72a2b6b..4de67b9 100644 --- a/uv.lock +++ b/uv.lock @@ -735,7 +735,7 @@ wheels = [ [[package]] name = "mcadb" -version = "0.3.1" +version = "0.4.0" source = { editable = "." } dependencies = [ { name = "fastmcp" },