Skip to main content
The Workflow Generation Tools enable agents to create, validate, and optimize automated workflows from natural language descriptions. These tools integrate with OpsHub’s workflow engine and Temporal Cloud for orchestrated, scalable execution.

​
Overview

Workflow generation transforms human intent into executable workflow definitions through intelligent parsing and template matching. The system manages the entire workflow lifecycle: from initial generation through validation to optimization and execution.

​
Key Capabilities

  • Natural Language Processing: Convert business descriptions to structured workflows
  • Template Library Integration: Leverage pre-built patterns for common operations
  • Dependency Resolution: Automatically manage task ordering and dependencies
  • Parallel Execution Planning: Identify and optimize concurrent execution opportunities
  • Temporal Cloud Integration: Deploy optimized workflows for production execution
  • Draft-Based Approval: All generated workflows require review before execution

​
Quick Reference

​
Generate from Natural Language

agent.generate_workflow(
    request="Ingest custodian data daily at 9am, validate it, then update NAV",
    context={"portfolio_ids": ["PORT-001", "PORT-002"]}
)

​
Validate Workflow Definition

agent.validate_workflow_tool(
    workflow_json='{"name": "daily_nav", "steps": [...]}'
)

​
Identify Optimization Opportunities

agent.optimize_workflow(
    workflow_json='{"name": "daily_nav", "steps": [...]}'
)

​
Detailed Reference

​
generate_workflow

Type: Tool (langchain_core.tools) Input Schema: GenerateWorkflowInput Returns: Dict[str, Any]

​
Description

Generates a workflow from natural language description using GPT-4-powered parsing. The system analyzes the request, selects appropriate template steps, configures dependencies, and identifies parallelization opportunities. All generated workflows are created as drafts requiring explicit approval before execution. The tool handles parameterized templates through variable substitution, enabling reusable workflow patterns across different contexts. Retry policies and timeout configurations are automatically applied based on step type.

​
Parameters

  • request (str, required): Natural language description of the desired workflow. Should specify:
    • Primary operations (ingest, validate, calculate, report)
    • Scheduling requirements (daily, weekly, on-demand)
    • Data sources and destinations
    • Approval or notification steps
    • Tolerance thresholds or business rules
  • context (Dict[str, Any], optional): Additional execution context:
    • portfolio_ids: List of portfolio identifiers to process
    • business_date: Reference date for calculations
    • data_sources: Available data ingestion endpoints
    • approval_groups: Teams requiring sign-off
    • notification_channels: Where to send results

​
Returns

Success response: 
{
  "success": True,
  "workflow": {
    "workflow_id": "wf-7a2f9c1e",
    "name": "Daily NAV Calculation",
    "description": "Ingest custodian data, validate, calculate NAV",
    "steps": [
      {
        "id": "step-1",
        "type": "ingest",
        "action": "fetch_custodian_data",
        "depends_on": [],
        "retry_policy": {"max_retries": 3, "backoff": "exponential"},
        "timeout_seconds": 600
      },
      {
        "id": "step-2",
        "type": "validate",
        "action": "validate_holdings",
        "depends_on": ["step-1"],
        "timeout_seconds": 300
      },
      {
        "id": "step-3",
        "type": "calculate",
        "action": "calculate_nav",
        "depends_on": ["step-2"],
        "timeout_seconds": 900
      }
    ],
    "schedule": "0 9 * * *",
    "requires_approval": True,
    "parameters": {
      "tolerance_pct": 0.1,
      "notification_on_failure": True
    }
  },
  "workflow_id": "wf-7a2f9c1e",
  "steps_count": 3,
  "schedule": "0 9 * * *",
  "requires_approval": True,
  "validation": {
    "valid": True,
    "errors": [],
    "warnings": []
  },
  "optimizations": [
    {
      "type": "parallelization",
      "description": "Steps 4-6 can run concurrently",
      "estimated_savings_minutes": 15
    }
  ],
  "db_format": { ... },
  "message": "Workflow generated successfully: Daily NAV Calculation. Created as DRAFT for approval.",
  "next_steps": [
    "Review the generated workflow steps",
    "Check optimization suggestions",
    "Approve the workflow to execute it"
  ]
}
Error response: 
{
  "success": False,
  "errors": ["Circular dependency detected: step-1 → step-3 → step-1"],
  "message": "Workflow validation failed"
}

​
Examples

