Stagehand Python SDK

repository·main·Indexed 19 days ago

https://github.com/browserbase/stagehand-python

An AI-powered browser automation framework that combines natural language instructions with code to control web browsers. The SDK provides a production-ready, self-healing automation tool featuring synchronous and asynchronous clients (AsyncStagehand), session management for observing, acting, and extracting data, and support for SSE streaming and Pydantic-based API responses.

Tokens
15.9K
Snippets
52
Records
59
Agent score
67%

What's inside stagehand

  1. Differentiate between null and missing fields in responses

    main

    In Stagehand API responses, both an explicit null in JSON and a missing key result in a value of None in Python. To distinguish between these two cases, use the .model_fields_set attribute on the response object to check if the key was actually present in the payload.

    if response.my_field is None:
        if "my_field" not in response.model_fields_set:
            print('Got json like {}, without a "my_field" key present at all.'.format(response.model_dump()))
        else:
            print('Got json like {"my_field": null}.')
  2. Handle Stagehand API responses

    main

    Stagehand API responses are returned as Pydantic models. This allows for easy serialization and manipulation.

    To work with responses:

    • Serialize to JSON: Use model.to_json().
    • Convert to dictionary: Use model.to_dict().
    • Create a modified copy: Since response objects are immutable, use model.model_copy(update={...}) (for Pydantic v2) instead of mutating the object in place.
  3. Understand Stagehand session operations

    main

    Stagehand operates through sessions. A typical workflow involves starting a session, navigating to a URL, and then performing actions like observing, acting, extracting, or executing instructions.

    Key session methods include:

    • client.sessions.start(...): Initializes a session with a specific model and browser type.
    • client.sessions.navigate(session_id, url=...): Navigates the session to a URL.
    • client.sessions.observe(...): Observes the page based on an instruction.
    • client.sessions.act(...): Performs an action (e.g., clicking) based on natural language input.
    • client.sessions.extract(...): Extracts structured data from the page using a provided schema.
    • client.sessions.execute(...): Runs an agentic workflow to complete a complex task.
    • client.sessions.end(session_id): Ends the session.
    import os
    from playwright.sync_api import sync_playwright
    from stagehand import Stagehand
    
    def main() -> None:
        with Stagehand(
            server="remote",
            browserbase_api_key=os.environ.get("BROWSERBASE_API_KEY"),
            model_api_key=os.environ.get("MODEL_API_KEY"),
        ) as client:
            session = client.sessions.start(
                model_name="anthropic/claude-sonnet-4-6",
                browser={"type": "browserbase"},
            )
    
            cdp_url = session.data.cdp_url
            with sync_playwright() as p:
                browser = p.chromium.connect_over_cdp(cdp_url)
                page = browser.new_page()
    
                client.sessions.navigate(session.id, url="https://news.ycombinator.com")
                
                # Example: Observe
                observe_stream = client.sessions.observe(
                    session.id,
                    instruction="find the link to view comments for the top post",
                    stream_response=True,
                    x_stream_response="true",
                )
                
                # Example: Act
                act_stream = client.sessions.act(
                    session.id,
                    input="Click the comments link for the top post",
                    stream_response=True,
                    x_stream_response="true",
                )
    
                # Example: Extract
                extract_stream = client.sessions.extract(
                    session.id,
                    instruction="extract the text of the top comment on this page",
                    schema={
                        "type": "object",
                        "properties": {
                            "commentText": {"type": "string"},
                            "author": {"type": "string"},
                        },
                        "required": ["commentText"],
                    },
                    stream_response=True,
                    x_stream_response="true",
                )
    
                # Example: Execute
                execute_stream = client.sessions.execute(
                    session.id,
                    execute_options={
                        "instruction": "Click the 'Learn more' link if available",
                        "max_steps": 3,
                    },
                    agent_config={
                        "model": {"model_name": "anthropic/claude-opus-4-6"},
                        "cua": False,
                    },
                    stream_response=True,
                    x_stream_response="true",
                )
    
            client.sessions.end(session.id)
  4. Make custom or undocumented API requests

    main

    If you need to interact with undocumented endpoints or send extra parameters, you can use the generic HTTP verb methods on the client (e.g., client.get, client.post).

    • Undocumented Endpoints: Use client.post("/path", ...).
    • Extra Parameters: Use extra_query, extra_body, or extra_headers options.
    • Undocumented Response Properties: Access them via response.unknown_prop or by inspecting response.model_extra (a dictionary of extra fields).
    import httpx
    
    # Requesting an undocumented endpoint
    response = client.post(
        "/foo",
        cast_to=httpx.Response,
        body={"my_param": True},
    )
    
    print(response.headers.get("x-foo"))
  5. Stream responses using Server-Sent Events (SSE)

    main

    To enable streaming responses, you must satisfy two requirements:

    1. Set the header x_stream_response="true" in the request.
    2. Set stream_response=True in the client call to instruct the client to parse the SSE stream.

    Streaming events are of type "system" or "log".

    import asyncio
    from stagehand import AsyncStagehand
    
    async def main() -> None:
        async with AsyncStagehand() as client:
            session = await client.sessions.start(model_name="anthropic/claude-sonnet-4-6")
    
            stream = await client.sessions.act(
                id=session.id,
                input="click the first link on the page",
                stream_response=True,
                x_stream_response="true",
            )
            async for event in stream:
                # event is a StreamEvent (type: "system" | "log")
                print(event.type, event.data)
    
    asyncio.run(main())
  6. Manage HTTP client resources and lifecycle

    main

    While the library automatically closes connections when the client is garbage collected, it is best practice to manage the lifecycle manually using a context manager or by calling .close(). This ensures HTTP resources are released promptly.

    from stagehand import Stagehand
    
    with Stagehand() as client:
      # make requests here
      ...
    
    # HTTP client is now closed
  7. Use AsyncStagehand for asynchronous execution

    main

    The Stagehand Python SDK is designed for asynchronous use. It is recommended to use AsyncStagehand and await every API call to ensure non-blocking execution.

    import asyncio
    from stagehand import AsyncStagehand
    
    async def main() -> None:
        client = AsyncStagehand()
        session = await client.sessions.start(model_name="anthropic/claude-sonnet-4-6")
        response = await session.act(input="click the first link on the page")
        print(response.data)
    
    asyncio.run(main())
  8. Configure Stagehand environment variables

    main

    Stagehand requires specific environment variables for authentication. You can set these in a .env file.

    Required variables:

    • MODEL_API_KEY: Your API key for the LLM provider.
    • BROWSERBASE_API_KEY: Your Browserbase API key.

    To set up your environment from the examples directory:

    cp examples/.env.example examples/.env
    # Edit examples/.env with your credentials.
  9. Use aiohttp as the HTTP backend for improved concurrency

    main

    By default, the async client uses httpx. For improved concurrency performance, you can install and use aiohttp as the HTTP backend.

    1. Install the extra dependency: uv pip install stagehand[aiohttp]

    2. Instantiate the client with DefaultAioHttpClient().

    import asyncio
    from stagehand import AsyncStagehand, DefaultAioHttpClient
    
    async def main() -> None:
        async with AsyncStagehand(http_client=DefaultAioHttpClient()) as client:
            session = await client.sessions.start(model_name="anthropic/claude-sonnet-4-6")
            response = await session.act(input="click the first link on the page")
            print(response.data)
    
    asyncio.run(main())
  10. Access raw and streaming HTTP responses

    main

    Standard SDK methods deserialize responses into Pydantic models, hiding headers and status codes. To access raw data, use the following patterns:

    Raw Responses

    Prefix any HTTP method call with with_raw_response. This eagerly reads the full body. Use .parse() to get the deserialized object.

    Streaming Response Bodies

    To stream the response body (not SSE), use with_streaming_response. This requires a context manager and only reads the body when you call methods like .read(), .text(), .json(), .iter_lines(), etc.

    # Raw response example
    async with AsyncStagehand() as client:
        response = await client.sessions.with_raw_response.start(model_name="...")
        print(response.headers.get("X-My-Header"))
        session = response.parse()
    
    # Streaming body example
    async with AsyncStagehand() as client:
        async with client.sessions.with_streaming_response.start(model_name="...") as response:
            async for line in response.iter_lines():
                print(line)