Amazon Bedrock AgentCore Browser demonstration - Shows secure, isolated browser environment for AI agents with screenshot capabilities and form filling

basic_browser.py

# basic_browser.py - Amazon Bedrock AgentCore Browser demonstration
"""
This example demonstrates how to use Amazon Bedrock AgentCore Browser to enable
AI agents to interact with websites in a secure, cloud-based environment.

AgentCore Browser provides:
- Secure, isolated browser environment
- Session isolation and ephemeral sessions
- Visual understanding through screenshots
- Human intervention capabilities
- Real-time observability and session replay
- Automatic scaling without infrastructure overhead
"""

import boto3
import json
import time
import asyncio
import base64
from typing import Dict, List, Any, Optional
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from contextlib import contextmanager

# Initialize the AgentCore app
app = BedrockAgentCoreApp()

# Initialize Browser client
browser_client = boto3.client('bedrock-agentcore-control', region_name='us-west-2')
browser_data_client = boto3.client('bedrock-agentcore', region_name='us-west-2')

class AgentCoreBrowserManager:
    """Manager for AgentCore Browser operations"""
    
    def __init__(self, region: str = 'us-west-2'):
        self.region = region
        self.control_client = boto3.client('bedrock-agentcore-control', region_name=region)
        self.data_client = boto3.client('bedrock-agentcore', region_name=region)
        self.browser_id = None
        self.browser_arn = None
        self.session_id = None
    
    def list_browsers(self) -> Dict[str, Any]:
        """List all existing browsers in the account"""
        try:
            response = self.control_client.list_browsers()
            
            browsers = response.get('browserSummaries', [])
            print(f"📋 Found {len(browsers)} existing browsers:")
            
            for browser in browsers:
                print(f"  - {browser['name']} (ID: {browser['browserId']}, Status: {browser['status']})")
            
            return response
            
        except Exception as e:
            print(f"❌ Failed to list browsers: {e}")
            return {}
    
    def find_browser_by_name(self, name: str) -> Dict[str, Any]:
        """Find a browser by name"""
        try:
            response = self.control_client.list_browsers()
            browsers = response.get('browserSummaries', [])
            
            for browser in browsers:
                if browser['name'] == name:
                    return browser
            
            return None
            
        except Exception as e:
            print(f"❌ Failed to find browser: {e}")
            return None
    
    def create_browser(self, name: str, description: str = None, execution_role_arn: str = None) -> Dict[str, Any]:
        """Create a new AgentCore Browser or use existing one"""
        try:
            # First check if browser already exists
            existing_browser = self.find_browser_by_name(name)
            
            if existing_browser:
                print(f"📌 Browser '{name}' already exists:")
                print(f"  Browser ID: {existing_browser['browserId']}")
                print(f"  Status: {existing_browser['status']}")
                
                # Use the existing browser
                self.browser_id = existing_browser['browserId']
                self.browser_arn = existing_browser['browserArn']
                
                return existing_browser
            
            # Create new browser if it doesn't exist
            create_params = {
                'name': name,
                'description': description or f"Browser tool for {name}",
                'networkConfiguration': {
                    'networkMode': 'PUBLIC'
                }
            }
            
            # Add execution role ARN if provided
            if execution_role_arn:
                create_params['executionRoleArn'] = execution_role_arn
            
            # Optional: Add recording configuration
            # create_params['recording'] = {
            #     'enabled': True,
            #     's3Location': {
            #         'bucket': 'your-recording-bucket',
            #         'prefix': 'browser-recordings/'
            #     }
            # }
            
            response = self.control_client.create_browser(**create_params)
            
            self.browser_id = response['browserId']
            self.browser_arn = response['browserArn']
            
            print(f"✅ Browser created successfully:")
            print(f"  Browser ID: {self.browser_id}")
            print(f"  Browser ARN: {self.browser_arn}")
            
            return response
            
        except Exception as e:
            print(f"❌ Failed to create browser: {e}")
            return None
    
    def create_or_get_browser(self, base_name: str, description: str = None, execution_role_arn: str = None) -> Dict[str, Any]:
        """Create a browser with unique name or get existing one"""
        try:
            # First try the base name
            result = self.create_browser(base_name, description, execution_role_arn)
            if result:
                return result
            
            # If base name exists, try with timestamp suffix
            import time
            timestamp = int(time.time())
            unique_name = f"{base_name}_{timestamp}"
            
            print(f"🔄 Trying unique name: {unique_name}")
            return self.create_browser(unique_name, description, execution_role_arn)
            
        except Exception as e:
            print(f"❌ Failed to create or get browser: {e}")
            return None
    
    def start_browser_session(self, session_name: str = "demo-session") -> str:
        """Start a new browser session"""
        try:
            response = self.data_client.start_browser_session(
                browserIdentifier=self.browser_id,
                name=session_name,
                sessionTimeoutSeconds=1800  # 30 minutes
            )
            
            self.session_id = response['sessionId']
            
            print(f"✅ Browser session started:")
            print(f"  Session ID: {self.session_id}")
            
            return self.session_id
            
        except Exception as e:
            print(f"❌ Failed to start browser session: {e}")
            return None
    
    def navigate_to_url(self, url: str) -> Dict[str, Any]:
        """Navigate to a URL using Playwright"""
        try:
            with self.playwright_session() as (browser, context, page):
                if not page:
                    return {'status': 'error', 'message': 'Failed to connect to browser'}
                
                print(f"🌐 Navigating to: {url}")
                response = page.goto(url, wait_until="domcontentloaded")
                
                # Wait a moment for page to load
                time.sleep(2)
                
                result = {
                    'status': 'success',
                    'url': page.url,
                    'title': page.title(),
                    'response_status': response.status if response else None,
                    'message': f'Successfully navigated to {url}'
                }
                
                print(f"✅ Navigation completed: {page.title()}")
                return result
                
        except Exception as e:
            print(f"❌ Failed to navigate to URL: {e}")
            return {'status': 'error', 'message': str(e)}
    
    def take_screenshot(self, filename: str = None) -> Dict[str, Any]:
        """Take a screenshot using Playwright CDP"""
        try:
            with self.playwright_session() as (browser, context, page):
                if not page:
                    return {'status': 'error', 'message': 'Failed to connect to browser'}
                
                # Generate filename if not provided
                if not filename:
                    timestamp = int(time.time())
                    filename = f"screenshot_{timestamp}.jpeg"
                
                print(f"📸 Taking screenshot...")
                
                # Use CDP for high-quality screenshot
                cdp_client = context.new_cdp_session(page)
                screenshot_data = cdp_client.send("Page.captureScreenshot", {
                    "format": "jpeg",
                    "quality": 80,
                    "captureBeyondViewport": True
                })
                
                # Decode and save screenshot
                image_data = base64.b64decode(screenshot_data['data'])
                with open(filename, "wb") as f:
                    f.write(image_data)
                
                result = {
                    'status': 'success',
                    'filename': filename,
                    'size': len(image_data),
                    'url': page.url,
                    'title': page.title(),
                    'message': f'Screenshot saved as {filename}'
                }
                
                print(f"✅ Screenshot saved: {filename} ({len(image_data)} bytes)")
                return result
                
        except Exception as e:
            print(f"❌ Failed to take screenshot: {e}")
            return {'status': 'error', 'message': str(e)}
    
    def click_element(self, selector: str) -> Dict[str, Any]:
        """Click an element using Playwright"""
        try:
            with self.playwright_session() as (browser, context, page):
                if not page:
                    return {'status': 'error', 'message': 'Failed to connect to browser'}
                
                print(f"🖱️ Clicking element: {selector}")
                
                # Wait for element to be visible and clickable
                try:
                    page.wait_for_selector(selector, timeout=5000)
                    
                    # Check if element exists
                    element = page.query_selector(selector)
                    if not element:
                        return {
                            'status': 'error',
                            'message': f'Element not found: {selector}'
                        }
                    
                    # Click the element using page method (proper Playwright way)
                    page.click(selector)
                    
                    # Wait a moment for any potential navigation
                    time.sleep(1)
                    
                    result = {
                        'status': 'success',
                        'selector': selector,
                        'current_url': page.url,
                        'current_title': page.title(),
                        'message': f'Successfully clicked element: {selector}'
                    }
                    
                    print(f"✅ Element clicked successfully")
                    return result
                    
                except Exception as e:
                    return {
                        'status': 'error',
                        'message': f'Failed to click element {selector}: {str(e)}'
                    }
                
        except Exception as e:
            print(f"❌ Failed to click element: {e}")
            return {'status': 'error', 'message': str(e)}
    
    def fill_form_field(self, selector: str, value: str) -> Dict[str, Any]:
        """Fill a form field using Playwright"""
        try:
            with self.playwright_session() as (browser, context, page):
                if not page:
                    return {'status': 'error', 'message': 'Failed to connect to browser'}
                
                print(f"📝 Filling form field '{selector}' with value: {value}")
                
                try:
                    # Wait for element to be visible
                    page.wait_for_selector(selector, timeout=5000)
                    
                    # Check if element exists
                    element = page.query_selector(selector)
                    if not element:
                        return {
                            'status': 'error',
                            'message': f'Form field not found: {selector}'
                        }
                    
                    # Use page methods to interact with the element (proper Playwright way)
                    page.fill(selector, '')  # Clear the field
                    page.fill(selector, value)  # Fill with new value
                    
                    # Verify the value was set
                    try:
                        current_value = page.input_value(selector)
                    except:
                        try:
                            current_value = page.inner_text(selector)
                        except:
                            current_value = "verification_failed"
                    
                    result = {
                        'status': 'success',
                        'selector': selector,
                        'value': value,
                        'current_value': current_value,
                        'message': f'Successfully filled form field: {selector}'
                    }
                    
                    print(f"✅ Form field filled successfully")
                    return result
                    
                except Exception as e:
                    return {
                        'status': 'error',
                        'message': f'Failed to fill form field {selector}: {str(e)}'
                    }
                
        except Exception as e:
            print(f"❌ Failed to fill form field: {e}")
            return {'status': 'error', 'message': str(e)}
    
    def extract_text(self, selector: str = None) -> Dict[str, Any]:
        """Extract text from page or specific elements using Playwright"""
        try:
            with self.playwright_session() as (browser, context, page):
                if not page:
                    return {'status': 'error', 'message': 'Failed to connect to browser'}
                
                if selector:
                    print(f"📝 Extracting text from selector: {selector}")
                    
                    # Check if elements exist
                    elements = page.query_selector_all(selector)
                    if not elements:
                        return {
                            'status': 'warning',
                            'message': f'No elements found for selector: {selector}',
                            'text': '',
                            'count': 0
                        }
                    
                    # Extract text from all matching elements
                    texts = []
                    for element in elements:
                        text = element.inner_text().strip()
                        if text:
                            texts.append(text)
                    
                    result = {
                        'status': 'success',
                        'selector': selector,
                        'text': '\n'.join(texts),
                        'texts': texts,
                        'count': len(texts),
                        'message': f'Extracted text from {len(texts)} elements'
                    }
                    
                else:
                    print(f"📝 Extracting all page text...")
                    
                    # Extract full page text - handle cases where body might not be accessible
                    try:
                        body_text = page.inner_text('body') or ""
                    except:
                        body_text = ""
                    
                    # Also get headings separately
                    headings = []
                    try:
                        for h_tag in ['h1', 'h2', 'h3', 'h4', 'h5', 'h6']:
                            h_elements = page.query_selector_all(h_tag)
                            for h in h_elements:
                                try:
                                    h_text = h.inner_text().strip()
                                    if h_text:  # Only add non-empty headings
                                        headings.append({
                                            'tag': h_tag,
                                            'text': h_text
                                        })
                                except:
                                    continue
                    except:
                        pass
                    
                    result = {
                        'status': 'success',
                        'text': body_text,
                        'text_length': len(body_text),
                        'headings': headings,
                        'headings_count': len(headings),
                        'url': page.url,
                        'title': page.title(),
                        'message': f'Extracted {len(body_text)} characters of text'
                    }
                
                print(f"✅ Text extraction completed")
                return result
                
        except Exception as e:
            print(f"❌ Failed to extract text: {e}")
            return {'status': 'error', 'message': str(e)}
    
    def get_page_info(self) -> Dict[str, Any]:
        """Get comprehensive page information using Playwright"""
        try:
            with self.playwright_session() as (browser, context, page):
                if not page:
                    return {'status': 'error', 'message': 'Failed to connect to browser'}
                
                print(f"📄 Getting page information...")
                
                # Get basic page info
                page_info = {
                    'status': 'success',
                    'url': page.url,
                    'title': page.title(),
                    'viewport': page.viewport_size,
                }
                
                # Get additional page metadata
                try:
                    # Get meta description
                    try:
                        meta_desc_element = page.query_selector('meta[name="description"]')
                        meta_desc = meta_desc_element.get_attribute('content') if meta_desc_element else None
                        if meta_desc:
                            page_info['meta_description'] = meta_desc
                    except:
                        pass
                    
                    # Get page content info - handle cases where elements might not be accessible
                    try:
                        body_text = page.inner_text('body') or ""
                        page_info['body_text_length'] = len(body_text)
                    except:
                        page_info['body_text_length'] = 0
                        
                    try:
                        page_info['links_count'] = len(page.query_selector_all('a'))
                    except:
                        page_info['links_count'] = 0
                        
                    try:
                        page_info['images_count'] = len(page.query_selector_all('img'))
                    except:
                        page_info['images_count'] = 0
                        
                    try:
                        page_info['forms_count'] = len(page.query_selector_all('form'))
                    except:
                        page_info['forms_count'] = 0
                    
                    # Get page load state
                    try:
                        page_info['ready_state'] = page.evaluate('document.readyState')
                    except:
                        page_info['ready_state'] = 'unknown'
                    
                except Exception as e:
                    page_info['metadata_error'] = str(e)
                
                print(f"✅ Page info retrieved: {page_info['title']}")
                return page_info
                
        except Exception as e:
            print(f"❌ Failed to get page info: {e}")
            return {'status': 'error', 'message': str(e)}
    
    def get_playwright_connection_info(self) -> Dict[str, Any]:
        """Get WebSocket connection info for Playwright integration"""
        try:
            session_info = self.data_client.get_browser_session(
                browserIdentifier=self.browser_id,
                sessionId=self.session_id
            )
            
            streams = session_info.get('streams', {})
            automation_endpoint = streams.get('automationStream', {}).get('streamEndpoint')
            live_view_endpoint = streams.get('liveViewStream', {}).get('streamEndpoint')
            
            return {
                'automation_endpoint': automation_endpoint,
                'live_view_endpoint': live_view_endpoint,
                'session_info': session_info
            }
            
        except Exception as e:
            print(f"❌ Failed to get connection info: {e}")
            return {}
    
    def _get_automation_headers(self):
        """Get authentication headers for WebSocket connection"""
        try:
            import boto3
            from botocore.auth import SigV4Auth
            from botocore.awsrequest import AWSRequest
            from urllib.parse import urlparse
            import datetime
            
            # Create a credentials provider
            session = boto3.Session()
            credentials = session.get_credentials()
            
            # Get the automation endpoint URL
            session_info = self.data_client.get_browser_session(
                browserIdentifier=self.browser_id,
                sessionId=self.session_id
            )
            
            automation_endpoint = session_info.get('streams', {}).get('automationStream', {}).get('streamEndpoint')
            if not automation_endpoint:
                return None, None
                
            # Parse the URL to get host and path
            parsed_url = urlparse(automation_endpoint)
            
            # Create AWS request for signing
            request = AWSRequest(
                method='GET',
                url=automation_endpoint,
                headers={
                    'host': parsed_url.netloc,
                    'x-amz-date': datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%dT%H%M%SZ')
                }
            )
            
            # Sign the request
            SigV4Auth(credentials, 'bedrock-agentcore', self.region).add_auth(request)
            
            # Convert headers for WebSocket
            ws_headers = {}
            for key, value in request.headers.items():
                if key.lower() in ['authorization', 'x-amz-date', 'x-amz-security-token']:
                    ws_headers[key] = value
                    
            return automation_endpoint, ws_headers
            
        except Exception as e:
            print(f"❌ Failed to generate authentication headers: {e}")
            return None, None

    @contextmanager
    def playwright_session(self):
        """Context manager for Playwright browser session with proper authentication"""
        try:
            from playwright.sync_api import sync_playwright, Playwright, BrowserType
        except ImportError:
            print("❌ Playwright not installed. Install with: pip install playwright")
            yield None, None, None
            return
        
        try:
            print("🔗 Getting WebSocket authentication...")
            
            # Get authentication headers for our existing browser session
            ws_url, headers = self._get_automation_headers()
            
            if not ws_url or not headers:
                print("❌ Failed to get authentication credentials")
                yield None, None, None
                return
                
            print(f"🔗 Connecting to existing session: {self.session_id}")
            
            # Connect to browser via Playwright with authentication headers
            with sync_playwright() as p:
                chromium = p.chromium
                browser = chromium.connect_over_cdp(ws_url, headers=headers)
                
                # Use existing context or create new one
                context = browser.contexts[0] if browser.contexts else browser.new_context()
                page = context.pages[0] if context.pages else context.new_page()
                
                print("✅ Connected to existing browser session")
                yield browser, context, page
                
        except Exception as e:
            print(f"❌ Failed to create Playwright session: {e}")
            print("💡 Falling back to bedrock_agentcore.tools.browser_client...")
            
            # Fallback to the official browser_session client
            try:
                from bedrock_agentcore.tools.browser_client import browser_session
                
                print("🔗 Creating new authenticated browser session...")
                with browser_session(self.region) as client:
                    ws_url, headers = client.generate_ws_headers()
                    
                    with sync_playwright() as p:
                        chromium = p.chromium
                        browser = chromium.connect_over_cdp(ws_url, headers=headers)
                        context = browser.contexts[0] if browser.contexts else browser.new_context()
                        page = context.pages[0] if context.pages else context.new_page()
                        
                        print("✅ Fallback connection established")
                        yield browser, context, page
                        
            except ImportError:
                print("❌ bedrock_agentcore not available. Install with: pip install bedrock-agentcore")
                yield None, None, None
            except Exception as e2:
                print(f"❌ Fallback also failed: {e2}")
                yield None, None, None
    
    def _process_browser_response(self, response: Dict[str, Any]) -> Dict[str, Any]:
        """Process browser response stream"""
        results = []
        
        for event in response.get('stream', []):
            if 'result' in event:
                results.append(event['result'])
        
        return {
            'results': results,
            'last_result': results[-1] if results else None
        }
    
    def stop_browser_session(self) -> bool:
        """Stop the current browser session"""
        try:
            if self.session_id:
                self.data_client.stop_browser_session(
                    browserIdentifier=self.browser_id,
                    sessionId=self.session_id
                )
                
                print(f"✅ Browser session stopped: {self.session_id}")
                self.session_id = None
                return True
            
        except Exception as e:
            print(f"❌ Failed to stop browser session: {e}")
            return False
    
    def get_browser_status(self) -> Dict[str, Any]:
        """Get browser status"""
        try:
            response = self.control_client.get_browser(
                browserId=self.browser_id
            )
            
            status = response.get('status', 'UNKNOWN')
            print(f"🔍 Browser status: {status}")
            
            return response
            
        except Exception as e:
            print(f"❌ Failed to get browser status: {e}")
            return {}