Daily Data Ingestion Workflow Natural language request: 
"Every day at 9am, fetch holdings from our custodian, validate the data against
our tolerance thresholds (0.1%), calculate NAV for all portfolios, and email
the report to the ops team. Mark as urgent if any validation fails."
Generated workflow structure:
  • Scheduled: 0 9 * * * (daily at 9am)
  • Steps: ingest → validate → calculate → notify
  • Retry: 3 attempts with exponential backoff on ingest failures
  • Parallel: None in this case (sequential dependencies)
Portfolio Reconciliation Workflow 
"Every Friday at 5pm, reconcile our portfolio records with administrator data
for all active portfolios, generate a difference report, and require approval
before posting adjustments. Timeout after 30 minutes."
Generated workflow:
  • Schedule: 0 17 * * 5 (Friday at 5pm)
  • Steps: fetch_admin_data (parallelized), reconcile, generate_report, approval_gate, post_adjustments
  • Timeout: 1800 seconds (30 minutes)
  • Approval: Manual review required
Event-Driven Adjustment Workflow 
"When portfolio composition changes are detected, validate the changes,
estimate NAV impact, notify traders, and await approval before executing."
Generated workflow:
  • Trigger: portfolio_change event
  • Conditional steps based on change magnitude
  • Parallelization: Validation and impact estimation run concurrently

​
Throws

  • ValueError: Invalid or incomplete natural language request
  • TemplateNotFound: No matching template for requested operation
  • ContextError: Missing required context variables

​
See Also

  • validate_workflow_tool: Verify generated workflow before approval
  • optimize_workflow: Identify efficiency improvements
  • Workflow Patterns: Common workflow templates

​
validate_workflow_tool

Type: Tool (langchain_core.tools) Input Schema: ValidateWorkflowInput Returns: Dict[str, Any]

​
Description

Validates a workflow definition for structural integrity and logical consistency. Checks include circular dependency detection, step reference validation, required field verification, and configuration error detection. This tool is automatically invoked during generation but can be called independently to verify manually-created or modified workflows.

​
Parameters

  • workflow_json (str, required): Complete workflow definition as JSON string. Must include:
    • name: Workflow identifier
    • steps: Array of workflow steps with dependencies
    • schedule: Cron expression (for scheduled workflows) or null (for on-demand)

​
Returns

{
  "success": True,
  "valid": True,
  "errors": [],
  "warnings": [
    "Step 'notify_traders' has no timeout defined, defaulting to 300 seconds"
  ],
  "message": "Workflow is valid"
}
Validation failure: 
{
  "success": True,
  "valid": False,
  "errors": [
    "Circular dependency: step-2 depends on step-4, step-4 depends on step-2",
    "Step 'calculate_nav' references undefined step 'load_data'",
    "Missing required field 'timeout_seconds' for step 'fetch_data'"
  ],
  "warnings": [],
  "message": "Workflow has errors"
}

​
Validation Rules

  1. Circular Dependencies: No step can transitively depend on itself
  2. Step References: All depends_on references must exist
  3. Required Fields: Every step must have name, type, and action
  4. Timeout Values: Must be positive integers (seconds)
  5. Schedule Format: Valid cron expression or null
  6. Retry Policy: max_retries >= 1, valid backoff strategy

​
optimize_workflow

Type: Tool (langchain_core.tools) Input Schema: OptimizeWorkflowInput Returns: Dict[str, Any]

​
Description

Analyzes workflow structure and identifies optimization opportunities for improved performance and resource efficiency. Provides specific recommendations with quantified time savings estimates.

​
Parameters

  • workflow_json (str, required): Workflow definition as JSON string

​
Returns

{
  "success": True,
  "optimizations": [
    {
      "type": "parallelization",
      "steps": ["validate_holding_1", "validate_holding_2", "validate_holding_3"],
      "description": "Three independent validation steps can execute concurrently",
      "estimated_time_savings_minutes": 45,
      "implementation": "Remove dependencies between steps, execute in parallel"
    },
    {
      "type": "caching",
      "step": "fetch_reference_data",
      "description": "Reference data loaded 4 times; implement caching",
      "estimated_time_savings_minutes": 8,
      "implementation": "Cache fetched data for 24 hours"
    },
    {
      "type": "batching",
      "steps": ["post_adjustment_1", "post_adjustment_2", ..., "post_adjustment_50"],
      "description": "50 individual posting operations can be batched",
      "estimated_time_savings_minutes": 25,
      "implementation": "Use bulk adjustment endpoint"
    }
  ],
  "optimization_count": 3,
  "current_estimated_duration_minutes": 120,
  "optimized_estimated_duration_minutes": 42,
  "estimated_time_savings_minutes": 78,
  "message": "Found 3 optimization opportunities. Potential time savings: 78 minutes."
}

