Aadb

English | 简体中文 | 繁體中文

aadb is an async-only Python 3.11+ client for the Android Debug Bridge server protocol. It talks directly to the ADB server smart socket and exposes ADB operations as native asyncio APIs.

This project does not wrap the adb command line executable and does not start adb server for you.

ADB Server Requirement

aadb connects to an already-running ADB server, usually 127.0.0.1:5037. ADB server startup, USB authorization, device pairing, and platform-tool installation are still handled by the official Android platform tools.

Before using aadb with real devices, make sure the official ADB server is running:

adb start-server
adb devices -l

If Android Studio or another ADB client has already started the server, no extra step is needed. If no ADB server is listening on the configured host and port, aadb raises AdbConnectionError when it tries to connect.

Use a custom server endpoint when needed:

from aadb import AdbClient

adb = AdbClient(host="127.0.0.1", port=5037)

Status

Version 2 is a breaking rewrite of the original project.

Implemented areas:

• ADB smart socket framing with OKAY and FAIL handling.
• Host services: version, features, host-features, server-status, devices, devices-l, track-devices, connect, disconnect, kill-server, and forward list.
• Device selection by serial or transport id.
• Shell v2 with stdout, stderr, stdin, and exit code.
• Legacy shell fallback when shell_v2 is not advertised.
• Sync v2 stat, list, push, and pull with sync v1 fallback.
• Device helpers for logcat, getprop, package listing, install, uninstall, forward, reverse, reboot, tcpip, usb, root, unroot, remount, and track-jdwp.
• Low-level escape hatches through adb.host_service(raw) and device.service(raw).

Not yet a full replacement for the official adb CLI:

• Sync v2 compression flags are negotiated, but payload compression requires optional dependencies and still falls back to uncompressed transfer when unavailable.
• Pairing, auth key management, mDNS discovery, bugreport, sideload, verity, reconnect, incremental install, fastdeploy, framebuffer, abb, abb_exec, track-app, and protobuf tracking APIs are not fully wrapped.
• Mock tests run by default; optional pytest -m emulator tests verify a real Android emulator through the official ADB server.

MCP Server

aadb can also be installed as an MCP server for local agents:

python -m pip install ".[mcp]"
aadb-mcp

HTTP transport is available for MCP Inspector or service-style debugging:

aadb-mcp --transport streamable-http --host 127.0.0.1 --port 8765 --http-token change-me

Example stdio client configuration:

{
  "mcpServers": {
    "aadb": {
      "command": "aadb-mcp",
      "args": ["--transport", "stdio"],
      "env": {
        "AADB_HOST": "127.0.0.1",
        "AADB_PORT": "5037",
        "AADB_MCP_PROFILE": "dev-open"
      }
    }
  }
}

The default MCP profile is dev-open. It allows agents to run shell commands, push and pull local files, install and uninstall packages, create forwards, reboot devices, switch adbd modes, and call raw ADB services. Use --profile safe-readonly in shared or production environments. HTTP dev-open mode requires a bearer token and is localhost-only unless --allow-remote-http is explicitly set.

Read the full MCP guide: English | 简体中文 | 繁體中文. Agents that need to install from a GitHub URL should read Agent MCP Installation Guide.

Installation

Install from the project root:

python -m pip install .

For editable development with test tools:

python -m pip install -e ".[dev]"

Python 3.11 or newer is required.

Quick Start

Start or verify the official ADB server first:

adb start-server
adb devices -l

import asyncio

from aadb import AdbClient


async def main() -> None:
    async with AdbClient() as adb:
        device = await adb.device()
        result = await device.shell("getprop ro.product.model")
        print(result.text.strip())


asyncio.run(main())

Use a non-default ADB server endpoint:

async with AdbClient(host="127.0.0.1", port=5037, timeout=10.0) as adb:
    print(await adb.version())

Host APIs

List devices:

async with AdbClient() as adb:
    for info in await adb.devices():
        print(info.serial, info.state, info.model, info.transport_id)

Select a device:

async with AdbClient() as adb:
    device = await adb.device("emulator-5554")

If no serial is provided, adb.device() selects the only online device. It raises DeviceSelectionError when multiple online devices are present.

Track device changes:

async with AdbClient() as adb:
    async for devices in adb.track_devices():
        print([device.serial for device in devices])

Connect to a TCP device:

async with AdbClient() as adb:
    message = await adb.connect("192.168.1.25", 5555)
    print(message)

Low-level host service call:

async with AdbClient() as adb:
    raw = await adb.host_service("host:version")
    print(raw)

Device APIs

Read device metadata:

