@Workflow

The @Workflow decorator defines a DAG-based multi-step pipeline that composes jobs with dependencies, conditions, and parallel execution.

Basic Usage

import { Workflow } from '@frontmcp/sdk';

@Workflow({
  name: 'data-pipeline',
  description: 'Extract, transform, and load data',
  steps: [
    { id: 'extract', jobName: 'extract-data' },
    { id: 'transform', jobName: 'transform-data', dependsOn: ['extract'] },
    { id: 'load', jobName: 'load-data', dependsOn: ['transform'] },
  ],
})
class DataPipelineWorkflow {}

Workflows are declarative — the class body is empty. All configuration goes in the decorator metadata.

Signature

function Workflow(opts: WorkflowMetadata): ClassDecorator

Configuration Options

Required Properties

Property	Type	Description
name	string	Unique workflow identifier
steps	WorkflowStep[]	Step definitions (minimum 1)

Optional Properties

Property	Type	Default	Description
description	string	—	Workflow description
id	string	name	Stable identifier for tracking
trigger	'manual' | 'webhook' | 'event'	'manual'	How the workflow is triggered
webhook	WorkflowWebhookConfig	—	Webhook configuration
timeout	number	600000	Max total execution time in ms (10 min)
maxConcurrency	number	5	Max parallel step concurrency
tags	string[]	—	Categorization tags
labels	Record<string, string>	—	Key-value labels
hideFromDiscovery	boolean	false	Hide from list-workflows
permissions	JobPermission[]	—	RBAC permission rules
inputSchema	ZodShape	—	Workflow-level input schema
outputSchema	ZodShape	—	Workflow-level output schema

WorkflowStep

Property	Type	Default	Description
id	string	—	Required. Unique step identifier
jobName	string	—	Required. Name of the registered job to execute
dependsOn	string[]	[]	Step IDs that must complete first
input	object | function	Workflow input	Static input or dynamic callback
condition	(steps) => boolean	—	Skip step if returns false
continueOnError	boolean	false	Continue workflow if step fails
timeout	number	Job default	Per-step timeout override in ms
retry	JobRetryConfig	Job default	Per-step retry override (maxAttempts, backoffMs, backoffMultiplier, maxBackoffMs)

WorkflowWebhookConfig

Property	Type	Default	Description
path	string	/workflows/webhook/{name}	Custom webhook path
secret	string	—	Webhook secret for validation. Load from environment variables or a secrets manager — never hardcode in source.
methods	('GET' | 'POST')[]	['POST']	Allowed HTTP methods

Examples

Linear Workflow

@Workflow({
  name: 'onboarding',
  description: 'User onboarding pipeline',
  steps: [
    { id: 'create-account', jobName: 'create-account' },
    {
      id: 'send-welcome',
      jobName: 'send-email',
      dependsOn: ['create-account'],
      input: (steps) => ({
        to: steps.get('create-account').outputs.email,
        subject: 'Welcome!',
        body: 'Welcome to our platform.',
      }),
    },
    {
      id: 'setup-defaults',
      jobName: 'setup-defaults',
      dependsOn: ['create-account'],
      input: (steps) => ({
        userId: steps.get('create-account').outputs.userId,
      }),
    },
  ],
})
class OnboardingWorkflow {}

Parallel Workflow

@Workflow({
  name: 'multi-analysis',
  description: 'Run multiple analyses in parallel',
  maxConcurrency: 3,
  steps: [
    { id: 'fetch', jobName: 'fetch-data' },
    { id: 'sentiment', jobName: 'analyze-sentiment', dependsOn: ['fetch'] },
    { id: 'entities', jobName: 'extract-entities', dependsOn: ['fetch'] },
    { id: 'topics', jobName: 'detect-topics', dependsOn: ['fetch'] },
    {
      id: 'report',
      jobName: 'generate-report',
      dependsOn: ['sentiment', 'entities', 'topics'],
      input: (steps) => ({
        sentiment: steps.get('sentiment').outputs,
        entities: steps.get('entities').outputs,
        topics: steps.get('topics').outputs,
      }),
    },
  ],
})
class MultiAnalysisWorkflow {}

Conditional Workflow

@Workflow({
  name: 'conditional-pipeline',
  description: 'Pipeline with conditional steps',
  steps: [
    { id: 'validate', jobName: 'validate-input' },
    {
      id: 'premium-processing',
      jobName: 'premium-process',
      dependsOn: ['validate'],
      condition: (steps) =>
        steps.get('validate').outputs.tier === 'premium',
    },
    {
      id: 'standard-processing',
      jobName: 'standard-process',
      dependsOn: ['validate'],
      condition: (steps) =>
        steps.get('validate').outputs.tier !== 'premium',
    },
    {
      id: 'finalize',
      jobName: 'finalize',
      dependsOn: ['premium-processing', 'standard-processing'],
    },
  ],
})
class ConditionalPipelineWorkflow {}

A step whose condition returns false is treated as skipped and satisfies downstream dependsOn dependencies. This means steps that depend on a skipped step will still run.

Function-Based Alternative

import { workflow } from '@frontmcp/sdk';

const DataPipeline = workflow({
  name: 'data-pipeline',
  steps: [
    { id: 'extract', jobName: 'extract-data' },
    { id: 'transform', jobName: 'transform-data', dependsOn: ['extract'] },
    { id: 'load', jobName: 'load-data', dependsOn: ['transform'] },
  ],
});

Related

WorkflowRegistry

Workflow registry API

Workflows Guide

Workflows documentation

@Job

Define jobs

DirectClient

Programmatic execution