Loading...
Loading...
Automate cloud browsers using the Scrapfly Cloud Browser API with Python Playwright. Covers connection options, sessions, anti-bot fingerprinting, and file downloads including streaming large files (up to 500 MB) out of the session with downloads_stream=1 and the CDP IO domain.
npx skill4agent add scrapfly/skills scrapfly-browserpip install playwrightSCRAPFLY_API_KEYwss://browser.scrapfly.io?api_key=YOUR_API_KEY| Parameter | Type | Default | Description |
|---|---|---|---|
| str | required | Scrapfly API key |
| str | | Proxy type: |
| str | random | OS fingerprint: |
| str | None | Proxy country (ISO 3166-1 alpha-2, e.g. |
| str | None | Session ID for browser state persistence |
| str | | Close session on disconnect. Set |
| int | 900 | Max session duration in seconds (900-1800) |
| str | | Enable session video recording |
| str | None | |
import os
from urllib.parse import urlencode
params = {
"api_key": os.environ["SCRAPFLY_API_KEY"],
"proxy_pool": "datacenter",
"os": "linux",
}
BROWSER_WS = f"wss://browser.scrapfly.io?{urlencode(params)}"from playwright.sync_api import sync_playwright
import os
API_KEY = os.environ["SCRAPFLY_API_KEY"]
BROWSER_WS = f"wss://browser.scrapfly.io?api_key={API_KEY}&proxy_pool=datacenter&os=linux"
with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
page.goto("https://web-scraping.dev")
print("Title:", page.title())
print("Content:", page.content()[:500])
finally:
browser.close()from playwright.async_api import async_playwright
import asyncio
import os
API_KEY = os.environ["SCRAPFLY_API_KEY"]
BROWSER_WS = f"wss://browser.scrapfly.io?api_key={API_KEY}&proxy_pool=datacenter&os=linux"
async def main():
async with async_playwright() as p:
browser = await p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else await context.new_page()
await page.goto("https://web-scraping.dev")
print("Title:", await page.title())
finally:
await browser.close()
asyncio.run(main())with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
page.goto("https://web-scraping.dev/login")
page.fill("input[name=username]", "user123")
page.fill("input[name=password]", "password")
page.click("button[type=submit]")
page.wait_for_selector("div#secret-message")
print("Logged in:", page.url)
finally:
browser.close()with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
page.goto("https://web-scraping.dev")
page.screenshot(path="screenshot.png", full_page=True)
finally:
browser.close()with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
page.goto("https://web-scraping.dev/products")
page.wait_for_selector("div.product")
products = page.query_selector_all("div.product")
for product in products:
name = product.query_selector("a").inner_text()
price = product.query_selector(".price").inner_text()
print(f"{name}: {price}")
finally:
browser.close()BROWSER_WS = (
f"wss://browser.scrapfly.io?"
f"api_key={API_KEY}"
f"&proxy_pool=residential"
f"&country=de"
f"&os=windows"
)
with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
page.goto("https://web-scraping.dev")
print("Page content from DE proxy:", page.title())
finally:
browser.close()LOGIN_URL = "https://web-scraping.dev/login"
USERNAME = "user123"
PASSWORD = "password"
# Step 1: Login and keep session alive
BROWSER_WS = (
f"wss://browser.scrapfly.io?"
f"api_key={API_KEY}"
f"&session=my-session"
f"&auto_close=false"
f"&proxy_pool=datacenter"
)
with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
page.goto(LOGIN_URL)
page.fill("input[name=username]", USERNAME)
page.fill("input[name=password]", PASSWORD)
page.click("button[type=submit]")
page.wait_for_selector("div#secret-message")
print("Login complete, session preserved")
finally:
browser.close() # Disconnects but session stays alive
# Step 2: Reconnect to same session (still authenticated with same credentials)
BROWSER_WS = (
f"wss://browser.scrapfly.io?"
f"api_key={API_KEY}"
f"&session=my-session"
f"&proxy_pool=datacenter"
)
with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
page.goto(LOGIN_URL)
page.wait_for_selector("div#secret-message")
print("Accessed protected page (still logged in):", page.title())
finally:
browser.close()with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
page.goto("https://web-scraping.dev/testimonials")
page.wait_for_selector("span.rating")
all_items = []
for _ in range(10): # Scroll 10 times
items = page.query_selector_all("span.rating")
all_items = items
page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
page.wait_for_timeout(2000)
print(f"Collected {len(all_items)} reviews")
finally:
browser.close()with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
page.goto("https://web-scraping.dev")
# Extract data using JavaScript
data = page.evaluate("""
() => {
return {
title: document.title,
links: Array.from(document.querySelectorAll('a')).map(a => ({
text: a.innerText,
href: a.href,
})),
};
}
""")
print(f"Found {len(data['links'])} links")
finally:
browser.close()with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
urls = [
"https://web-scraping.dev/products",
"https://web-scraping.dev/products?page=2",
"https://web-scraping.dev/products?page=3",
]
results = []
for url in urls:
page.goto(url)
results.append({"url": url, "content": content})
print(f"Scraped {len(results)} pages")
finally:
browser.close()ScrapiumBrowser.*| Command | Params | Returns |
|---|---|---|
| none | |
| none | |
| | |
| | |
| | every file's content, carried twice: |
Browser.downloadProgressstate == "completed"getDownloadsMetadatasgetDownloads{}KeyErrorimport base64
import time
def wait_for_download(cdp, filename, timeout=60):
"""Poll until the file exists and its size is stable across two reads."""
deadline = time.monotonic() + timeout
last = -1
while time.monotonic() < deadline:
size = cdp.send("ScrapiumBrowser.getDownloadsMetadatas")["metadata"].get(filename)
if size is not None and size == last:
return size
last = size if size is not None else -1
time.sleep(0.5)
raise TimeoutError(f"{filename} did not finish downloading")
with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
cdp = context.new_cdp_session(page)
page.goto("https://web-scraping.dev/file-download")
page.click("#download-btn")
size = wait_for_download(cdp, "download-sample.pdf")
# One file by name: content is carried once, so the reply is ~1.34x the file.
result = cdp.send("ScrapiumBrowser.getDownload", {"filename": "download-sample.pdf"})
open("download-sample.pdf", "wb").write(base64.b64decode(result["data"]))
finally:
browser.close()delete: TruegetDownloadgetDownloadsgetDownloads1009Target closedSession closedgetDownloadsfilesfiles_by_guidgetDownloadgetDownloadsdownload too large: "export.csv" (620000000 bytes) exceeds the 524288000 byte limit per filegetDownloadsMetadatasgetDownloadgetDownloadsMetadatasdownloads_stream=1downloads_stream=1getDownloadsfiles_by_guidguid_namesstreams{
"files": {},
"files_by_guid": {"9F2C1A4E-...": "JVBERi0xLjQKJe..."},
"guid_names": {"9F2C1A4E-...": "invoice.pdf", "B71D0E88-...": "export.csv"},
"streams": {
"B71D0E88-...": {"stream": "scrapfly-download:1", "size": 91250688, "name": "export.csv"}
}
}streamIO.readIO.closeimport base64
from urllib.parse import urlencode
from playwright.sync_api import sync_playwright
params = {"api_key": API_KEY, "proxy_pool": "datacenter", "downloads_stream": "1"}
BROWSER_WS = f"wss://browser.scrapfly.io?{urlencode(params)}"
CHUNK = 8 * 1024 * 1024 # base64 characters, not raw bytes (32 MB is the cap)
def drain(cdp, handle, destination):
"""Read a scrapfly-download: handle to eof. Sequential only, no offset."""
encoded = []
try:
while True:
read = cdp.send("IO.read", {"handle": handle, "size": CHUNK})
if read.get("data"):
encoded.append(read["data"])
if read.get("eof"):
break
finally:
try:
cdp.send("IO.close", {"handle": handle})
except Exception:
pass
# Chunks are cut on base64 group boundaries, so they concatenate as text
# and decode once.
with open(destination, "wb") as out:
out.write(base64.b64decode("".join(encoded)))
with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(BROWSER_WS)
try:
context = browser.contexts[0]
page = context.pages[0] if context.pages else context.new_page()
cdp = context.new_cdp_session(page)
page.goto("https://web-scraping.dev/file-download")
page.click("#download-btn")
wait_for_download(cdp, "download-sample.pdf")
result = cdp.send("ScrapiumBrowser.getDownloads")
# Small files still arrive inline, large ones as handles. Handle both.
for guid, b64 in result.get("files_by_guid", {}).items():
open(result["guid_names"][guid], "wb").write(base64.b64decode(b64))
for guid, stream in result.get("streams", {}).items():
drain(cdp, stream["stream"], stream["name"])
finally:
browser.close()offsetsizesizeeofdataeofIO.closeeofscrapfly-download:<n>blob:Page.printToPDFCHUNK = 10 * 1024 * 1024 # raw bytes here; the reply inflates ~1.34x
evaluated = cdp.send("Runtime.evaluate", {
"expression": f"fetch({url!r}).then(r => r.blob())",
"awaitPromise": True,
})
resolved = cdp.send("IO.resolveBlob", {"objectId": evaluated["result"]["objectId"]})
handle = "blob:" + resolved["uuid"]
# Drain with IO.read / IO.close, decoding each chunk when base64Encoded is true.Page.printToPDFtransferMode: "ReturnAsStream"Network.getResponseBodyRequest content was evicted from inspector cachecountryconnect_over_cdp()connect()browser.contexts[0]try/finallygetDownloadsdownloads_stream=1proxy_pool=datacenterresidentialdebug=true