#!/usr/bin/env python3 """ Zapier MCP Setup v5 - Minimal, robust Google OAuth + MCP URL extraction """ import time import re import sys from playwright.sync_api import sync_playwright EMAIL = 'kinan@agileadapt.com' PASSWORD = 'Systema55' SCREENSHOT_DIR = '/mnt/e/genesis-system/scripts' def ss(page, name): path = f'{SCREENSHOT_DIR}/zv5_{name}.png' try: page.screenshot(path=path, full_page=False) 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) # Filter generic endpoints filtered = [u for u in found if '/servers' not in u and u != 'https://mcp.zapier.com/mcp'] return list(dict.fromkeys(filtered)) def run(): with sync_playwright() as p: # Use Firefox which has less bot detection issues with Google try: browser = p.firefox.launch(headless=False) print("Using Firefox") except: browser = p.chromium.launch( headless=False, args=['--no-sandbox', '--disable-blink-features=AutomationControlled'] ) print("Using Chromium") context = browser.new_context( viewport={'width': 1280, 'height': 900}, ) page = context.new_page() try: # ======================================================== # STEP 1: Zapier Login Page # ======================================================== print("\n[1] Going to Zapier login page...") page.goto('https://zapier.com/app/login', wait_until='domcontentloaded', timeout=30000) time.sleep(2) ss(page, '01_login') print(f" URL: {page.url}") # Click "Continue with Google" print("[2] Clicking 'Continue with Google'...") page.click('text=Continue with Google', timeout=10000) time.sleep(3) ss(page, '02_after_google') print(f" URL: {page.url}") # ======================================================== # STEP 2: Google OAuth - Email # ======================================================== if 'accounts.google.com' in page.url or 'google.com' in page.url: print("[3] On Google OAuth - entering email...") # Wait for email field email_inp = page.wait_for_selector('input[type="email"]', timeout=15000) email_inp.click() time.sleep(0.5) email_inp.fill(EMAIL) time.sleep(0.5) ss(page, '03_email_filled') # Click Next print(" Clicking Next...") page.click('#identifierNext, button:has-text("Next")', timeout=5000) time.sleep(4) ss(page, '04_after_email_next') print(f" URL: {page.url}") # ====================================================== # STEP 3: Google OAuth - Password # ====================================================== print("[4] Entering password...") # Wait for password field to appear try: pass_inp = page.wait_for_selector('input[type="password"]', timeout=15000, state='visible') time.sleep(1) pass_inp.click() time.sleep(0.3) pass_inp.fill(PASSWORD) time.sleep(0.5) ss(page, '05_password_filled') # Click Next print(" Clicking Next on password...") try: page.click('#passwordNext, button:has-text("Next")', timeout=5000) except: pass_inp.press('Enter') time.sleep(6) ss(page, '06_after_password') print(f" URL: {page.url}") except Exception as e: print(f" Password field error: {e}") ss(page, '05_error') # Maybe it navigated to a 2FA page or challenge time.sleep(5) ss(page, '05b_after_wait') print(f" URL after wait: {page.url}") # ======================================================== # STEP 4: Handle post-auth state # ======================================================== print("[5] Handling post-auth state...") time.sleep(5) current_url = page.url print(f" URL: {current_url}") ss(page, '07_post_auth') # Check for various scenarios page_text = '' try: page_text = page.inner_text('body')[:1000] except: pass # 2FA / challenge if 'challenge' in current_url or '2-step' in page_text.lower(): print(" 2FA/Challenge detected - waiting 45s for manual entry...") time.sleep(45) ss(page, '07b_after_2fa_wait') print(f" URL after wait: {page.url}") # Consent/approval page elif 'consent' in current_url: print(" Consent page - looking for Allow button...") try: allow_btn = page.locator('text=Allow, text=Continue, text=Grant').first if allow_btn.is_visible(timeout=3000): allow_btn.click() time.sleep(3) except: pass # Account chooser elif 'selectaccount' in current_url or 'choose an account' in page_text.lower(): print(" Account chooser - selecting agileadapt account...") try: account = page.locator(f'text={EMAIL}').first if account.is_visible(timeout=3000): account.click() time.sleep(3) except: try: # Click first account first = page.locator('[data-identifier]').first if first.is_visible(): first.click() time.sleep(3) except: pass time.sleep(5) print(f" Final auth URL: {page.url}") ss(page, '08_final_auth') # ======================================================== # STEP 5: Navigate to MCP # ======================================================== print("\n[6] Navigating to Zapier MCP page...") # Try mcp.zapier.com first try: page.goto('https://mcp.zapier.com', wait_until='domcontentloaded', timeout=25000) time.sleep(4) print(f" URL: {page.url}") ss(page, '09_mcp_zapier') except Exception as e: print(f" mcp.zapier.com failed: {e}") # Try actions.zapier.com/settings/mcp page.goto('https://actions.zapier.com/settings/mcp/', wait_until='domcontentloaded', timeout=25000) time.sleep(4) print(f" URL: {page.url}") ss(page, '09_actions_mcp') # Check if authenticated if 'login' in page.url.lower(): print(" Still not logged in! Trying again...") # Go back and try another login method page.goto('https://zapier.com/app/login', wait_until='domcontentloaded', timeout=20000) time.sleep(2) ss(page, '10_back_to_login') # This time try email/password directly (below Google button) try: # Fill email in the form at the bottom of Zapier's login email_inp = page.locator('input[name="email"], input[placeholder*="Email" i]').first if email_inp.is_visible(timeout=5000): email_inp.fill(EMAIL) time.sleep(1) page.click('text=Continue', timeout=5000) time.sleep(3) # Password pass_inp = page.locator('input[type="password"]').first if pass_inp.is_visible(timeout=8000): pass_inp.fill(PASSWORD) pass_inp.press('Enter') time.sleep(5) ss(page, '10b_after_email_login') print(f" After email login: {page.url}") # Navigate to MCP page.goto('https://mcp.zapier.com', wait_until='domcontentloaded', timeout=25000) time.sleep(4) print(f" MCP URL: {page.url}") ss(page, '10c_mcp_after_email_login') except Exception as e: print(f" Email login error: {e}") # ======================================================== # STEP 6: Inspect MCP page and get URL # ======================================================== print(f"\n[7] On MCP page: {page.url}") content = page.content() with open(f'{SCREENSHOT_DIR}/zv5_mcp_page.html', 'w', encoding='utf-8') as f: f.write(content) print(f" Title: {page.title()}") # Search for MCP URL mcp_urls = find_mcp_urls(content) if mcp_urls: print(f" Found MCP URLs: {mcp_urls}") return mcp_urls[0] # Dump elements try: els = page.eval_on_selector_all('*', 'els => els.filter(e => e.children.length === 0 && e.textContent.trim()).map(e => e.textContent.trim()).filter(t => t.length > 3 && t.length < 100)') print(" Page text elements:") for el in els[:30]: print(f" {el}") except: pass # ======================================================== # STEP 7: Look for New Server button # ======================================================== print("\n[8] Looking for New Server / Create button...") ss(page, '11_before_create') for sel in [ 'text=New server', 'text=Create server', 'text=New MCP server', 'text=Add server', 'button:has-text("New")', 'button:has-text("Create")', '[data-testid*="new"]', ]: try: btn = page.locator(sel).first if btn.is_visible(timeout=1500): print(f" Clicking: {sel}") btn.click() time.sleep(4) ss(page, '12_after_create') content = page.content() mcp_urls = find_mcp_urls(content) if mcp_urls: return mcp_urls[0] # Look for Connect/URL tab for tab_sel in ['text=Connect', 'text=MCP URL', 'text=Connection']: try: tab = page.locator(tab_sel).first if tab.is_visible(timeout=1500): tab.click() time.sleep(2) content = page.content() mcp_urls = find_mcp_urls(content) if mcp_urls: return mcp_urls[0] except: continue break except: continue # ======================================================== # STEP 8: Try API approach # ======================================================== print("\n[9] Trying API approach to get MCP URL...") # Try calling the Zapier NLA API for api_url in [ 'https://actions.zapier.com/api/v2/configuration/', 'https://mcp.zapier.com/api/mcp/servers', 'https://mcp.zapier.com/mcp/servers', ]: try: page.goto(api_url, wait_until='domcontentloaded', timeout=15000) time.sleep(2) print(f" {api_url} -> {page.url}") content = page.content() body = page.inner_text('body')[:1000] print(f" Content: {body[:300]}") mcp_urls = find_mcp_urls(content) if mcp_urls: return mcp_urls[0] # Check if it's JSON with a URL try: import json data = json.loads(body) print(f" JSON data keys: {list(data.keys()) if isinstance(data, dict) else 'list'}") # Look for URL fields def find_in_json(obj, depth=0): if depth > 5: return None if isinstance(obj, str) and 'mcp' in obj and 'zapier' in obj: return obj if isinstance(obj, dict): for v in obj.values(): r = find_in_json(v, depth+1) if r: return r if isinstance(obj, list): for item in obj: r = find_in_json(item, depth+1) if r: return r return None found = find_in_json(data) if found: print(f" Found in JSON: {found}") return found except: pass except Exception as e: print(f" Error: {e}") # Final screenshot ss(page, '99_final') content = page.content() with open(f'{SCREENSHOT_DIR}/zv5_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}") try: ss(page, 'ERROR') except: pass raise finally: try: time.sleep(3) browser.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] Review screenshots:") import subprocess result_ls = subprocess.run(['ls', '-la', f'{SCREENSHOT_DIR}/zv5_*.png'], capture_output=True, text=True) print(result_ls.stdout)