#!/usr/bin/env python3 """ Zapier MCP v6 - Human-like Google OAuth with persistent browser context """ import time import re import sys import os import json from playwright.sync_api import sync_playwright EMAIL = 'kinan@agileadapt.com' PASSWORD = 'Systema55' SCREENSHOT_DIR = '/mnt/e/genesis-system/scripts' USER_DATA_DIR = '/tmp/zapier_browser_profile' def ss(page, name): path = f'{SCREENSHOT_DIR}/zv6_{name}.png' try: page.screenshot(path=path) print(f" [ss] {name}") except Exception as e: print(f" [ss failed] {name}: {e}") def find_mcp_urls(content): patterns = [ r'https://actions\.zapier\.com/mcp/[a-zA-Z0-9/_?=&.-]+', r'https://mcp\.zapier\.com/mcp/[a-zA-Z0-9/_?=&.-]+', ] found = [] for pattern in patterns: matches = re.findall(pattern, content) found.extend(matches) filtered = [u for u in found if '/servers' not in u and u != 'https://mcp.zapier.com/mcp'] return list(dict.fromkeys(filtered)) def human_type(element, text, delay=0.08): """Type text character by character to simulate human input""" element.click() time.sleep(0.3) for char in text: element.type(char) time.sleep(delay + (0.03 * (hash(char) % 3))) # Variable delay def run(): with sync_playwright() as p: # Use persistent context to keep cookies/session across runs context = p.chromium.launch_persistent_context( user_data_dir=USER_DATA_DIR, headless=False, slow_mo=80, args=[ '--no-sandbox', '--disable-blink-features=AutomationControlled', '--disable-features=IsolateOrigins,site-per-process', ], viewport={'width': 1280, 'height': 900}, user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36', ) try: page = context.new_page() # Override webdriver flag to reduce bot detection page.add_init_script(""" Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5], }); window.chrome = { runtime: {}, }; """) # ============================================================ # STEP 1: Go directly to Google login for the account first # ============================================================ print("\n[1] Logging into Google directly...") page.goto('https://accounts.google.com', wait_until='domcontentloaded', timeout=30000) time.sleep(2) ss(page, '01_google') print(f" URL: {page.url}") # Check if already logged in if 'myaccount.google.com' in page.url or 'accounts.google.com/b' in page.url: print(" Already logged into Google!") elif 'accounts.google.com' in page.url: # Look for sign-in button sign_in_found = False for sel in ['text=Sign in', 'a:has-text("Sign in")', '[data-action="sign in"]']: try: btn = page.locator(sel).first if btn.is_visible(timeout=3000): btn.click() time.sleep(2) sign_in_found = True break except: continue # Handle the Google sign-in flow if 'accounts.google.com/signin' in page.url or 'accounts.google.com/v3/signin' in page.url or not sign_in_found: print(" Entering email...") # Navigate directly to sign-in page.goto('https://accounts.google.com/signin/v2/identifier', wait_until='domcontentloaded', timeout=20000) time.sleep(2) ss(page, '02_google_signin') # Enter email with human-like typing email_inp = page.wait_for_selector('#identifierId', timeout=15000) time.sleep(1) human_type(email_inp, EMAIL) time.sleep(1) ss(page, '03_email_typed') # Press Enter instead of clicking button (less bot-like) page.keyboard.press('Enter') time.sleep(4) ss(page, '04_after_email') print(f" After email URL: {page.url}") # Wait for password field try: pass_inp = page.wait_for_selector('input[type="password"]', timeout=15000, state='visible') time.sleep(1.5) human_type(pass_inp, PASSWORD) time.sleep(1) ss(page, '05_password_typed') # Press Enter page.keyboard.press('Enter') time.sleep(6) ss(page, '06_after_password') print(f" After password URL: {page.url}") except Exception as e: print(f" Password error: {e}") ss(page, '05_error') # Check final Google login state time.sleep(3) print(f" Google URL: {page.url}") ss(page, '07_google_state') # Handle 2FA if present page_text = '' try: page_text = page.inner_text('body')[:1000] except: pass if '2-step' in page_text.lower() or 'verify' in page_text.lower() or 'phone' in page_text.lower(): print(" 2FA detected - waiting 60s for manual entry...") time.sleep(60) ss(page, '07b_after_2fa') # ============================================================ # STEP 2: Now go to Zapier login (Google should recognize us) # ============================================================ print("\n[2] Navigating to Zapier login...") page.goto('https://zapier.com/app/login', wait_until='domcontentloaded', timeout=30000) time.sleep(3) ss(page, '08_zapier_login') print(f" URL: {page.url}") # Check if already logged into Zapier if 'zapier.com/app/' in page.url and 'login' not in page.url: print(" Already logged into Zapier!") else: # Click Continue with Google print(" Clicking Continue with Google...") try: page.click('text=Continue with Google', timeout=10000) time.sleep(4) ss(page, '09_after_google_click') print(f" URL: {page.url}") # If Google asks for account selection if 'accounts.google.com' in page.url: page_text = page.inner_text('body')[:1000] if 'choose an account' in page_text.lower() or 'select' in page_text.lower(): print(" Account chooser - selecting our account...") try: account = page.locator(f'text={EMAIL}').first if account.is_visible(timeout=5000): account.click() time.sleep(4) except: # Click first account try: first = page.locator('[data-identifier]').first first.click() time.sleep(4) except: pass elif 'password' in page_text.lower() or 'Enter your password' in page_text: print(" Google asking for password again...") pass_inp = page.wait_for_selector('input[type="password"]', timeout=10000, state='visible') time.sleep(1) human_type(pass_inp, PASSWORD) time.sleep(1) page.keyboard.press('Enter') time.sleep(5) elif 'identifier' in page.url: print(" Google asking for email again...") email_inp = page.wait_for_selector('#identifierId', timeout=10000, state='visible') time.sleep(0.5) human_type(email_inp, EMAIL) time.sleep(0.5) page.keyboard.press('Enter') time.sleep(4) pass_inp = page.wait_for_selector('input[type="password"]', timeout=10000, state='visible') time.sleep(1) human_type(pass_inp, PASSWORD) time.sleep(1) page.keyboard.press('Enter') time.sleep(5) # Handle consent/OAuth approval time.sleep(3) if 'consent' in page.url: print(" Handling OAuth consent...") for allow_sel in ['text=Allow', 'text=Continue', '[id*="submit"]', 'button[type="submit"]']: try: btn = page.locator(allow_sel).first if btn.is_visible(timeout=3000): btn.click() time.sleep(4) break except: continue except Exception as e: print(f" Google click error: {e}") # Check login status time.sleep(5) print(f" Zapier URL: {page.url}") ss(page, '10_zapier_state') # ============================================================ # STEP 3: Navigate to MCP # ============================================================ print("\n[3] Going to MCP page...") # Navigate to the Zapier MCP page page.goto('https://mcp.zapier.com', wait_until='domcontentloaded', timeout=30000) time.sleep(5) print(f" URL: {page.url}") ss(page, '11_mcp_page') if 'login' in page.url.lower(): print(" Still redirecting to login, checking page...") # Maybe we need to re-authenticate ss(page, '11_login_redirect') # Try the old actions.zapier.com approach page.goto('https://actions.zapier.com', wait_until='domcontentloaded', timeout=20000) time.sleep(3) print(f" actions.zapier.com URL: {page.url}") ss(page, '12_actions_zapier') # Inspect the current page content = page.content() with open(f'{SCREENSHOT_DIR}/zv6_current_page.html', 'w', encoding='utf-8') as f: f.write(content) print(f" Title: {page.title()}") # Find MCP URLs mcp_urls = find_mcp_urls(content) if mcp_urls: print(f" Found MCP URLs: {mcp_urls}") return mcp_urls[0] # Dump page structure try: buttons = page.eval_on_selector_all('button, [role="button"], a', 'els => els.map(e => e.textContent.trim()).filter(t => t.length > 0 && t.length < 100)') print(" Page buttons/links:") for b in buttons[:30]: print(f" {b}") except: pass # ============================================================ # STEP 4: Look for and create MCP server # ============================================================ print("\n[4] Looking for MCP server creation...") for sel in [ 'text=New server', 'text=Create server', 'text=New MCP server', 'button:has-text("New")', 'button:has-text("Create")', 'button:has-text("Add")', '[data-testid="new-server-button"]', '[data-testid="create-server"]', ]: try: btn = page.locator(sel).first if btn.is_visible(timeout=1500): print(f" Clicking: {sel}") btn.click() time.sleep(4) ss(page, '13_after_create') content = page.content() mcp_urls = find_mcp_urls(content) if mcp_urls: return mcp_urls[0] break except: continue # ============================================================ # STEP 5: Try to find MCP URL via Connect tab # ============================================================ print("\n[5] Looking for Connect tab...") for tab_sel in ['text=Connect', 'text=MCP URL', 'text=Connection URL', '[role="tab"]:has-text("Connect")']: try: tab = page.locator(tab_sel).first if tab.is_visible(timeout=1500): tab.click() time.sleep(3) ss(page, '14_connect_tab') content = page.content() mcp_urls = find_mcp_urls(content) if mcp_urls: return mcp_urls[0] # Look for URL in code blocks try: codes = page.eval_on_selector_all('code, pre, input[readonly]', 'els => els.map(e => e.value || e.textContent)') for code in codes: if 'zapier' in code.lower() and 'mcp' in code.lower(): print(f" Found in code: {code[:200]}") match = re.search(r'https://[^\s"\'<>]+mcp[^\s"\'<>]+', code) if match: return match.group() except: pass except: continue # ============================================================ # STEP 6: Try API to get the MCP server URL # ============================================================ print("\n[6] Trying Zapier API...") api_urls = [ 'https://mcp.zapier.com/api/mcp/servers', 'https://actions.zapier.com/api/v2/configuration/', 'https://mcp.zapier.com/mcp/servers', ] for api_url in api_urls: try: page.goto(api_url, wait_until='domcontentloaded', timeout=15000) time.sleep(2) print(f" {api_url}") print(f" -> {page.url}") if 'login' not in page.url: body = page.inner_text('body')[:2000] print(f" Content: {body[:500]}") # Try to parse as JSON try: import json data = json.loads(body) print(f" JSON: {json.dumps(data, indent=2)[:1000]}") # Look for MCP URL in JSON json_str = json.dumps(data) mcp_matches = re.findall(r'https://[^"\\]+mcp[^"\\]+', json_str) if mcp_matches: print(f" Found in JSON: {mcp_matches}") valid = [u for u in mcp_matches if '/servers' not in u] if valid: return valid[0] except: pass mcp_urls = find_mcp_urls(page.content()) if mcp_urls: return mcp_urls[0] except Exception as e: print(f" Error for {api_url}: {e}") # Final state ss(page, '99_final') content = page.content() with open(f'{SCREENSHOT_DIR}/zv6_final.html', 'w', encoding='utf-8') as f: f.write(content) print(f"\nFinal URL: {page.url}") return None except Exception as e: print(f"[ERROR] {e}") import traceback traceback.print_exc() try: ss(page, 'ERROR') except: pass return None finally: try: time.sleep(3) context.close() except: pass if __name__ == '__main__': result = run() if result: print(f"\n{'='*60}") print(f"ZAPIER MCP URL: {result}") print(f"{'='*60}") with open(f'{SCREENSHOT_DIR}/zapier_mcp_url.txt', 'w') as f: f.write(result) print(f"Saved to: {SCREENSHOT_DIR}/zapier_mcp_url.txt") else: print("\n[INCOMPLETE] No MCP URL found") print(f"Screenshots in: {SCREENSHOT_DIR}/zv6_*.png")