MAS SDK Reference: Python Android Automation API

This page is the mas SDK reference: every public name in the mas package that ships with Macro Automation Studio (MAS), with the signature as written in the source, each parameter with its default, what comes back, what can raise, and a short example. Read the SDK overview first if you have not written a macro yet. The namespace pages explain when to use each call; this page is the lookup table. If you describe the task to MAS Agent, it writes this code for you.

Every signature below is copied from the SDK source. mas.__version__ is "0.1.0".

Namespace map of the mas SDK reference

Everything is importable from mas directly, for example from mas import Region, KeyCode. Only the runtime UI lives in its own namespace, mas.ui.

Types

The types are plain dataclasses, enums and type aliases. Import them from mas.

Coordinates

Coordinates = tuple[int, int]

An (x, y) pair in pixels, measured from the top-left corner of the screen. swipe, zoom_in and zoom_out take one; ObjectMatch.center and ElementMatch.center return one.

import mas

start: mas.Coordinates = (540, 1600)
end: mas.Coordinates = (540, 600)
mas.swipe(start, end, duration_ms=400)

Region

class Region

A rectangle on screen given by two corners. find_object, find_objects and find_any_object take one as search_region; read_text takes one as region.

import mas
from mas import Region

score_box = Region(x1=800, y1=10, x2=1050, y2=60)
result = mas.read_text(region=score_box)

ColorConversion

class ColorConversion(Enum)

Preprocessing applied before OCR in read_text. The values match OpenCV’s color conversion codes; two are MAS additions. Members that share a value are aliases of each other.

There is no GRAYSCALE or BINARY member. Use BGR_TO_GRAY for grayscale and BLACK_WHITE for a binary image.

import mas
from mas import ColorConversion, Region

word = mas.read_text(region=Region(100, 200, 400, 260), psm=8,
                     color_conversion=ColorConversion.BLACK_WHITE)

SearchStrategy

SearchStrategy = Literal["first_match", "best_match", "priority_order"]

How find_any_object chooses among several templates.

DeviceType

DeviceType = Literal["android", "desktop"]

The kind of device bound to the run, as reported in DeviceInfo.type.

DeviceInfo

class DeviceInfo

Returned by get_device_info.

ScreenSize

class ScreenSize

Returned by get_screen_size.

ObjectMatch

class ObjectMatch

The result of a template match. x and y are the center of the matched area. There is no confidence, width or height field: the host applies threshold before it answers, so a returned match already passed it.

Properties: center returns (x, y) as Coordinates; template_id is an alias for matched_template_id.

import mas

match = mas.find_object(42)
if match:
    mas.click(*match.center)
    print(match.x, match.y, match.template_id)

Screenshot

class Screenshot

Returned by take_screenshot. The image stays in memory as base64-encoded PNG.

TextRecognitionResult

class TextRecognitionResult

Returned by read_text.

KeyCode

class KeyCode(Enum)

Android hardware key codes for key_press. Each member’s value is the Android key code string. There are 33 members; there is no ESCAPE.

import mas
from mas import KeyCode

mas.key_press(KeyCode.BACK)
mas.key_press(KeyCode.DELETE, repeat=10)

Modifier

class Modifier(Enum)

Modifier keys for key_press: SHIFT ("shift"), CTRL ("ctrl"), ALT ("alt"), META ("meta").

LogLevel

class LogLevel(Enum)

Severity for log: DEBUG ("debug"), INFO ("info"), WARNING ("warning"), ERROR ("error"). log also accepts the plain strings.

ExecutionResult

class ExecutionResult

A generic command result. It is exported for typing; no function on this page returns it.

Interaction

Touch and keyboard input. Coordinates are pixels on the device screen. Each call blocks until the host confirms the command and then returns None. A failure on the device surfaces as an RPCError subclass (see Exceptions): DeviceNotConnectedError when no device is bound to the run, CommandFailedError when the device rejects the command.

click

def click(x: int, y: int, delay_ms: int = 1000) -> None

Tap once at (x, y), then pause for delay_ms so the app can react.

Returns: None.

Raises: an RPCError subclass when the host reports a failure.

import mas

mas.click(540, 1200)                # tap, then wait 1000 ms
mas.click(540, 1200, delay_ms=0)    # no pause, for tight loops
match = mas.find_object_retry(42)
if match:
    mas.click(match.x, match.y, delay_ms=1500)

swipe

def swipe(from_coords: Coordinates, to_coords: Coordinates, duration_ms: int = 1000) -> None

Drag from one point to another. A short duration is a flick; a long one is a controlled drag.

Returns: None.

Raises: an RPCError subclass when the host reports a failure.

import mas

mas.swipe((500, 1500), (500, 500), duration_ms=400)    # swipe up to scroll down
mas.swipe((100, 500), (900, 500), duration_ms=300)     # swipe right
mas.swipe((200, 300), (800, 300), duration_ms=1000)    # slow drag

input_text

def input_text(text: str, delay_ms: int = 0, clear: bool = False) -> None

Type into the field that currently has focus. Tap the field first. With clear=True the cursor moves to the end of the field and the existing characters are deleted before typing, so the field does not need to be empty.

Returns: None.

Raises: an RPCError subclass when the host reports a failure.

import mas

mas.click(300, 450)                       # focus the field
mas.input_text("user@example.com")
mas.input_text("new value", clear=True)   # replace whatever is there
mas.input_text("done", delay_ms=500)

key_press

