Usage

Structure

Tasks in Compozy follow a consistent structure that supports both simple operations and complex flow control. For detailed information about task types and their specific structures, see the Task Types documentation.

tasks:
  - id: string           # Unique identifier for the task
    name: string         # Optional human-readable name
    type: string         # Task type (basic, parallel, collection, etc.)
    use: string|object   # Task implementation reference
    import: string       # Path to external task definition
    config: object       # Task type-specific configuration
    memory: string       # Memory store identifier
    schema:             # Optional schema definition
      input: object     # Input schema for task parameters
    with: object        # Task-specific configuration
    env: object         # Task-specific environment variables
    on_start: object    # Task start handler
    on_success: object  # Success handler
    on_error: object    # Error handler
    on_complete: object # Completion handler (runs after success/error)
    retry:             # Retry configuration
      attempts: number
      backoff: string
      interval: duration

Task Types

Basic

Simple, single-operation tasks for API calls, transformations, and processing steps.

Parallel

Execute multiple operations simultaneously with configurable concurrency limits.

Decision

Route execution based on conditions with support for complex business logic.

Collection

Process arrays of items sequentially or in parallel with advanced execution patterns.

Wait

Pause execution until specific conditions are met or timeout occurs.

Map

Transform data between tasks with type-safe field mapping.

Composite

Group multiple operations into reusable units.

Event Handlers

Tasks support various event handlers to control flow and handle state transitions:

tasks:
  - id: process_document
    type: basic
    use: tool(document-processor)
    on_start:
      next: log_start
      with:
        task: "{{ id }}"
        timestamp: "{{ now() }}"
    on_success:
      next: handle_success
      with:
        result: "{{ output }}"
    on_error:
      next: handle_error
      with:
        error: "{{ error }}"
    on_complete:
      next: cleanup
      with:
        status: "{{ status }}"
        duration: "{{ duration }}"

tasks:
  - id: process_batch
    type: collection
    input: "{{ trigger.items }}"
    config:
      mode: parallel
      batch_size: 10
    execute:
      - use: tool(processor)
    on_item_start:
      next: log_item_start
      with:
        index: "{{ index }}"
    on_item_success:
      next: update_progress
      with:
        progress: "{{ (index + 1) / total * 100 }}"
    on_item_error:
      next: handle_item_error
      with:
        item: "{{ item }}"
        error: "{{ error }}"
    on_batch_complete:
      next: process_batch_results
      with:
        batch: "{{ batch_index }}"
        results: "{{ batch_output }}"

tasks:
  - id: route_content
    type: decision
    condition: "{{ tasks.analyze.output.type }}"
    on_condition_match:
      next: log_route
      with:
        route: "{{ matched_route }}"
    on_condition_nomatch:
      next: handle_unknown
      with:
        type: "{{ tasks.analyze.output.type }}"
    routes:
      article: process_article
      video: process_video
      default: unknown_type

Error Handling & Recovery

Tasks support sophisticated error handling and recovery mechanisms:

tasks:
  - id: api_call
    type: basic
    use: tool(api-client)
    retry:
      attempts: 3
      backoff: exponential
      interval: 1s
    on_error:
      next: handle_failure
      with:
        attempts: "{{ retry_count }}"
        error: "{{ error }}"

tasks:
  - id: process_document
    type: basic
    use: tool(document-processor)
    retry:
      attempts: 5
      backoff: exponential
      interval: 2s
      max_interval: 30s
      jitter: true
    on_error:
      condition: "{{ error.code }}"
      routes:
        rate_limited:
          next: handle_rate_limit
          retry: true
          with:
            wait_time: "{{ error.retry_after }}"
        invalid_input:
          next: handle_validation
          retry: false
        default:
          next: general_error
          retry: "{{ retry_count < 3 }}"

tasks:
  - id: batch_process
    type: parallel
    config:
      max_concurrent: 3
      fail_fast: false
    execute:
      - use: tool(processor-1)
        retry:
          attempts: 3
        on_error:
          next: handle_processor1_error
      - use: tool(processor-2)
        retry:
          attempts: 2
        on_error:
          continue: true
    on_error:
      next: handle_batch_error
      with:
        failed: "{{ failed_operations }}"
        succeeded: "{{ successful_operations }}"

