Advanced Workflow Patterns for Virtual MCP Composite Tools

April 21, 2026 · View on GitHub

Overview

This guide covers advanced workflow patterns and best practices for Virtual MCP Composite Tools, including parallel execution, dependency management, error handling strategies, and state management.

Parallel Execution with DAG
Step Dependencies
Advanced Error Handling
Workflow State Management
Performance Optimization
Best Practices
Common Patterns
ForEach Iteration Patterns

Parallel Execution with DAG

Virtual MCP Composite Tools use a Directed Acyclic Graph (DAG) execution model that automatically executes independent steps in parallel while respecting dependencies.

How DAG Execution Works

Execution Levels: Steps are organized into levels based on dependencies
Parallel Within Levels: All steps in the same level execute concurrently
Sequential Across Levels: Each level waits for the previous level to complete
Automatic Optimization: The system automatically determines optimal parallelization

Example: Parallel Data Fetching

apiVersion: toolhive.stacklok.dev/v1beta1
kind: VirtualMCPCompositeToolDefinition
metadata:
  name: incident-investigation
spec:
  name: investigate_incident
  description: Investigate incident by gathering logs, metrics, and traces in parallel
  parameters:
    type: object
    properties:
      incident_id:
        type: string
        description: The incident identifier
      time_range:
        type: string
        description: Time range for data collection
    required:
      - incident_id
      - time_range
  steps:
    # Level 1: These three steps run in parallel (no dependencies)
    - id: fetch_logs
      type: tool
      tool: splunk.fetch_logs
      arguments:
        incident_id: "{{.params.incident_id}}"
        time_range: "{{.params.time_range}}"

    - id: fetch_metrics
      type: tool
      tool: datadog.fetch_metrics
      arguments:
        incident_id: "{{.params.incident_id}}"
        time_range: "{{.params.time_range}}"

    - id: fetch_traces
      type: tool
      tool: jaeger.fetch_traces
      arguments:
        incident_id: "{{.params.incident_id}}"
        time_range: "{{.params.time_range}}"

    # Level 2: Waits for all Level 1 steps to complete
    - id: correlate
      type: tool
      tool: analysis.correlate_data
      dependsOn: [fetch_logs, fetch_metrics, fetch_traces]
      arguments:
        logs: "{{.steps.fetch_logs.output}}"
        metrics: "{{.steps.fetch_metrics.output}}"
        traces: "{{.steps.fetch_traces.output}}"

    # Level 3: Waits for Level 2
    - id: create_report
      type: tool
      tool: jira.create_issue
      dependsOn: [correlate]
      arguments:
        title: "Incident {{.params.incident_id}} Analysis"
        body: "{{.steps.correlate.output.summary}}"

Execution Timeline:

Time    Level 1 (Parallel)              Level 2         Level 3
0ms     fetch_logs    ─┐
0ms     fetch_metrics ─┼─> correlate ──> create_report
0ms     fetch_traces  ─┘

Performance: Fetching 3 data sources takes ~1x time instead of 3x (sequential).

Step Dependencies

Use the dependsOn field to define explicit dependencies between steps.

Syntax

steps:
  - id: step_name
    dependsOn: [dependency1, dependency2, ...]
    # ... rest of step config

Dependency Rules

Multiple Dependencies: Step waits for ALL dependencies to complete
Transitive Dependencies: Automatically handled (A→B→C works as expected)
Cycle Detection: Circular dependencies are detected and rejected at validation time
Missing Dependencies: Referencing non-existent steps fails validation

Example: Diamond Pattern

steps:
  # Level 1
  - id: fetch_data
    type: tool
    tool: api.fetch

  # Level 2: Both depend on fetch_data, can run in parallel
  - id: process_left
    type: tool
    tool: transform.left
    dependsOn: [fetch_data]

  - id: process_right
    type: tool
    tool: transform.right
    dependsOn: [fetch_data]

  # Level 3: Waits for both Level 2 steps
  - id: merge_results
    type: tool
    tool: combine.merge
    dependsOn: [process_left, process_right]

Execution Graph:

       fetch_data
       /        \
