Python SDK Overview

Alpha Release - This SDK is in alpha and actively under development. Method signatures may change.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

Coming Soon - The Python package is currently being prepared for release on PyPI.
Once available, install the latest alpha version with:
pip install cloudcruise --pre
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)

2. Processing Events via Webhook (Most Reliable)

Webhooks are the most reliable way to receive updates.
from fastapi import FastAPI, HTTPException, Request
from cloudcruise.webhook.types import VerificationError

app = FastAPI()

@app.post("/webhook")
async def cloudcruise_webhook(request: Request):
    signature = request.headers.get("x-hmac-signature")
    body = await request.body()

    try:
        payload = client.webhook.verify_signature(
            body,
            signature,
            "your-webhook-secret",
        )

        event = payload.get("event")
        if event == "execution.success":
            print("Workflow completed successfully")
        elif event == "execution.failed":
            print("Workflow failed")

        return {"status": "received"}
    except VerificationError as error:
        raise HTTPException(status_code=error.statusCode, detail=str(error))

3. Event Handler Alternative

If you are unable to receive webhooks, you can stream events from the run handle:
from cloudcruise import StartRunRequest


def on_run_event(message):
    event = (message.get("data") or {}).get("event")
    print("Event:", event)
    print("Payload:", (message.get("data") or {}).get("payload"))


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

# Or iterate over streaming events
for message in run:
    event = (message.get("data") or {}).get("event")
    print("Received event:", event)

    if event == "execution.success":
        break

4. Wait for Completion

You can block until the workflow completes. For long‑running workflows this may take a while, so use cautiously.
from cloudcruise import StartRunRequest


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)

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