> ## 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.

# Workflows

> Explore and validate workflows with the Python SDK

# Workflows Client

The Workflows client lets you explore workflow definitions, fetch metadata, and validate input payloads before executing a run. Use it to inspect schemas and prevent invalid submissions.

## Setup

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

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

## Listing All Workflows

Retrieve all workflows accessible by your API key:

```python theme={null}
workflows = client.workflows.get_all_workflows()

for workflow in workflows:
    print(f"ID: {workflow.id}")
    print(f"Name: {workflow.name}")
    print(f"Description: {workflow.description}")
    print("---")
```

### Workflow Object

| Field                          | Type   | Description                              |
| ------------------------------ | ------ | ---------------------------------------- |
| `id`                           | `str`  | Unique workflow identifier               |
| `name`                         | `str`  | Workflow name                            |
| `description`                  | `str`  | Workflow description                     |
| `created_at`                   | `str`  | Creation timestamp                       |
| `updated_at`                   | `str`  | Last update timestamp                    |
| `workspace_id`                 | `str`  | Associated workspace ID                  |
| `created_by`                   | `str`  | Creator user ID                          |
| `enable_popup_handling`        | `bool` | Whether popup handling is enabled        |
| `enable_xpath_recovery`        | `bool` | Whether XPath recovery is enabled        |
| `enable_error_code_generation` | `bool` | Whether error code generation is enabled |

## Getting Workflow Metadata

Fetch detailed metadata for a specific workflow, including its input schema:

```python theme={null}
metadata = client.workflows.get_workflow_metadata("workflow-123")

input_schema = metadata.input_schema
properties = input_schema.properties or {}
required = input_schema.required or []

print("Input Schema Properties:")
for name, schema in properties.items():
    is_required = "required" if name in required else "optional"
    print(f"  {name} ({is_required}): {schema}")
```

### Metadata Object

| Field          | Type                  | Description                                   |
| -------------- | --------------------- | --------------------------------------------- |
| `input_schema` | `WorkflowInputSchema` | JSON Schema defining required input variables |

### Input Schema Structure

The `input_schema` follows JSON Schema format:

```python theme={null}
{
    "type": "object",
    "properties": {
        "url": {"type": "string"},
        "max_attempts": {"type": "integer"},
        "USER": {"type": "string"},
    },
    "required": ["url", "USER"],
    "additionalProperties": False
}
```

## Validating Workflow Input

Validate input payloads against a workflow's schema before starting a run:

```python theme={null}
from cloudcruise.workflows.types import InputValidationError

payload = {
    "url": "https://example.com",
    "USER": "user-123",
}

try:
    client.workflows.validate_workflow_input("workflow-123", payload)
    print("✓ Payload is valid")
except InputValidationError as exc:
    print(f"✗ Validation failed: {exc}")
    print(f"  Missing required: {exc.missingRequired}")
    print(f"  Invalid types: {exc.invalidTypes}")
    print(f"  Unknown keys: {exc.unknownKeys}")
```

<Note>
  When you call `client.runs.start()`, the SDK automatically validates the input against the workflow's schema. Manual validation is useful when you want to check inputs before making the API call.
</Note>

### Validation Checks

The validator checks:

| Check                     | Description                                                           |
| ------------------------- | --------------------------------------------------------------------- |
| **Required fields**       | All fields listed in `required` must be present                       |
| **Type matching**         | Values must match their declared type (string, number, boolean, etc.) |
| **Additional properties** | If `additionalProperties: false`, unknown keys are rejected           |

### InputValidationError

When validation fails, an `InputValidationError` is raised with details:

| Property          | Type                      | Description                                           |
| ----------------- | ------------------------- | ----------------------------------------------------- |
| `missingRequired` | `List[str]`               | Fields that are required but missing                  |
| `invalidTypes`    | `List[InvalidTypeDetail]` | Fields with incorrect value types                     |
| `unknownKeys`     | `List[str]`               | Fields not defined in schema (when extras disallowed) |

## Combining with Runs

Use workflow metadata to build dynamic UIs or validate inputs before execution:

```python theme={null}
from cloudcruise import StartRunRequest
from cloudcruise.workflows.types import InputValidationError

workflow_id = "workflow-123"

# Get metadata to understand required inputs
metadata = client.workflows.get_workflow_metadata(workflow_id)
schema = metadata.input_schema
required_fields = schema.required or []

print(f"Required inputs: {required_fields}")

# Build a form with required_fields...

# Build your payload
payload = {
    "url": "https://example.com",
    "USER": "user-123",
}

# Validate before running
try:
    client.workflows.validate_workflow_input(workflow_id, payload)
except InputValidationError as exc:
    print(f"Fix these issues: {exc}")
    raise

# Start the run
run = client.runs.start(
    StartRunRequest(
        workflow_id=workflow_id,
        run_input_variables=payload,
    )
)

print(f"Started run: {run.sessionId}")
```