def key_press(key_code: KeyCode, modifiers: list[Modifier] | None = None, duration_ms: int = 100, repeat: int = 1) -> None

Press a hardware key, optionally with modifiers. A duration_ms of 500 or more sends a long press; smaller values send a normal press. The Android input bridge has no arbitrary hold, so the exact number of milliseconds is not honored beyond that distinction. All repeat presses go to the device in one command, which is much faster than a Python loop. Holding DELETE does not clear a field; use repeat to delete several characters, or input_text(..., clear=True).

Returns: None.

Raises: an RPCError subclass when the host reports a failure.

import mas
from mas import KeyCode, Modifier

mas.key_press(KeyCode.BACK)
mas.key_press(KeyCode.ENTER)
mas.key_press(KeyCode.POWER, duration_ms=3000)        # long press
mas.key_press(KeyCode.DELETE, repeat=10)              # delete 10 characters
mas.key_press(KeyCode.TAB, modifiers=[Modifier.SHIFT])

zoom_in

def zoom_in(center: Coordinates | None = None, percent: int = 50, duration_ms: int = 0, steps: int = 10) -> None

A two-finger pinch that spreads the fingers apart. The gesture is synthesized through the emulator’s touch input device, which is detected automatically.

Returns: None.

Raises: an RPCError subclass when the host reports a failure.

import mas

mas.zoom_in()                                   # at the screen center
mas.zoom_in(center=(540, 960), percent=80)      # strong zoom at a point

zoom_out

def zoom_out(center: Coordinates | None = None, percent: int = 50, duration_ms: int = 0, steps: int = 10) -> None

A two-finger pinch that brings the fingers together. Same mechanics and parameters as zoom_in; percent is how far apart the fingers start.

Returns: None.

Raises: an RPCError subclass when the host reports a failure.

import mas

mas.zoom_out()
mas.zoom_out(center=(540, 960), percent=30)     # gentle zoom out at a point

Vision

A macro decides what to do by looking at the screen. Template images live in your Image Library in the cloud, each with an integer ID; you crop them in Asset Lab. Vision calls take that ID, or an ImageRef from images. Every find_* call returns None (or an empty list) on a miss and never raises for a miss. Take one screenshot and pass it to several searches when you need to check more than one thing on the same frame.

take_screenshot

def take_screenshot() -> Screenshot

Capture the device screen as an in-memory base64 PNG.

Returns: a Screenshot with base64, width, height and timestamp.

Raises: an RPCError subclass when the capture fails.

import mas
import base64

shot = mas.take_screenshot()
play = mas.find_object(42, screenshot=shot)
coins = mas.find_object(43, screenshot=shot)
with open("screen.png", "wb") as f:
    f.write(base64.b64decode(shot.base64))

find_object

def find_object(image_id: int | ImageRef, *, threshold: float = 0.8, screenshot: Screenshot | None = None, continuous_mode: bool = False, timeout_ms: int = 1000, capture_interval_ms: int = 500, max_matches: int = 1, search_region: Region | None = None) -> ObjectMatch | None

Find one template on screen by template matching. All options after image_id are keyword-only.

Returns: an ObjectMatch when a match reaches threshold, otherwise None.

Raises: ImageNotFoundError when the ID is not in your Image Library or cannot be loaded; TypeError when image_id is neither an int nor an ImageRef; other RPCError subclasses on host failures.

import mas
from mas import Region

images = mas.images({"login_btn": 123, "badge": 124})
match = mas.find_object(images.login_btn)
if match:
    mas.click(*match.center)
top_left = mas.find_object(images.badge, search_region=Region(0, 0, 500, 400))

find_objects

def find_objects(image_id: int | ImageRef, *, threshold: float = 0.8, screenshot: Screenshot | None = None, continuous_mode: bool = False, timeout_ms: int = 1000, capture_interval_ms: int = 500, max_matches: int = 10, search_region: Region | None = None) -> list[ObjectMatch]

Find every occurrence of one template. The parameters match find_object; only the max_matches default differs.

Returns: a list of ObjectMatch sorted by confidence, best first; an empty list when nothing matches.

Raises: ImageNotFoundError, TypeError, or another RPCError subclass, as for find_object.

import mas

images = mas.images({"collectible": 88})
for star in mas.find_objects(images.collectible, max_matches=20):
    mas.click(*star.center, delay_ms=300)

find_any_object

def find_any_object(image_ids: list[int | ImageRef], *, threshold: float = 0.8, screenshot: Screenshot | None = None, continuous_mode: bool = False, timeout_ms: int = 1000, capture_interval_ms: int = 500, max_matches: int = 1, search_strategy: SearchStrategy = "first_match", search_region: Region | None = None) -> ObjectMatch | None

Search for several templates at once. Use it when one control has variants: languages, themes or states. The returned match’s matched_template_id tells you which template hit.

Returns: an ObjectMatch with x, y and matched_template_id, or None.

Raises: ImageNotFoundError, TypeError, or another RPCError subclass.

import mas

images = mas.images({"claim_en": 123, "claim_fr": 124, "claim_es": 125})
match = mas.find_any_object([images.claim_en, images.claim_fr, images.claim_es],
                            search_strategy="best_match")
if match:
    print(f"Matched image {match.template_id} at {match.center}")
    mas.click(*match.center)

find_object_retry

def find_object_retry(image_id: int | ImageRef, *, total_tries: int = 3, time_sleep: float = 2.0, **kwargs) -> ObjectMatch | None