# Browser-enabled agent
@app.entrypoint
async def browser_agent(request):
    """Agent that uses Browser to interact with websites"""
    user_input = request.get("prompt", "")
    user_id = request.get("user_id", "demo_user")
    
    print(f"🌐 Browser Agent processing: {user_input}")
    
    # Simulate web interaction based on user input
    if "navigate" in user_input.lower() or "visit" in user_input.lower():
        url = "https://example.com"  # Extract URL from input in production
        response = f"Using AgentCore Browser to navigate to {url}. Browser provides secure, isolated web interaction with screenshot capabilities."
    
    elif "screenshot" in user_input.lower():
        response = "Taking screenshot using AgentCore Browser. Screenshots help agents understand web content visually, similar to human browsing."
    
    elif "form" in user_input.lower() or "fill" in user_input.lower():
        response = "Using AgentCore Browser to fill forms. Browser can interact with web elements like buttons, forms, and links in a secure environment."
    
    elif "search" in user_input.lower():
        response = "Using AgentCore Browser to perform web search. Browser enables agents to search websites and extract information dynamically."
    
    elif "extract" in user_input.lower():
        response = "Using AgentCore Browser to extract text from web pages. Browser can parse both static and dynamic content."
    
    elif "capabilities" in user_input.lower() or "features" in user_input.lower():
        response = """AgentCore Browser capabilities:
• Secure, isolated browser environment with session isolation
• Screenshot capture for visual understanding
• Form filling and button clicking
• Text extraction from web pages
• Real-time observability and session replay
• Automatic scaling without infrastructure management
• Support for complex web applications and dynamic content"""
    
    else:
        response = """Hello! I'm a Browser-enabled agent using AgentCore Browser.
I can help you with:
• Navigate to websites: "navigate to example.com"
• Take screenshots: "take a screenshot"
• Fill forms: "fill out the form"
• Extract text: "extract text from the page"
• Web search: "search for information"
• Show capabilities: "what are your browser capabilities?"

AgentCore Browser provides secure, cloud-based web interaction for AI agents."""
    
    return {
        "response": response,
        "browser_enabled": True,
        "capabilities": ["navigate", "screenshot", "form_fill", "text_extract", "search"],
        "status": "success"
    }

