Python SDK

`pip install ventus-sessions-sdk`. Async, Playwright-style, type-checked.

The official Python client wraps the REST + CDP surface in an idiomatic async API. It speaks CDP via Playwright under the hood, so anything Playwright can do, this SDK can do.

A TypeScript / Node SDK is planned, not yet shipped. For now, call the REST API directly with fetch + drive CDP via playwright from Node.

Install

pip install ventus-sessions-sdk

Requires Python 3.11+.

Authenticate

export VENTUS_API_KEY=vs_01J….…

Or pass explicitly:

from ventus_sessions_sdk import Sessions

client = Sessions(api_key="vs_01J….…")

One-shot session

import asyncio
from ventus_sessions_sdk import Sessions

async def main() -> None:
    async with Sessions() as client:
        async with client.session(region="us-east1") as session:
            await session.page.goto("https://example.com")
            title = await session.page.title()
            print(title)

asyncio.run(main())

async with client.session(...) does three things:

  1. POST /v1/sessions with a fresh Idempotency-Key (UUIDv4 — the SDK generates it). Response is 202 Accepted; the SDK polls until status is RUNNING.
  2. playwright.connect_over_cdp(session.connect_url).
  3. On exit (clean or exception): PATCH /v1/sessions/<id> {"status":"REQUEST_RELEASE"} so you don't pay for a hung Pod. (Admin force-stop via DELETE is a different surface; SDKs use the owner-PATCH form.)

Long-lived session

Sometimes you want the session outside of async with — e.g., to share across tasks:

session = await client.create_session(region="us-east1")
try:
    await session.page.goto("...")
    # ...
finally:
    await session.release()

Persistent login (profiles)

# 1. Drive the login flow once.
async with client.session() as session:
    await session.page.goto("https://customer.portal/login")
    await session.page.fill("input[name=user]", "...")
    await session.page.fill("input[name=pass]", "...")
    await session.page.click("button[type=submit]")
    profile_id = await session.save_profile(label="customer-portal")

# 2. Reuse it next time — no login flow.
async with client.session(profile_id=profile_id) as session:
    await session.page.goto("https://customer.portal/")
    # already signed in

Profiles store cookies + localStorage + sessionStorage but not filesystem state. They survive Chrome upgrades. See POST /v1/profiles for the raw API.

Human handoff (live view, interactive)

async with client.session(live_view={"interactive": True}) as session:
    print(f"Open {session.live_view_url} and complete the CAPTCHA.")
    await session.wait_for_human_input(timeout_s=300)
    # automation continues after the human clicks Done in the viewer

wait_for_human_input polls a session-scoped event the live viewer posts when the human clicks Done in the floating toolbar. Use it for MFA prompts, CAPTCHAs, signature pads.

Errors

The SDK maps every RFC-7807 problem doc to a typed exception:

from ventus_sessions_sdk import errors

try:
    await client.create_session(region="atlantis-1")
except errors.UnknownRegion as e:
    print(e.detail, e.status, e.type)
except errors.InsufficientCapacity:
    # …retry with a different region
    ...
except errors.RateLimit as e:
    await asyncio.sleep(e.retry_after_s)

errors.VentusError is the base; everything else extends it.

Type checking

Every public symbol is typed; py.typed is shipped. Run:

mypy --strict your_code.py

CLI

The package ships a small CLI for one-off probes:

ventus sessions list
ventus sessions create --region us-east1
ventus sessions release ses_01JC…
ventus profiles list

ventus --help enumerates everything.