The house wait primitive. Call find_object up to total_tries times, sleeping time_sleep seconds between failed attempts (never after the last one), and return the first match, or None when every attempt fails. Discrete tries with a flat pause read plainly, cost a bounded and predictable amount of time, and behave the same on every host. Every other keyword argument (threshold, screenshot, max_matches, search_region and so on) is forwarded to find_object by name. A caller-supplied screenshot is a frozen frame, so the call collapses to one attempt.

Returns: the first ObjectMatch, or None.

Raises: whatever find_object raises.

import mas

images = mas.images({"play": 784})
button = mas.find_object_retry(images.play, total_tries=5, time_sleep=1.5, threshold=0.85)
if button is None:
    mas.log("Play button not found, will try again later", level="warning")
else:
    mas.click(button.x, button.y, delay_ms=1000)

find_any_object_retry

def find_any_object_retry(image_ids: list[int | ImageRef], *, total_tries: int = 3, time_sleep: float = 2.0, **kwargs) -> ObjectMatch | None

find_object_retry over a list of templates. Identical semantics: up to total_tries attempts, a flat time_sleep pause between failed ones, every other keyword forwarded to find_any_object by name, and a caller-supplied screenshot collapsing to one attempt. Use it where several variants of one control are acceptable.

Returns: the first ObjectMatch, or None.

Raises: whatever find_any_object raises.

import mas

images = mas.images({"ok_light": 201, "ok_dark": 202})
ok = mas.find_any_object_retry([images.ok_light, images.ok_dark], total_tries=4)
if ok:
    mas.click(*ok.center)

read_text

def read_text(region: Region | None = None, screenshot: Screenshot | None = None, model: str = "eng_best", psm: int = 7, color_conversion: ColorConversion = ColorConversion.NONE, timeout_ms: int = 30000) -> TextRecognitionResult

Read text off the screen with OCR (Tesseract). Always pass a region when you can: OCR on a tight box is faster and far more accurate than reading the whole screen. psm tells the engine the shape of the text.

Returns: a TextRecognitionResult with text, confidence and region.

Raises: an RPCError subclass when OCR fails or the host times out.

import mas
from mas import ColorConversion, Region

result = mas.read_text(region=Region(420, 90, 660, 140))
score = int(result.text) if result.text.strip().isdigit() else 0
word = mas.read_text(region=Region(100, 200, 400, 260), psm=8,
                     color_conversion=ColorConversion.BLACK_WHITE)
print(word.text, word.confidence)

wait_for_object

def wait_for_object(image_id: int | ImageRef, timeout_ms: int = 10000, threshold: float = 0.8) -> ObjectMatch | None

Wait on the host until a template appears or timeout_ms elapses. This call is kept for existing macros. Prefer find_object_retry: discrete tries with a flat pause read plainly and behave identically on every host.

Returns: an ObjectMatch when the template appears in time, otherwise None.

Raises: ImageNotFoundError, TypeError, or another RPCError subclass.

import mas

images = mas.images({"home_screen": 456})
mas.click(150, 300)
home = mas.wait_for_object(images.home_screen, timeout_ms=30000)
if home is None:
    raise RuntimeError("Home screen did not load")

Image registry

Template images do not live in your project folder. They live in your Image Library in the cloud, where each image has a numeric ID. You could pass raw integers to every vision call, but that scatters meaningless numbers through the code. Declare the IDs once with images and refer to them by name. The publisher reads that call to know which images to bundle with the macro, so keep every image in one images call near the top of src/app.py. When a script uses images and still passes a raw integer somewhere, the SDK prints a hint on exit listing the IDs to add.

images

def images(mapping: dict[str, int]) -> ImageMap

Declare the image IDs your script uses under readable names.

Returns: an ImageMap with one ImageRef per entry, reached by dot access.

Raises: TypeError when a name is not a string or an ID is not an integer; ValueError when a name is not a valid identifier.

import mas

images = mas.images({
    "login_btn": 123,
    "home_screen": 456,
    "daily_reward": 789,
})
match = mas.find_object(images.login_btn)

ImageRef

class ImageRef

A reference to one registered image. images creates them; you do not construct one by hand. Any vision call accepts an ImageRef in place of an integer. It compares equal to another ImageRef with the same ID and to the plain integer, and int(ref) returns the ID.

import mas

images = mas.images({"login_btn": 123})
ref = images.login_btn
print(ref)            # ImageRef('login_btn', id=123)
print(int(ref) == 123, ref == 123)

ImageMap

class ImageMap

The read-only container returned by images. Each key of the mapping becomes an attribute holding an ImageRef. Assigning a new attribute raises AttributeError; define every image in the images call instead.

import mas

images = mas.images({"login_btn": 123, "home": 456})
print(images.get_all_ids())    # [123, 456]
print(images)                  # ImageMap(login_btn=123, home=456)

Elements

The accessibility tree is the second way to see the screen. Where vision matches pixels, elements read the widget tree the app exposes to Android: resource IDs, text, content descriptions and bounds. The device side is one wire method, a hierarchy dump; matching, ranking and scrolling run inside the SDK, so every host that can dump the tree serves the whole API.

A selector is a plain dict. Every key is optional, and the same dict shape is what MAS Agent harvests as <NAME>_SELECTOR constants.

{"rid": "com.app:id/claim", "text": "Claim", "desc": "Claim reward",
 "cls": "Button", "index": 0, "clickable": True}

Ranking is rid over desc over text over cls. The match_count field on a result says how many nodes matched at all: 1 is unambiguous, more means the pick was by rank and index.