process_left  process_right
       \        /
      merge_results

Accessing Dependency Outputs

Use template syntax to access outputs from dependencies:

- id: analyze
  dependsOn: [fetch_logs, fetch_metrics]
  arguments:
    # Access specific fields from dependency outputs
    log_count: "{{.steps.fetch_logs.output.count}}"
    metric_avg: "{{.steps.fetch_metrics.output.average}}"

    # Pass entire output object
    raw_data: "{{.steps.fetch_logs.output}}"

Template System Overview

Workflows use Go's text/template with these additional context variables and functions:

Context Variables:

.params.* - Input parameters
.steps.<id>.output - Step outputs
.steps.<id>.status - Step status (completed, failed, skipped, running)
.steps.<id>.error - Step error messages (if failed)
.vars.* - Workflow-scoped variables

Custom Functions:

json - JSON encode a value
fromJson - Parse a JSON string into a value (useful when MCP servers return JSON as text content)
quote - Quote a string value

Built-in Functions: All Go template built-ins are available (eq, ne, lt, le, gt, ge, and, or, not, index, len, range, with, printf, etc.)

Example with Advanced Features:

- id: conditional_step
  dependsOn: [fetch_data]
  condition: "{{and (eq .steps.fetch_data.status \"completed\") (gt (len .steps.fetch_data.output.items) 0)}}"
  arguments:
    message: "{{printf \"Found %d items\" (len .steps.fetch_data.output.items)}}"
    data: "{{json .steps.fetch_data.output}}"

Step Output Format

Backend tools can return results in two formats:

Structured Content (Object Response): When a tool returns structured content (an object), fields are directly accessible via .steps.<id>.output.<field>:

# Tool returns: {"user": {"name": "Alice", "email": "alice@example.com"}, "status": "active"}
arguments:
  name: "{{.steps.get_user.output.user.name}}"
  email: "{{.steps.get_user.output.user.email}}"
  status: "{{.steps.get_user.output.status}}"

Unstructured Content (Text Response): When a tool returns text content, it is stored under the text key:

# Tool returns: "Operation completed successfully"
arguments:
  result: "{{.steps.run_command.output.text}}"

Note: Structured content must be an object. Arrays, primitives, or other non-object types fall back to unstructured content handling.

Numeric Comparisons

All numeric values from JSON are float64. Use float literals in comparisons:

# Correct: float literal
condition: '{{if gt .steps.get_count.output.total 100.0}}true{{else}}false{{end}}'

# Incorrect: integer literal causes type mismatch
condition: '{{if gt .steps.get_count.output.total 100}}true{{else}}false{{end}}'

Advanced Error Handling

Configure sophisticated error handling at both workflow and step levels.

Workflow-Level Failure Modes

Set the workflow's failureMode to control global error behavior:

spec:
  name: resilient_workflow
  failureMode: continue  # Options: abort, continue
  steps:
    # ...

Failure Modes:

Mode	Behavior	Use Case
`abort`	Stop immediately on first error (default)	Critical workflows where partial completion is dangerous
`continue`	Log errors but continue executing remaining steps	Data collection where some failures are acceptable

Step-Level Error Handling

Override workflow-level behavior for specific steps:

steps:
  - id: optional_notification
    type: tool
    tool: slack.notify
    onError:
      action: continue  # Don't fail workflow if Slack is down

  - id: critical_payment
    type: tool
    tool: stripe.charge
    # Inherits workflow failureMode (defaults to abort)

Retry Logic with Exponential Backoff

Configure automatic retries for transient failures:

steps:
  - id: fetch_external_api
    type: tool
    tool: external.fetch_data
    onError:
      action: retry
      maxRetries: 3           # Maximum 3 retries (4 total attempts)

Retry Behavior:

Exponential Backoff: Delay increases by 1.5x each retry with ±50% randomization (1s → ~1.5s → ~2.25s → ~3.4s...), capped at 60 seconds
Maximum Retries: Capped at 10 (configurable per step)
Context Aware: Respects workflow timeout (won't retry if timeout exceeded)
Error Propagation: Final error includes retry count in metadata

Example: Combining Error Strategies

apiVersion: toolhive.stacklok.dev/v1beta1
kind: VirtualMCPCompositeToolDefinition
metadata:
  name: robust-deployment
spec:
  name: deploy_with_resilience
  failureMode: abort  # Fail fast by default
  steps:
    # Retry transient network issues
    - id: fetch_artifact
      type: tool
      tool: s3.download
      onError:
        action: retry
        maxRetries: 3

    - id: deploy
      type: tool
      tool: kubernetes.apply
      dependsOn: [fetch_artifact]
      # Critical: uses workflow failureMode (abort)

    # Optional post-deployment tasks
    - id: notify_slack
      type: tool
      tool: slack.notify
      dependsOn: [deploy]
      onError:
        action: continue  # Don't fail if notification fails

    - id: update_dashboard
      type: tool
      tool: grafana.update
      dependsOn: [deploy]
      onError:
        action: continue

Workflow State Management

Virtual MCP tracks workflow execution state for monitoring, debugging, and cancellation.

State Tracking

The workflow engine automatically maintains state including:

Workflow ID: Unique identifier (UUID) for each execution
Status: Current state (pending, running, completed, failed, cancelled, timed_out)
Completed Steps: List of successfully completed steps
Step Results: Outputs and timing for each step
Pending Elicitations: User interactions awaiting response
Timestamps: Start time, end time, last update time

Workflow Timeout

Configure maximum execution time to prevent runaway workflows:

spec:
  name: time_sensitive_workflow
  timeout: 30m  # 30 minutes maximum
  steps:
    - id: long_running_task
      type: tool
      tool: data.process
      timeout: 5m  # Individual step timeout

Timeout Behavior:

Workflow timeout applies to entire execution
Step timeouts apply to individual steps
Timeouts trigger graceful cancellation (context.DeadlineExceeded)
State is saved with timed_out status

Timeout Precedence:

Workflow Timeout: 30m
  ├─ Step 1 (5m timeout)   ✓ Respects both
  ├─ Step 2 (10m timeout)  ✓ Respects both
  └─ Step 3 (40m timeout)  ✗ Limited by workflow timeout

State Persistence

In-Memory State Store (Default):

Suitable for single-instance deployments
Automatic cleanup of completed workflows (configurable)
Thread-safe for parallel step execution
Workflow status available programmatically via the Composer Go API

Future: Distributed State Store (Redis/Database):

For multi-instance deployments
Workflow resumption after restart
Cross-instance workflow visibility

Monitoring Workflow State

Workflow status is currently available programmatically through the Composer Go API:

// Get workflow status
status, err := composer.GetWorkflowStatus(ctx, workflowID)
if err != nil {
    // Handle error
}

// Check workflow state
fmt.Printf("Workflow ID: %s\n", status.WorkflowID)
fmt.Printf("Status: %s\n", status.Status)
fmt.Printf("Started: %s\n", status.StartTime)
fmt.Printf("Duration: %s\n", status.Duration)
fmt.Printf("Completed Steps: %v\n", status.CompletedSteps)

Note: HTTP REST API endpoints for external workflow monitoring are planned for a future release.

Performance Optimization

Concurrency Limits

The DAG executor limits parallel execution to prevent resource exhaustion:

// Default: 10 concurrent steps maximum
// Configurable in workflow engine initialization

Tuning Recommendations:

I/O-bound workflows: Higher concurrency (10-20 steps)
CPU-bound workflows: Lower concurrency (2-5 steps)
Memory-intensive: Monitor and adjust based on capacity

Execution Statistics

The system tracks execution metrics:

stats := {
  "total_levels":      3,     // Number of execution levels
  "total_steps":       8,     // Total steps in workflow
  "max_parallelism":   3,     // Max steps in any level
  "sequential_steps":  2,     // Steps that run alone
}

Optimization Strategies

Minimize Dependencies: Reduce dependsOn where possible
Group Related Steps: Steps with similar execution time work well in same level
Split Large Steps: Break monolithic steps into parallel sub-steps
Use Conditional Execution: Skip unnecessary steps with condition field

Example: Optimized Data Pipeline

# Before: Sequential (9 seconds total)
steps:
  - id: fetch1     # 3s
  - id: fetch2     # 3s
  - id: fetch3     # 3s

# After: Parallel (3 seconds total)
steps:
  - id: fetch1     # 3s ─┐
  - id: fetch2     # 3s ─┼─ All run in parallel
  - id: fetch3     # 3s ─┘

Best Practices

1. Design for Parallelism

✅ DO: Identify independent operations

steps:
  - id: notify_slack
  - id: notify_email
  - id: notify_pagerduty
  # All independent, run in parallel

❌ DON'T: Create unnecessary dependencies

steps:
  - id: notify_slack
  - id: notify_email
    dependsOn: [notify_slack]  # Unnecessary!
  - id: notify_pagerduty
    dependsOn: [notify_email]  # Creates false sequencing

2. Declare All Dependencies Explicitly

✅ DO: Be explicit about data dependencies

- id: aggregate
  dependsOn: [fetch_logs, fetch_metrics]  # Clear intent
  arguments:
    logs: "{{.steps.fetch_logs.output}}"
    metrics: "{{.steps.fetch_metrics.output}}"

❌ DON'T: Rely on implicit ordering

# This will fail! process_data tries to access fetch_data output,
# but they run in parallel without depends_on
- id: fetch_data
  type: tool
  tool: api.fetch

- id: process_data  # ERROR: fetch_data may not have completed!
  type: tool
  tool: transform.process
  arguments:
    data: "{{.steps.fetch_data.output}}"

3. Use Appropriate Error Handling

✅ DO: Match error handling to business requirements

steps:
  # Critical: must succeed
  - id: charge_payment
    type: tool
    tool: stripe.charge
    # Uses default abort behavior

  # Optional: nice to have
  - id: send_receipt
    type: tool
    tool: email.send
    dependsOn: [charge_payment]
    onError:
      action: continue

4. Set Realistic Timeouts

✅ DO: Set timeouts based on SLAs

spec:
  timeout: 5m  # API SLA: 5 minutes
  steps:
    - id: external_api
      timeout: 30s  # Individual operation: 30 seconds
      onError:
        action: retry
        maxRetries: 3

5. Keep Steps Focused

✅ DO: One responsibility per step

steps:
  - id: fetch_user
    tool: db.query_user
  - id: validate_permissions
    tool: auth.check_permissions
    dependsOn: [fetch_user]
  - id: perform_action
    tool: api.execute
    dependsOn: [validate_permissions]

❌ DON'T: Combine unrelated operations

steps:
  - id: do_everything
    tool: monolith.execute  # Hard to parallelize, test, debug

Common Patterns

Pattern 1: Fan-Out / Fan-In

Parallel execution followed by aggregation.

steps:
  # Fan-out: Parallel data collection
  - id: fetch_source_a
    type: tool
    tool: api.fetch_a

  - id: fetch_source_b
    type: tool
    tool: api.fetch_b

  - id: fetch_source_c
    type: tool
    tool: api.fetch_c

  # Fan-in: Aggregate results
  - id: aggregate
    type: tool
    tool: analysis.combine
    dependsOn: [fetch_source_a, fetch_source_b, fetch_source_c]

Use Cases: Data aggregation, multi-source reporting, distributed search

Pattern 2: Pipeline with Parallel Stages

Sequential stages with parallel operations within each stage.

steps:
  # Stage 1: Fetch raw data
  - id: fetch
    type: tool
    tool: api.fetch

  # Stage 2: Parallel transformations
  - id: transform_format_a
    type: tool
    tool: transform.to_format_a
    dependsOn: [fetch]

  - id: transform_format_b
    type: tool
    tool: transform.to_format_b
    dependsOn: [fetch]

  # Stage 3: Parallel storage
  - id: store_warehouse
    type: tool
    tool: warehouse.store
    dependsOn: [transform_format_a]

  - id: store_cache
    type: tool
    tool: cache.store
    dependsOn: [transform_format_b]

Use Cases: ETL pipelines, data transformation, multi-target deployments

Pattern 3: Conditional Parallel Execution

Use conditions to selectively enable parallel branches.

steps:
  - id: fetch_user
    type: tool
    tool: db.query_user

  # Parallel conditional branches
  - id: notify_slack
    type: tool
    tool: slack.notify
    dependsOn: [fetch_user]
    condition: "{{.steps.fetch_user.output.preferences.slack_enabled}}"

  - id: notify_email
    type: tool
    tool: email.send
    dependsOn: [fetch_user]
    condition: "{{.steps.fetch_user.output.preferences.email_enabled}}"

  - id: notify_sms
    type: tool
    tool: sms.send
    dependsOn: [fetch_user]
    condition: "{{.steps.fetch_user.output.preferences.sms_enabled}}"

Use Cases: Multi-channel notifications, feature flags, A/B testing

Pattern 4: Retry with Fallback

Try primary service, retry on failure, fall back to secondary.

steps:
  - id: try_primary
    type: tool
    tool: primary_api.call
    onError:
      action: retry
      maxRetries: 2

  - id: use_fallback
    type: tool
    tool: fallback_api.call
    dependsOn: [try_primary]
    condition: "{{ne .steps.try_primary.status \"completed\"}}"

Use Cases: High availability, disaster recovery, service degradation

Pattern 5: Default Results for Skippable Steps

Use defaultResults to provide fallback values when conditional or error-prone steps may not produce output.

steps:
  - id: fetch_core_data
    type: tool
    tool: db.query
    arguments:
      id: "{{.params.entity_id}}"

  # Optional enrichment - may be skipped based on condition
  - id: enrich_data
    type: tool
    tool: enrichment.service
    dependsOn: [fetch_core_data]
    condition: "{{.params.enable_enrichment}}"
    arguments:
      data: "{{.steps.fetch_core_data.output.text}}"
    # Fallback when step is skipped
    defaultResults:
      text: "{\"enriched\": false, \"source\": \"none\"}"

  # External API that may fail
  - id: external_lookup
    type: tool
    tool: external.api
    dependsOn: [fetch_core_data]
    onError:
      action: continue
    # Fallback when step fails
    defaultResults:
      text: "{\"available\": false}"

  # Aggregate results - works regardless of whether optional steps ran
  - id: aggregate
    type: tool
    tool: processor.combine
    dependsOn: [fetch_core_data, enrich_data, external_lookup]
    arguments:
      core: "{{.steps.fetch_core_data.output.text}}"
      enrichment: "{{.steps.enrich_data.output.text}}"
      external: "{{.steps.external_lookup.output.text}}"

Key Points:

defaultResults provides fallback output when a step is skipped or fails with continue
Keys must match the output fields referenced by downstream templates
Backend tools return text under the text key
Validation ensures defaultResults is specified when required

Use Cases: Graceful degradation, optional features, resilient pipelines

Pattern 6: Parallel Validation

Validate multiple aspects concurrently before proceeding.

steps:
  # Parallel validations
  - id: validate_schema
    type: tool
    tool: validation.check_schema

  - id: validate_permissions
    type: tool
    tool: auth.check_permissions

  - id: validate_quota
    type: tool
    tool: billing.check_quota

  # Proceed only if all validations pass
  - id: execute_action
    type: tool
    tool: api.execute
    dependsOn: [validate_schema, validate_permissions, validate_quota]

Use Cases: Pre-flight checks, authorization, resource validation

Troubleshooting

Debugging Parallel Execution

Problem: Step fails with "output not found" error

Solution: Add dependency to ensure step completes first

# Before (broken)
- id: process
  arguments:
    data: "{{.steps.fetch.output}}"  # May run before fetch completes!

# After (fixed)
- id: process
  dependsOn: [fetch]  # Explicit dependency
  arguments:
    data: "{{.steps.fetch.output}}"

Detecting Circular Dependencies

Problem: Workflow validation fails with "circular dependency detected"

Solution: Review dependsOn chains for cycles

# Circular dependency (invalid)
- id: step_a
  dependsOn: [step_b]
- id: step_b
  dependsOn: [step_a]  # ❌ Cycle!

# Fixed (valid)
- id: step_a
- id: step_b
  dependsOn: [step_a]  # ✓ Linear dependency

Performance Issues

Problem: Workflow slower than expected despite parallel execution

Checklist:

Verify steps actually run in parallel (check execution levels)
Check for unnecessary dependsOn constraints
Review concurrency limits (may be throttling)
Profile individual step execution times
Consider network/external service bottlenecks

Migration from Sequential to Parallel

If you have existing sequential workflows, here's how to migrate:

Step 1: Identify Independent Steps

Review your workflow and identify steps that:

Don't use outputs from other steps
Access different external services
Perform independent validations or checks

Step 2: Remove Unnecessary Dependencies

# Before: Implicit sequential execution
steps:
  - id: step1
  - id: step2
  - id: step3

# After: Explicit independence (parallel)
steps:
  - id: step1  # No depends_on = runs in parallel
  - id: step2  # No depends_on = runs in parallel
  - id: step3  # No depends_on = runs in parallel

Step 3: Add Required Dependencies

# If step3 actually needs step1's output:
steps:
  - id: step1
  - id: step2
  - id: step3
    dependsOn: [step1]  # Explicit data dependency
    arguments:
      data: "{{.steps.step1.output}}"

Step 4: Test Incrementally

Start with one parallel group
Validate outputs and timing
Gradually parallelize more steps
Monitor for race conditions or dependency issues

The forEach step type iterates over a collection produced by a previous step, executing an inner tool step for each item. The forEach step is a single node in the DAG -- its internal parallelism is self-managed.

Basic forEach: Vulnerability Scanning

steps:
  - id: get_packages
    type: tool
    tool: oci-registry.get_image_config
    arguments:
      image_ref: "{{.params.image}}"

  - id: check_each_vuln
    type: forEach
    collection: "{{json .steps.get_packages.output.packages}}"
    itemVar: pkg
    maxParallel: 5
    step:
      type: tool
      tool: osv.query_vulnerability
      arguments:
        package_name: "{{.forEach.pkg.name}}"
        ecosystem: "{{.forEach.pkg.ecosystem}}"
        version: "{{.forEach.pkg.version}}"
    dependsOn: [get_packages]
    onError:
      action: continue    # Skip failed items, don't abort

  - id: summarize
    type: tool
    tool: reporter.summarize
    arguments:
      total: "{{.steps.check_each_vuln.output.count}}"
      failed: "{{.steps.check_each_vuln.output.failed}}"
      results: "{{json .steps.check_each_vuln.output.iterations}}"
    dependsOn: [check_each_vuln]

forEach with Error Abort

When any iteration fails, abort immediately and fail the workflow:

- id: deploy_each
  type: forEach
  collection: "{{json .steps.get_targets.output.targets}}"
  itemVar: target
  maxParallel: 1            # Sequential deployment
  step:
    type: tool
    tool: kubectl.apply
    arguments:
      cluster: "{{.forEach.target.cluster}}"
      manifest: "{{.params.manifest}}"
  dependsOn: [get_targets]
  # Default onError is abort -- any failure stops remaining iterations

forEach Limits and Safety

Setting	Default	Hard Cap	Description
`maxIterations`	100	1000	Max collection items
`maxParallel`	10 (DAG default)	50	Concurrent iterations

The forEach step's timeout (inherited from step-level timeout) applies to the entire iteration set.

Additional Resources

VirtualMCPCompositeToolDefinition Guide - Basic workflow concepts
Architecture Documentation - System architecture and design
Operator Guide - Kubernetes deployment

Summary

Key takeaways for advanced workflows:

✅ Embrace Parallelism: Design workflows for concurrent execution
✅ Explicit Dependencies: Always declare data dependencies with dependsOn
✅ Error Resilience: Use retry for transient failures, continue for optional steps
✅ Set Timeouts: Prevent runaway workflows with appropriate timeouts
✅ Monitor State: Track workflow execution for debugging and optimization

The DAG execution model provides automatic parallelization while maintaining correctness through dependency management. Follow these patterns and practices to build efficient, reliable, and maintainable workflows.