Task Types

Common Structure

All task types in Compozy share a common base structure with the following fields:

- id: string           # Unique identifier for the task
  name: string         # Optional human-readable name
  type: string         # Task type identifier
  import: string       # Path to external task definition
  schema:             # Optional schema definition
    input: object     # Input schema for task parameters
  env: object         # Task-specific environment variables
  config: object      # Optional configuration
  retry:             # Retry configuration
    attempts: number
    backoff: string
    interval: duration
    max_interval: duration
    jitter: boolean
  on_start: object    # Task start handler
  on_success: object  # Success handler
  on_error: object    # Error handler
  on_complete: object # Completion handler

Each task type extends this common structure with its specific fields and configurations. Tasks provide access to state information through event handlers:

status: string        # Current task status (running, success, error)
duration: number      # Task duration in milliseconds
start_time: string    # Task start timestamp
end_time: string      # Task end timestamp
retry_count: number   # Number of retry attempts made
error: object        # Error information if failed
output: any          # Task output data

Basic

This is the default if you don’t specify a type.

Structure
Basic Usage
Advanced

# Extends common structure with:
use: string|object   # Tool or agent reference
with: object        # Input parameters

- id: process_document
  type: basic
  use: tool(document-processor)
  with:
    format: pdf
    quality: high
  on_success:
    next: save_results

- id: process_document
  type: basic
  use: tool(document-processor)
  config:
    timeout: 30s
  retry:
    attempts: 3
    backoff: exponential
    interval: 1s
    max_interval: 30s
    jitter: true
  on_start:
    next: log_start
    with:
      task: "{{ id }}"
      timestamp: "{{ now() }}"
  on_success:
    next: handle_success
    with:
      result: "{{ output }}"
      duration: "{{ duration }}"
  on_error:
    condition: "{{ error.code }}"
    routes:
      rate_limited:
        next: handle_rate_limit
        retry: true
      validation_failed:
        next: handle_validation
        retry: false
      default:
        next: general_error
        retry: "{{ retry_count < 3 }}"
  on_complete:
    next: cleanup
    with:
      status: "{{ status }}"
      duration: "{{ duration }}"
      retries: "{{ retry_count }}"

Best Used For:

Single operations
API calls
Data transformations
Simple processing steps

JSON Schema

https://compozy.dev/schemas/task-basic.json

{
  "type": "object",
  "required": ["id", "type", "use"],
  "properties": {
    "id": { "type": "string" },
    "type": { "const": "basic" },
    "use": {
      "oneOf": [
        { "type": "string" },
        {
          "type": "object",
          "required": ["repo", "package"],
          "properties": {
            "repo": { "type": "string" },
            "package": { "type": "string" },
            "version": { "type": "string" }
          }
        }
      ]
    },
    "with": { "type": "object" }
  }
}

Parallel

Executes multiple operations simultaneously to improve performance

Structure
Basic Usage
Advanced

# Extends common structure with:
config:             # Parallel-specific configuration
  max_concurrent: number
  fail_fast: boolean
  timeout: duration
execute:            # List of operations
  - use: string     # Tool/agent reference
    with: object    # Operation parameters
    retry: object   # Operation-specific retry
    on_error: object # Operation error handler

- id: enrich_data
  type: parallel
  config:
    max_concurrent: 3
  execute:
    - use: tool(metadata-extractor)
    - use: tool(thumbnail-generator)
    - use: tool(keyword-analyzer)

- id: process_batch
  type: parallel
  config:
    max_concurrent: 5
    fail_fast: false
    timeout: 60s
  execute:
    - use: tool(metadata-extractor)
      retry:
        attempts: 3
      on_error:
        next: handle_metadata_error
        continue: true
    - use: tool(thumbnail-generator)
      retry:
        attempts: 2
      on_error:
        continue: true
    - use: tool(keyword-analyzer)
      retry:
        attempts: 3
      on_error:
        next: handle_analyzer_error
  on_operation_start:
    next: log_operation
    with:
      operation: "{{ operation_id }}"
  on_operation_complete:
    next: update_progress
    with:
      completed: "{{ completed_operations }}"
      total: "{{ total_operations }}"
  on_error:
    next: handle_batch_error
    with:
      failed: "{{ failed_operations }}"
      succeeded: "{{ successful_operations }}"
  on_complete:
    next: finalize
    with:
      status: "{{ status }}"
      duration: "{{ duration }}"
      success_rate: "{{ successful_operations.length / total_operations * 100 }}"

Best Used For:

Independent operations
Performance optimization
Batch processing
Multiple API calls