The tri-state rule: a tree that cannot be read is never the same as an element that is absent. A game drawing into a canvas, a secure screen, a desktop host, a dump that failed or an RPC timeout all raise ElementsUnavailable. Catch it and fall back to vision or retry; never treat it as “already done”. A readable tree with no match returns None.

The host answers from one of two channels. The rich dump carries text and content descriptions but needs the screen to hold still. The structural dump is fast and works while the screen animates, but it carries no text or descriptions. The SDK tells the host whether the selector needs text. When only the structural tree could be read and the selector offers nothing but text or desc, the call raises ElementsUnavailable with a message that says so; anchor such selectors on a resource ID, or use a vision asset for that screen.

ElementMatch

class ElementMatch

One matched element. x and y are the center, which is the tap target.

The center property returns (x, y) as Coordinates, mirroring ObjectMatch.

import mas

btn = mas.find_element({"text": "Start"})
if btn:
    print(btn.rid, btn.bounds, btn.match_count)
    mas.click(*btn.center)

dump_hierarchy

def dump_hierarchy() -> str

Return the current screen’s accessibility tree as uiautomator XML. Useful while authoring, to see which resource IDs and texts exist.

Returns: the XML as a string; empty when the host returned nothing.

Raises: ElementsUnavailable when this host cannot provide a dump.

import mas

xml = mas.dump_hierarchy()
print(xml[:500])

host_capabilities

def host_capabilities() -> list[str]

The RPC methods the connected host dispatches. Use it to prove a host’s surface instead of assuming it: a method one host wires and another lacks fails mid-run with MethodNotFoundError.

Returns: a list of method names; an empty list on older hosts that do not answer.

Raises: nothing; MethodNotFoundError is swallowed and yields [].

import mas

methods = mas.host_capabilities()
if "dump_hierarchy" in methods:
    btn = mas.find_element({"rid": "com.app:id/start"})

find_element

def find_element(selector: dict) -> ElementMatch | None

Find one element on the current screen by selector and return the best-ranked match.

Returns: an ElementMatch, or None when the tree is readable but nothing matches.

Raises: ElementsUnavailable when the tree cannot describe this screen.

import mas

try:
    btn = mas.find_element({"rid": "com.app:id/start"})
except mas.ElementsUnavailable:
    btn = None
if btn:
    mas.click(*btn.center)

find_elements

def find_elements(selector: dict, *, max_matches: int = 10) -> list[ElementMatch]

Every element matching the selector, best-ranked first.

Returns: a list of ElementMatch; empty when nothing matches.

Raises: ElementsUnavailable when the tree cannot describe this screen.

import mas

rows = mas.find_elements({"cls": "TextView", "clickable": True}, max_matches=5)
for row in rows:
    print(row.text, row.center)

find_element_retry

def find_element_retry(selector: dict, *, total_tries: int = 3, time_sleep: float = 2.0) -> ElementMatch | None

find_element with discrete retries, mirroring find_object_retry. A transiently unreadable tree counts as a failed attempt; ElementsUnavailable propagates only when the final attempt still cannot read the tree. One flaky dump never fails a run, and a treeless screen still refuses to look like “element absent”.

Returns: the first ElementMatch, or None.

Raises: ElementsUnavailable when the last attempt could not read the tree.

import mas

claim = mas.find_element_retry({"rid": "com.app:id/claim"}, total_tries=5, time_sleep=1.0)
if claim:
    mas.click(*claim.center)

read_element

def read_element(selector: dict) -> str

The text of the best-matching element: text first, then the content description. This is an exact string from the UI toolkit, where OCR would approximate.

Returns: the string, or "" when nothing matches.

Raises: ElementsUnavailable when the tree cannot describe this screen.

import mas

balance = mas.read_element({"rid": "com.app:id/gold_balance"})
if balance.replace(",", "").isdigit():
    print(int(balance.replace(",", "")))

scroll_to_element

def scroll_to_element(selector: dict, *, container: dict | None = None, max_swipes: int = 8) -> ElementMatch | None

Scroll until the selector matches or the swipes run out. Each step swipes inside the container’s bounds (or the biggest scrollable element, or the visible extent of the tree) from 75 percent of its height to 25 percent, then waits for the fling to settle before the next dump.

Returns: the ElementMatch once visible, or None.

Raises: ElementsUnavailable when the tree cannot describe this screen.

import mas

row = mas.scroll_to_element({"text": "Advanced settings"}, max_swipes=6)
if row:
    mas.click(*row.center)

read_page

def read_page(container: dict | None = None, *, fields: list[str] | None = None, max_swipes: int = 8, row_cap: int = 200) -> list[dict]

Scroll through a list screen and collect its repeated rows. The call finds the container (by selector, or the biggest scrollable element), detects the repeated row structure, scrolls while collecting, and returns the rows deduplicated in reading order. It stops early when a full swipe surfaces nothing new.

Returns: a list of {"cells": [...]} dicts. Each cell is {"text", "desc", "rid", "center"} for one content-bearing element of the row, in document order.

Raises: ElementsUnavailable when the tree cannot describe this screen.

import mas

rows = mas.read_page(fields=["text", "center"], max_swipes=10)
for row in rows:
    print([cell["text"] for cell in row["cells"]])

App

Launch, close and inspect apps by package name. Package names look like com.android.settings.

open_app

def open_app(package_name: str, timeout_ms: int = 2000) -> None

Launch an app by package name.

Returns: None.