@app.ping
def health_check():
    """Health check for Browser agent"""
    return {
        "status": "Healthy",
        "browser_enabled": True,
        "time_of_last_update": int(time.time())
    }

def demo_browser_setup():
    """Demonstrate Browser setup and configuration"""
    print("🚀 AgentCore Browser Setup Demo")
    print("=" * 50)
    
    # Initialize Browser manager
    browser_manager = AgentCoreBrowserManager()
    
    # Create browser
    print("\n1. Creating Browser...")
    browser_result = browser_manager.create_browser(
        name="DemoBrowser",
        description="Demo browser for AgentCore tutorial"
    )
    
    if not browser_result:
        print("❌ Browser creation failed, exiting demo")
        return None
    
    # Start browser session
    print("\n2. Starting Browser session...")
    session_id = browser_manager.start_browser_session("demo-session")
    
    if not session_id:
        print("❌ Failed to start browser session")
        return None
    
    # Check browser status
    print("\n3. Checking Browser status...")
    status = browser_manager.get_browser_status()
    
    print("\n✅ Browser setup complete!")
    print(f"Browser ID: {browser_manager.browser_id}")
    print(f"Session ID: {browser_manager.session_id}")
    
    return browser_manager

def demo_browser_usage():
    """Demonstrate Browser usage with web interactions"""
    print("\n🌐 Testing Browser Web Interactions...")
    print("=" * 50)
    
    # Initialize Browser manager
    browser_manager = AgentCoreBrowserManager()
    
    # Create browser and start session
    browser_result = browser_manager.create_browser("TestBrowser")
    if not browser_result:
        return
    
    session_id = browser_manager.start_browser_session("test-session")
    if not session_id:
        return
    
    try:
        # Get WebSocket connection info for Playwright
        print("\n1. Getting WebSocket connection info...")
        connection_info = browser_manager.get_playwright_connection_info()
        
        if connection_info.get('automation_endpoint'):
            print(f"✅ Automation endpoint: {connection_info['automation_endpoint']}")
            print(f"✅ Live view endpoint: {connection_info['live_view_endpoint']}")
        
        # Demo actual web interactions using Playwright in a single session
        print("\n2. Complete web interaction workflow...")
        
        # Use a single Playwright session for all operations
        with browser_manager.playwright_session() as (browser, context, page):
            if not page:
                print("❌ Failed to establish browser connection")
                return
                
            try:
                # Step 1: Navigate to test website
                print("   🌐 Navigating to test website...")
                response = page.goto("https://httpbin.org", wait_until="domcontentloaded")
                time.sleep(2)  # Wait for page to fully load
                print(f"   📍 Navigation completed: {page.title()}")
                print(f"   🔗 Current URL: {page.url}")
                
                # Step 2: Take screenshot
                print("   📸 Taking screenshot...")
                try:
                    cdp_client = context.new_cdp_session(page)
                    screenshot_data = cdp_client.send("Page.captureScreenshot", {
                        "format": "jpeg",
                        "quality": 80,
                        "captureBeyondViewport": True
                    })
                    
                    image_data = base64.b64decode(screenshot_data['data'])
                    timestamp = int(time.time())
                    filename = f"httpbin_screenshot_{timestamp}.jpeg"
                    
                    with open(filename, "wb") as f:
                        f.write(image_data)
                    
                    print(f"   ✅ Screenshot saved: {filename} ({len(image_data)} bytes)")
                except Exception as e:
                    print(f"   ❌ Screenshot failed: {e}")
                
                # Step 3: Get page information
                print("   📄 Getting page information...")
                try:
                    page_title = page.title()
                    page_url = page.url
                    body_text = page.inner_text('body') or ""
                    links_count = len(page.query_selector_all('a'))
                    images_count = len(page.query_selector_all('img'))
                    forms_count = len(page.query_selector_all('form'))
                    
                    print(f"   📄 Title: {page_title}")
                    print(f"   🌐 URL: {page_url}")
                    print(f"   📝 Body text length: {len(body_text)} characters")
                    print(f"   🔗 Links found: {links_count}")
                    print(f"   🖼️ Images found: {images_count}")
                    print(f"   📋 Forms found: {forms_count}")
                except Exception as e:
                    print(f"   ❌ Page info failed: {e}")
                
                # Step 4: Extract text
                print("   📝 Extracting page text...")
                try:
                    body_text = page.inner_text('body') or ""
                    
                    # Get headings
                    headings = []
                    for h_tag in ['h1', 'h2', 'h3']:
                        elements = page.query_selector_all(h_tag)
                        for elem in elements:
                            text = elem.inner_text().strip()
                            if text:
                                headings.append({'tag': h_tag, 'text': text})
                    
                    print(f"   📝 Text extracted: {len(body_text)} characters")
                    print(f"   🔤 Headings found: {len(headings)}")
                    
                    # Show first heading
                    if headings:
                        print(f"   🎯 First heading: {headings[0]['text']}")
                    
                    # Show text preview
                    text_preview = body_text[:200] + "..." if len(body_text) > 200 else body_text
                    print(f"   📖 Text preview: {text_preview}")
                    
                except Exception as e:
                    print(f"   ❌ Text extraction failed: {e}")
                
                # Step 5: Test element interaction
                print("   🖱️ Testing element interaction...")
                try:
                    # Look for links to click
                    links = page.query_selector_all('a[href]')
                    if links:
                        first_link = links[0]
                        link_text = first_link.inner_text().strip()
                        if link_text:
                            print(f"   🔗 Found clickable link: {link_text}")
                        else:
                            print("   🔗 Found links but no visible text")
                    else:
                        print("   🔗 No clickable links found")
                        
                except Exception as e:
                    print(f"   ❌ Element interaction test failed: {e}")
                    
            except Exception as e:
                print(f"   ❌ Browser workflow failed: {e}")
                
            print("   ✅ Single-session workflow completed!")
        
        print("\n✅ All Playwright browser interactions completed successfully!")
        print("\n💡 Features demonstrated:")
        print("   ✅ Real navigation with Playwright")
        print("   ✅ High-quality screenshot capture via CDP")
        print("   ✅ Comprehensive page information extraction") 
        print("   ✅ Text extraction with CSS selectors")
        print("   ✅ Element clicking and form filling capabilities")
        
    except Exception as e:
        print(f"❌ Browser interaction failed: {e}")
    
    finally:
        # Clean up
        print("\n6. Stopping browser session...")
        browser_manager.stop_browser_session()