JSON Schema

https://compozy.dev/schemas/task-parallel.json

{
  "type": "object",
  "required": ["id", "type", "execute"],
  "properties": {
    "id": { "type": "string" },
    "type": { "const": "parallel" },
    "config": {
      "type": "object",
      "properties": {
        "max_concurrent": { "type": "number" },
        "fail_fast": { "type": "boolean" },
        "timeout": { "type": "string" }
      }
    },
    "execute": {
      "type": "array",
      "items": {
        "type": "object",
        "required": ["use"],
        "properties": {
          "use": { "type": "string" },
          "with": { "type": "object" },
          "retry": { "type": "object" },
          "on_error": { "type": "object" }
        }
      }
    }
  }
}

Decision

Routes execution based on conditions or data values.

Structure
Basic Usage
Advanced

# Extends common structure with:
condition: string    # Single condition
conditions: array   # Multiple conditions
routes: object      # Route mapping
default: object     # Default route

- id: route_document
  type: decision
  condition: "{{ tasks.analyze.output.type }}"
  routes:
    invoice: process_invoice
    contract: process_contract
    default: unknown_document

- id: complex_routing
  type: decision
  conditions:
    - condition: {{ tasks.analyze.output.confidence > 0.9 }}
      next: high_confidence
    - condition: {{ tasks.analyze.output.type == 'urgent' }}
      next: priority_processing
  default:
    next: manual_review
  on_condition_match:
    next: log_route
    with:
      condition: "{{ matched_condition }}"
      route: "{{ matched_route }}"
  on_condition_nomatch:
    next: handle_nomatch
    with:
      conditions: "{{ evaluated_conditions }}"
  on_error:
    next: handle_routing_error
    with:
      error: "{{ error }}"
  on_complete:
    next: log_decision
    with:
      route: "{{ selected_route }}"
      matched: "{{ condition_matched }}"

Best Used For:

Business logic implementation
Content-based routing
Error handling flows
Multi-path processing

JSON Schema

https://compozy.dev/schemas/task-decision.json

{
  "type": "object",
  "required": ["id", "type"],
  "properties": {
    "id": { "type": "string" },
    "type": { "const": "decision" },
    "condition": { "type": "string" },
    "conditions": {
      "type": "array",
      "items": {
        "type": "object",
        "required": ["condition", "next"],
        "properties": {
          "condition": { "type": "string" },
          "next": { "type": "string" }
        }
      }
    },
    "routes": { "type": "object" },
    "default": {
      "type": "object",
      "properties": {
        "next": { "type": "string" }
      }
    }
  }
}

Collection

Processes arrays of items efficiently, with support for batching and parallel execution.

Structure
Basic Usage
Advanced

# Extends common structure with:
input: array        # Input array
config:             # Collection-specific config
  mode: string      # parallel/sequential
  batch_size: number
  max_concurrent: number
  continue_on_error: boolean
  timeout_per_batch: duration
  timeout_per_item: duration
execute:            # Item operation
  - use: string     # Tool/agent reference
    with: object    # Operation parameters
on_item_start: object    # Item start handler
on_item_success: object  # Item success handler
on_item_error: object    # Item error handler
on_batch_start: object   # Batch start handler
on_batch_complete: object # Batch completion handler

- id: process_items
  type: collection
  input: "{{ trigger.items }}"
  config:
    mode: parallel
    batch_size: 10
  execute:
    - use: tool(item-processor)
      with:
        item: "{{ item }}"

- id: process_items
  type: collection
  input: "{{ trigger.items }}"
  config:
    mode: parallel
    batch_size: 5
    max_concurrent: 2
    continue_on_error: true
    timeout_per_item: 10s
    timeout_per_batch: 60s
  execute:
    - use: tool(item-processor)
      with:
        item: "{{ item }}"
        index: "{{ index }}"
        batch: "{{ batch_index }}"
  retry:
    attempts: 3
    backoff: exponential
    interval: 1s
  on_item_start:
    next: log_item_start
    with:
      item_index: "{{ index }}"
      batch: "{{ batch_index }}"
  on_item_success:
    next: update_progress
    with:
      progress: "{{ (index + 1) / total * 100 }}"
      success: true
  on_item_error:
    next: handle_item_error
    with:
      error: "{{ error }}"
      retry_count: "{{ retry_count }}"
      item: "{{ item }}"
  on_batch_start:
    next: log_batch_start
    with:
      batch: "{{ batch_index }}"
      size: "{{ batch_size }}"
  on_batch_complete:
    next: process_batch_results
    with:
      batch: "{{ batch_index }}"
      results: "{{ batch_output }}"
      success_rate: "{{ batch_success_rate }}"
  on_complete:
    next: finalize
    with:
      processed: "{{ processed_items }}"
      failed: "{{ failed_items }}"
      success_rate: "{{ success_rate }}"
      duration: "{{ duration }}"