Raises: an RPCError subclass when the host reports a failure.

import mas

mas.open_app("com.android.settings")
mas.open_app("com.android.chrome", timeout_ms=5000)

close_app

def close_app(package_name: str) -> None

Force stop an app.

Returns: None.

Raises: an RPCError subclass when the host reports a failure.

import mas

mas.open_app("com.example.game")
mas.log("automation runs here")
mas.close_app("com.example.game")

get_current_app

def get_current_app() -> str

The package name of the foreground app.

Returns: the package name as a string.

Raises: an RPCError subclass when the host reports a failure.

import mas
import time

mas.open_app("com.android.chrome")
while "chrome" not in mas.get_current_app().lower():
    time.sleep(0.5)
print("Chrome is active")

get_app_state

def get_app_state(package_name: str) -> int

The state of an app as an integer. Compare it with the module constants.

Returns: one of the five integers above.

Raises: an RPCError subclass when the host reports a failure.

import mas

if mas.get_app_state("com.example.game") == mas.NOT_RUNNING:
    mas.open_app("com.example.game")
elif mas.get_app_state("com.example.game") == mas.NOT_INSTALLED:
    mas.log("Game is not installed", level="error")

is_app_focused

def is_app_focused(package_name: str) -> bool

Whether the app currently has input focus, according to the Android window manager. A good guard before a tap or typing.

Returns: True when the app is focused, otherwise False.

Raises: an RPCError subclass when the host reports a failure.

import mas
import time

mas.open_app("com.example.game")
while not mas.is_app_focused("com.example.game"):
    time.sleep(0.5)

Device

Facts about the device and the computer the macro runs on.

get_screen_size

def get_screen_size() -> ScreenSize

The device screen dimensions in pixels. Use it to compute coordinates so a macro survives a resolution change.

Returns: a ScreenSize with width and height.

Raises: an RPCError subclass when the host reports a failure.

import mas

