Python SDK Overview

Contributions are welcome at https://github.com/CloudCruise/cloudcruise-python. Report bugs via GitHub issues or Discord. The CloudCruise Python SDK lets you launch browser automation workflows, stream run events, verify webhooks, and manage vault credentials directly from Python services.

Installation

pip install cloudcruise

Set the required credentials in code or via environment variables (CLOUDCRUISE_API_KEY, CLOUDCRUISE_ENCRYPTION_KEY). Retrieve them from CloudCruise Settings.

from cloudcruise import CloudCruise, CloudCruiseParams

client = CloudCruise(
    CloudCruiseParams(
        api_key="your-api-key",
        encryption_key="your-encryption-key",
    )
)

1. Creating a Run

from cloudcruise import StartRunRequest

request = StartRunRequest(
    workflow_id="workflow-123",
    run_input_variables={
        "variable_1": "https://example.com",
        "variable_2": "john_doe",
    },
)

run = client.runs.start(request)
print("Session ID:", run.sessionId)

By default a run executes the latest version on the workflow’s default branch. Set at most one of workflow_version (an exact version or semver range such as "1.2.3", "^1.2.0" or "1.x"; the highest matching release runs) or workflow_branch_id (that branch’s latest version runs) to pin it. Workflows shows how to list versions and branches, and Workflow Versioning explains what each version bump means and which pin to choose. The server validates run_input_variables against the input schema of the version that runs. Invalid input raises InputValidationError.

2. Listening to run events

Attach per-event handlers for an attribute-access experience. Each handler receives a FlattenedRunEvent with type, payload, timestamp, expires_at, and raw.

def on_execution_step(event):
    payload = event.payload
    current_step = payload.get("current_step")
    next_step = payload.get("next_step")
    print(f"[SESSION: {handle.sessionId}] STEP: {current_step} -> {next_step} @ {event.timestamp}")


def on_execution_start(event):
    workflow_id = event.payload.get("workflow_id")
    print(f"[SESSION: {handle.sessionId}] START: {workflow_id} @ {event.timestamp}")


handle = client.runs.start(
    StartRunRequest(
        workflow_id="workflow-123",
        run_input_variables={"target": "https://example.com"},
    )
)

handle.on("execution.step", on_execution_step)
handle.on("execution.start", on_execution_start)

handle.on("error", lambda err: print(f"[SESSION: {handle.sessionId}] ERROR: {err}"))
handle.on("end", lambda info: print(f"[SESSION: {handle.sessionId}] Workflow completed with status: {info.get('type')}"))

You can also subscribe to run.event once and switch on event.type:

def on_run_event(event):
    if event.type == "execution.step":
        payload = event.payload
        print("Execution step:", payload.get("current_step"), "->", payload.get("next_step"))
    else:
        print("Event:", event.type)


handle.on("run.event", on_run_event)

Async iteration (streaming)

Prefer a loop? Iterate over the handle to process the underlying SSE messages. Each item is the original SSE envelope.

for message in handle:
    data = (message.get("data") or {})
    event_type = data.get("event")
    if event_type != "execution.success":
        print("Received:", event_type)
        continue

    print("Workflow completed successfully")
    break

Other available events

open: connection established
ping: heartbeat messages
reconnect: reconnect attempts with {"attemptDelayMs": ...}
message: catch-all mirror of run.event and ping
error: SSE errors
end: the run reached a final status (execution.success, execution.failed or execution.stopped; see Run Statuses)

Example:

handle.on("ping", lambda payload: print("ping:", payload))
handle.on("reconnect", lambda payload: print("reconnect in", payload.get("attemptDelayMs"), "ms"))

3. Wait for Completion

You can block until the workflow completes. For long-running workflows this may take a while, so use cautiously. The returned value is a RunResult dataclass.

run = client.runs.start(
    StartRunRequest(
        workflow_id="workflow-123",
        run_input_variables={"target": "https://example.com"},
    )
)
result = run.wait()
print("Run completed:", result.status)
print("Output data:", result.data)

Watching a Live Session

While a session is running, fetch a viewer URL to watch its browser stream:

connection = client.runs.get_live_view_connection(run.sessionId)
print("Watch live:", connection.url)

The auth token embedded in connection.url is single-use — reopening a previously used viewer link (reloading the tab, or opening it later) fails to connect. Call get_live_view_connection again to mint a fresh link instead of reusing the old one. This only works while the session is still active.

Next Steps

Explore the Run API for detailed workflow execution options
Learn about Vault API for secure credential management
Check out Workflow API to manage your workflows
Set up Webhooks for event notifications

Getting Started

Concepts

API Reference

SDK

Integrations

Python Overview

Python SDK Overview

Installation

1. Creating a Run

2. Listening to run events

Async iteration (streaming)

Other available events

3. Wait for Completion

Watching a Live Session

Next Steps

​Python SDK Overview

​Installation

​1. Creating a Run

​2. Listening to run events

​Async iteration (streaming)

​Other available events

​3. Wait for Completion

​Watching a Live Session

​Next Steps

Python SDK Overview

Installation

1. Creating a Run

2. Listening to run events

Async iteration (streaming)

Other available events

3. Wait for Completion

Watching a Live Session

Next Steps