diff --git a/.gitignore b/.gitignore index 6bf1f06..226ba55 100644 --- a/.gitignore +++ b/.gitignore @@ -9,4 +9,5 @@ wheels/ # Virtual environments .venv -screenshots/ \ No newline at end of file +screenshots/ +appium_profiles.yaml diff --git a/README.md b/README.md index 4fd7d1c..375b56c 100644 --- a/README.md +++ b/README.md @@ -1,19 +1,90 @@ # Appium FastMCP stdio MCP (Auto-Wait + Retry) This MCP allows Claude Code to control Android/iOS Appium sessions via stdio. -It supports: + +## Features - Auto-wait - Retry heuristics -- Screenshot + XML return after each action +- Screenshot + XML after each action +- External Appium profile config (no hardcoded server/capabilities) ## Install -- pip install --upgrade uv -- uv sync -- source .venv/bin/activate # macOS / Linux -- .venv\Scripts\activate # Windows +- `pip install --upgrade uv` +- `uv sync` +- `source .venv/bin/activate` (macOS/Linux) +- `.venv\Scripts\activate` (Windows) ## Run -python server.py +`python server.py` + +## Register (without starting manually) +`claude mcp add --transport stdio appium-mcp -- python3 /abs/path/server.py` + +--- + +## Decoupled Appium configuration + +### 1) Create your local config + +```bash +cp appium_profiles.example.yaml appium_profiles.yaml +``` + +Then edit `appium_profiles.yaml` for your device farm or local Appium server. + +### 2) Start sessions from profiles + +- List profiles: `list_appium_profiles(config_path=None)` +- Start by profile: `start_appium_session_with_profile(profile_name, config_path=None, capabilities_override=None)` +- Backward-compatible direct start: `start_appium_session(platform, server_url, capabilities)` + +Default tools read profiles by name: +- `start_default_ios_appium_session()` → defaults to `ios-local` +- `start_default_android_appium_session()` → defaults to `android-local` + +You can override default profile names with env vars: +- `APPIUM_MCP_DEFAULT_IOS_PROFILE` +- `APPIUM_MCP_DEFAULT_ANDROID_PROFILE` + +### 3) Switch config file by environment + +```bash +export APPIUM_MCP_CONFIG=/abs/path/to/profiles.yaml +``` + +`appium_profiles.yaml` is ignored by git, so local UDID/secrets stay uncommitted. + +### 4) Advanced profile capabilities + +The config loader supports: +- **Profile inheritance** with `extends` +- **Environment placeholder expansion** (e.g. `${IOS_UDID}`) + +See `appium_profiles.example.yaml` for both patterns. + +--- + +## Improvements to handle more scenarios + +### A. Multi-device and concurrency +- Replace singleton driver with session pool (`session_id -> driver`) +- Require `session_id` in action tools +- Add `list_sessions`, `close_session(session_id)`, and session metadata + +### B. Cross-platform UI abstraction +- Add Android XML parser and normalize fields with iOS +- Unified semantic query layer: role/text/state/bounds + +### C. Semantic action APIs +- Add high-level tools like `tap_text`, `input_into`, and `assert_text_visible` +- Reduce dependence on brittle raw locators + +### D. Observability and reliability +- Structured JSON logs per action (duration, retries, error type) +- Action traces with before/after screenshot + XML change summary +- Stable error codes (`ELEMENT_NOT_FOUND`, `ACTION_TIMEOUT`, etc.) -## Register (Don't require Run) -claude mcp add --transport stdio appium-mcp -- python3 /abs/path/server.py +### E. Quality gates +- Unit tests for config parsing, profile resolution, and retries +- Contract tests for MCP tool return schemas +- CI checks (`pytest`, lint, static checks) diff --git a/appium_config.py b/appium_config.py new file mode 100644 index 0000000..4efc671 --- /dev/null +++ b/appium_config.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +import json +import os +import re +from pathlib import Path +from typing import Any + +import yaml + + +DEFAULT_CONFIG_PATH = Path(__file__).resolve().parent / "appium_profiles.yaml" +ENV_CONFIG_PATH = "APPIUM_MCP_CONFIG" + +_ENV_VAR_PATTERN = re.compile(r"\$\{([A-Z0-9_]+)\}") + + +class AppiumConfigError(ValueError): + """Raised when Appium MCP configuration is invalid.""" + + +def _load_config_file(config_path: Path) -> dict[str, Any]: + if not config_path.exists(): + raise AppiumConfigError( + f"Config file not found: {config_path}. " + "Create one based on appium_profiles.example.yaml." + ) + + raw = config_path.read_text(encoding="utf-8") + + try: + if config_path.suffix.lower() == ".json": + data = json.loads(raw) + else: + data = yaml.safe_load(raw) + except Exception as exc: # parse errors from json/yaml + raise AppiumConfigError(f"Failed to parse config file '{config_path}': {exc}") from exc + + if not isinstance(data, dict): + raise AppiumConfigError("Config root must be a JSON/YAML object") + + return data + + +def _get_default_path() -> Path: + configured_path = os.getenv(ENV_CONFIG_PATH) + if configured_path: + return Path(configured_path).expanduser().resolve() + return DEFAULT_CONFIG_PATH + + +def _resolve_placeholders(value: Any) -> Any: + if isinstance(value, str): + return _ENV_VAR_PATTERN.sub(lambda m: os.getenv(m.group(1), m.group(0)), value) + if isinstance(value, dict): + return {k: _resolve_placeholders(v) for k, v in value.items()} + if isinstance(value, list): + return [_resolve_placeholders(v) for v in value] + return value + + +def _merge_profile(parent: dict[str, Any], child: dict[str, Any]) -> dict[str, Any]: + merged = dict(parent) + merged.update({k: v for k, v in child.items() if k != "capabilities"}) + + parent_caps = parent.get("capabilities") if isinstance(parent.get("capabilities"), dict) else {} + child_caps = child.get("capabilities") if isinstance(child.get("capabilities"), dict) else {} + merged["capabilities"] = {**parent_caps, **child_caps} + return merged + + +def _normalize_profiles(raw_profiles: dict[str, Any]) -> dict[str, dict[str, Any]]: + normalized: dict[str, dict[str, Any]] = {} + + def resolve(name: str, stack: set[str]) -> dict[str, Any]: + if name in normalized: + return normalized[name] + if name in stack: + chain = " -> ".join([*stack, name]) + raise AppiumConfigError(f"Circular profile inheritance detected: {chain}") + + value = raw_profiles.get(name) + if not isinstance(value, dict): + raise AppiumConfigError(f"Profile '{name}' must be an object") + + stack.add(name) + parent_name = value.get("extends") + + if parent_name is not None and not isinstance(parent_name, str): + raise AppiumConfigError(f"Profile '{name}' field 'extends' must be a string") + + if parent_name: + if parent_name not in raw_profiles: + raise AppiumConfigError( + f"Profile '{name}' extends unknown profile '{parent_name}'" + ) + parent = resolve(parent_name, stack) + merged = _merge_profile(parent, value) + else: + merged = dict(value) + + stack.remove(name) + + platform = merged.get("platform") + server_url = merged.get("server_url") + capabilities = merged.get("capabilities") + + if not isinstance(platform, str) or not platform: + raise AppiumConfigError(f"Profile '{name}' missing 'platform'") + if not isinstance(server_url, str) or not server_url: + raise AppiumConfigError(f"Profile '{name}' missing 'server_url'") + if not isinstance(capabilities, dict): + raise AppiumConfigError(f"Profile '{name}' missing dict 'capabilities'") + + normalized[name] = { + "platform": platform, + "server_url": server_url, + "capabilities": _resolve_placeholders(capabilities), + } + return normalized[name] + + for profile_name in raw_profiles: + resolve(profile_name, set()) + + return normalized + + +def load_profiles(config_path: str | None = None) -> dict[str, dict[str, Any]]: + path = Path(config_path).expanduser().resolve() if config_path else _get_default_path() + data = _load_config_file(path) + profiles = data.get("profiles") + + if not isinstance(profiles, dict) or not profiles: + raise AppiumConfigError("Config must include non-empty 'profiles' mapping") + + return _normalize_profiles(profiles) + + +def get_profile(profile_name: str, config_path: str | None = None) -> dict[str, Any]: + profiles = load_profiles(config_path=config_path) + + if profile_name not in profiles: + available = ", ".join(sorted(profiles.keys())) + raise AppiumConfigError( + f"Unknown profile '{profile_name}'. Available profiles: {available}" + ) + + return profiles[profile_name] diff --git a/appium_profiles.example.yaml b/appium_profiles.example.yaml new file mode 100644 index 0000000..92efb67 --- /dev/null +++ b/appium_profiles.example.yaml @@ -0,0 +1,22 @@ +profiles: + cloud-base: + platform: iOS + server_url: ${APPIUM_SERVER_URL} + capabilities: + gads:clientSecret: ${GADS_CLIENT_SECRET} + + ios-local: + extends: cloud-base + platform: iOS + capabilities: + platformName: iOS + automationName: XCUITest + udid: ${IOS_UDID} + + android-local: + platform: Android + server_url: ${APPIUM_SERVER_URL} + capabilities: + platformName: Android + automationName: UiAutomator2 + udid: ${ANDROID_UDID} diff --git a/server.py b/server.py index 0b0d862..cfb9225 100644 --- a/server.py +++ b/server.py @@ -1,4 +1,5 @@ -from pathlib import Path +from pathlib import Path +import os import base64 from io import BytesIO import time @@ -11,6 +12,7 @@ get_driver, stop_session ) +from appium_config import load_profiles, get_profile, AppiumConfigError from utils import ( resize_image, get_current_timestamp, @@ -23,6 +25,7 @@ mcp = FastMCP("appium-stdio-mcp") +LAST_UI_TREE = None def collect_artifacts(driver): @@ -47,45 +50,68 @@ def get_bounds_by_element_id(element_id): return None @mcp.tool() -def start_default_ios_appium_session(): - _platform = "iOS" - _server_url = "http://10.160.13.112:8080/grid" - _capabilities = { - "udid": "f67d7ce40691d9ab546d7362a4cc7a6182870de2", - "gads:clientSecret": "YmLZZlF6PnduXxSZvF3sTEeHIjT2XKKuA2UBoaqT4E0=", - "platformName": "iOS", - "automationName": "XCUITest" - } - start_session(_platform, _server_url, _capabilities) - return "default ios appium session started" +def list_appium_profiles(config_path: str | None = None): + """List available Appium profiles from config file.""" + try: + profiles = load_profiles(config_path=config_path) + return { + "status": "ok", + "profiles": sorted(profiles.keys()), + } + except AppiumConfigError as e: + return {"status": "failed", "reason": str(e)} @mcp.tool() -def start_default_android_appium_session(): - _platform = "Android" - _server_url = "http://10.160.13.112:8080/grid" - _capabilities = { - "udid": "8BEX18XP6", - "gads:clientSecret": "YmLZZlF6PnduXxSZvF3sTEeHIjT2XKKuA2UBoaqT4E0=", - "platformName": "Android", - "automationName": "UiAutomator2" +def start_appium_session_with_profile( + profile_name: str, + config_path: str | None = None, + capabilities_override: dict | None = None, +): + """Start session from named profile, with optional capability overrides.""" + try: + profile = get_profile(profile_name, config_path=config_path) + except AppiumConfigError as e: + return {"status": "failed", "reason": str(e)} + + capabilities = dict(profile["capabilities"]) + if capabilities_override: + capabilities.update(capabilities_override) + + start_session(profile["platform"], profile["server_url"], capabilities) + return { + "status": "ok", + "profile": profile_name, + "platform": profile["platform"], + "server_url": profile["server_url"], } - start_session(_platform, _server_url, _capabilities) - return "default android appium session started" + + +@mcp.tool() +def start_default_ios_appium_session(config_path: str | None = None): + profile_name = os.getenv("APPIUM_MCP_DEFAULT_IOS_PROFILE", "ios-local") + return start_appium_session_with_profile(profile_name, config_path=config_path) + + +@mcp.tool() +def start_default_android_appium_session(config_path: str | None = None): + profile_name = os.getenv("APPIUM_MCP_DEFAULT_ANDROID_PROFILE", "android-local") + return start_appium_session_with_profile(profile_name, config_path=config_path) @mcp.tool() def start_appium_session(platform: str, server_url: str, capabilities: dict): start_session(platform, server_url, capabilities) - return "session started" + return {"status": "ok", "platform": platform, "server_url": server_url} @mcp.tool() def stop_appium_session(): stop_session() - return "session stopped" + return {"status": "ok"} @mcp.tool() def get_screenshot(region: str = "full"): driver = get_driver() ts = get_current_timestamp() + focused_id = LAST_UI_TREE.get("focused") if LAST_UI_TREE else None base_dir = Path(__file__).resolve().parent out_dir = base_dir / "screenshots" @@ -95,7 +121,6 @@ def get_screenshot(region: str = "full"): img = Image.open(BytesIO(png_bytes)) if region == "focused": - focused_id = LAST_UI_TREE.get("focused") if LAST_UI_TREE else None bounds = get_bounds_by_element_id(focused_id) if bounds: @@ -106,7 +131,7 @@ def get_screenshot(region: str = "full"): ) img = img.crop(bounds) else: - region = "full" # fallback,永不炸 + region = "full" # fallback: do not fail when no focused node is available img = resize_image(img, max_long=768, max_short=384) @@ -129,7 +154,7 @@ def get_ui_tree(): driver = get_driver() source = driver.page_source # XML - elements, ui_w, ui_h = parse_ios_xml(source) # 你可以先只抽 button / textfield + elements, ui_w, ui_h = parse_ios_xml(source) # currently iOS-oriented parser focused = get_focused_element_id(elements) LAST_UI_TREE = { diff --git a/tests/test_appium_config.py b/tests/test_appium_config.py new file mode 100644 index 0000000..bb60c65 --- /dev/null +++ b/tests/test_appium_config.py @@ -0,0 +1,60 @@ +import os +import tempfile +import textwrap +import unittest +from pathlib import Path + +from appium_config import get_profile, load_profiles + + +class AppiumConfigTests(unittest.TestCase): + def write_config(self, content: str) -> str: + tmp = tempfile.NamedTemporaryFile("w", suffix=".yaml", delete=False) + tmp.write(textwrap.dedent(content)) + tmp.flush() + tmp.close() + self.addCleanup(lambda: Path(tmp.name).unlink(missing_ok=True)) + return tmp.name + + def test_load_basic_profiles(self): + path = self.write_config( + """ + profiles: + ios-local: + platform: iOS + server_url: http://127.0.0.1:4723 + capabilities: + platformName: iOS + """ + ) + profiles = load_profiles(path) + self.assertIn("ios-local", profiles) + self.assertEqual(profiles["ios-local"]["platform"], "iOS") + + def test_profile_inheritance_and_env_placeholder(self): + old = os.environ.get("IOS_UDID") + os.environ["IOS_UDID"] = "udid-from-env" + self.addCleanup(lambda: os.environ.__setitem__("IOS_UDID", old) if old is not None else os.environ.pop("IOS_UDID", None)) + + path = self.write_config( + """ + profiles: + base: + platform: iOS + server_url: http://farm/grid + capabilities: + appium:automationName: XCUITest + ios-local: + extends: base + capabilities: + udid: ${IOS_UDID} + """ + ) + profile = get_profile("ios-local", path) + self.assertEqual(profile["server_url"], "http://farm/grid") + self.assertEqual(profile["capabilities"]["udid"], "udid-from-env") + self.assertEqual(profile["capabilities"]["appium:automationName"], "XCUITest") + + +if __name__ == "__main__": + unittest.main()