size = mas.get_screen_size()
mas.click(size.width // 2, size.height // 2)    # tap the center

get_device_info

def get_device_info() -> DeviceInfo

Details about the bound device.

Returns: a DeviceInfo with id, name, type, screen_width, screen_height and connected.

Raises: an RPCError subclass when the host reports a failure.

import mas

info = mas.get_device_info()
print(f"{info.name} ({info.type}) {info.screen_width}x{info.screen_height}")

get_host_machine_id

def get_host_machine_id() -> str

A stable UUID for the computer the macro runs on. It is generated once, survives restarts and is shared by every macro on that machine. It is the default machine_id for storage, so you rarely call it yourself.

Returns: the UUID as a string.

Raises: RPCError when the host returns no usable ID. The SDK refuses to fall back to an empty string, because storage is keyed by (machine_id, port, task_name) and an empty ID would corrupt every later save.

import mas

machine_id = mas.get_host_machine_id()
print(machine_id)    # 550e8400-e29b-41d4-a716-446655440000

Storage

save and retrieve give a macro memory across runs: a streak counter, a checkpoint, a “last processed” marker. Data is any JSON-serializable dict. Every entry is keyed by the triple (machine_id, port, task_name), so two emulators running the same macro keep separate state. Leave machine_id and port at their defaults; they resolve to this computer and the current device. Passing a port that is not the current device’s port raises ValueError.

save

def save(task_name: str, data: Dict[str, Any], machine_id: Optional[str] = None, port: int = 0) -> bool

Persist a dict for a task. Saving the same key again updates the entry rather than duplicating it.

Returns: True when the save succeeded.

Raises: ValueError when data is not JSON-serializable or port does not match the current device; RPCError when the machine ID cannot be read.

import mas

state = mas.retrieve("daily_login")
streak = state.get("streak", 0) + 1
mas.save("daily_login", {"streak": streak, "last_run": "2026-09-04"})

retrieve

def retrieve(task_name: str, machine_id: Optional[str] = None, port: int = 0) -> Dict[str, Any]

Read an entry back.

Returns: the stored dict, or an empty dict when nothing was stored.

Raises: ValueError when port does not match the current device; RPCError when the machine ID cannot be read.

import mas

state = mas.retrieve("harvest")    # {} on the first run
last_field = state.get("last_field", 0)

retrieve_all

def retrieve_all(task_name: str) -> List[Dict[str, Any]]

Every entry stored under a task name, across all machines and ports. Use it to roll up results from many devices running the same macro.

Returns: a list of dicts, each with machine_id, port, data and updated_at.

Raises: an RPCError subclass when the host reports a failure.

import mas

entries = mas.retrieve_all("harvest")
total = sum(e["data"].get("collected", 0) for e in entries)
mas.log(f"Collected {total} across {len(entries)} devices")

clear

def clear(task_name: str, machine_id: Optional[str] = None, port: int = 0) -> bool

Delete a stored entry.

Returns: True when the entry was cleared.

Raises: ValueError when port does not match the current device; RPCError when the machine ID cannot be read.

import mas

if mas.retrieve("daily_login").get("streak", 0) > 30:
    mas.clear("daily_login")

get_current_device_port

def get_current_device_port() -> int

The port of the device this run is bound to, for example 5556 for emulator-5556. The value is fetched once and cached. Useful for labelling per-device output.

Returns: the port as an integer.

Raises: RPCError when no device is connected.

import mas

port = mas.get_current_device_port()
mas.log(f"Running on port {port}")

Utilities and webhooks

Logging, the clipboard, custom webhook events, and the JSON-RPC client that every call uses underneath.

log

def log(message: str, level: Union[LogLevel, str] = LogLevel.INFO) -> None

Send a leveled line to the run console in Studio. Unlike print, the line carries a severity and is surfaced in the app’s log view. No device is required.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.log("Run started")
mas.log("retries running low", level="warning")
mas.log("step failed", level=mas.LogLevel.ERROR)

get_clipboard

def get_clipboard() -> str

Read the device clipboard as text.

Returns: the clipboard contents.

Raises: an RPCError subclass when the host reports a failure.

import mas

mas.click(300, 500, delay_ms=1500)    # select text in the app
mas.click(400, 300)                   # tap the app's Copy control
copied = mas.get_clipboard()
print(copied)

webhook

def webhook(event: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

Send a custom webhook event to your configured endpoints. Use it to signal things only the script knows: a level reached, a captcha hit, an inventory full. MAS’s servers deliver the event, so it is signed, retried on failure and recorded in the delivery log next to macro.started and macro.failed. Your event name is namespaced under custom., so "level.reached" arrives as custom.level.reached; a script can never impersonate a lifecycle event. The run and device are attached automatically. Set up an endpoint first on the Webhooks page of the app, subscribed to custom.* or *; see Webhooks.

Returns: a dict with event_id (the receiver’s dedupe key), event_type (the namespaced name) and queued (how many endpoints it went to). queued of 0 is not an error; no endpoint is subscribed to this event.

Raises: RuntimeError when the event could not be queued.

import mas

result = mas.webhook("level.reached", {"level": 40, "account": "alt-3"})
print(result["event_type"], result["queued"])

RPCClient

class RPCClient

The JSON-RPC 2.0 client that every mas.* call uses. You rarely construct one yourself; get_client returns the shared instance. The constructor discovers the transport from the environment: MAS_RPC_SOCKET (a unix socket, used inside the cloud sandbox) takes precedence over TCP; otherwise MAS_RPC_HOST (default localhost) and MAS_RPC_PORT, then the port file the app writes, and finally ConnectionError when no port can be found. On connect it registers the session from MAS_SESSION_TOKEN, or binds a device from MAS_DEVICE_ID when a script was started outside Studio.

def __init__(self, host: Optional[str] = None, port: Optional[int] = None, connect_timeout: float = DEFAULT_CONNECT_TIMEOUT, read_timeout: float = DEFAULT_READ_TIMEOUT)

Methods:

The client is a context manager: with RPCClient() as client: connects on entry and disconnects on exit. Calls are serialized with a lock, so one client is safe to share across threads. A read that exceeds read_timeout raises RPCError and marks the client disconnected; the next call reconnects.

import mas
from mas import RPCClient

client = RPCClient(read_timeout=60.0)
try:
    methods = client.call("host_capabilities", {})
    print(methods.get("methods", []))
finally:
    client.disconnect()

get_client

def get_client() -> RPCClient

Get or create the global RPCClient that the SDK functions share.

Returns: the shared RPCClient.

Raises: ConnectionError on first use when no RPC port can be discovered.

import mas

client = mas.get_client()
size = client.call("get_screen_size", {})
print(size["width"], size["height"])

Runtime UI (mas.ui)

mas.ui drives the runtime dashboard, the live panel shown next to a running macro. You design the dashboard in the UI Builder as a .uibrt file, give every widget a name in the Inspector, and address it by that name from the script. A name that does not exist in the dashboard updates nothing and raises nothing, so a typo is the most common dashboard bug. Each call is one round trip to the app; wrap several updates in batch to send them together. See Runtime UI for the widget-by-widget guide and UI Builder for the designer.

UIEvent

class UIEvent

A user interaction reported by wait_for_event or dispatched to a handler. It is a frozen dataclass.

import mas

event = mas.ui.wait_for_event(timeout=60)
if event and event.widget_id == "start_btn" and event.event_type == "click":
    print("Start pressed")

set_text

def set_text(name: str, text: str) -> None

Set the text of a label, a text area, a button, or the caption of a progress bar. The value is converted with str.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.set_text("status", "Logging in...")
mas.ui.set_text("status", f"Step {3}/{10}")

set_progress

def set_progress(name: str, value: Union[int, float]) -> None

Set the value of a progress bar. The app clamps it to the widget’s [min, max].

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

for i in range(101):
    mas.ui.set_progress("main_bar", i)

add_data_point

def add_data_point(name: str, value: Union[int, float], label: str, series: Optional[str] = None) -> None

Append one point to a streaming chart. The app drops points beyond the widget’s maxDataPoints.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.add_data_point("cpu_chart", value=42.5, label="t=12", series="CPU")
mas.ui.add_data_point("cpu_chart", value=61.0, label="t=13", series="CPU")

set_chart_data

def set_chart_data(name: str, data: List[Dict[str, Any]]) -> None

Replace a chart’s whole dataset. Each entry needs label (a string) and value (a number); series is optional.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.set_chart_data("scores", [
    {"label": "Gold", "value": 1250},
    {"label": "XP", "value": 45000},
])

clear_chart

def clear_chart(name: str) -> None

Remove every point from a chart.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.clear_chart("scores")

append_text

def append_text(name: str, line: str) -> None

Append a line to a text area. The panel auto-scrolls, and lines beyond the widget’s maxLines drop off the top. The SDK keeps a local copy of the text so it can compose the next append without a round trip.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.append_text("log", "Run started")
mas.ui.append_text("log", "Step 1 complete")

set_textarea

def set_textarea(name: str, text: str) -> None

Replace the whole content of a text area.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.set_textarea("log", "fresh contents")

clear_text

def clear_text(name: str) -> None

Clear a text area.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.clear_text("log")

set_table_data

def set_table_data(name: str, rows: List[Dict[str, Any]]) -> None

Replace every row of a table. Each row is a dict whose keys match the widget’s columns.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.set_table_data("results", [
    {"Name": "Gold", "Value": "1,250", "Status": "OK"},
    {"Name": "Stone", "Value": "12", "Status": "low"},
])

append_table_row

def append_table_row(name: str, row: Dict[str, Any]) -> None

Append one row to a table. The SDK keeps the rows locally and re-sends the full table.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.append_table_row("results", {"Name": "Wood", "Value": "300", "Status": "OK"})

clear_table

def clear_table(name: str) -> None

Remove every row from a table.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.clear_table("results")

cell_button

def cell_button(*, id: str, label: str, variant: str = "default", disabled: bool = False) -> Dict[str, Any]

Build a button to place inside a table cell. Put the returned dict in a row passed to set_table_data or append_table_row. A click emits a UI event whose widget_id equals id, so the script handles it with the same on_click or wait_for_event channel as a standalone button. All parameters are keyword-only.

Returns: a dict describing the button.

Raises: ValueError when variant is not one of the three values.

import mas

mas.ui.append_table_row("orders", {
    "Name": "Gold",
    "Action": mas.ui.cell_button(id="refund_gold", label="Refund", variant="destructive"),
})
mas.ui.on_click("refund_gold", lambda: mas.log("Refund requested"))
mas.ui.start_listener()

set_cell_button_enabled

def set_cell_button_enabled(table_name: str, button_id: str, enabled: bool) -> None

Enable or disable a button cell in a table that was already rendered. The SDK finds the cell by button_id in its local copy of the table, flips the flag and re-sends the table. Nothing happens when no cell has that id or the flag already matches.

Returns: None.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.set_cell_button_enabled("orders", "refund_gold", False)

wait_for_event

def wait_for_event(timeout: float = 30.0) -> Optional[UIEvent]

Block until the user clicks a button or changes an input, or until timeout seconds pass. This is the blocking model: the script stops and waits.

Returns: a UIEvent, or None when the timeout elapsed with no event.

Raises: an RPCError subclass when the app cannot be reached.

import mas

mas.ui.set_text("status", "Press Start when the game is on the home screen")
event = mas.ui.wait_for_event(timeout=120)
if event is None or event.widget_id != "start_btn":
    raise SystemExit(1)

get_input_value

def get_input_value(name: str) -> Optional[str]

The most recent value typed into an input widget. No listener is needed.

Returns: the value, or None when the widget has none.

Raises: an RPCError subclass when the app cannot be reached.

import mas

query = mas.ui.get_input_value("search_box")
if query:
    mas.input_text(query)

on_click

def on_click(name: str, handler: Callable[[], None]) -> None

Register a callback for clicks on a button or a cell_button. The callback fires only while the listener thread runs; call start_listener once after registering every handler. Registering a name again replaces the previous handler.

Returns: None.

Raises: nothing.

import mas

stop = False

def on_stop():
    global stop
    stop = True

mas.ui.on_click("stop_btn", on_stop)
mas.ui.start_listener()

on_change

def on_change(name: str, handler: Callable[[Any], None]) -> None

Register a callback for value changes on an input widget. The callback receives the new value as its only argument and fires only while the listener thread runs.

Returns: None.

Raises: nothing.

import mas

def on_delay_change(value):
    mas.log(f"Delay set to {value}")

mas.ui.on_change("delay_input", on_delay_change)
mas.ui.start_listener()

start_listener

def start_listener() -> None

Start the daemon thread that dispatches UI events to on_click and on_change handlers. It is idempotent: a second call while the thread is alive does nothing. The thread opens its own RPC connection so its long waits never block mas.* calls on the main thread, and it is a daemon, so it does not keep the interpreter alive after the script returns. A handler that raises is caught and its traceback printed; the listener keeps running.

Returns: None, immediately.

Raises: ConnectionError when the listener’s connection cannot be opened.

import mas

paused = False

def toggle():
    global paused
    paused = not paused

mas.ui.on_click("pause_btn", toggle)
mas.ui.start_listener()

stop_listener

def stop_listener() -> None

Signal the listener thread to exit and close its connection. Mostly useful in tests; a finished script does not need it.

Returns: None.

Raises: nothing.

import mas

mas.ui.start_listener()
mas.ui.set_text("status", "Running")
mas.ui.stop_listener()

batch

def batch() -> Iterator[None]

A context manager that coalesces every update inside the with block into one RPC call. This is the difference between a smooth dashboard and a flickering one. Nesting is not supported. The buffer is process-global, not thread-local: updates from other threads while a batch is open are pulled into it, so finish the batch before letting listener handlers fire their own updates.

Returns: a context manager; the block yields None.

Raises: RuntimeError when a batch is opened inside another batch.

import mas

with mas.ui.batch():
    mas.ui.set_text("status", "Step 3/10")
    mas.ui.set_progress("main_bar", 30)
    mas.ui.add_data_point("cpu_chart", value=42, label="t=3", series="CPU")

Exceptions

Rule of thumb: None means “not found, and that is fine”, so handle it with if. An exception means something is wrong, so let it stop the macro unless you have a recovery. Every SDK error descends from MASError. Errors that come back from the app carry a JSON-RPC code and are raised as the matching RPCError subclass; an unknown code raises plain RPCError. ElementsUnavailable is the one exception that is not an RPC error. The Errors page shows handling patterns.

MASError
  RPCError                    .code, .message, .data
    ConnectionError           could not reach the app (code -1)
    ParseError                -32700
    InvalidRequestError       -32600
    MethodNotFoundError       -32601
    InvalidParamsError        -32602
    InternalError             -32603
    DeviceNotConnectedError   -32001
    DeviceNotFoundError       -32002
    CommandFailedError        -32003
    ImageNotFoundError        -32004
    TimeoutError              -32005
    AuthenticationError       -32010
    SubscriptionError         -32011
  ElementsUnavailable         the accessibility tree could not be read

ConnectionError and TimeoutError shadow the Python built-ins of the same name inside the mas namespace. Refer to them as mas.ConnectionError and mas.TimeoutError to keep the two apart.

MASError

class MASError(Exception)

The base class for every SDK error. Catch it to handle any SDK failure in one place.

import mas

try:
    mas.open_app("com.example.game")
except mas.MASError as e:
    mas.log(f"SDK error: {e}", level="error")
    raise

RPCError

class RPCError(MASError)

Raised when an RPC call fails. It is the base class of every coded error below, and the class you get for a code that has no subclass, an invalid response, a socket error, or a response that took longer than the client’s read timeout.

def __init__(self, message: str, code: int = -1, data: Optional[Any] = None)

str(error) renders as RPC Error (<code>): <message>.

import mas

try:
    mas.click(100, 100)
except mas.RPCError as e:
    print(e.code, e.message, e.data)

ConnectionError

class ConnectionError(RPCError)

Raised when the SDK cannot reach the app: no RPC port could be discovered, the connection was refused, or connecting timed out. The message says to make sure MAS is running. Its code is -1.

import mas

try:
    size = mas.get_screen_size()
except mas.ConnectionError:
    print("Start Macro Automation Studio, then run the script again")

ParseError

class ParseError(RPCError)

Raised when the app received invalid JSON (code -32700). The SDK builds every request itself, so this indicates a transport problem, not a script bug.

import mas

try:
    mas.log("hello")
except mas.ParseError as e:
    print(e.message)

InvalidRequestError

class InvalidRequestError(RPCError)

Raised when the JSON-RPC request is malformed (code -32600).

import mas

try:
    mas.get_current_app()
except mas.InvalidRequestError as e:
    print(e.message)

MethodNotFoundError

class MethodNotFoundError(RPCError)

Raised when the requested method does not exist on this host (code -32601). Check host_capabilities before relying on a method that not every host provides.

import mas

try:
    xml = mas.dump_hierarchy()
except (mas.MethodNotFoundError, mas.ElementsUnavailable):
    xml = ""

InvalidParamsError

class InvalidParamsError(RPCError)

Raised when the method parameters are invalid (code -32602), for example a value outside the range the host accepts.

import mas

try:
    mas.key_press(mas.KeyCode.DELETE, repeat=500)
except mas.InvalidParamsError as e:
    print(e.message)

InternalError

class InternalError(RPCError)

Raised when the app hits an internal error (code -32603).

import mas

try:
    shot = mas.take_screenshot()
except mas.InternalError as e:
    mas.log(f"Host error: {e.message}", level="error")
    raise

DeviceNotConnectedError

class DeviceNotConnectedError(RPCError)

Raised when a device operation runs with no device bound to this connection (code -32001). The emulator stopped, the cloud device went away, or the script was started without a device.

import mas

try:
    mas.click(540, 960)
except mas.DeviceNotConnectedError:
    mas.log("No device connected", level="error")
    raise SystemExit(1)

DeviceNotFoundError

class DeviceNotFoundError(RPCError)

Raised when the requested device does not exist or is not available (code -32002), for example when MAS_DEVICE_ID names a device the app cannot see.

import mas

try:
    info = mas.get_device_info()
except mas.DeviceNotFoundError as e:
    print(e.message)

CommandFailedError

class CommandFailedError(RPCError)

Raised when a device command such as a click, swipe or screenshot fails to complete on the device (code -32003).

import mas

try:
    mas.swipe((500, 1500), (500, 500))
except mas.CommandFailedError as e:
    mas.log(f"Swipe failed: {e.message}", level="warning")

ImageNotFoundError

class ImageNotFoundError(RPCError)

Raised when a template image cannot be used (code -32004): the ID does not exist in your Image Library, or the file cannot be loaded. A template that exists but does not appear on screen is not an error; find_object returns None for that.

import mas

try:
    match = mas.find_object(999999)
except mas.ImageNotFoundError:
    mas.log("Image 999999 is not in your Image Library", level="error")
    raise

TimeoutError

class TimeoutError(RPCError)

Raised when an operation on the host exceeds its time budget (code -32005). Note that wait_for_object returns None on a timeout instead of raising; this error comes from the host when a command itself times out.

import mas

try:
    text = mas.read_text(timeout_ms=5000)
except mas.TimeoutError:
    mas.log("OCR took longer than 5 seconds", level="warning")

AuthenticationError

class AuthenticationError(RPCError)

Raised when you are not logged in to MAS (code -32010). Log in and run the script again.

import mas

try:
    mas.log("start")
except mas.AuthenticationError:
    print("Log in to Macro Automation Studio, then run the script again")
    raise SystemExit(1)

SubscriptionError

class SubscriptionError(RPCError)