def demo_browser_agent():
    """Demonstrate Browser-enabled agent"""
    print("\n🧪 Testing Browser-enabled Agent...")
    print("=" * 50)
    
    # Test requests
    test_requests = [
        {"prompt": "Hello, what can you do?", "user_id": "test_user"},
        {"prompt": "Navigate to example.com", "user_id": "test_user"},
        {"prompt": "Take a screenshot of the page", "user_id": "test_user"},
        {"prompt": "Fill out the form", "user_id": "test_user"},
        {"prompt": "Extract text from the page", "user_id": "test_user"},
        {"prompt": "What are your browser capabilities?", "user_id": "test_user"}
    ]
    
    for i, request in enumerate(test_requests, 1):
        print(f"\nTest {i}: {request['prompt']}")
        
        # Run the agent
        result = asyncio.run(browser_agent(request))
        print(f"Response: {result['response']}")
        
        time.sleep(1)  # Brief pause between tests
    
    print("\n✅ Browser agent demo completed!")

def demo_complete_browser_workflow():
    """Demonstrate a complete browser workflow with all features in a single session"""
    print("\n🎯 Complete Browser Workflow Demo")
    print("=" * 50)
    
    browser_manager = AgentCoreBrowserManager()
    
    try:
        # Setup
        print("\n1. Setting up browser...")
        browser_result = browser_manager.create_browser("WorkflowDemo")
        if not browser_result:
            return
            
        session_id = browser_manager.start_browser_session("workflow-session")
        if not session_id:
            return
            
        # Complete workflow in single session
        print("\n2. Complete web automation workflow...")
        
        # Use a single Playwright session for all operations to maintain state
        try:
            from bedrock_agentcore.tools.browser_client import browser_session
            print("🔗 Creating single persistent browser session...")
            
            with browser_session(browser_manager.region) as client:
                ws_url, headers = client.generate_ws_headers()
                
                from playwright.sync_api import sync_playwright
                with sync_playwright() as p:
                    browser = p.chromium.connect_over_cdp(ws_url, headers=headers)
                    context = browser.contexts[0] if browser.contexts else browser.new_context()
                    page = context.pages[0] if context.pages else context.new_page()
                    
                    try:
                        # Step 1: Navigate to forms page
                        print("   🌐 Navigating to forms page...")
                        response = page.goto("https://httpbin.org/forms/post", wait_until="domcontentloaded")
                        time.sleep(2)
                        
                        page_title = page.title() or "Forms Page"
                        print(f"   ✅ Navigation completed: {page_title}")
                        print(f"   🔗 Current URL: {page.url}")
                        
                        # Step 2: Take initial screenshot
                        print("   📸 Taking initial screenshot...")
                        cdp_client = context.new_cdp_session(page)
                        screenshot_data = cdp_client.send("Page.captureScreenshot", {
                            "format": "jpeg",
                            "quality": 80,
                            "captureBeyondViewport": True
                        })
                        
                        image_data = base64.b64decode(screenshot_data['data'])
                        with open("form_initial.jpg", "wb") as f:
                            f.write(image_data)
                        print(f"   ✅ Initial screenshot saved: form_initial.jpg ({len(image_data)} bytes)")
                        
                        # Step 3: Extract form information
                        print("   📝 Extracting form information...")
                        form_elements = page.query_selector_all('form')
                        if form_elements:
                            form_text = page.inner_text('form')
                            print(f"   ✅ Form found with {len(form_text)} characters of text")
                            
                            # Check what form fields are actually available
                            inputs = page.query_selector_all('form input[type="text"], form input[name]')
                            print(f"   🔍 Found {len(inputs)} input fields:")
                            
                            field_names = []
                            for i, input_elem in enumerate(inputs):
                                try:
                                    name_attr = input_elem.get_attribute('name')
                                    type_attr = input_elem.get_attribute('type') or 'text'
                                    if name_attr:
                                        field_names.append(name_attr)
                                        print(f"     - Field {i+1}: name='{name_attr}', type='{type_attr}'")
                                except:
                                    pass
                            
                            # Step 4: Fill form fields with actual field names
                            print("   📝 Filling form fields...")
                            filled_successfully = []
                            
                            if field_names:
                                # Try to fill the first two fields we find
                                test_values = ["John Doe", "555-1234", "john@example.com"]
                                
                                for i, field_name in enumerate(field_names[:2]):  # Only fill first 2 fields
                                    try:
                                        selector = f'input[name="{field_name}"]'
                                        value = test_values[i] if i < len(test_values) else f"test_value_{i+1}"
                                        
                                        print(f"     🖊️ Filling field '{field_name}' with '{value}'...")
                                        page.fill(selector, value)
                                        
                                        # Verify it was filled
                                        current_value = page.input_value(selector)
                                        if current_value == value:
                                            print(f"     ✅ Successfully filled '{field_name}' = '{current_value}'")
                                            filled_successfully.append(field_name)
                                        else:
                                            print(f"     ⚠️ Field '{field_name}' has unexpected value: '{current_value}'")
                                            
                                    except Exception as e:
                                        print(f"     ❌ Failed to fill field '{field_name}': {str(e)}")
                                
                                # Step 5: Take screenshot if any fields were filled
                                if filled_successfully:
                                    print(f"   📸 Taking screenshot after filling {len(filled_successfully)} fields...")
                                    screenshot_data = cdp_client.send("Page.captureScreenshot", {
                                        "format": "jpeg",
                                        "quality": 80,
                                        "captureBeyondViewport": True
                                    })
                                    
                                    image_data = base64.b64decode(screenshot_data['data'])
                                    with open("form_filled.jpg", "wb") as f:
                                        f.write(image_data)
                                    print(f"   ✅ Form filled screenshot saved: form_filled.jpg ({len(image_data)} bytes)")
                                else:
                                    print("   ❌ No fields were successfully filled, skipping filled form screenshot")
                            else:
                                print("   ❌ No form fields with 'name' attributes found")
                        else:
                            print("   ❌ No forms found on the page")
                        
                        # Step 6: Get final page information
                        print("   📊 Getting final page information...")
                        forms_count = len(page.query_selector_all('form'))
                        inputs_count = len(page.query_selector_all('input'))
                        links_count = len(page.query_selector_all('a'))
                        
                        print(f"   📄 Page title: {page.title()}")
                        print(f"   🌐 Current URL: {page.url}")
                        print(f"   📋 Forms found: {forms_count}")
                        print(f"   ⌨️ Input fields: {inputs_count}")
                        print(f"   🔗 Links: {links_count}")
                        
                        print("\n✅ Single-session workflow completed successfully!")
                        
                    except Exception as e:
                        print(f"   ❌ Workflow error: {e}")
                        import traceback
                        traceback.print_exc()
                        
        except ImportError:
            print("❌ bedrock_agentcore not available for single session. Using individual method calls...")
            # Fallback to individual method calls (less reliable due to session issues)
            nav_result = browser_manager.navigate_to_url("https://httpbin.org/forms/post")
            if nav_result.get('status') == 'success':
                print(f"✅ Navigation fallback successful")
        
    except Exception as e:
        print(f"❌ Workflow demo failed: {e}")
        import traceback
        traceback.print_exc()
    finally:
        print(f"\n🧹 Cleaning up...")
        browser_manager.stop_browser_session()

