import gevent.monkey gevent.monkey.patch_all(asyncio=True) # Keep this at the very top import asyncio from flask import Flask, request, jsonify from proxy_lite import Runner, RunnerConfig import os import logging from datetime import datetime # Load environment variables from .env file try: from dotenv import load_dotenv load_dotenv() # This loads .env from current directory # Also try loading from subdirectory if it exists if os.path.exists('proxy-lite-demo-v2/.env'): load_dotenv('proxy-lite-demo-v2/.env') print("✅ Environment variables loaded from .env file") except ImportError: print("⚠️ python-dotenv not installed. Install with: pip install python-dotenv") except Exception as e: print(f"⚠️ Could not load .env file: {e}") logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) app = Flask(__name__) _runner: Runner | None = None def create_agent_task(target_url: str, request_task_instruction: str) -> str: """Create the agent task with mandatory new tab navigation.""" return f""" CRITICAL FIRST STEP - MANDATORY: Your VERY FIRST action must be to use the open_new_tab_and_go_to tool to navigate to {target_url} DO NOT skip this step. DO NOT use goto. You MUST use: open_new_tab_and_go_to(url='{target_url}') This is necessary because direct navigation to this URL gets stuck loading. The new tab approach bypasses this issue. STEP 1: Use open_new_tab_and_go_to(url='{target_url}') STEP 2: Wait for the page to be fully loaded (no loading spinners visible) STEP 3: {request_task_instruction} CRITICAL WORKFLOW - FOLLOW THESE EXACT STEPS IN SEQUENCE: STEP A: Select Permission Set - Use select_option_by_text tool to find and select the target permission set from Available list - Wait for "[ACTION COMPLETED]" response before proceeding STEP B: Click Add Button - After successful selection, immediately click the "Add" button to move permission set to Enabled list - Do NOT repeat the selection - proceed directly to Add button STEP C: Click Save Button - After clicking Add, immediately click "Save" to persist the changes - After Save, Salesforce redirects to User page indicating SUCCESS CRITICAL: Do NOT repeat actions. Each step should happen exactly once in sequence. GENERAL INSTRUCTIONS: - You must EXECUTE all actions immediately - do NOT just describe what you plan to do - Do NOT wait for user input or ask "what should I do next?" - Complete the entire task autonomously using the available tools - After completing all steps, use the return_value tool to provide your final response - If you make a plan, IMMEDIATELY execute it step by step using the appropriate tools """ async def initialize_runner_with_single_browser_login(username: str, password: str, target_url: str): """Initialize Proxy-lite Runner with single-browser login approach.""" global _runner logger.info("Initializing Proxy-lite Runner with single-browser login approach...") # Check for required API keys with debugging gemini_api_key = os.environ.get("GEMINI_API_KEY") hf_api_token = os.environ.get("HF_API_TOKEN") logger.info(f"🔍 Environment check: GEMINI_API_KEY={'SET' if gemini_api_key else 'NOT SET'}") logger.info(f"🔍 Environment check: HF_API_TOKEN={'SET' if hf_api_token else 'NOT SET'}") if not gemini_api_key and not hf_api_token: logger.error("Neither GEMINI_API_KEY nor HF_API_TOKEN environment variable is set") raise ValueError("Either GEMINI_API_KEY or HF_API_TOKEN must be set") # Prefer Gemini if both are available if gemini_api_key: client_config = { "name": "gemini", "model_id": "gemini-2.0-flash-001", "api_key": gemini_api_key, } logger.info("🤖 Using Gemini API for inference") else: client_config = { "name": "convergence", "model_id": "convergence-ai/proxy-lite-3b", "api_base": "https://convergence-ai-demo-api.hf.space/v1", "api_key": hf_api_token, "http_timeout": 50.0, "http_concurrent_connections": 50, } logger.info("🤖 Using Convergence AI for inference") config_dict = { "environment": { "name": "webbrowser", "homepage": "about:blank", # Will be skipped due to login "headless": True, "launch_args": ["--no-sandbox", "--disable-setuid-sandbox"], "screenshot_delay": 0.5, "include_html": True, "include_poi_text": True, "record_pois": True, "viewport_width": 1280, "viewport_height": 720, "browserbase_timeout": 7200, "keep_original_image": False, "no_pois_in_image": False, # --- SINGLE-BROWSER LOGIN CONFIG --- "perform_login": True, "salesforce_login_url": "https://login.salesforce.com/", "salesforce_username": username, "salesforce_password": password, "target_url": target_url # --- END SINGLE-BROWSER LOGIN CONFIG --- }, "solver": { "name": "simple", "agent": { "name": "proxy_lite", "client": client_config, "history_messages_limit": { "screenshot": 1 }, "history_messages_include": None, } }, "environment_timeout": 1800.0, "action_timeout": 1800.0, "task_timeout": 18000.0, "max_steps": 150, "logger_level": "DEBUG", "save_every_step": True, "detailed_logger_name": False } config = RunnerConfig.from_dict(config_dict) logger.info(f"DEBUG: app.py - Initializing Proxy-lite Runner with single-browser login approach") _runner = Runner(config=config) logger.info("Proxy-lite Runner initialized successfully with single-browser login") return _runner @app.route('/run_proxy_task', methods=['POST']) async def run_proxy_task_endpoint(): data = request.json if not data: return jsonify({"error": "No JSON data provided"}), 400 request_task_instruction = data.get('task') target_url = data.get('url') if not request_task_instruction: logger.warning("Received request without 'task' field. Returning 400.") return jsonify({"error": "No 'task' provided in request body"}), 400 if not target_url: logger.warning("Received request without 'url' field. Returning 400.") return jsonify({"error": "No 'url' provided in request body"}), 400 logger.info(f"Received user request task: '{request_task_instruction}'") logger.info(f"Target URL: '{target_url}'") # Check if this is a Salesforce URL is_salesforce_url = "salesforce.com" in target_url or "force.com" in target_url try: if is_salesforce_url: # Salesforce automation - requires login salesforce_username = os.environ.get("SALESFORCE_USERNAME") salesforce_password = os.environ.get("SALESFORCE_PASSWORD") if not salesforce_username or not salesforce_password: logger.error("Salesforce credentials (SALESFORCE_USERNAME, SALESFORCE_PASSWORD) environment variables not set.") return jsonify({"error": "Salesforce credentials not configured. Please set SALESFORCE_USERNAME and SALESFORCE_PASSWORD as Space secrets."}), 500 runner = await initialize_runner_with_single_browser_login(salesforce_username, salesforce_password, target_url) logger.info("Proxy-lite Runner initialized with Salesforce login." if salesforce_username and salesforce_password else "Proxy-lite Runner initialized for general web browsing.") logger.info("Agent will use mandatory new tab tool to bypass loading issues.") # Create the agent task using the centralized function agent_task = create_agent_task(target_url, request_task_instruction) logger.info("Executing agent task with mandatory new tab navigation...") result = await runner.run(task=agent_task) # Extract the actual result value from the Run object task_result = str(getattr(result, "value", None) or getattr(result, "result", None) or result) logger.info(f"Proxy-lite task completed. Output (truncated for log): {task_result[:500]}...") # Structure response for LWC integration response = { "status": "success", "message": "Task completed successfully", "data": { "task_result": task_result, "steps_completed": [ "Salesforce login completed", "Browser session initialized", "New tab navigation executed", "Target Salesforce setup page accessed", "Task execution completed successfully" ], "environment": { "target_url": target_url, "navigation_method": "new_tab_bypass" } }, "timestamp": datetime.now().isoformat(), "task_request": request_task_instruction } return jsonify(response) else: # General web browsing - no login required logger.info("Non-Salesforce URL detected. Skipping Salesforce login.") runner = await initialize_runner_with_single_browser_login("", "", target_url) logger.info("Proxy-lite Runner initialized for general web browsing.") logger.info("Agent will use mandatory new tab tool to bypass loading issues.") # Create the agent task using the centralized function agent_task = create_agent_task(target_url, request_task_instruction) logger.info("Executing agent task with mandatory new tab navigation...") result = await runner.run(task=agent_task) # Extract the actual result value from the Run object task_result = str(getattr(result, "value", None) or getattr(result, "result", None) or result) logger.info(f"Proxy-lite task completed. Output (truncated for log): {task_result[:500]}...") # Structure response for LWC integration response = { "status": "success", "message": "Task completed successfully", "data": { "task_result": task_result, "steps_completed": [ "Browser session initialized", "New tab navigation executed", "Target page accessed", "Task execution completed successfully" ], "environment": { "target_url": target_url, "navigation_method": "new_tab_bypass" } }, "timestamp": datetime.now().isoformat(), "task_request": request_task_instruction } return jsonify(response) except ValueError as e: logger.exception(f"Configuration error: {e}") error_response = { "status": "error", "error_type": "configuration_error", "message": "System configuration issue", "data": { "error_details": str(e), "suggested_action": "Check environment variables and system configuration", "steps_completed": ["Configuration validation failed"] }, "timestamp": datetime.now().isoformat(), "task_request": request_task_instruction } return jsonify(error_response), 500 except Exception as e: logger.exception(f"Unexpected error processing Salesforce task: {e}") error_response = { "status": "error", "error_type": "unexpected_error", "message": "An unexpected error occurred during task execution", "data": { "error_details": str(e), "error_class": type(e).__name__, "suggested_action": "Check logs for detailed error information and retry", "steps_completed": ["Login attempted", "Error occurred during execution"] }, "timestamp": datetime.now().isoformat(), "task_request": request_task_instruction } return jsonify(error_response), 500 @app.route('/') def root(): logger.info("Root endpoint accessed.") return "Proxy-lite API is running. Send POST requests to /run_proxy_task with a 'task' in JSON body." @app.route('/health', methods=['GET']) def health_check(): """Health check endpoint for monitoring and debugging""" logger.info("Health check endpoint accessed.") # Check environment variables env_status = { "GEMINI_API_KEY": "✓" if os.environ.get("GEMINI_API_KEY") else "✗", "HF_API_TOKEN": "✓" if os.environ.get("HF_API_TOKEN") else "✗", "SALESFORCE_USERNAME": "✓" if os.environ.get("SALESFORCE_USERNAME") else "✗", "SALESFORCE_PASSWORD": "✓" if os.environ.get("SALESFORCE_PASSWORD") else "✗" } health_response = { "status": "healthy", "message": "Proxy-lite API is running", "environment_variables": env_status, "endpoints": { "POST /run_proxy_task": "Execute Salesforce automation tasks (requires 'task' and 'url' parameters)", "GET /health": "Health check and status", "GET /": "API information" }, "supported_pages": [ "Warranty Lifecycle Management", "Account Forecasting Settings", "Sales Agreements", "Account Manager Targets", "Any Salesforce Setup page" ], "timestamp": datetime.now().isoformat() } return jsonify(health_response) if __name__ == '__main__': if not os.environ.get("GEMINI_API_KEY") and not os.environ.get("HF_API_TOKEN"): logger.error("Neither GEMINI_API_KEY nor HF_API_TOKEN environment variable is set. Please set at least one for local testing.") logger.info("Starting Flask development server on 0.0.0.0:6101...") app.run(host='0.0.0.0', port=6101, debug=True)