Schema Definition

Schemas in Compozy use JSON Schema format to validate task inputs. Note that outputs are handled by the executed components (agents, MCPs, or tools):

tasks:
  - id: analyze_content
    type: basic
    use: tool(content-analyzer)
    schema:
      input:
        type: object
        properties:
          language:
            type: string
            enum: [en, es, fr]
            default: en
    with:
      model: basic
      language: "{{ input.language }}"

Environment Variables

Configure task-specific settings and secrets:

tasks:
  - id: api_call
    type: basic
    use: tool(api-client)
    env:
      API_KEY: "{{ env.SERVICE_KEY }}"
      TIMEOUT: "30"
    with:
      endpoint: /data

Task Imports

Reuse and customize tasks from external files:

tasks:
  - id: custom_processor
    import: ./tasks/content-processor.yaml
    env:
      OPTIMIZATION_LEVEL: low
    with:
      platform: blog

State Management

Tasks maintain state information that can be accessed in event handlers:

tasks:
  - id: process_data
    type: basic
    use: tool(data-processor)
    on_complete:
      next: log_metrics
      with:
        task_id: "{{ id }}"
        status: {{ status }}        # success, error, timeout
        duration: {{ duration }}    # Task duration in ms
        start_time: "{{ start_time }}"
        end_time: "{{ end_time }}"
        retries: "{{ retry_count }}"
        error: {{ error }}          # Error object if failed

Memory Integration

Tasks can interact with workflow-level memory systems for state persistence and context management. First, define memory at the workflow level, then reference it in tasks:

# Define memory in workflow
memory:
  chat_memory:
    id: chat_memory
    storage:
      use: compozy/storage:postgres
    vector:
      use: compozy/vector:pinecone
      dimensions: 1536
    options:
      lastMessages: 10
      semanticRecall:
        enabled: true

# Use memory in tasks
tasks:
  - id: chat_task
    type: basic
    use: tool(chat-processor)
    memory: chat_memory    # Reference to workflow memory
    with:
      context: "{{ memory.chat_task.current_thread }}"
      history: "{{ memory.chat_task.last_messages }}"
      relevant: "{{ memory.chat_task.semantic_search }}"

Memory operations in tasks support:

Reading from and writing to memory stores
Semantic search across stored content
Thread and conversation management
Context windowing and retrieval
Vector similarity search
Memory persistence and cleanup

Best Practices

Task Design

Choose appropriate task types for your needs
Keep tasks focused and single-purpose
Use parallel tasks judiciously

Configuration

Set reasonable defaults for concurrent operations
Use environment variables for sensitive data
Document task requirements

Event Handling

Implement appropriate handlers for critical events
Use on_complete for cleanup operations
Log important state transitions

Error Management

Configure retries for transient failures
Use conditional error routing
Implement graceful degradation

State Management

Monitor task state changes
Implement appropriate logging
Use state information for debugging

Recovery Strategies

Configure appropriate retry policies
Implement fallback mechanisms
Handle partial failures in parallel tasks

Next Steps

Types

Learn about different task types and their use cases

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

Usage

Structure

Task Types

Basic

Parallel

Decision

Collection

Wait

Map

Composite

Event Handlers

Error Handling & Recovery

Schema Definition

Environment Variables

Task Imports

State Management

Memory Integration

Best Practices

Next Steps

Types

Usage

API reference

Registry

​Structure

​Task Types

Basic

Parallel

Decision

Collection

Wait

Map

Composite

​Event Handlers

​Error Handling & Recovery

​Schema Definition

​Environment Variables

​Task Imports

​State Management

​Memory Integration

​Best Practices

​Next Steps

Types

Usage

API reference

Registry

Structure

Task Types

Event Handlers

Error Handling & Recovery

Schema Definition

Environment Variables

Task Imports

State Management

Memory Integration

Best Practices

Next Steps