Browserbase integrations

pip install browserbase

from langchain_community.document_loaders import BrowserbaseLoader

pip install deepagents browserbase stagehand langchain-openai python-dotenv

export BROWSERBASE_API_KEY="bb_..."

# Optional: model overrides
export DEEPAGENT_MODEL="gpt-5.4"
export DEEPAGENT_BASE_URL="https://<your-openai-compatible-gateway>"
export STAGEHAND_MODEL="google/gemini-3-flash-preview"
export STAGEHAND_AGENT_MODEL="anthropic/claude-sonnet-4-6"

import asyncio
import os
from browserbase import Browserbase
from bs4 import BeautifulSoup
from langchain.tools import tool
from stagehand import AsyncStagehand

@tool
def browserbase_search(query: str, num_results: int = 5) -> str:
    """Search the web with Browserbase. Use this first for discovery before opening pages."""
    bb = Browserbase(api_key=os.environ["BROWSERBASE_API_KEY"])
    response = bb.search.web(query=query, num_results=max(1, min(num_results, 10)))
    results = [
        {"title": r.title, "url": r.url}
        for r in getattr(response, "results", [])
    ]
    return str({"query": query, "results": results})


@tool
def browserbase_fetch(url: str, use_proxy: bool = False, max_chars: int = 12000) -> str:
    """Fetch page content without a browser session. Best for static pages and quick reads."""
    bb = Browserbase(api_key=os.environ["BROWSERBASE_API_KEY"])
    response = bb.fetch_api.create(url=url, proxies=use_proxy)
    content = str(getattr(response, "content", ""))
    soup = BeautifulSoup(content, "html.parser")
    for tag in soup(["script", "style"]):
        tag.decompose()
    text = (soup.body or soup).get_text("\n", strip=True)[:max_chars]
    return str({"url": url, "status_code": response.status_code, "text": text})


@tool
def browserbase_rendered_extract(start_url: str, instruction: str) -> str:
    """Open a full Browserbase browser session and extract rendered content with Stagehand."""
    return asyncio.run(_rendered_extract(start_url, instruction))


@tool
def browserbase_interactive_task(start_url: str, task: str) -> str:
    """Open a Browserbase-hosted Stagehand session and execute a multi-step browser task."""
    return asyncio.run(_interactive_task(start_url, task))


async def _rendered_extract(start_url: str, instruction: str) -> str:
    client = AsyncStagehand(browserbase_api_key=os.environ["BROWSERBASE_API_KEY"])
    resp = await client.sessions.start(model_name=os.getenv("STAGEHAND_MODEL", "google/gemini-3-flash-preview"))
    session_id = resp.data.session_id
    try:
        await client.sessions.navigate(id=session_id, url=start_url, frame_id="")
        result = await client.sessions.extract(id=session_id, instruction=instruction)
        return str({"session_id": session_id, "result": getattr(getattr(result, "data", None), "result", None)})
    finally:
        await client.sessions.end(id=session_id)


async def _interactive_task(start_url: str, task: str) -> str:
    model = os.getenv("STAGEHAND_AGENT_MODEL", "anthropic/claude-sonnet-4-6")
    client = AsyncStagehand(browserbase_api_key=os.environ["BROWSERBASE_API_KEY"])
    resp = await client.sessions.start(model_name=model)
    session_id = resp.data.session_id
    try:
        await client.sessions.navigate(id=session_id, url=start_url, frame_id="")
        result = await client.sessions.execute(
            id=session_id,
            execute_options={"instruction": task, "max_steps": 20},
            agent_config={"model": model, "instructions": "Execute the browser task precisely and stop when done."},
            timeout=300.0,
        )
        return str(result)
    finally:
        await client.sessions.end(id=session_id)

import uuid
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

BROWSER_SUBAGENT = {
    "name": "browser-specialist",
    "description": "Handles JS-heavy browsing, rendered extraction, and interactive browser tasks through Browserbase.",
    "system_prompt": (
        "You are a Browserbase browsing specialist. "
        "Use browserbase_rendered_extract for read-only work on rendered pages. "
        "Use browserbase_interactive_task only for stateful actions such as clicking, typing, or submitting forms. "
        "Return concise summaries with the relevant page URL and whether the task succeeded."
    ),
    "tools": [browserbase_rendered_extract, browserbase_interactive_task],
}

agent = create_deep_agent(
    model=ChatOpenAI(model="gpt-5.4", api_key=os.environ["OPENAI_API_KEY"]),
    tools=[browserbase_search, browserbase_fetch],
    subagents=[BROWSER_SUBAGENT],
    system_prompt=(
        "You are a research-oriented Deep Agent with Browserbase tools. "
        "Start with browserbase_search for discovery. "
        "Prefer browserbase_fetch for quick reads of static pages. "
        "Delegate JS-heavy or interactive work to the browser-specialist subagent."
    ),
    # Gate interactive browser actions behind human approval
    interrupt_on={
        "browserbase_interactive_task": {
            "allowed_decisions": ["approve", "edit", "reject"]
        }
    },
    checkpointer=MemorySaver(),
)

from langgraph.types import Command

config = {"configurable": {"thread_id": str(uuid.uuid4())}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Research the Browserbase Fetch API and explain when to escalate to a full browser session."}]},
    config=config,
    version="v2",
)

# Resume after each interrupt (approve / edit / reject)
while result.interrupts:
    interrupt_value = result.interrupts[0].value
    for action in interrupt_value["action_requests"]:
        print(f"Pending: {action['name']}\nArgs: {action['args']}")
    decision = input("Decision [approve/edit/reject]: ").strip().lower()
    result = agent.invoke(
        Command(resume={"decisions": [{"type": decision}]}),
        config=config,
        version="v2",
    )

User query
  └─ Is a specific URL already known?
       ├─ No  → browserbase_search (discover URLs first)
       └─ Yes ↓
            Is the page JavaScript-rendered?
              ├─ No  → browserbase_fetch (fast, stateless)
              └─ Yes ↓
                   Does the task require interaction (click / login / form)?
                     ├─ No  → browserbase_rendered_extract (read-only full browser)
                     └─ Yes → browserbase_interactive_task (Stagehand agent, triggers interrupt_on)