> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cloudcruise.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Python Overview

> Get started with the CloudCruise Python SDK

# Python SDK Overview

Contributions are welcome at [https://github.com/CloudCruise/cloudcruise-python](https://github.com/CloudCruise/cloudcruise-python).

Report bugs via [GitHub issues](https://github.com/CloudCruise/cloudcruise-python/issues) or [Discord](https://discord.com/invite/MHjbUqedZF).

The CloudCruise Python SDK lets you launch browser automation workflows, stream run events, verify webhooks, and manage vault credentials directly from Python services.

## Installation

```bash theme={null}
pip install cloudcruise
```

Set the required credentials in code or via environment variables (`CLOUDCRUISE_API_KEY`, `CLOUDCRUISE_ENCRYPTION_KEY`). Retrieve them from [CloudCruise Settings](https://app.cloudcruise.com/settings/api-keys).

```python theme={null}
from cloudcruise import CloudCruise, CloudCruiseParams

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

## 1. Creating a Run

```python theme={null}
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. 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`.

```python theme={null}
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`:

```python theme={null}
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.

```python theme={null}
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`: terminal status for the session

Example:

```python theme={null}
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.

```python theme={null}
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:

```python theme={null}
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](/run-api/start-a-run) for detailed workflow execution options
* Learn about [Vault API](/vault-api/create-vault-entry) for secure credential management
* Check out [Workflow API](/workflow-api/get-workflows) to manage your workflows
* Set up [Webhooks](/concepts/webhooks) for event notifications