Stagehand

pip install stagehand anchorbrowser

from __future__ import annotations

import asyncio
import os
import sys
from typing import Any, Optional

from anchorbrowser import AsyncAnchorbrowser

from stagehand import AsyncStagehand


async def main() -> None:
    # 1. Check for required environment variables
    anchor_api_key = os.environ.get("ANCHORBROWSER_API_KEY")
    if not anchor_api_key:
        sys.exit(
            "❌ ANCHORBROWSER_API_KEY environment variable is required. "
            "Get your API key from https://anchorbrowser.io"
        )

    model_api_key = os.environ.get("MODEL_API_KEY") or os.environ.get("OPENAI_API_KEY") or os.environ.get("GOOGLE_API_KEY")
    if not model_api_key:
        sys.exit("❌ MODEL_API_KEY, OPENAI_API_KEY, or GOOGLE_API_KEY environment variable is required.")

    # Detect model type from API key format
    if model_api_key.startswith("AIza"):
        model_name = "google/gemini-2.0-flash"
        print(f"   Detected Google API key, using model: {model_name}")
    else:
        model_name = "openai/gpt-4o-mini"
        print(f"   Detected OpenAI API key, using model: {model_name}")

    anchorbrowser_session_id: Optional[str] = None
    stagehand_session_id: Optional[str] = None

    # Initialize Anchorbrowser client
    anchor_client = AsyncAnchorbrowser(api_key=anchor_api_key)

    try:
        # 2. Create Anchorbrowser session using the SDK
        print("🌐 Creating Anchorbrowser session...")
        anchor_session = await anchor_client.sessions.create()
        
        if not anchor_session.data or not anchor_session.data.cdp_url:
            raise ValueError(f"Anchorbrowser session did not return a CDP URL. Response: {anchor_session}")
        
        anchorbrowser_session_id = anchor_session.data.id
        cdp_url = anchor_session.data.cdp_url
        
        print("✅ Anchorbrowser session created!")
        print(f"   Session ID: {anchorbrowser_session_id}")
        print(f"   CDP URL: {cdp_url}")

        # 3. Initialize Stagehand with the CDP URL
        print("\n🚀 Initializing Stagehand with CDP URL...")
        async with AsyncStagehand(
            server="local",
            model_api_key=model_api_key,
            local_openai_api_key=model_api_key,
            local_ready_timeout_s=30.0,
        ) as client:
            # Start session with CDP URL pointing to Anchorbrowser
            # Note: cdp_url must be inside launch_options for connecting to external browsers
            session_response = await client.sessions.start(
                model_name=model_name,
                browser={
                    "type": "local",
                    "launch_options": {
                        "cdp_url": cdp_url,
                    },
                },
            )
            stagehand_session_id = session_response.data.session_id
            print("✅ Stagehand initialized successfully!")
            print(f"   Stagehand Session ID: {stagehand_session_id}")

            try:
                # 4. Navigate to Hacker News
                print("\n📍 Navigating to Hacker News...")
                await client.sessions.navigate(
                    id=stagehand_session_id,
                    url="https://news.ycombinator.com",
                )
                print("✅ Navigation complete")

                # 5. Use Stagehand's AI-powered extraction
                print("\n🔍 Extracting top stories using AI...")
                extract_response = await client.sessions.extract(
                    id=stagehand_session_id,
                    instruction="Extract the titles and URLs of the first 5 stories on the page",
                    schema={
                        "type": "object",
                        "properties": {
                            "stories": {
                                "type": "array",
                                "items": {
                                    "type": "object",
                                    "properties": {
                                        "title": {"type": "string"},
                                        "url": {"type": "string"},
                                    },
                                    "required": ["title"],
                                },
                            },
                        },
                        "required": ["stories"],
                    },
                )

                print("\n📰 Extracted stories:")
                result: Any = extract_response.data.result
                if isinstance(result, dict) and "stories" in result:
                    stories = result["stories"]
                    if isinstance(stories, list):
                        for i, story in enumerate(stories, 1):
                            if isinstance(story, dict):
                                title = story.get("title", "N/A")
                                url = story.get("url")
                                print(f"   {i}. {title}")
                                if url:
                                    print(f"      URL: {url}")
                else:
                    print(f"   Raw result: {result}")

                # 6. Use Stagehand's observe to find clickable elements
                print("\n👁️  Observing page for login link...")
                observe_response = await client.sessions.observe(
                    id=stagehand_session_id,
                    instruction="find the login link",
                )
                actions = observe_response.data.result
                print(f"   Found {len(actions)} possible actions")
                if actions:
                    print(f"   First action: {actions[0].description}")

                # 7. Use Stagehand's act to perform an action
                print("\n🖱️  Clicking on the first story...")
                act_response = await client.sessions.act(
                    id=stagehand_session_id,
                    input="click on the first story title",
                )
                print(f"   Act completed: {act_response.data.result.message}")

                # Wait a moment for navigation
                await asyncio.sleep(2)

                print("\n✅ Test completed successfully!")

            finally:
                # 8. End the Stagehand session
                if stagehand_session_id:
                    print("\n🛑 Ending Stagehand session...")
                    await client.sessions.end(id=stagehand_session_id)
                    print("✅ Stagehand session ended")

    finally:
        # 9. Delete the Anchorbrowser session using the SDK
        if anchorbrowser_session_id:
            print("\n🧹 Deleting Anchorbrowser session...")
            try:
                await anchor_client.sessions.delete(anchorbrowser_session_id)
                print("✅ Anchorbrowser session deleted")
            except Exception as e:
                print(f"⚠️  Failed to delete Anchorbrowser session: {e}")

        # Close the Anchorbrowser client
        await anchor_client.close()


if __name__ == "__main__":
    asyncio.run(main())