feat: add Android ADB MCP server#17
Conversation
There was a problem hiding this comment.
Code Review
This pull request introduces an Android ADB control layer for OpenChronicle, enabling agents to interact with Android devices through a secure MCP tool surface. The implementation includes a safety policy to block destructive commands, a memory adapter to log all operations into daily event files, and a suite of tools for screen interaction, app management, and log retrieval. The review feedback identifies several critical improvement opportunities: strengthening security against flag injection in logcat and keyevent bypasses, increasing the robustness of device list parsing and UI dumping, and optimizing performance when querying the current application state.
| if not re.match(r"^[A-Za-z0-9_.*:\-\s]+$", filter_expr): | ||
| exc = safety.ADBSafetyError("filter_expr contains unsupported characters") | ||
| display = self.client.command_for_display(command, device_id) | ||
| return self._blocked_payload( | ||
| tool_name=tool, device_id=device_id, command=display, params=params, exc=exc | ||
| ) | ||
| command.extend(filter_expr.split()) |
There was a problem hiding this comment.
The filter_expr validation allows the character -, which enables an agent to pass arbitrary flags to logcat. For example, a filter of -c would clear the device logs, which is a destructive operation. You should ensure that tokens in the filter expression do not start with a dash to prevent flag injection.
| if not re.match(r"^[A-Za-z0-9_.*:\-\s]+$", filter_expr): | |
| exc = safety.ADBSafetyError("filter_expr contains unsupported characters") | |
| display = self.client.command_for_display(command, device_id) | |
| return self._blocked_payload( | |
| tool_name=tool, device_id=device_id, command=display, params=params, exc=exc | |
| ) | |
| command.extend(filter_expr.split()) | |
| if filter_expr: | |
| if not re.match(r"^[A-Za-z0-9_.*:\-\s]+$", filter_expr): | |
| exc = safety.ADBSafetyError("filter_expr contains unsupported characters") | |
| display = self.client.command_for_display(command, device_id) | |
| return self._blocked_payload( | |
| tool_name=tool, device_id=device_id, command=display, params=params, exc=exc | |
| ) | |
| tokens = filter_expr.split() | |
| if any(t.startswith("-") for t in tokens): | |
| exc = safety.ADBSafetyError("filter_expr cannot contain logcat flags") | |
| display = self.client.command_for_display(command, device_id) | |
| return self._blocked_payload( | |
| tool_name=tool, device_id=device_id, command=display, params=params, exc=exc | |
| ) | |
| command.extend(tokens) |
| key = shell_tokens[idx + 1].removeprefix("keycode_") | ||
| original = shell_tokens[idx + 1] | ||
| if original in _BLOCKED_KEYEVENTS or key in _BLOCKED_KEYEVENTS: | ||
| raise ADBSafetyError(f"blocked high-risk keyevent: {shell_tokens[idx + 1]}") |
There was a problem hiding this comment.
The safety check for keyevents can be bypassed by using alternative representations of the blocked keycodes, such as hexadecimal (e.g., 0x1a for 26). It is safer to attempt to parse the token as an integer and check its value against the blocked list.
raw_key = shell_tokens[idx + 1].lower().removeprefix("keycode_")
is_blocked = shell_tokens[idx + 1].lower() in _BLOCKED_KEYEVENTS or raw_key in _BLOCKED_KEYEVENTS
if not is_blocked:
try:
# Check if the numeric value matches a blocked keycode
if str(int(raw_key, 0)) in _BLOCKED_KEYEVENTS:
is_blocked = True
except ValueError:
pass
if is_blocked:
raise ADBSafetyError(f"blocked high-risk keyevent: {shell_tokens[idx + 1]}")| for raw_line in output.splitlines()[1:]: | ||
| line = raw_line.strip() |
There was a problem hiding this comment.
Parsing adb devices by skipping the first line is fragile. If the ADB daemon is not running, it may output status messages (e.g., "* daemon started successfully") before the header, causing the parser to incorrectly identify these messages as device serials.
| for raw_line in output.splitlines()[1:]: | |
| line = raw_line.strip() | |
| lines = output.splitlines() | |
| try: | |
| # Find the header and start parsing from the next line | |
| header_idx = next(i for i, line in enumerate(lines) if "List of devices attached" in line) | |
| lines_to_parse = lines[header_idx + 1:] | |
| except StopIteration: | |
| return [] | |
| for raw_line in lines_to_parse: |
|
|
||
| def dump_ui(self, device_id: str | None = None) -> dict[str, Any]: | ||
| tool = "adb_dump_ui" | ||
| dump_path = "/sdcard/window.xml" |
There was a problem hiding this comment.
Using /sdcard/window.xml for the UI dump can fail on devices where the external storage is not writable or is restricted by scoped storage. /data/local/tmp/ is a more reliable location for ADB-driven scripts as it is generally writable by the shell user across all Android versions.
| dump_path = "/sdcard/window.xml" | |
| dump_path = "/data/local/tmp/window.xml" |
|
|
||
| def current_app(self, device_id: str | None = None) -> dict[str, Any]: | ||
| tool = "adb_current_app" | ||
| command = ["shell", "dumpsys", "window"] |
There was a problem hiding this comment.
Running dumpsys window without arguments is extremely heavy, as it dumps the state of all windows and can produce several megabytes of text. This impacts performance and memory usage. Using dumpsys window windows or dumpsys window displays is significantly more efficient for retrieving the focused application.
| command = ["shell", "dumpsys", "window"] | |
| command = ["shell", "dumpsys", "window", "windows"] |
|
I might need to make some changes using Gemini Relay. I need to rest now; I've run out of tokens. |
Summary
Adds a separate Android ADB Control MCP server for OpenChronicle.
Adds
openchronicle adb-mcpstdio MCP server.Exposes ADB tools for device listing, screenshot, UI dump, tap, swipe, text input, keyevent,
current app, app launch, and logcat.
Writes every ADB tool attempt into OpenChronicle daily event memory.
Adds
safety.pydenylist for destructive or privileged ADB commands.Adds Windows 11 + WSL2 setup docs in
README_ADB_AGENT.md.adb_input_textadb_keyeventadb_current_appadb_open_appadb_read_logcatValidation
uv run pytest --basetemp .pytest_cache\full-test-tmp-2uv run python -m compileall src\openchronicle\adb src\openchronicle\mcp\adb_server.pyE:\ai\product\.tooling\platform-tools\adb.exe devices -llist_tools+adb_list_devicessuccessfulResult:
105 passed, 1 skipped.