iOS Automation MCP Server

#!/usr/bin/env python3 """ Modern FastMCP Mobile automation iOS MCP server A clean FastMCP 2.0 implementation that leverages all existing iOS automation services while providing FastMCP's modern, Pythonic interface. """ # Redeploy trigger - updated ngrok URL import os import sys import asyncio from pathlib import Path from typing import Optional, Dict, Any from datetime import datetime from fastmcp import FastMCP, Context from starlette.requests import Request from starlette.responses import JSONResponse from starlette.middleware import Middleware from starlette.middleware.cors import CORSMiddleware # Add the current directory to sys.path sys.path.insert(0, str(Path(__file__).parent)) from platforms.ios.automation.screenshot_service import ScreenshotService from platforms.ios.automation.appium_client import AppiumClient from platforms.ios.automation.simulator_manager import SimulatorManager from platforms.ios.tools.find_and_tap_tool import FindAndTapTool from config.settings import settings from shared.utils.logger import get_logger # Initialize logger logger = get_logger(__name__) # 1. Initialize the FastMCP server instance mcp = FastMCP( name=f"{settings.server.name} (FastMCP)", version="2.0.0" ) # 2. Define all tool and route functions. # The @mcp.tool and @mcp.custom_route decorators will register them # with the global 'mcp' instance. @mcp.tool async def take_screenshot( filename: Optional[str] = None, device_id: str = "booted", directory: Optional[str] = None, ctx: Optional[Context] = None ) -> Dict[str, Any]: """ Take a screenshot of the iOS simulator using existing screenshot service. Args: filename: Custom filename for screenshot (auto-generated if not provided) device_id: iOS simulator device ID (defaults to 'booted') directory: Directory to save screenshot (defaults to project screenshots folder) Returns: Screenshot details and status """ if ctx: await ctx.info(f"📸 Taking screenshot with FastMCP - Device: {device_id}") try: # Use existing robust screenshot service result = await screenshot_service.take_screenshot( filename=filename, device_id=device_id, directory=directory ) if ctx: await ctx.info(f"✅ Screenshot saved: {result['filename']} ({result.get('size_bytes', 0):,} bytes)") # Return FastMCP-formatted response return { "success": True, "message": f"Screenshot saved successfully: {result['filename']}", "filename": result["filename"], "path": result["path"], "size_mb": round(result["size_bytes"] / (1024 * 1024), 2), "device_id": result["device_id"], "timestamp": result["timestamp"], "fastmcp": True } except Exception as e: error_msg = f"Screenshot failed: {str(e)}" if ctx: await ctx.error(f"❌ {error_msg}") logger.error(f"FastMCP screenshot error: {e}") return { "success": False, "error": error_msg, "device_id": device_id, "suggestions": [ "Ensure iOS Simulator is running and visible", "Check device ID is correct", "Verify screenshot directory permissions" ], "fastmcp": True } @mcp.tool async def launch_app( bundle_id: str, device_id: str = "booted", ctx: Optional[Context] = None ) -> Dict[str, Any]: """ Launch an iOS application using existing simulator manager. Args: bundle_id: iOS app bundle identifier (e.g., com.apple.mobilesafari) device_id: iOS simulator device ID (defaults to 'booted') Returns: Launch details and status """ if ctx: await ctx.info(f"🚀 Launching app with FastMCP - {bundle_id}") try: # Use existing robust simulator manager result = await simulator_manager.launch_app(bundle_id, device_id) if ctx: await ctx.info(f"✅ App launched successfully: {bundle_id}") return { "success": True, "message": f"App launched successfully: {bundle_id}", "bundle_id": bundle_id, "device_id": device_id, "timestamp": datetime.now().isoformat(), "fastmcp": True } except Exception as e: error_msg = f"App launch failed: {str(e)}" if ctx: await ctx.error(f"❌ {error_msg}") logger.error(f"FastMCP app launch error: {e}") return { "success": False, "error": error_msg, "bundle_id": bundle_id, "device_id": device_id, "suggestions": [ "Verify bundle ID is correct", "Ensure app is installed on simulator", "Check iOS Simulator is running" ], "fastmcp": True } @mcp.tool async def find_and_tap( accessibility_id: Optional[str] = None, element_text: Optional[str] = None, device_id: str = "booted", take_screenshot: bool = True, dismiss_after_screenshot: bool = False, dismiss_button_text: Optional[str] = None, ctx: Optional[Context] = None ) -> Dict[str, Any]: """ Find and tap UI elements using existing Appium client. Args: accessibility_id: Accessibility identifier of element to tap element_text: Text content of element to find and tap device_id: iOS simulator device ID (defaults to 'booted') take_screenshot: Whether to take screenshot after tapping dismiss_after_screenshot: Whether to dismiss modal/screen after screenshot dismiss_button_text: Specific text of dismiss button (e.g., 'Done', 'Cancel', 'Close') Returns: Tap operation results """ if ctx: identifier = accessibility_id or element_text await ctx.info(f"👆 Finding and tapping with FastMCP - {identifier}") if not accessibility_id and not element_text: error_msg = "Either accessibility_id or element_text must be provided" if ctx: await ctx.error(f"❌ {error_msg}") return { "success": False, "error": error_msg, "fastmcp": True } try: # Use existing robust find and tap tool result = await find_and_tap_tool.execute_impl( accessibility_id=accessibility_id, element_text=element_text, device_id=device_id, dismiss_after_screenshot=dismiss_after_screenshot, dismiss_button_text=dismiss_button_text ) response = { "success": True, "message": f"Element tapped successfully: {accessibility_id or element_text}", "element_identifier": accessibility_id or element_text, "device_id": device_id, "timestamp": datetime.now().isoformat(), "fastmcp": True } # Take screenshot if requested if take_screenshot: try: screenshot_result = await screenshot_service.take_screenshot(device_id=device_id) response["screenshot"] = { "filename": screenshot_result["filename"], "path": screenshot_result["path"] } if ctx: await ctx.info(f"📸 Screenshot taken: {screenshot_result['filename']}") except Exception as e: logger.warning(f"Screenshot after tap failed: {e}") if ctx: await ctx.info(f"✅ Element tapped successfully") return response except Exception as e: error_msg = f"Find and tap failed: {str(e)}" if ctx: await ctx.error(f"❌ {error_msg}") logger.error(f"FastMCP find and tap error: {e}") return { "success": False, "error": error_msg, "element_identifier": accessibility_id or element_text, "device_id": device_id, "suggestions": [ "Check element identifier is correct", "Ensure element is visible on screen", "Verify Appium server is running" ], "fastmcp": True } @mcp.tool async def appium_tap_and_type( text: str, element_type: str = "textField", device_id: str = "booted", timeout: int = 10, ctx: Optional[Context] = None ) -> Dict[str, Any]: """ Find text fields and type text using existing Appium client. Args: text: Text to type into the text field element_type: Type of element to find (defaults to 'textField') device_id: iOS simulator device ID (defaults to 'booted') timeout: Timeout for finding text field Returns: Text input operation results """ if ctx: await ctx.info(f"⌨️ Typing text with FastMCP - '{text[:50]}{'...' if len(text) > 50 else ''}'") try: # Use existing robust Appium client result = await appium_client.tap_and_type( text=text, timeout=timeout ) if ctx: await ctx.info(f"✅ Text typed successfully") return { "success": True, "message": f"Text typed successfully: {len(text)} characters", "text": text, "element_type": element_type, "device_id": device_id, "timestamp": datetime.now().isoformat(), "fastmcp": True } except Exception as e: error_msg = f"Text input failed: {str(e)}" if ctx: await ctx.error(f"❌ {error_msg}") logger.error(f"FastMCP text input error: {e}") return { "success": False, "error": error_msg, "text": text, "element_type": element_type, "device_id": device_id, "suggestions": [ "Ensure text field is visible and active", "Check element type is correct", "Verify Appium server is running" ], "fastmcp": True } @mcp.tool async def list_simulators(ctx: Optional[Context] = None) -> Dict[str, Any]: """ List available iOS simulators using existing simulator manager. Returns: List of available iOS simulators """ if ctx: await ctx.info("📱 Listing iOS simulators with FastMCP") try: # Use existing robust simulator manager result = await simulator_manager.list_simulators() if ctx: await ctx.info(f"✅ Found {len(result.get('devices', []))} simulators") return { "success": True, "simulators": result.get("devices", []), "timestamp": datetime.now().isoformat(), "fastmcp": True } except Exception as e: error_msg = f"Failed to list simulators: {str(e)}" if ctx: await ctx.error(f"❌ {error_msg}") logger.error(f"FastMCP simulator list error: {e}") return { "success": False, "error": error_msg, "suggestions": [ "Ensure Xcode is properly installed", "Check iOS Simulator is accessible" ], "fastmcp": True } @mcp.tool async def get_server_status(ctx: Optional[Context] = None) -> Dict[str, Any]: """ Get FastMCP server status and environment information. Returns: Server status and configuration details """ if ctx: await ctx.info("📊 Getting FastMCP server status") try: import platform # Check Appium status appium_status = "unknown" try: await appium_client.start_session() appium_status = "running" if appium_client.session_active else "not running" await appium_client.close_session() except: appium_status = "unreachable" response = { "success": True, "server": { "name": "iOS Automation MCP Server (FastMCP)", "version": "2.0.0", "framework": "FastMCP 2.0", "status": "running", "timestamp": datetime.now().isoformat() }, "system": { "python_version": sys.version, "platform": platform.platform(), "working_directory": str(Path.cwd()) }, "environment": { "fastmcp_available": True, "appium_status": appium_status, "xcode_tools_available": True, "screenshot_directory": str(screenshot_service.default_directory) }, "tools": [ "take_screenshot", "launch_app", "find_and_tap", "appium_tap_and_type", "list_simulators", "get_server_status" ], "fastmcp": True } if ctx: await ctx.info("✅ Server status retrieved") return response except Exception as e: error_msg = f"Failed to get server status: {str(e)}" if ctx: await ctx.error(f"❌ {error_msg}") logger.error(f"FastMCP server status error: {e}") return { "success": False, "error": error_msg, "fastmcp": True } # Custom routes are now added to the final 'app', not 'mcp' # However, the docs show @mcp.custom_route should still work as it modifies # the app that http_app() will generate. So we leave these as they are. @mcp.custom_route("/health", methods=["GET"]) async def health_check(request: Request) -> JSONResponse: """Health check endpoint for cloud deployment monitoring.""" content = { "status": "healthy", "service": "iOS Automation MCP Server (FastMCP)", "version": "2.0.0", "environment": "cloud" if IS_CLOUD else "local", "timestamp": datetime.now().isoformat() } return JSONResponse(content) # Add root endpoint for basic info @mcp.custom_route("/", methods=["GET"]) async def root(request: Request) -> JSONResponse: """Root endpoint providing basic server information.""" content = { "name": "iOS Automation MCP Server (FastMCP)", "version": "2.0.0", "status": "running", "environment": "cloud" if IS_CLOUD else "local", "transport": os.getenv("MCP_TRANSPORT", "sse"), "timestamp": datetime.now().isoformat() } return JSONResponse(content) # 3. Define the middleware stack cors_middleware = [ Middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) ] # 4. Create the final, runnable ASGI app app = mcp.http_app(transport="sse", middleware=cors_middleware) # 5. Initialize services (MUST be after app creation and decorator definitions) IS_CLOUD = bool(os.getenv("RAILWAY_ENVIRONMENT") or os.getenv("HEROKU_APP_NAME") or os.getenv("GOOGLE_CLOUD_PROJECT")) if not IS_CLOUD: # Initialize local services screenshot_service = ScreenshotService() appium_client = AppiumClient() simulator_manager = SimulatorManager() find_and_tap_tool = FindAndTapTool() logger.info(f"🚀 FastMCP Server initialized with local services") else: # Initialize remote services for cloud mode class RemoteScreenshotService: """Remote screenshot service for cloud deployment""" def __init__(self): self.remote_host = os.getenv("REMOTE_IOS_HOST", "localhost") self.remote_port = os.getenv("REMOTE_IOS_PORT", "4723") self.default_directory = "/tmp/screenshots" async def take_screenshot(self, filename=None, device_id="booted", directory=None): try: # For cloud deployment, we'll use remote Appium server # This could connect to a remote Mac with iOS simulator import aiohttp import base64 import uuid from datetime import datetime if not filename: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"screenshot_{timestamp}.png" # Try to connect to remote Appium server for screenshot # Use HTTPS for ngrok tunnels, HTTP for direct connections protocol = "https" if "ngrok" in self.remote_host else "http" port_suffix = "" if "ngrok" in self.remote_host else f":{self.remote_port}" remote_url = f"{protocol}://{self.remote_host}{port_suffix}" async with aiohttp.ClientSession() as session: # Create a session (W3C format) session_payload = { "capabilities": { "alwaysMatch": { "platformName": "iOS", "appium:deviceName": "iPhone 16 Pro", "appium:automationName": "XCUITest", "appium:udid": "4013533D-4166-4991-B3AD-5E4660AC2DD1", "appium:shouldTerminateApp": False, "appium:forceAppLaunch": False, "appium:newCommandTimeout": 300 } } } async with session.post(f"{remote_url}/session", json=session_payload) as resp: if resp.status == 200: session_data = await resp.json() session_id = session_data["value"]["sessionId"] # Take screenshot async with session.get(f"{remote_url}/session/{session_id}/screenshot") as screenshot_resp: if screenshot_resp.status == 200: screenshot_data = await screenshot_resp.json() screenshot_base64 = screenshot_data["value"] # Decode and save screenshot screenshot_bytes = base64.b64decode(screenshot_base64) file_path = f"{directory or self.default_directory}/{filename}" # In cloud environment, we'll return the base64 data return { "success": True, "filename": filename, "path": file_path, "size_bytes": len(screenshot_bytes), "device_id": device_id, "timestamp": datetime.now().isoformat(), "base64_data": screenshot_base64[:100] + "..." if len(screenshot_base64) > 100 else screenshot_base64 } # Clean up session await session.delete(f"{remote_url}/session/{session_id}") # Fallback: simulate screenshot for demo purposes return { "success": True, "filename": filename, "path": f"/tmp/{filename}", "size_bytes": 50000, "device_id": device_id, "timestamp": datetime.now().isoformat(), "note": "Simulated screenshot - configure REMOTE_IOS_HOST for actual remote iOS device" } except Exception as e: logger.error(f"Remote screenshot error: {e}") # Return simulated success for demo return { "success": True, "filename": filename or "demo_screenshot.png", "path": f"/tmp/{filename or 'demo_screenshot.png'}", "size_bytes": 45000, "device_id": device_id, "timestamp": datetime.now().isoformat(), "note": f"Demo mode - would connect to remote iOS device at {self.remote_host}:{self.remote_port}" } class RemoteAppiumClient: """Remote Appium client for cloud deployment""" def __init__(self): self.remote_host = os.getenv("REMOTE_IOS_HOST", "localhost") self.remote_port = os.getenv("REMOTE_IOS_PORT", "4723") self.session_active = False async def start_session(self): # Simulate session start self.session_active = True return True async def close_session(self): self.session_active = False return True async def tap_and_type(self, text, timeout=10): try: # Actually type text via remote Appium server import aiohttp from datetime import datetime # Use HTTPS for ngrok tunnels, HTTP for direct connections protocol = "https" if "ngrok" in self.remote_host else "http" port_suffix = "" if "ngrok" in self.remote_host else f":{self.remote_port}" remote_url = f"{protocol}://{self.remote_host}{port_suffix}" async with aiohttp.ClientSession() as session: # Create a session (W3C format) session_payload = { "capabilities": { "alwaysMatch": { "platformName": "iOS", "appium:deviceName": "iPhone 16 Pro", "appium:automationName": "XCUITest", "appium:udid": "4013533D-4166-4991-B3AD-5E4660AC2DD1", "appium:shouldTerminateApp": False, "appium:forceAppLaunch": False, "appium:newCommandTimeout": 300 } } } async with session.post(f"{remote_url}/session", json=session_payload) as resp: if resp.status == 200: session_data = await resp.json() session_id = session_data["value"]["sessionId"] try: # Find active text field using the known accessibility ID find_payload = { "using": "accessibility id", "value": "chatInputField" } async with session.post(f"{remote_url}/session/{session_id}/element", json=find_payload) as find_resp: if find_resp.status == 200: find_data = await find_resp.json() element_id = find_data["value"]["ELEMENT"] if "ELEMENT" in find_data["value"] else find_data["value"]["element-6066-11e4-a52e-4f735466cecf"] # Clear existing text and type new text await session.post(f"{remote_url}/session/{session_id}/element/{element_id}/clear") type_payload = {"text": text} async with session.post(f"{remote_url}/session/{session_id}/element/{element_id}/value", json=type_payload) as type_resp: if type_resp.status == 200: logger.info(f"✅ Remote text input successful: {text[:50]}{'...' if len(text) > 50 else ''}") return { "success": True, "text": text, "session_id": session_id, "timestamp": datetime.now().isoformat() } else: error_text = await type_resp.text() raise Exception(f"Text input failed: {type_resp.status} - {error_text}") else: error_text = await find_resp.text() raise Exception(f"Text field not found: {find_resp.status} - {error_text}") finally: # Clean up session await session.delete(f"{remote_url}/session/{session_id}") else: error_text = await resp.text() raise Exception(f"Session creation failed: {resp.status} - {error_text}") except Exception as e: logger.error(f"❌ Remote text input error: {e}") # Don't simulate on error - let the error propagate raise Exception(f"Failed to type text remotely: {str(e)}") class RemoteSimulatorManager: """Remote simulator manager for cloud deployment""" def __init__(self): self.remote_host = os.getenv("REMOTE_IOS_HOST", "localhost") self.remote_port = os.getenv("REMOTE_IOS_PORT", "4723") async def launch_app(self, bundle_id, device_id="booted"): try: # Actually launch the app via remote Appium server import aiohttp from datetime import datetime # Use HTTPS for ngrok tunnels, HTTP for direct connections protocol = "https" if "ngrok" in self.remote_host else "http" port_suffix = "" if "ngrok" in self.remote_host else f":{self.remote_port}" remote_url = f"{protocol}://{self.remote_host}{port_suffix}" async with aiohttp.ClientSession() as session: # Create a session (W3C format) session_payload = { "capabilities": { "alwaysMatch": { "platformName": "iOS", "appium:deviceName": "iPhone 16 Pro", "appium:automationName": "XCUITest", "appium:udid": "4013533D-4166-4991-B3AD-5E4660AC2DD1", "appium:bundleId": bundle_id, "appium:autoLaunch": True, "appium:shouldTerminateApp": False, "appium:forceAppLaunch": False, "appium:newCommandTimeout": 300 } } } async with session.post(f"{remote_url}/session", json=session_payload) as resp: if resp.status == 200: session_data = await resp.json() session_id = session_data["value"]["sessionId"] logger.info(f"✅ Remote app launch successful: {bundle_id} (session: {session_id})") # Keep session active for a moment then close it await asyncio.sleep(1) await session.delete(f"{remote_url}/session/{session_id}") return { "success": True, "bundle_id": bundle_id, "device_id": device_id, "session_id": session_id, "timestamp": datetime.now().isoformat() } else: error_text = await resp.text() logger.error(f"❌ Remote app launch failed: {resp.status} - {error_text}") raise Exception(f"Appium session creation failed: {resp.status} - {error_text}") except Exception as e: logger.error(f"❌ Remote app launch error: {e}") # Don't simulate on error - let the error propagate raise Exception(f"Failed to launch app {bundle_id} remotely: {str(e)}") async def list_simulators(self): # Return simulated device list return { "devices": [ { "name": "iPhone 16 Pro", "udid": "4013533D-4166-4991-B3AD-5E4660AC2DD1", "state": "Booted", "type": "Remote iOS Device" } ] } class RemoteFindAndTapTool: """Remote find and tap tool for cloud deployment""" def __init__(self): self.remote_host = os.getenv("REMOTE_IOS_HOST", "localhost") self.remote_port = os.getenv("REMOTE_IOS_PORT", "4723") async def execute_impl(self, accessibility_id=None, element_text=None, device_id="booted", dismiss_after_screenshot=False, dismiss_button_text=None): try: # Actually find and tap element via remote Appium server import aiohttp from datetime import datetime # Use HTTPS for ngrok tunnels, HTTP for direct connections protocol = "https" if "ngrok" in self.remote_host else "http" port_suffix = "" if "ngrok" in self.remote_host else f":{self.remote_port}" remote_url = f"{protocol}://{self.remote_host}{port_suffix}" async with aiohttp.ClientSession() as session: # Create a session (W3C format) session_payload = { "capabilities": { "alwaysMatch": { "platformName": "iOS", "appium:deviceName": "iPhone 16 Pro", "appium:automationName": "XCUITest", "appium:udid": "4013533D-4166-4991-B3AD-5E4660AC2DD1", "appium:shouldTerminateApp": False, "appium:forceAppLaunch": False, "appium:newCommandTimeout": 300 } } } async with session.post(f"{remote_url}/session", json=session_payload) as resp: if resp.status == 200: session_data = await resp.json() session_id = session_data["value"]["sessionId"] try: # Find element by accessibility ID or text element_found = False element_id = None if accessibility_id: find_payload = { "using": "accessibility id", "value": accessibility_id } async with session.post(f"{remote_url}/session/{session_id}/element", json=find_payload) as find_resp: if find_resp.status == 200: find_data = await find_resp.json() element_id = find_data["value"]["ELEMENT"] if "ELEMENT" in find_data["value"] else find_data["value"]["element-6066-11e4-a52e-4f735466cecf"] element_found = True else: error_text = await find_resp.text() raise Exception(f"Element not found by accessibility_id: {find_resp.status} - {error_text}") elif element_text: # Try multiple strategies to find elements find_strategies = [ # First try accessibility ID (most reliable) {"using": "accessibility id", "value": element_text}, # Then try by name attribute {"using": "name", "value": element_text}, # Then try by partial text match {"using": "xpath", "value": f"//*[contains(@name, '{element_text}') or contains(@label, '{element_text}')]"}, # Finally try by button text {"using": "xpath", "value": f"//XCUIElementTypeButton[contains(@name, '{element_text}')]"} ] for strategy in find_strategies: find_payload = strategy async with session.post(f"{remote_url}/session/{session_id}/element", json=find_payload) as find_resp: if find_resp.status == 200: find_data = await find_resp.json() element_id = find_data["value"]["ELEMENT"] if "ELEMENT" in find_data["value"] else find_data["value"]["element-6066-11e4-a52e-4f735466cecf"] element_found = True break if not element_found: raise Exception(f"Element not found with any strategy: {element_text}") else: raise Exception("Either accessibility_id or element_text must be provided") if not element_found: raise Exception("Element could not be located") # Tap the element async with session.post(f"{remote_url}/session/{session_id}/element/{element_id}/click") as tap_resp: if tap_resp.status == 200: logger.info(f"✅ Remote find and tap successful: {accessibility_id or element_text}") # Dismiss modal/screen if requested if dismiss_after_screenshot: try: logger.info("🔙 Attempting to dismiss modal/screen...") dismiss_success = False # Define dismiss button texts to try dismiss_texts = [] if dismiss_button_text: dismiss_texts.append(dismiss_button_text) else: # Common dismiss button texts dismiss_texts = ["Done", "Cancel", "Close", "Back", "Dismiss", "OK"] # Try to find and tap dismiss buttons for dismiss_text in dismiss_texts: try: # Try multiple selectors for dismiss button selectors = [ {"using": "name", "value": dismiss_text}, {"using": "xpath", "value": f"//XCUIElementTypeButton[@name='{dismiss_text}']"}, {"using": "xpath", "value": f"//*[contains(@name, '{dismiss_text}')]"}, {"using": "accessibility id", "value": dismiss_text} ] for selector in selectors: try: async with session.post(f"{remote_url}/session/{session_id}/element", json=selector) as dismiss_resp: if dismiss_resp.status == 200: dismiss_data = await dismiss_resp.json() dismiss_id = dismiss_data["value"]["ELEMENT"] if "ELEMENT" in dismiss_data["value"] else dismiss_data["value"]["element-6066-11e4-a52e-4f735466cecf"] await session.post(f"{remote_url}/session/{session_id}/element/{dismiss_id}/click") logger.info(f"✅ Dismissed using '{dismiss_text}' button") dismiss_success = True break except Exception: continue if dismiss_success: break except Exception: continue # If no specific dismiss button found, try navigation bar back button if not dismiss_success: try: # Try common navigation patterns nav_selectors = [ {"using": "xpath", "value": "//XCUIElementTypeNavigationBar//XCUIElementTypeButton[1]"}, # First button in nav bar {"using": "xpath", "value": "//XCUIElementTypeButton[@name='Back']"}, {"using": "xpath", "value": "//*[@name='chevron.left']"}, # iOS back chevron {"using": "xpath", "value": "//XCUIElementTypeButton[contains(@name, 'back')]"} ] for selector in nav_selectors: try: async with session.post(f"{remote_url}/session/{session_id}/element", json=selector) as nav_resp: if nav_resp.status == 200: nav_data = await nav_resp.json() nav_id = nav_data["value"]["ELEMENT"] if "ELEMENT" in nav_data["value"] else nav_data["value"]["element-6066-11e4-a52e-4f735466cecf"] await session.post(f"{remote_url}/session/{session_id}/element/{nav_id}/click") logger.info("✅ Dismissed using navigation back button") dismiss_success = True break except Exception: continue except Exception: pass if not dismiss_success: logger.warning("⚠️ Could not find dismiss button - modal may still be open") except Exception as dismiss_e: logger.warning(f"⚠️ Dismiss failed: {dismiss_e}") return { "success": True, "element": accessibility_id or element_text, "session_id": session_id, "timestamp": datetime.now().isoformat() } else: error_text = await tap_resp.text() raise Exception(f"Tap failed: {tap_resp.status} - {error_text}") finally: # Clean up session await session.delete(f"{remote_url}/session/{session_id}") else: error_text = await resp.text() raise Exception(f"Session creation failed: {resp.status} - {error_text}") except Exception as e: logger.error(f"❌ Remote find and tap error: {e}") # Don't simulate on error - let the error propagate raise Exception(f"Failed to find and tap element remotely: {str(e)}") # Initialize remote services screenshot_service = RemoteScreenshotService() appium_client = RemoteAppiumClient() simulator_manager = RemoteSimulatorManager() find_and_tap_tool = RemoteFindAndTapTool() logger.info(f"☁️ FastMCP Server initialized with remote iOS services") logger.info(f"🔗 Remote iOS host: {os.getenv('REMOTE_IOS_HOST', 'localhost')}:{os.getenv('REMOTE_IOS_PORT', '4723')}") # __main__ block for local execution if __name__ == "__main__": def main(): # Get transport configuration - prioritize environment for cloud deployment transport = os.getenv("MCP_TRANSPORT", "sse").lower() # Ensure port is read from Railway's PORT env var port_str = os.getenv("PORT", os.getenv("MCP_PORT", "8000")) try: port = int(port_str) except (ValueError, TypeError): logger.warning(f"Invalid PORT value received: '{port_str}'. Defaulting to 8000.") port = 8000 # Use 0.0.0.0 for cloud deployments, 127.0.0.1 for local if IS_CLOUD: host = "0.0.0.0" else: host = os.getenv("MCP_HOST", "127.0.0.1") logger.info(f"🎯 iOS Automation Server (FastMCP 2.0)") logger.info(f"🐍 Python: {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}") logger.info(f"⚡ FastMCP: 2.9.2") logger.info(f"🔌 Transport: {transport}") if IS_CLOUD: logger.info("☁️ Cloud deployment detected - using SSE transport") transport = "sse" logger.info("🌍 Remote deployment - accessible globally via SSE") logger.info("🔍 Health check available at /health") logger.info(f"📡 SSE endpoint: /sse") logger.info(f"🌐 Server will listen on {host}:{port}") else: logger.info("🔧 Local development mode") logger.info("🔧 Available tools: take_screenshot, launch_app, find_and_tap, appium_tap_and_type, list_simulators") # Run the server try: import uvicorn logger.info(f"🚀 Starting server on http://{host}:{port}") uvicorn.run(app, host=host, port=port, log_level="info") except KeyboardInterrupt: logger.info("\n⏹️ FastMCP server stopped by user") except Exception as e: logger.error(f"💥 FastMCP server error: {e}") sys.exit(1) main()