Best Used For:

Processing arrays of items
Batch processing
Parallel data processing
Rate-limited operations

JSON Schema

https://compozy.dev/schemas/task-collection.json

{
  "type": "object",
  "required": ["id", "type", "input", "execute"],
  "properties": {
    "id": { "type": "string" },
    "type": { "const": "collection" },
    "input": { "type": "array" },
    "config": {
      "type": "object",
      "properties": {
        "mode": {
          "type": "string",
          "enum": ["parallel", "sequential"]
        },
        "batch_size": { "type": "number" },
        "max_concurrent": { "type": "number" },
        "continue_on_error": { "type": "boolean" },
        "timeout_per_batch": { "type": "string" },
        "timeout_per_item": { "type": "string" }
      }
    },
    "execute": {
      "type": "array",
      "items": {
        "type": "object",
        "required": ["use"],
        "properties": {
          "use": { "type": "string" },
          "with": { "type": "object" }
        }
      }
    }
  }
}

Wait

Pauses execution until specific conditions are met or a timeout occurs.

Structure
Basic Usage
Advanced

# Extends common structure with:
config:             # Wait-specific configuration
  timeout: duration
  check_interval: duration
  max_checks: number
condition: string    # Wait condition
on_timeout: object  # Timeout handler
on_check: object    # Condition check handler

- id: await_approval
  type: wait
  config:
    timeout: 24h
    check_interval: 5m
  condition: "{{ external.approval_status == 'approved' }}"

- id: await_approval
  type: wait
  config:
    timeout: 24h
    check_interval: 5m
    max_checks: 288
  condition: "{{ external.approval_status == 'approved' && external.approver_role == 'admin' }}"
  on_start:
    next: notify_pending
    with:
      item: "{{ input.item_id }}"
  on_check:
    next: log_check
    with:
      check_count: "{{ check_count }}"
      status: "{{ external.approval_status }}"
  on_timeout:
    next: handle_timeout
    with:
      waited_for: "{{ elapsed_time }}"
      checks: "{{ check_count }}"
      last_status: "{{ external.approval_status }}"
  on_success:
    next: process_approved
    with:
      approver: "{{ external.approver }}"
      timestamp: "{{ now() }}"
  on_complete:
    next: cleanup_wait
    with:
      final_status: "{{ status }}"
      duration: "{{ duration }}"

Best Used For:

Human approval flows
External service integration
Time-based operations
Rate limiting

JSON Schema

https://compozy.dev/schemas/task-wait.json

{
  "type": "object",
  "required": ["id", "type", "config", "condition"],
  "properties": {
    "id": { "type": "string" },
    "type": { "const": "wait" },
    "config": {
      "type": "object",
      "required": ["timeout", "check_interval"],
      "properties": {
        "timeout": { "type": "string" },
        "check_interval": { "type": "string" },
        "max_checks": { "type": "number" }
      }
    },
    "condition": { "type": "string" }
  }
}

Map

Transforms data between tasks while maintaining type safety.

Structure
Basic Usage
Advanced

# Extends common structure with:
config:             # Map-specific configuration
  validate_schema: boolean
  strict_mode: boolean
input: object       # Input data
mapping: object     # Output mapping
on_validation_error: object # Schema validation error handler

- id: transform_data
  type: map
  input: "{{ tasks.fetch_data.output }}"
  mapping:
    title: "{{ input.name }}"
    created_at: "{{ now() }}"

- id: transform_data
  type: map
  input: "{{ tasks.fetch_data.output }}"
  config:
    validate_schema: true
    strict_mode: true
  mapping:
    title: "{{ input.name | uppercase }}"
    slug: "{{ input.name | lowercase | replace(' ', '-') }}"
    created_at: "{{ now() }}"
    metadata:
      source: "{{ input.origin || 'unknown' }}"
      type: "{{ input.doc_type }}"
  on_start:
    next: log_transform
    with:
      input_schema: "{{ input_schema }}"
  on_validation_error:
    next: handle_validation
    with:
      errors: "{{ validation_errors }}"
  on_error:
    next: handle_mapping_error
    with:
      error: "{{ error }}"
      field: "{{ error.field }}"
  on_success:
    next: save_transformed
    with:
      result: "{{ output }}"

