diff --git a/samples/cdp-tests/.env.example b/samples/cdp-tests/.env.example new file mode 100644 index 0000000..7b26d57 --- /dev/null +++ b/samples/cdp-tests/.env.example @@ -0,0 +1,11 @@ +# Microsoft Playwright Service - Environment Variables +# Copy this file to .env and fill in your values + +# Playwright Service (Required for all samples) +PLAYWRIGHT_SERVICE_URL= +PLAYWRIGHT_SERVICE_ACCESS_TOKEN= + +# Azure OpenAI (Required for browser_use_remote.py only) +AZURE_OPENAI_API_KEY= +AZURE_OPENAI_ENDPOINT= +AZURE_OPENAI_API_VERSION= diff --git a/samples/cdp-tests/Browser-Use-Remote.py b/samples/cdp-tests/Browser-Use-Remote.py new file mode 100644 index 0000000..4d7549d --- /dev/null +++ b/samples/cdp-tests/Browser-Use-Remote.py @@ -0,0 +1,160 @@ +""" +Amazon Product Search with Browser-Use + Hosted LLM (Azure OpenAI) + +This script lets you search Amazon for products using a remote Playwright browser +(Microsoft Playwright Cloud) and a hosted LLM (Azure OpenAI). + +---------------------------------------- +๐Ÿ“Œ Prerequisites +---------------------------------------- +1๏ธโƒฃ Python environment with the following packages installed: + pip install aiohttp pydantic browser-use python-dotenv + +2๏ธโƒฃ Hosted LLM Setup (Azure OpenAI) + - Create an Azure OpenAI resource in the Azure Portal. + - Deploy a model (e.g., gpt-35-turbo). + - Set the following environment variables: + + export AZURE_OPENAI_API_KEY="your_api_key_here" + export AZURE_OPENAI_ENDPOINT="https://.openai.azure.com/" + export AZURE_OPENAI_API_VERSION="2023-07-01-preview" + +3๏ธโƒฃ Playwright Remote Browser Setup + - Sign up for Microsoft Playwright Cloud. + - Obtain your service URL and access token. + - Set the following environment variables: + + export PLAYWRIGHT_SERVICE_URL="wss://.api.playwright.microsoft.com/playwrightworkspaces//browsers" + export PLAYWRIGHT_SERVICE_ACCESS_TOKEN="your_access_token" + +---------------------------------------- +๐Ÿ“Œ How to Use +---------------------------------------- +1๏ธโƒฃ Run the script: + python Browser-Use-Remote.py + +2๏ธโƒฃ Enter product keywords when prompted. + (Default is "wireless mouse" if you press Enter.) + +3๏ธโƒฃ The script will connect to the remote browser and hosted LLM to perform + the search and print structured results in the terminal. +""" + +import asyncio +import os +from pydantic import BaseModel +from dotenv import load_dotenv +from browser_use import Agent +from browser_use.llm import AzureChatOpenAI +from browser_use.browser.session import BrowserSession +from browser_use.browser.profile import BrowserProfile + +# Load environment variables from .env file +load_dotenv() + +# Import the shared module +from playwright_service_client import get_cdp_endpoint + + +# --- Azure OpenAI Setup --- +def get_llm(): + """Initialize the hosted Azure OpenAI LLM.""" + return AzureChatOpenAI( + model_name="gpt-35-turbo", + openai_api_key=os.environ["AZURE_OPENAI_API_KEY"], + azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"], + deployment_name="gpt-35-turbo", + max_tokens=3000, + api_version=os.environ["AZURE_OPENAI_API_VERSION"], + ) + + +# --- Remote Playwright Browser --- +async def create_remote_browser_session() -> BrowserSession: + """ + Create a remote Playwright browser session. + Returns a BrowserSession configured for browser-use. + """ + cdp_url = await get_cdp_endpoint() + print(f"๐Ÿ”— Connected to Playwright Service") + + profile = BrowserProfile(cdp_url=cdp_url) + return BrowserSession(browser_profile=profile) + + +# --- Data Models --- +class Product(BaseModel): + name: str + price: str + rating: str | None = None + reviews: str | None = None + url: str + + +class ProductSearchResults(BaseModel): + items: list[Product] = [] + + +# --- Amazon Search Function --- +async def search_amazon_remote(keywords: str = 'wireless mouse'): + """Search Amazon using remote browser + hosted Azure OpenAI LLM""" + print(f"๐ŸŽฏ Searching Amazon for: {keywords}") + print("๐Ÿ”ง Using: Remote Playwright Browser + Azure OpenAI LLM") + + browser_session = await create_remote_browser_session() + llm = get_llm() + + agent = Agent( + task=f""" + Go to Amazon (https://www.amazon.com) and search for "{keywords}". + Collect the top 5 product results. + For each product, extract: + - Product name + - Price + - Rating (stars) + - Number of reviews + - Product page URL + Return structured output. + """, + llm=llm, + browser_session=browser_session, + output_model_schema=ProductSearchResults, + ) + + print("๐ŸŽฏ Starting Amazon search...") + result = await agent.run() + print("โœ… Search completed successfully!") + return result + + +# --- Main --- +async def main(): + print("๐Ÿ›’ Amazon Product Search with Browser-Use + Azure OpenAI") + print("=" * 50) + + keywords = input('Enter product keywords (default "wireless mouse"): ').strip() or 'wireless mouse' + + try: + result = await search_amazon_remote(keywords) + + if result and result.structured_output: + products = result.structured_output + print(f'\n{"=" * 70}') + print(f'Amazon Search Results for "{keywords}"') + print(f'{"=" * 70}\n') + for i, item in enumerate(products.items, 1): + print(f'{i}. Name: {item.name}') + print(f' Price: {item.price}') + if item.rating: print(f' Rating: {item.rating}') + if item.reviews: print(f' Reviews: {item.reviews}') + print(f' URL: {item.url}') + print(f'{"-" * 70}') + else: + print("โŒ No products found or search failed") + + except Exception as e: + print(f"โŒ Error: {e}") + + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/samples/cdp-tests/README.md b/samples/cdp-tests/README.md new file mode 100644 index 0000000..a3abd97 --- /dev/null +++ b/samples/cdp-tests/README.md @@ -0,0 +1,81 @@ +# Microsoft Playwright Service - Python CDP Samples + +Python samples for connecting to Microsoft Playwright Service via CDP. + +## ๐Ÿ“ Files + +| File | Use Case | Description | +|------|----------|-------------| +| `playwright_service_client.py` | Core Module | Shared client for all samples | +| `test_runner.py` | **Testing** | Test runner with helpers | +| `connectOverCDPScript.py` | **Manual** | Simple connect_over_cdp example | +| `Browser-Use-Remote.py` | **AI Agent** | Browser-Use + Azure OpenAI | + +## ๐Ÿš€ Quick Start + +```bash +# Install dependencies +pip install -r requirements.txt + +# Copy .env.example to .env and fill in your values +cp .env.example .env + +# Run samples +python connectOverCDPScript.py # Basic CDP connection +python test_runner.py # Run example tests +pytest test_runner.py -v # With pytest +python Browser-Use-Remote.py # AI agent (requires Azure OpenAI) +``` + +## ๐Ÿ“– Usage Examples + +### Basic CDP Connection +```python +from playwright.async_api import async_playwright +from playwright_service_client import get_cdp_endpoint + +cdp_url = await get_cdp_endpoint() +async with async_playwright() as p: + browser = await p.chromium.connect_over_cdp(cdp_url) + page = await browser.new_page() + await page.goto("https://example.com") +``` + +### Test Automation +```python +from test_runner import remote_page + +async with remote_page() as page: + await page.goto("https://example.com") + assert await page.title() == "Example Domain" +``` + +### AI Agent +```python +from playwright_service_client import get_cdp_endpoint +from browser_use.browser.profile import BrowserProfile + +cdp_url = await get_cdp_endpoint() +profile = BrowserProfile(cdp_url=cdp_url) +``` + +## ๐Ÿ”ง Environment Variables + +Copy `.env.example` to `.env` and fill in your values: + +```bash +PLAYWRIGHT_SERVICE_URL=wss://.api.playwright.microsoft.com/playwrightworkspaces//browsers +PLAYWRIGHT_SERVICE_ACCESS_TOKEN=your_access_token + +# For AI agent example only +AZURE_OPENAI_API_KEY=your_api_key +AZURE_OPENAI_ENDPOINT=https://.openai.azure.com/ +AZURE_OPENAI_API_VERSION=2023-07-01-preview +``` + +## ๐Ÿ“š Resources + +- [Microsoft Playwright Service](https://learn.microsoft.com/azure/playwright-testing/) +- [Playwright Python](https://playwright.dev/python/) +- [Browser-Use](https://github.com/browser-use/browser-use) + diff --git a/samples/cdp-tests/connectOverCDPScript.js b/samples/cdp-tests/connectOverCDPScript.js new file mode 100644 index 0000000..bc2707d --- /dev/null +++ b/samples/cdp-tests/connectOverCDPScript.js @@ -0,0 +1,93 @@ +""" +Connect Over CDP - Microsoft Playwright Service + +Simple example showing how to connect to a remote browser via CDP. +This demonstrates a NON-TESTING scenario for manual browser automation. + +---------------------------------------- +๐Ÿ“Œ Prerequisites +---------------------------------------- +1๏ธโƒฃ Python environment with the following packages installed: + pip install playwright aiohttp python-dotenv + +2๏ธโƒฃ Playwright Remote Browser Setup + - Sign up for Microsoft Playwright Service. + - Obtain your service URL and access token. + - Copy .env.example to .env and fill in your values: + + PLAYWRIGHT_SERVICE_URL=wss://.api.playwright.microsoft.com/playwrightworkspaces//browsers + PLAYWRIGHT_SERVICE_ACCESS_TOKEN=your_access_token + +---------------------------------------- +๐Ÿ“Œ How to Use +---------------------------------------- +1๏ธโƒฃ Run the script: + python connectOverCDPScript.py + +2๏ธโƒฃ The script will: + - Connect to the remote browser + - Navigate to example.com + - Take a screenshot + - Extract page content + - Click a link +""" + +import asyncio +from playwright.async_api import async_playwright +from dotenv import load_dotenv + +# Load environment variables from .env file +load_dotenv() + +from playwright_service_client import get_cdp_endpoint + + +async def main(): + """Connect to a remote browser via CDP and perform basic operations.""" + + print("๐Ÿ”— Connecting to Microsoft Playwright Service...") + + # Step 1: Get CDP endpoint from the service + cdp_url = await get_cdp_endpoint() + print(f"โœ… Got CDP endpoint") + + # Step 2: Connect to remote browser using Playwright + async with async_playwright() as p: + browser = await p.chromium.connect_over_cdp( + cdp_url, + headers={"User-Agent": "Chrome-DevTools-Protocol/1.3"} + ) + print(f"โœ… Connected to remote browser") + + # Step 3: Use the browser + context = await browser.new_context() + page = await context.new_page() + + # Example: Navigate and take screenshot + print("๐Ÿ“„ Navigating to example.com...") + await page.goto("https://example.com") + + title = await page.title() + print(f"๐Ÿ“Œ Page title: {title}") + + # Take a screenshot + await page.screenshot(path="screenshot.png") + print("๐Ÿ“ธ Screenshot saved to screenshot.png") + + # Example: Extract content + heading = await page.locator("h1").text_content() + print(f"๐Ÿ“ Page heading: {heading}") + + # Example: Click a link + await page.click("a") + await page.wait_for_load_state("networkidle") + print(f"๐Ÿ”— Navigated to: {page.url}") + + # Cleanup + await context.close() + await browser.close() + print("โœ… Done!") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/samples/cdp-tests/playwright_service_client.py b/samples/cdp-tests/playwright_service_client.py new file mode 100644 index 0000000..ca4815d --- /dev/null +++ b/samples/cdp-tests/playwright_service_client.py @@ -0,0 +1,119 @@ +""" +Microsoft Playwright Service - Python Client + +Get a CDP endpoint URL to connect to remote browsers. + +---------------------------------------- +๐Ÿ“Œ Prerequisites +---------------------------------------- +pip install aiohttp python-dotenv + +---------------------------------------- +๐Ÿ“Œ Environment Variables +---------------------------------------- +PLAYWRIGHT_SERVICE_URL=wss://.api.playwright.microsoft.com/playwrightworkspaces//browsers +PLAYWRIGHT_SERVICE_ACCESS_TOKEN=your_access_token + +---------------------------------------- +๐Ÿ“Œ How to Use +---------------------------------------- + from playwright_service_client import get_cdp_endpoint + + cdp_url = await get_cdp_endpoint() + browser = await playwright.chromium.connect_over_cdp(cdp_url) +""" + +import re +import os +import aiohttp +from dotenv import load_dotenv + +load_dotenv() + + +class PlaywrightServiceError(Exception): + """Exception for Playwright Service errors.""" + pass + + +# URL pattern: wss://.api.playwright.microsoft.com/playwrightworkspaces//browsers +_URL_PATTERN = re.compile( + r'wss://(\w+)\.api\.playwright\.microsoft\.com/playwrightworkspaces/([^/]+)/browsers' +) + + +def _parse_url(url: str) -> tuple[str, str]: + """Extract region and workspace ID from service URL.""" + match = _URL_PATTERN.match(url) + if not match: + raise PlaywrightServiceError( + f"Invalid PLAYWRIGHT_SERVICE_URL format: {url}\n" + f"Expected: wss://.api.playwright.microsoft.com/playwrightworkspaces//browsers" + ) + return match.group(1), match.group(2) + + +async def get_cdp_endpoint( + service_url: str | None = None, + access_token: str | None = None, + os_name: str = "Linux", + api_version: str = "2025-09-01" +) -> str: + """ + Get a CDP endpoint URL from Microsoft Playwright Service. + + Args: + service_url: Service URL (defaults to PLAYWRIGHT_SERVICE_URL env var) + access_token: Access token (defaults to PLAYWRIGHT_SERVICE_ACCESS_TOKEN env var) + os_name: Browser OS - "Linux" or "Windows" (default: Linux) + api_version: API version (default: 2025-09-01) + + Returns: + WebSocket URL for CDP connection + + Example: + cdp_url = await get_cdp_endpoint() + browser = await playwright.chromium.connect_over_cdp(cdp_url) + """ + # Get credentials from env vars if not provided + service_url = service_url or os.getenv("PLAYWRIGHT_SERVICE_URL") + access_token = access_token or os.getenv("PLAYWRIGHT_SERVICE_ACCESS_TOKEN") + + if not service_url: + raise PlaywrightServiceError( + "PLAYWRIGHT_SERVICE_URL environment variable is not set.\n" + "Expected: wss://.api.playwright.microsoft.com/playwrightworkspaces//browsers" + ) + if not access_token: + raise PlaywrightServiceError( + "PLAYWRIGHT_SERVICE_ACCESS_TOKEN environment variable is not set." + ) + + # Parse URL to get region and workspace ID + region, workspace_id = _parse_url(service_url) + + # Build API URL + api_url = ( + f"https://{region}.api.playwright.microsoft.com" + f"/playwrightworkspaces/{workspace_id}/browsers" + f"?api-version={api_version}&os={os_name}" + ) + + # Make request + headers = { + "Authorization": f"Bearer {access_token}", + "Accept": "application/json", + } + + async with aiohttp.ClientSession() as session: + async with session.get(api_url, headers=headers) as response: + if response.status == 401: + raise PlaywrightServiceError("Authentication failed. Check your access token.") + if response.status == 403: + raise PlaywrightServiceError("Access forbidden. Check your permissions.") + if response.status != 200: + text = await response.text() + raise PlaywrightServiceError(f"Failed to get browser endpoint: HTTP {response.status}\n{text}") + + data = await response.json() + return data["endpoint"] diff --git a/samples/cdp-tests/requirements.txt b/samples/cdp-tests/requirements.txt new file mode 100644 index 0000000..cad986f --- /dev/null +++ b/samples/cdp-tests/requirements.txt @@ -0,0 +1,14 @@ +# Microsoft Playwright Service - Python Samples Dependencies + +# Core (required for all samples) +aiohttp>=3.8.0 +playwright>=1.40.0 +python-dotenv>=1.0.0 + +# For test_runner.py (testing scenario) +pytest>=7.0.0 +pytest-asyncio>=0.21.0 + +# For browser_use_remote.py (AI agent scenario) +pydantic>=2.0.0 +browser-use>=0.1.0 diff --git a/samples/cdp-tests/test_runner.py b/samples/cdp-tests/test_runner.py new file mode 100644 index 0000000..c36480e --- /dev/null +++ b/samples/cdp-tests/test_runner.py @@ -0,0 +1,180 @@ +""" +Playwright Testing Example - Microsoft Playwright Service + +Run Playwright tests on remote browsers via CDP (Chrome DevTools Protocol). + +---------------------------------------- +๐Ÿ“Œ Prerequisites +---------------------------------------- +1๏ธโƒฃ Python environment with the following packages installed: + pip install playwright aiohttp pytest pytest-asyncio python-dotenv + +2๏ธโƒฃ Playwright Remote Browser Setup + - Sign up for Microsoft Playwright Service. + - Obtain your service URL and access token. + - Copy .env.example to .env and fill in your values: + + PLAYWRIGHT_SERVICE_URL=wss://.api.playwright.microsoft.com/playwrightworkspaces//browsers + PLAYWRIGHT_SERVICE_ACCESS_TOKEN=your_access_token + +---------------------------------------- +๐Ÿ“Œ How to Use +---------------------------------------- +1๏ธโƒฃ Run as a standalone script: + python test_runner.py + +2๏ธโƒฃ Run with pytest: + pytest test_runner.py -v + +3๏ธโƒฃ Import and use in your own tests: + from test_runner import remote_page + + async with remote_page() as page: + await page.goto("https://example.com") + assert await page.title() == "Example Domain" +""" + +import asyncio +from contextlib import asynccontextmanager +from typing import AsyncGenerator +from playwright.async_api import async_playwright, Browser, Page +from dotenv import load_dotenv + +# Load environment variables from .env file +load_dotenv() + +# Import the shared module +from playwright_service_client import get_cdp_endpoint + + +# ============================================================================ +# Context Managers for Easy Test Setup +# ============================================================================ + +@asynccontextmanager +async def remote_browser() -> AsyncGenerator[Browser, None]: + """ + Context manager for quick access to a remote browser. + + Example: + async with remote_browser() as browser: + page = await browser.new_page() + await page.goto("https://example.com") + """ + cdp_url = await get_cdp_endpoint() + async with async_playwright() as p: + browser = await p.chromium.connect_over_cdp( + cdp_url, + headers={"User-Agent": "Chrome-DevTools-Protocol/1.3"} + ) + try: + yield browser + finally: + await browser.close() + + +@asynccontextmanager +async def remote_page() -> AsyncGenerator[Page, None]: + """ + Context manager for quick access to a remote page. + + Example: + async with remote_page() as page: + await page.goto("https://example.com") + assert await page.title() == "Example Domain" + """ + async with remote_browser() as browser: + context = await browser.new_context() + page = await context.new_page() + try: + yield page + finally: + await context.close() + + +# ============================================================================ +# Example Tests +# ============================================================================ + +async def test_example_domain(): + """Test that example.com loads correctly.""" + async with remote_page() as page: + await page.goto("https://example.com") + + title = await page.title() + assert title == "Example Domain", f"Expected 'Example Domain', got '{title}'" + + heading = await page.locator("h1").text_content() + assert heading == "Example Domain" + + print("โœ… test_example_domain passed!") + + +async def test_navigation(): + """Test basic navigation functionality.""" + async with remote_page() as page: + await page.goto("https://example.com") + initial_url = page.url + + # Click the "More information..." link + await page.click("a") + await page.wait_for_load_state("networkidle") + + assert page.url != initial_url, "URL should have changed after clicking link" + print("โœ… test_navigation passed!") + + +async def test_screenshot(): + """Test taking a screenshot.""" + async with remote_page() as page: + await page.goto("https://example.com") + + screenshot = await page.screenshot() + assert len(screenshot) > 0, "Screenshot should not be empty" + + print(f"โœ… test_screenshot passed! ({len(screenshot)} bytes)") + + +# ============================================================================ +# Test Runner +# ============================================================================ + +async def run_all_tests(): + """Run all example tests.""" + print("=" * 50) + print("๐Ÿงช Running Playwright Service Tests") + print("=" * 50) + + tests = [ + ("Example Domain", test_example_domain), + ("Navigation", test_navigation), + ("Screenshot", test_screenshot), + ] + + passed = failed = 0 + + for name, test_func in tests: + print(f"\n๐Ÿ“‹ {name}") + print("-" * 30) + try: + await test_func() + passed += 1 + except Exception as e: + print(f"โŒ Failed: {e}") + failed += 1 + + print(f"\n{'=' * 50}") + print(f"๐Ÿ“Š Results: {passed} passed, {failed} failed") + print("=" * 50) + + return failed == 0 + + +# ============================================================================ +# Main +# ============================================================================ + +if __name__ == "__main__": + print("๐Ÿงช Playwright Testing - Microsoft Playwright Service\n") + success = asyncio.run(run_all_tests()) + exit(0 if success else 1)