Python SDK
Control headless browsers from Python. Create sessions, interact with pages, and extract structured data.
Installation #
Install the package from PyPI.
pip install browserbeam
Requirements
| Dependency | Version |
|---|---|
| Python | 3.9+ |
| httpx | Installed automatically |
Browserbeam) and async (AsyncBrowserbeam) clients ship in the same package. No extra dependencies needed.
Quick Start #
Create a session, click a button, and extract data — all in under 10 lines.
from browserbeam import Browserbeam
client = Browserbeam(api_key="sk_live_abc123")
session = client.sessions.create(url="https://example.com")
# Click a button by its ref
session.click(ref="e3")
# Extract structured data from the page
result = session.extract(
title="h1 >> text",
links=[{"_parent": "a", "href": ">> href", "text": ">> text"}]
)
print(result.extraction)
# {"title": "Example Domain", "links": [{"href": "https://...", "text": "More info"}]}
# Close the session when done
session.close()
Every method returns a SessionEnvelope with the latest page state, extraction results, and media. Close the session explicitly when you're done to release resources.
Configuration #
API Key
Pass your API key directly or set the BROWSERBEAM_API_KEY environment variable.
# Option 1: Pass directly
client = Browserbeam(api_key="sk_live_abc123")
# Option 2: Environment variable
# export BROWSERBEAM_API_KEY=sk_live_abc123
client = Browserbeam()
Client Options
| Param | Type | Default | Description |
|---|---|---|---|
api_key |
str |
None |
Your Browserbeam API key. Falls back to BROWSERBEAM_API_KEY env var. |
base_url |
str |
https://api.browserbeam.com |
API base URL. Override for self-hosted or staging environments. |
timeout |
float |
120.0 |
HTTP request timeout in seconds. |
client = Browserbeam(
api_key="sk_live_abc123",
base_url="https://api.browserbeam.com",
timeout=60.0,
)
client.close() when you're done to release the underlying HTTP connection pool.
Creating Sessions #
Call client.sessions.create() to launch a browser session. It returns a Session object you use for all subsequent actions.
session = client.sessions.create(
url="https://example.com",
viewport={"width": 1280, "height": 720},
locale="en-US",
timezone="America/New_York",
)
Parameters
| Param | Type | Required | Default | Description |
|---|---|---|---|---|
url |
str |
no | — | URL to navigate to after session creation. |
viewport |
dict |
no | 1280×800 |
Browser viewport size. Keys: width, height. |
locale |
str |
no | en-US |
Browser locale (BCP 47 tag). |
timezone |
str |
no | UTC |
IANA timezone identifier. |
proxy |
str |
no | — | HTTP proxy URL for the session. |
block_resources |
list[str] |
no | [] |
Resource types to block: "image", "stylesheet", "font", "media". |
auto_dismiss_blockers |
bool |
no | False |
Auto-dismiss cookie banners and overlays. |
cookies |
list[dict] |
no | [] |
Cookies to inject before navigation. Each dict: name, value, domain. |
timeout |
int |
no | 300 |
Session lifetime in seconds (10–1800). |
idempotency_key |
str |
no | — | Prevent duplicate session creation on retries. |
Session Properties
After creation and after every method call, the session object exposes the latest state.
| Property | Type | Description |
|---|---|---|
session_id | str | Unique session identifier. |
expires_at | str | ISO 8601 expiration timestamp. |
page | PageState | Current page URL, title, markdown, interactive elements, scroll state. |
extraction | dict | None | Extracted data from the last extract call. |
media | list[MediaItem] | Screenshots, PDFs from the last call. |
error | StepError | None | Step-level error if the last action failed. |
completed | int | Number of steps completed in the last call. |
Session Methods #
Every method returns a SessionEnvelope with updated page state. Methods map 1:1 to API step types.
| Method | Description |
|---|---|
goto(url) | Navigate to a URL. |
observe() | Get page content and interactive elements. |
click() | Click an element by ref, text, or label. |
fill(value) | Clear and fill an input field. |
type(value) | Type text character by character. |
select(value) | Select a dropdown option. |
check() | Toggle a checkbox or radio button. |
scroll() | Scroll the page or an element. |
scroll_collect() | Scroll and collect all page content. |
screenshot() | Capture a screenshot. |
wait() | Wait for time, selector, or text. |
extract(**schema) | Extract structured data from the page. |
fill_form(fields) | Fill multiple form fields at once. |
upload(files) | Upload files to a file input. |
pdf() | Generate a PDF of the page. |
execute_js(code) | Run JavaScript in the browser. |
close() | Close the session and release resources. |
goto(url)
#
Navigate the session to a new URL. Waits for the page to load before returning.
| Param | Type | Required | Description |
|---|---|---|---|
url | str | yes | Target URL. |
wait_for | str | no | CSS selector to wait for after navigation. |
wait_timeout | int | no | Timeout in ms for wait_for. |
wait_until | str | no | JavaScript expression that must become truthy after navigation. |
session.goto("https://news.ycombinator.com")
print(session.page.url) # "https://news.ycombinator.com"
print(session.page.title) # "Hacker News"
observe()
#
Get the current page content and interactive elements. Use this to discover element refs before clicking or filling.
| Param | Type | Required | Description |
|---|---|---|---|
scope | str | no | CSS selector to scope observation to a subtree. |
format | str | no | Content format: "markdown" or "text". |
include_links | bool | no | Include links in markdown output. |
max_text_length | int | no | Truncate content to this character count. |
result = session.observe(scope="nav", format="markdown")
# Page content
print(result.page.markdown.content)
# Interactive elements with refs
for el in result.page.interactive_elements:
print(f"{el.ref}: [{el.tag}] {el.label}")
Response example:
{
"page": {
"url": "https://example.com",
"title": "Example",
"markdown": {"content": "# Welcome\nThis is an example page."},
"interactive_elements": [
{"ref": "e1", "tag": "a", "role": "link", "label": "Home"},
{"ref": "e2", "tag": "button", "role": "button", "label": "Sign In"}
]
}
}
click()
#
Click an element. Target it by ref (from observe), visible text, or accessible label.
| Param | Type | Required | Description |
|---|---|---|---|
ref | str | no* | Element ref from observation. |
text | str | no* | Visible text to match. |
label | str | no* | Accessible label to match. |
ref, text, or label.
# By ref (most precise)
session.click(ref="e2")
# By visible text
session.click(text="Sign In")
# By accessible label
session.click(label="Submit form")
fill(value)
#
Clear an input field and set its value. Target by ref, text, or label.
| Param | Type | Required | Description |
|---|---|---|---|
value | str | yes | Text to fill into the field. |
ref | str | no | Element ref from observation. |
text | str | no | Visible text to match the element. |
label | str | no | Accessible label to match. |
# Fill by ref
session.fill("user@example.com", ref="e5")
# Fill by label
session.fill("hunter2", label="Password")
# Fill by visible text
session.fill("hello@example.com", text="Email")
fill_form(fields)
#
Fill multiple form fields in a single call. Fields are matched by label. Optionally submit the form.
| Param | Type | Required | Default | Description |
|---|---|---|---|---|
fields | dict[str, str] | yes | — | Map of label → value pairs. |
submit | bool | no | False | Submit the form after filling. |
session.fill_form(
{
"Email": "user@example.com",
"Password": "s3cur3p@ss",
"Company": "Acme Inc.",
},
submit=True,
)
extract(**schema)
#
Extract structured data from the page. Define your schema as keyword arguments. Supports nested objects and arrays.
# Simple extraction
result = session.extract(
title="h1 >> text",
description="meta[name=description] >> content",
)
print(result.extraction)
# {"title": "Example Domain", "description": "An example page."}
# Array extraction with nested schema
result = session.extract(
products=[{
"_parent": ".product-card",
"name": "h2 >> text",
"price": ".price >> text",
"url": "a >> href",
}]
)
for product in result.extraction["products"]:
print(f"{product['name']}: {product['price']}")
Response example:
{
"extraction": {
"products": [
{"name": "Widget Pro", "price": "$29.99", "url": "/products/widget-pro"},
{"name": "Gadget Plus", "price": "$49.99", "url": "/products/gadget-plus"}
]
}
}
screenshot()
#
Capture a screenshot of the current viewport or the full page.
| Param | Type | Required | Default | Description |
|---|---|---|---|---|
full_page | bool | no | False | Capture the full scrollable page. |
format | str | no | "png" | "png", "jpeg", or "webp". |
quality | int | no | 80 | JPEG/WebP quality (1–100). |
selector | str | no | — | CSS selector to screenshot a specific element. |
import base64
result = session.screenshot(full_page=True, format="png")
# Save to file
image_data = base64.b64decode(result.media[0].data)
with open("screenshot.png", "wb") as f:
f.write(image_data)
scroll_collect()
#
Scroll through the entire page and collect all content. Useful for infinite-scroll or long pages.
| Param | Type | Required | Default | Description |
|---|---|---|---|---|
max_scrolls | int | no | 10 | Maximum number of scrolls. |
wait_ms | int | no | 500 | Wait between scrolls in milliseconds. |
timeout_ms | int | no | 30000 | Total timeout for the scroll operation. |
max_text_length | int | no | — | Truncate collected content. |
result = session.scroll_collect(max_scrolls=20, wait_ms=1000)
print(result.page.markdown.content) # Full page content
print(result.page.scroll.percent) # 100
close()
#
Close the session and release browser resources. Always close sessions when you're done.
session.close()
type(value)
#
Type text character by character. Useful for inputs with autocomplete or client-side filtering. Target by ref, text, or label.
session.type("search query", ref="e5", delay=50)
session.type("hello", label="Search")
select(value)
#
Select a dropdown option by value. Target the select element by ref, text, or label.
session.select("us", label="Country")
check()
#
Toggle a checkbox or radio button. Target by ref, text, or label.
session.check(label="I agree to the terms")
session.check(ref="cb3", checked=False)
scroll()
#
Scroll the page or scroll an element into view. Target by ref, text, or label.
session.scroll(to="bottom")
session.scroll(direction="down", amount=500)
session.scroll(text="Footer section")
wait()
#
Wait for a duration, CSS selector, visible text, or a JavaScript expression to become truthy.
session.wait(ms=2000)
session.wait(selector=".loaded", timeout=10000)
session.wait(until_js="document.querySelectorAll('.item').length > 5")
upload(files)
#
Upload files to a file input. Target by ref, text, or label.
session.upload(["https://example.com/file.pdf"], ref="file1")
session.upload(["https://example.com/photo.jpg"], label="Upload photo")
pdf()
#
Generate a PDF of the page. Supports format, landscape, background, scale, and margin options.
import base64
result = session.pdf(format="A4", landscape=True, scale=0.8,
margin={"top": "1cm", "bottom": "1cm"})
pdf_data = base64.b64decode(result.media[0].data)
with open("page.pdf", "wb") as f:
f.write(pdf_data)
execute_js(code)
#
Run JavaScript code in the browser. The result is stored in the extraction under the specified key.
result = session.execute_js("return document.title")
print(result.extraction["js_result"])
# With custom result key and timeout
result = session.execute_js(
"return document.querySelectorAll('a').length",
result_key="link_count", timeout=5000
)
Async Support #
Use AsyncBrowserbeam for async/await workflows. The API is identical to the sync client — every method is an async coroutine.
import asyncio
from browserbeam import AsyncBrowserbeam
async def main():
client = AsyncBrowserbeam(api_key="sk_live_abc123")
session = await client.sessions.create(url="https://example.com")
await session.click(ref="e3")
result = await session.extract(
title="h1 >> text",
links=[{"_parent": "a", "href": ">> href"}]
)
print(result.extraction)
await session.close()
await client.close()
asyncio.run(main())
Session Management #
List, inspect, and destroy sessions through the client.sessions namespace.
List Sessions
result = client.sessions.list(status="active", limit=10)
for s in result.sessions:
print(f"{s.session_id} — {s.status} — {s.started_at}")
| Param | Type | Required | Description |
|---|---|---|---|
status | str | no | Filter by status: "active", "closed". |
limit | int | no | Max sessions to return. |
after | str | no | Cursor for pagination (from next_cursor). |
Get Session
info = client.sessions.get("ses_abc123")
print(info.status) # "active"
print(info.duration_seconds) # 42
print(info.expires_at) # "2026-03-24T15:30:00Z"
Destroy Session
client.sessions.destroy("ses_abc123")
destroy() is different from session.close(). The close() method on a session sends a close step, while destroy() is a direct API call that terminates the session immediately.
Error Handling #
All API errors raise typed exceptions that extend BrowserbeamError. Catch specific errors or the base class.
| Exception | HTTP Status | When |
|---|---|---|
AuthenticationError | 401 | Missing or invalid API key. |
InvalidRequestError | 400 | Malformed request body. |
SessionNotFoundError | 404 | Session expired or does not exist. |
RateLimitError | 429 | Too many requests. Check retry_after. |
QuotaExceededError | 429 | Runtime quota exhausted for the billing period. |
EngineUnavailableError | 503 | Browser engine temporarily unavailable. Retry after retry_after. |
StepExecutionError | — | A step failed during execution. Includes step_index and action. |
Error Properties
Every exception carries structured context.
| Property | Type | Description |
|---|---|---|
message | str | Human-readable error message. |
code | str | Machine-readable error code. |
status_code | int | HTTP status code. |
request_id | str | Request ID for support debugging. |
context | dict | Additional error context from the API. |
from browserbeam import Browserbeam, AuthenticationError, RateLimitError, SessionNotFoundError, BrowserbeamError
import time
client = Browserbeam(api_key="sk_live_abc123")
try:
session = client.sessions.create(url="https://example.com")
session.click(ref="e1")
result = session.extract(title="h1 >> text")
print(result.extraction)
session.close()
except AuthenticationError:
print("Invalid API key. Check your BROWSERBEAM_API_KEY.")
except RateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after} seconds.")
time.sleep(e.retry_after or 5)
except SessionNotFoundError:
print("Session expired or not found.")
except BrowserbeamError as e:
print(f"API error [{e.code}]: {e.message}")
print(f"Request ID: {e.request_id}")
Full Example #
Extract the top stories from Hacker News, search for a topic, and save a screenshot.
import base64
from browserbeam import Browserbeam, BrowserbeamError
client = Browserbeam(api_key="sk_live_abc123")
try:
# Step 1: Navigate to Hacker News
session = client.sessions.create(
url="https://news.ycombinator.com/",
auto_dismiss_blockers=True,
viewport={"width": 1440, "height": 900},
)
# Step 2: Extract top stories from the front page
result = session.extract(
stories=[{
"_parent": ".titleline",
"title": "a >> text",
"url": "a >> href",
}]
)
for story in result.extraction["stories"][:5]:
print(f"{story['title']} — {story['url']}")
# Step 3: Fill the search form and submit
session.fill_form({"q": "browser automation"}, submit=True)
# Step 4: Take a screenshot of search results
screenshot = session.screenshot(full_page=True)
image_data = base64.b64decode(screenshot.media[0].data)
with open("hn-search.png", "wb") as f:
f.write(image_data)
print("Screenshot saved to hn-search.png")
# Step 5: Close the session
session.close()
except BrowserbeamError as e:
print(f"Error [{e.code}]: {e}")
print(f"Request ID: {e.request_id}")