from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from fastmcp import FastMCP
from appcrawler.config import AppiumConfig, AzureConfig, CrawlerSettings
from appcrawler.crawler import AppCrawler
from appcrawler.docker_runner import build_and_run_docker
from appcrawler.examples import CALCULATOR_EXAMPLE_TEST
from appcrawler.prompts import GENERATE_TEST_CASE_PROMPT, TASK_PROMPT
from appcrawler.ui_dump import get_ui_dump as fetch_ui_dump
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
)
mcp = FastMCP("AppCrawler MCP Server")
_crawler: Optional[AppCrawler] = None
def _ensure_crawler() -> AppCrawler:
if _crawler is None:
raise RuntimeError(
"AppCrawler has not been initialized. Call initialize_crawler first.",
)
return _crawler
@mcp.tool
def initialize_crawler(
desired_capabilities: Dict[str, Any],
output_dir: str,
api_key: str,
azure_endpoint: str,
platform: str = "android",
wait_time: int = 10,
test_example: Optional[str] = None,
server_url: str = "http://127.0.0.1:4723",
) -> str:
"""Instantiate the AppCrawler with the provided configuration."""
global _crawler
settings = CrawlerSettings(
appium=AppiumConfig(
desired_capabilities=desired_capabilities,
server_url=server_url,
),
azure=AzureConfig(
api_key=api_key,
azure_endpoint=azure_endpoint,
),
output_dir=output_dir,
platform=platform,
wait_time=wait_time,
test_example=test_example,
)
_crawler = AppCrawler(settings)
return "AppCrawler initialized successfully."
@mcp.tool
def get_screen_hash() -> str:
"""Generate a unique hash for the current screen's XML."""
crawler = _ensure_crawler()
return crawler.get_screen_hash()
@mcp.tool
def save_screen_xml(screen_hash: str, screen_name: Optional[str] = None) -> str:
"""Persist the current screen XML (and screenshot) to disk."""
crawler = _ensure_crawler()
return crawler.save_screen_xml(screen_hash, screen_name)
@mcp.tool
def query_llm_for_next_action(
xml: str,
task_description: str,
task_prompt: str,
) -> Dict[str, Any]:
"""Query the LLM for the next action given a screen XML."""
crawler = _ensure_crawler()
return crawler.query_llm_for_next_action(xml, task_description, task_prompt)
@mcp.tool
def get_mobile_by(locator_strategy: str) -> str:
"""Map a locator strategy string to Appium's MobileBy constant."""
crawler = _ensure_crawler()
return crawler.get_mobile_by(locator_strategy)
@mcp.tool
def perform_click(locator_strategy: str, locator_value: str) -> str:
"""Perform a click action on a UI element."""
crawler = _ensure_crawler()
crawler.perform_click(locator_strategy, locator_value)
return "Click action executed."
@mcp.tool
def perform_send_keys(
locator_strategy: str,
locator_value: str,
text: str,
) -> str:
"""Send keys to a UI element."""
crawler = _ensure_crawler()
crawler.perform_send_keys(locator_strategy, locator_value, text)
return "Send keys action executed."
@mcp.tool
def save_test_case(task_description: str, test_case_prompt: str) -> str:
"""Generate and save the test case using the recorded steps."""
crawler = _ensure_crawler()
crawler.save_test_case(task_description, test_case_prompt)
return "Test case saved."
@mcp.tool
def process_flow(
task_description: str,
task_prompt: Optional[str] = None,
test_case_prompt: Optional[str] = None,
) -> str:
"""Run the LLM-guided automation loop."""
crawler = _ensure_crawler()
crawler.process_flow(
task_description,
task_prompt or TASK_PROMPT,
test_case_prompt or GENERATE_TEST_CASE_PROMPT,
)
return "Process flow completed."
@mcp.tool
def get_example_test_case() -> str:
"""Return the bundled calculator example test case."""
return CALCULATOR_EXAMPLE_TEST
@mcp.tool
def get_task_prompt() -> str:
"""Return the default task automation prompt."""
return TASK_PROMPT
@mcp.tool
def get_generate_test_case_prompt() -> str:
"""Return the default test case generation prompt."""
return GENERATE_TEST_CASE_PROMPT
@mcp.tool
def get_default_prompts() -> Dict[str, str]:
"""Return both default prompt templates."""
return {
"task_prompt": TASK_PROMPT,
"generate_test_case_prompt": GENERATE_TEST_CASE_PROMPT,
}
@mcp.tool
def get_ui_dump(
device_name: str = "emulator-5554",
output_dir: str = "output",
appium_server_url: str = "http://localhost:4723/wd/hub",
settle_seconds: int = 5,
) -> Dict[str, Optional[str]]:
"""Retrieve the UiAutomator hierarchy from the connected device."""
path = fetch_ui_dump(
device_name=device_name,
output_dir=output_dir,
appium_server_url=appium_server_url,
settle_seconds=settle_seconds,
)
return {"path": path}
@mcp.tool
def run_docker_automation(
output_dir: str = "output",
image_name: str = "appium-ui-dump",
dockerfile_dir: Optional[str] = None,
) -> Dict[str, bool]:
"""Build and execute the Dockerized automation workflow."""
success = build_and_run_docker(
output_dir=output_dir,
image_name=image_name,
dockerfile_dir=dockerfile_dir,
)
return {"success": success}
@mcp.tool
def cleanup() -> str:
"""Terminate the Appium session and release resources."""
global _crawler
crawler = _ensure_crawler()
crawler.cleanup()
_crawler = None
return "AppCrawler resources cleaned up."
if __name__ == "__main__":
mcp.run()