​
Optimization Types

  • Parallelization: Independent steps executing sequentially can run concurrently
  • Caching: Repeated data fetches can be cached to avoid redundant operations
  • Batching: Multiple similar operations can be combined into single batch operations
  • Resource Allocation: Step-specific resource requirements can be optimized
  • Retry Policy: Aggressive retry policies on flaky steps can be tuned

​
Workflow Patterns

​
Data Ingestion Pattern

{
  "name": "daily_custodian_ingestion",
  "steps": [
    {
      "id": "fetch",
      "type": "ingest",
      "action": "fetch_custodian_data",
      "timeout_seconds": 600,
      "retry_policy": {"max_retries": 3}
    },
    {
      "id": "validate",
      "type": "validate",
      "action": "validate_holdings",
      "depends_on": ["fetch"],
      "timeout_seconds": 300
    },
    {
      "id": "store",
      "type": "store",
      "action": "persist_holdings",
      "depends_on": ["validate"],
      "timeout_seconds": 120
    }
  ],
  "schedule": "0 9 * * *"
}

​
Reconciliation Pattern with Approval

{
  "name": "weekly_reconciliation",
  "steps": [
    {
      "id": "fetch_admin",
      "type": "ingest",
      "action": "fetch_admin_records"
    },
    {
      "id": "fetch_internal",
      "type": "ingest",
      "action": "fetch_internal_records"
    },
    {
      "id": "reconcile",
      "type": "calculate",
      "action": "reconcile_holdings",
      "depends_on": ["fetch_admin", "fetch_internal"]
    },
    {
      "id": "approval",
      "type": "approval",
      "action": "manual_review",
      "depends_on": ["reconcile"],
      "approval_groups": ["ops_team"]
    },
    {
      "id": "post",
      "type": "execute",
      "action": "post_adjustments",
      "depends_on": ["approval"]
    }
  ],
  "schedule": "0 17 * * 5",
  "requires_approval": true
}

​
Temporal Cloud Integration

Generated workflows deploy to Temporal Cloud for production execution:
  • Durability: Workflow state persists across server restarts
  • Scalability: Horizontal scaling for high-throughput operations
  • Monitoring: Built-in observability with execution history
  • Retry Semantics: Automatic retry with exponential backoff
  • Task Routing: Route tasks to specific worker pools (portfolio data, risk calculations)
Workflow execution traces are available in the OpsHub dashboard, enabling investigation of failures and performance bottlenecks.

​
Error Handling

All tools return structured responses with success flag: 
if response["success"]:
  # Process workflow
  approve_workflow(response["workflow_id"])
else:
  # Handle error
  log_error(response["error"])
  notify_user(response["message"])
Common error scenarios:
  • Generation Failure: Request too ambiguous or unsupported operations
  • Validation Failure: Circular dependencies, missing references
  • Optimization Timeout: Complex workflows may require manual optimization
  • Template Not Found: Requested operation not in template library

​
Configuration Reference

​
Workflow Definition Schema

{
  "workflow_id": "string (auto-generated)",
  "name": "string (required)",
  "description": "string",
  "steps": [
    {
      "id": "string (required)",
      "type": "enum: ingest|validate|calculate|execute|approve|notify|store",
      "action": "string (required)",
      "parameters": "object",
      "depends_on": ["string[]"],
      "timeout_seconds": "integer",
      "retry_policy": {
        "max_retries": "integer (default: 1)",
        "backoff": "enum: exponential|linear|constant"
      }
    }
  ],
  "schedule": "cron expression or null",
  "requires_approval": "boolean (default: false)",
  "parameters": "object (template variables)"
}

​
Return Format Envelope

All tools return consistent envelope: 
{
  "success": bool,           # Operation completed without exception
  "valid": bool,             # (validate/optimize) Definition is valid
  "message": str,            # Human-readable summary
  "error": str,              # (if success=false) Exception message
  "errors": list,            # (if applicable) Detailed error list
  "warnings": list           # (if applicable) Non-fatal issues
}

​
See Also