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

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:

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
`enable_service_unavailable_recovery`	`bool`	Whether transient-failure requeue is enabled
`enable_action_timing_recovery`	`bool`	Whether page-loading recovery is enabled
`enable_incorrect_form_input_recovery`	`bool`	Deprecated, no effect
`enable_password_update_recovery`	`bool`	Whether password rotation recovery is enabled
`enable_tfa_setup_recovery`	`bool`	Deprecated, no effect
`enable_node_description_enrichment`	`bool`	Whether diagnostic step descriptions are generated
`allow_requeue`	`bool`	Master switch for automatic retries
`deny_requeue_past_dry_run_marker`	`bool`	Restrict retries to before the dry-run marker

Getting Workflow Metadata

Fetch detailed metadata for a specific workflow, including its effective input schema and Vault credential configuration:

metadata = client.workflows.get_workflow_metadata("workflow-123")

input_schema = metadata.input_schema
vault_schema = metadata.vault_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`	Effective JSON Schema used to validate `run_input_variables`, including Vault credential aliases
`vault_schema`	`dict`	Vault credential aliases and the domain configured for each alias
`workspace_id`	`str`	ID of the workspace that owns the workflow
`version_id`	`str`	UUID of the workflow version these schemas belong to
`version`	`str`	Semantic version of the workflow version these schemas belong to
`branch_id`	`str`	Branch the described version was saved on

Input Schema Structure

The input_schema follows JSON Schema format:

{
    "type": "object",
    "properties": {
        "url": {"type": "string"},
        "max_attempts": {"type": "integer"},
        "USER": {"type": "string"},
    },
    "required": ["url", "USER"],
    "additionalProperties": False
}

Vault Credential Inputs

A Vault Alias in Workflow is a required run input. Pass the matching Vault entry’s permissioned_user_id under the exact alias:

run_input_variables={
    "axxess_credentials": "customer-42",
}

The entry’s domain must match the configured Vault Domain. For credential pooling, pass a list of permissioned_user_id values. Pass an empty list to pool every vault entry in your workspace for that domain.

Validating Workflow Input

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

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}")

client.runs.start() does not call this method; the server validates run_input_variables against the version that runs and raises the same InputValidationError. Manual validation is useful when you want to check inputs without starting a run. Pass the same workflow_version or workflow_branch_id you will start the run with to validate against that version:

client.workflows.validate_workflow_input("workflow-123", payload, workflow_version="^1.2.0")

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)

Versions and Branches

See Workflow Versioning for how versions are numbered and which pin to choose. Metadata describes the latest version on the workflow’s default branch unless you select another one. get_workflow_metadata and validate_workflow_input take the same optional keyword arguments as the run pins; set at most one (setting both raises ValueError):

# Highest release matching an exact version or semver range
metadata = client.workflows.get_workflow_metadata("workflow-123", workflow_version="^1.2.0")
print(metadata.version, metadata.version_id)

# Latest version on a branch
metadata = client.workflows.get_workflow_metadata(
    "workflow-123", workflow_branch_id="3f6c2a8e-6b1d-4c3e-9a51-2d7f0e8b4c19"
)

# Every version across all branches, newest first
for version in client.workflows.list_workflow_versions("workflow-123"):
    print(version.version, version.branch_id, version.version_note)

# Branches, oldest first; is_default marks the branch unpinned runs use
for branch in client.workflows.list_workflow_branches("workflow-123"):
    print(branch.id, branch.name, branch.is_default)

WorkflowVersion Object

Field	Type	Description
`version_id`	`str`	UUID of the version
`version`	`str`	Semantic version, unique per workflow; pass it as `workflow_version`
`branch_id`	`str`	Branch the version was saved on
`version_note`	`str`	Optional note describing what changed
`created_by`	`str`	ID of the user who created the version
`created_at`	`str`	Creation timestamp

WorkflowBranch Object

Field	Type	Description
`id`	`str`	UUID of the branch; pass it as `workflow_branch_id`
`name`	`str`	Branch name, unique per workflow
`is_default`	`bool`	Whether unpinned runs use this branch
`head_version_id`	`str`	UUID of the branch’s latest version
`workflow_id`	`str`	ID of the workflow
`created_by`	`str`	ID of the user who created the branch
`created_at`	`str`	Creation timestamp
`updated_at`	`str`	Timestamp of the branch’s last change

Combining with Runs

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

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}")

Getting Started

Concepts

API Reference

SDK

Integrations

Workflows

Workflows Client

Setup

Listing All Workflows

Workflow Object

Getting Workflow Metadata

Metadata Object

Input Schema Structure

Vault Credential Inputs

Validating Workflow Input

Validation Checks

InputValidationError

Versions and Branches

WorkflowVersion Object

WorkflowBranch Object

Combining with Runs

​Workflows Client

​Setup

​Listing All Workflows

​Workflow Object

​Getting Workflow Metadata

​Metadata Object

​Input Schema Structure

​Vault Credential Inputs

​Validating Workflow Input

​Validation Checks

​InputValidationError

​Versions and Branches

​WorkflowVersion Object

​WorkflowBranch Object

​Combining with Runs

Workflows Client

Setup

Listing All Workflows

Workflow Object

Getting Workflow Metadata

Metadata Object

Input Schema Structure

Vault Credential Inputs

Validating Workflow Input

Validation Checks

InputValidationError

Versions and Branches

WorkflowVersion Object

WorkflowBranch Object

Combining with Runs