def demo_web_scraping():
    """Demonstrate web scraping with AgentCore Browser"""
    print("🔍 Web Scraping Demo with AgentCore Browser")
    print("=" * 50)
    
    browser_manager = AgentCoreBrowserManager()
    
    try:
        # Setup
        print("\n1. Setting up browser for web scraping...")
        browser_result = browser_manager.create_browser("WebScraper")
        if not browser_result:
            return
            
        session_id = browser_manager.start_browser_session("scraping-session")
        if not session_id:
            return
        
        # Define websites to scrape
        websites = [
            "https://example.com",
            "https://httpbin.org"
        ]
        
        scraped_data = []
        
        print(f"\n2. Scraping {len(websites)} websites...")
        
        for i, url in enumerate(websites, 1):
            print(f"\n   🌐 [{i}/{len(websites)}] Scraping: {url}")
            
            try:
                # Use single session for all operations
                with browser_manager.playwright_session() as (browser, context, page):
                    if not page:
                        print(f"   ❌ Failed to connect to browser for {url}")
                        continue
                    
                    # Navigate to the website
                    print(f"   📍 Navigating to {url}...")
                    response = page.goto(url, wait_until="domcontentloaded")
                    time.sleep(2)
                    
                    # Get page info
                    page_title = page.title()
                    page_url = page.url
                    
                    # Take screenshot
                    print(f"   📸 Taking screenshot...")
                    try:
                        cdp_client = context.new_cdp_session(page)
                        screenshot_data = cdp_client.send("Page.captureScreenshot", {
                            "format": "jpeg",
                            "quality": 80,
                            "captureBeyondViewport": True
                        })
                        
                        image_data = base64.b64decode(screenshot_data['data'])
                        timestamp = int(time.time())
                        filename = f"scrape_{i}_{timestamp}.jpeg"
                        
                        with open(filename, "wb") as f:
                            f.write(image_data)
                        print(f"   ✅ Screenshot saved: {filename}")
                    except Exception as e:
                        print(f"   ⚠️ Screenshot failed: {e}")
                        filename = None
                    
                    # Extract text content
                    print(f"   📝 Extracting text content...")
                    try:
                        body_text = page.inner_text('body') or ""
                        text_length = len(body_text)
                    except:
                        body_text = ""
                        text_length = 0
                    
                    # Extract headlines
                    print(f"   🔤 Extracting headlines...")
                    headlines = []
                    try:
                        for h_tag in ['h1', 'h2', 'h3']:
                            elements = page.query_selector_all(h_tag)
                            for elem in elements:
                                text = elem.inner_text().strip()
                                if text:
                                    headlines.append({'tag': h_tag, 'text': text})
                    except Exception as e:
                        print(f"   ⚠️ Headlines extraction failed: {e}")
                    
                    # Extract links
                    print(f"   🔗 Extracting links...")
                    links = []
                    try:
                        link_elements = page.query_selector_all('a[href]')
                        for link in link_elements[:10]:  # Limit to first 10 links
                            href = link.get_attribute('href')
                            text = link.inner_text().strip()
                            if href and text:
                                links.append({'href': href, 'text': text})
                    except Exception as e:
                        print(f"   ⚠️ Links extraction failed: {e}")
                    
                    # Get page metadata
                    try:
                        links_count = len(page.query_selector_all('a'))
                        images_count = len(page.query_selector_all('img'))
                        forms_count = len(page.query_selector_all('form'))
                    except:
                        links_count = images_count = forms_count = 0
                    
                    # Store scraped data
                    result = {
                        "url": url,
                        "title": page_title,
                        "text_length": text_length,
                        "headlines": headlines,
                        "links": links,
                        "links_count": links_count,
                        "images_count": images_count,
                        "forms_count": forms_count,
                        "screenshot": filename,
                        "status": "success"
                    }
                    
                    scraped_data.append(result)
                    
                    print(f"   ✅ Scraping completed for {url}")
                    print(f"      📄 Title: {page_title}")
                    print(f"      📝 Text length: {text_length} characters")
                    print(f"      🔤 Headlines found: {len(headlines)}")
                    print(f"      🔗 Links found: {len(links)}")
                    print(f"      📊 Page elements: {links_count} links, {images_count} images, {forms_count} forms")
                    
            except Exception as e:
                print(f"   ❌ Failed to scrape {url}: {e}")
                scraped_data.append({
                    "url": url,
                    "error": str(e),
                    "status": "error"
                })
        
        # Summary
        print(f"\n3. Scraping Results Summary:")
        print(f"   📊 Total websites: {len(websites)}")
        successful = len([d for d in scraped_data if d.get('status') == 'success'])
        failed = len(scraped_data) - successful
        print(f"   ✅ Successfully scraped: {successful}")
        print(f"   ❌ Failed: {failed}")
        
        # Show details for successful scrapes
        for data in scraped_data:
            if data.get('status') == 'success':
                print(f"\n   📋 {data['url']}:")
                print(f"      Title: {data['title']}")
                print(f"      Text: {data['text_length']} chars")
                if data['headlines']:
                    print(f"      First headline: {data['headlines'][0]['text'][:100]}...")
                if data['links']:
                    print(f"      First link: {data['links'][0]['text'][:50]}...")
        
        print(f"\n✅ Web scraping demo completed!")
        
    except Exception as e:
        print(f"❌ Web scraping demo failed: {e}")
        import traceback
        traceback.print_exc()
    
    finally:
        # Clean up
        print(f"\n🧹 Cleaning up...")
        browser_manager.stop_browser_session()