async with AdbClient() as adb:
    device = await adb.device()
    print(await device.get_serialno())
    print(await device.get_state())
    print(await device.features())

Run a shell command:

result = await device.shell("id")
print(result.stdout)
print(result.stderr)
print(result.exit_code)

ShellResult.text decodes stdout as UTF-8 with replacement:

result = await device.shell("getprop ro.build.version.release")
print(result.text.strip())

Stream shell output line by line:

async for line in device.shell_stream("logcat -d"):
    print(line)

Send stdin to shell v2:

result = await device.shell("cat", stdin=b"hello\n")
print(result.text)

Run exec-out style commands:

png = await device.exec_out("screencap -p")

Logcat

device.logcat() returns an async iterator of decoded lines:

async for line in device.logcat("-d", "*:I"):
    print(line)

For long-running streams, no operation timeout is applied by default after the connection has been opened. Pass timeout=... only when the whole stream should be bounded.

Files

Push a file:

await device.push("local.txt", "/data/local/tmp/local.txt")

Push a directory recursively:

await device.push("fixtures", "/data/local/tmp/fixtures")

Pull a file:

await device.pull("/data/local/tmp/local.txt", "downloaded.txt")

Pull a directory recursively:

await device.pull("/data/local/tmp/fixtures", "fixtures-copy")

Inspect remote files:

info = await device.stat("/data/local/tmp/local.txt")
print(info.path, info.size, info.kind)

for entry in await device.list("/data/local/tmp"):
    print(entry.path, entry.kind)

Packages

List packages:

packages = await device.list_packages()
print(packages)

Find package APK paths:

paths = await device.package_path("com.example.app")
print(paths)

Install one APK:

from aadb import InstallOptions

await device.install(
    "app-debug.apk",
    options=InstallOptions(reinstall=True, grant_all_permissions=True),
)

Install split APKs:

await device.install(
    ["base.apk", "config.arm64_v8a.apk", "config.xxhdpi.apk"],
    options=InstallOptions(reinstall=True),
)

Uninstall a package:

removed = await device.uninstall("com.example.app")
print(removed)

Forward and Reverse

Forward a host port to the device:

await device.forward("tcp:9000", "tcp:9000")

List and remove forwards:

print(await device.list_forward())
await device.kill_forward("tcp:9000")
await device.kill_forward_all()

Reverse a device port to the host:

await device.reverse("tcp:9000", "tcp:9000")
print(await device.list_reverse())
await device.kill_reverse("tcp:9000")
await device.kill_reverse_all()

Device Control

await device.reboot()
await device.reboot("bootloader")
await device.tcpip(5555)
await device.usb()
await device.root()
await device.unroot()
await device.remount()

These operations depend on device build type, authorization state, and adbd permissions. Unsupported operations raise typed aadb exceptions.

Timeouts and Cancellation

AdbClient(timeout=30.0) sets the default timeout for connection setup and one-shot operations. Individual calls can override it:

await device.shell("sleep 1", timeout=2.0)

Long-running streams such as track_devices(), shell_stream(), and logcat() keep running until the iterator is cancelled or closed unless an explicit timeout is passed.

Errors

All project-specific exceptions inherit from AdbError:

from aadb import AdbError, AdbServerError, DeviceSelectionError

try:
    async with AdbClient() as adb:
        device = await adb.device()
        await device.shell("false")
except DeviceSelectionError:
    print("Choose a serial explicitly")
except AdbServerError as exc:
    print("ADB server rejected the request:", exc)
except AdbError as exc:
    print("ADB operation failed:", exc)

Low-Level Services

Use low-level service calls when a high-level wrapper is not available yet:

async with AdbClient() as adb:
    device = await adb.device()
    raw = await device.service("exec:getprop ro.serialno")
    print(raw.decode().strip())

For host-scoped services:

async with AdbClient() as adb:
    raw = await adb.host_service("host:server-status")
    print(raw.decode(errors="replace"))

Testing

Run the test suite:

python -m pytest -q

Run lint and type checks:

python -m ruff check .
python -m ruff format --check .
python -m mypy aadb

The test suite uses aadb.testing.MockAdbServer. It validates protocol framing, device selection, shell v2, sync file transfer, FAIL handling, timeouts, and cancellation without requiring a physical Android device.

Project Layout

Important modules:

• aadb.client: host-level AdbClient APIs.
• aadb.device: device-scoped APIs.
• aadb.connection: one smart socket connection.
• aadb.protocol: smart socket framing helpers.
• aadb.shell: shell v2 and legacy shell helpers.
• aadb.sync: sync service implementation.
• aadb.types: public dataclasses.
• aadb.errors: typed exception hierarchy.
• aadb.testing: mock ADB server for tests.