"""Connectivity mixin for Android ADB MCP Server. Provides tools for managing ADB network connections and device properties. """ import re from fastmcp import Context from fastmcp.contrib.mcp_mixin import mcp_tool from ..config import is_developer_mode from ..models import ConnectResult, DevicePropertiesResult, TcpipResult 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, ) -> ConnectResult: """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 ConnectResult( 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, ) -> ConnectResult: """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 ConnectResult( 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, ) -> TcpipResult: """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 TcpipResult( 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 TcpipResult( 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 TcpipResult( 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 TcpipResult( 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 TcpipResult( 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, ) -> ConnectResult: """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 ConnectResult( 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, ) -> DevicePropertiesResult: """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: 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"), ], } categories: dict[str, dict[str, str]] = {} 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: categories[category] = category_data # Check if we got anything at all if not categories: return DevicePropertiesResult( success=False, error="No properties returned. Is the device connected?", ) return DevicePropertiesResult( success=True, identity=categories.get("identity"), software=categories.get("software"), hardware=categories.get("hardware"), system=categories.get("system"), )