if __name__ == "__main__":
    import sys
    
    if len(sys.argv) > 1:
        command = sys.argv[1]
        
        if command == "setup":
            # Demo Browser setup
            browser_manager = demo_browser_setup()
        elif command == "test":
            # Test Browser usage
            demo_browser_usage()
        elif command == "agent":
            # Test Browser-enabled agent
            demo_browser_agent()
        elif command == "workflow":
            # Demo complete browser workflow
            demo_complete_browser_workflow()
        elif command == "scraping":
            # Demo web scraping
            demo_web_scraping()
        else:
            print("Usage: python basic_browser.py [setup|test|agent|workflow|scraping]")
            print("Commands:")
            print("  setup    - Demo browser setup and configuration")
            print("  test     - Test browser usage with Playwright integration")
            print("  agent    - Test browser-enabled agent responses")
            print("  workflow - Demo complete browser automation workflow")
            print("  scraping - Demo web scraping with actual execution")
    else:
        print("Usage: python basic_browser.py [setup|test|agent|workflow|scraping]")
        print("Commands:")
        print("  setup    - Demo browser setup and configuration")
        print("  test     - Test browser usage with Playwright integration")
        print("  agent    - Test browser-enabled agent responses")
        print("  workflow - Demo complete browser automation workflow")
        print("  scraping - Demo web scraping with actual execution")