Best Used For:

Data transformation
Schema validation
Field mapping
Response formatting

JSON Schema

https://compozy.dev/schemas/task-map.json

{
  "type": "object",
  "required": ["id", "type", "input", "mapping"],
  "properties": {
    "id": { "type": "string" },
    "type": { "const": "map" },
    "config": {
      "type": "object",
      "properties": {
        "validate_schema": { "type": "boolean" },
        "strict_mode": { "type": "boolean" }
      }
    },
    "input": { "type": "object" },
    "mapping": { "type": "object" }
  }
}

Composite

Groups multiple operations into a single, reusable unit.

Structure
Basic Usage
Advanced

# Extends common structure with:
config:             # Composite-specific configuration
  error_handling: string  # continue/stop
  timeout: duration
tasks:              # Subtasks list
  - id: string      # Subtask identifier
    type: string    # Subtask type
    ...            # Subtask configuration
on_task_start: object    # Subtask start handler
on_task_complete: object # Subtask completion handler

- id: process_document
  type: composite
  tasks:
    - id: validate
      type: basic
      use: tool(document-validator)
    - id: process
      type: basic
      use: tool(content-processor)

- id: process_document
  type: composite
  config:
    error_handling: continue
    timeout: 5m
  tasks:
    - id: validate
      type: basic
      use: tool(document-validator)
      retry:
        attempts: 3
      on_error:
        next: handle_validation_error

    - id: process
      type: parallel
      config:
        max_concurrent: 2
      execute:
        - use: tool(text-extractor)
        - use: tool(metadata-extractor)

    - id: classify
      type: decision
      condition: "{{ tasks.validate.output.type }}"
      routes:
        invoice: process_invoice
        contract: process_contract

  on_task_start:
    next: log_task_start
    with:
      task: "{{ task_id }}"
      type: "{{ task_type }}"

  on_task_complete:
    next: log_task_complete
    with:
      task: "{{ task_id }}"
      status: "{{ task_status }}"
      duration: "{{ task_duration }}"

  on_error:
    next: handle_composite_error
    with:
      failed_task: "{{ failed_task_id }}"
      error: "{{ error }}"

  on_complete:
    next: finalize_processing
    with:
      status: "{{ status }}"
      duration: "{{ duration }}"
      results: "{{ tasks_output }}"

Best Used For:

Reusable task sequences
Complex workflows
Modular task design
Process encapsulation

JSON Schema

https://compozy.dev/schemas/task-composite.json

{
  "type": "object",
  "required": ["id", "type", "tasks"],
  "properties": {
    "id": { "type": "string" },
    "type": { "const": "composite" },
    "config": {
      "type": "object",
      "properties": {
        "error_handling": {
          "type": "string",
          "enum": ["continue", "stop"]
        },
        "timeout": { "type": "string" }
      }
    },
    "tasks": {
      "type": "array",
      "items": {
        "type": "object",
        "required": ["id", "type"],
        "properties": {
          "id": { "type": "string" },
          "type": { "type": "string" }
        }
      }
    }
  }
}

Best Practices

Choose the Right Type

Select task types based on your specific needs:

Use Basic for simple operations
Use Parallel for independent concurrent operations
Use Decision for conditional flows
Use Collection/Foreach for iterative processing
Use Wait for external dependencies
Use Map for data transformations
Use Composite for reusable workflows

Event Handling

Implement appropriate event handlers:

Use on_start for initialization
Use on_success/on_error for outcomes
Use on_complete for cleanup
Implement type-specific handlers
Log important state changes

Error Management

Implement proper error handling:

Configure retry policies
Use conditional error routing
Handle partial failures
Implement fallback mechanisms
Log error details

State Management

Monitor and manage task state:

Track execution progress
Monitor performance metrics
Log state transitions
Handle state persistence
Implement recovery mechanisms

Next Steps

Types

Learn about different task types and their use cases

Usage

Understand how to use tasks in your workflows

API reference

Review the task configuration API reference

Registry

Learn about publishing tasks to the registry

Getting Started

Core Concepts

Common Structure

Basic

Parallel

Decision

Collection

Wait

Map

Composite

Best Practices

Next Steps

Types

Usage

API reference

Registry

​Common Structure

​Basic

​Parallel

​Decision

​Collection

​Wait

​Map

​Composite

​Best Practices

​Next Steps

Types

Usage

API reference

Registry

Common Structure

Basic

Parallel

Decision

Collection

Wait

Map

Composite

Best Practices

Next Steps