TurfAI Workflow Integration Guide

Synced from the source repositories. Do not edit by hand.

Version: 1.0 Last Updated: November 2025 Target Audience: External developers integrating workflow builders with TurfAI

Overview
Architecture
Authentication
API Reference
Complete Workflow Example
Error Handling
Best Practices

Overview

TurfAI provides a complete workflow orchestration system for document processing. External applications can:

Discover available task types dynamically
Build workflow definitions programmatically
Validate workflows before execution
Execute workflows with real-time status monitoring
Pause/Resume workflows for user interaction
Analyze workflow performance and errors

✅ 13 Task Types: Classification, Extraction, LLM, Summarization, OCR, RTIO, and more ✅ User Interaction: Pause workflows for document/prompt input ✅ Real-time Monitoring: Poll execution status and get step-by-step timeline ✅ Analytics: Metrics, performance data, and error tracking ✅ Multi-tenant: System-level and user-level workflows

Architecture

┌──────────────────┐
│  Your Frontend   │
│  (React/Vue/etc) │
└────────┬─────────┘
         │ HTTP/REST
         ▼
┌──────────────────┐
│   DMS (Strapi)   │ ← All API calls go here
│   Port: 1337     │
└────────┬─────────┘
         │ Redis Queue
         ▼
┌──────────────────┐
│  Router Service  │
│   Port: 8000     │
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│    Processors    │ ← Backend execution
│  (Python/FastAPI)│
└──────────────────┘

Important: Your frontend only interacts with DMS. Internal services (Router, Processors) are hidden.

Authentication

All API endpoints require JWT authentication.

Getting a JWT Token

# Login to get JWT
curl -X POST http://localhost:1337/api/auth/local \
  -H "Content-Type: application/json" \
  -d '{
    "identifier": "user@example.com",
    "password": "yourpassword"
  }'

# Response:
{
  "jwt": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "user": {
    "id": 1,
    "username": "user",
    "email": "user@example.com"
  }
}

Using JWT in Requests

curl http://localhost:1337/api/workflows/task-types \
  -H "Authorization: Bearer YOUR_JWT_TOKEN"

API Reference

Base URL

Development: http://localhost:1337/api
Production: https://your-domain.com/api

Endpoints

1. Task Types Discovery

GET /workflows/task-types

Discover all available task processors.

curl http://localhost:1337/api/workflows/task-types \
  -H "Authorization: Bearer $JWT"

Response:

{
  "data": {
    "task_types": [
      {
        "id": "classification_task",
        "name": "Document Classification",
        "category": "AI Operations",
        "description": "Classify documents into categories",
        "icon": "FileText",
        "processor": "classification",
        "config_schema": {
          "type": "object",
          "properties": {
            "classification_type": {
              "type": "string",
              "enum": ["document_type", "sentiment", "topic"],
              "default": "document_type"
            }
          }
        },
        "input_schema": {
          "type": "object",
          "required": ["file_url"],
          "properties": {
            "file_url": { "type": "string" }
          }
        },
        "output_schema": {
          "type": "object",
          "properties": {
            "prompt_title": { "type": "string" },
            "confidence": { "type": "number" }
          }
        }
      }
      // ... 12 more task types
    ]
  }
}

2. Workflow Validation

POST /workflows/validate

Validate a workflow definition before saving.

curl -X POST http://localhost:1337/api/workflows/validate \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_definition": {
      "nodes": [
        {
          "id": "start-1",
          "type": "start",
          "data": {
            "label": "Start",
            "task_type": "custom_task",
            "config": {}
          }
        },
        {
          "id": "classify-1",
          "type": "task",
          "data": {
            "label": "Classify Document",
            "task_type": "classification_task",
            "config": {
              "classification_type": "document_type"
            }
          }
        }
      ],
      "edges": [
        {
          "id": "edge-1",
          "source": "start-1",
          "target": "classify-1"
        }
      ]
    }
  }'

Response:

{
  "data": {
    "valid": true,
    "errors": []
  }
}

Error Response (invalid workflow):

{
  "data": {
    "valid": false,
    "errors": [
      "Workflow contains cycles, which are not allowed",
      "Edge at index 2 references non-existent target node: invalid-node"
    ]
  }
}

3. Create Activity (Save Workflow)

POST /activities

Save a workflow definition as an activity.

curl -X POST http://localhost:1337/api/activities \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {
      "name": "Invoice Processing Pipeline",
      "description": "Classify and extract invoice data",
      "workflow_type": "document_processing",
      "status": "active",
      "workflow_definition": {
        "nodes": [...],
        "edges": [...],
        "input_schema": {
          "type": "object",
          "required": ["file_url"],
          "properties": {
            "file_url": { "type": "string" }
          }
        }
      }
    }
  }'

Response:

{
  "data": {
    "id": 123,
    "name": "Invoice Processing Pipeline",
    "status": "active",
    "workflow_definition": {...},
    "createdAt": "2025-11-17T...",
    "updatedAt": "2025-11-17T..."
  }
}

4. List Activities

GET /activities

Get all workflows (system + user-owned).

curl http://localhost:1337/api/activities \
  -H "Authorization: Bearer $JWT"

Response:

{
  "data": [
    {
      "id": 123,
      "name": "Invoice Processing",
      "level": "user",
      "workflow_type": "document_processing"
    }
  ],
  "meta": {
    "pagination": {
      "page": 1,
      "pageSize": 25,
      "total": 5
    }
  }
}

5. Create Workflow Execution

POST /workflow-executions

Create an execution instance (does NOT start execution).

curl -X POST http://localhost:1337/api/workflow-executions \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {
      "activity": 123,
      "workflow_definition": {...},
      "inputs": {
        "file_url": "gs://bucket/invoice.pdf"
      }
    }
  }'

Response:

{
  "data": {
    "id": 456,
    "status": "queued",
    "activity": {...},
    "started_at": "2025-11-17T..."
  }
}

6. Execute Workflow

POST /workflow-executions/:id/execute

Start workflow execution.

curl -X POST http://localhost:1337/api/workflow-executions/456/execute \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "inputs": {
      "file_url": "gs://bucket/invoice.pdf",
      "document_name": "invoice.pdf"
    }
  }'

Response:

{
  "data": {
    "id": 456,
    "status": "running",
    "started_at": "2025-11-17T10:00:00Z"
  }
}

7. Get Execution Status

GET /workflow-executions/:id/status

Poll execution status (call every 2-5 seconds).

curl http://localhost:1337/api/workflow-executions/456/status \
  -H "Authorization: Bearer $JWT"

Response (running):

{
  "data": {
    "id": 456,
    "status": "running",
    "task_states": {
      "classify-1": {
        "status": "completed",
        "outputs": {
          "prompt_title": "Invoice",
          "confidence": 0.92
        }
      },
      "extract-1": {
        "status": "running"
      }
    }
  }
}

Response (awaiting user input):

{
  "data": {
    "id": 456,
    "status": "awaiting_user_input",
    "awaiting_node": "document-input-1",
    "correlation_token": "abc123",
    "awaiting_input_schema": {
      "type": "object",
      "properties": {
        "file_url": { "type": "string" },
        "document_name": { "type": "string" }
      }
    }
  }
}

Response (completed):

{
  "data": {
    "id": 456,
    "status": "completed",
    "outputs": {
      "classification": {...},
      "extraction_result": {...}
    },
    "completed_at": "2025-11-17T10:00:45Z"
  }
}

8. Resume Workflow

POST /workflow-executions/:id/resume

Resume paused workflow with user input.

curl -X POST http://localhost:1337/api/workflow-executions/456/resume \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "correlation_token": "abc123",
    "payload": {
      "file_url": "gs://bucket/user-uploaded.pdf",
      "document_name": "uploaded.pdf"
    }
  }'

Response:

{
  "data": {
    "id": 456,
    "status": "running"
  }
}

9. Workflow Metrics Overview

GET /workflows/metrics/overview

Get dashboard metrics for all workflows.

curl "http://localhost:1337/api/workflows/metrics/overview?period=last_7_days" \
  -H "Authorization: Bearer $JWT"

Response:

{
  "data": {
    "period": "last_7_days",
    "total_executions": 150,
    "successful": 142,
    "failed": 5,
    "awaiting_input": 3,
    "success_rate": 94.7,
    "avg_execution_time_seconds": 34.5,
    "by_workflow": [
      {
        "activity_id": 123,
        "workflow_name": "Invoice Processing",
        "executions": 89,
        "success_rate": 95.5,
        "avg_time": 32.1
      }
    ],
    "by_day": [
      {"date": "2025-11-10", "executions": 25, "success": 24},
      {"date": "2025-11-11", "executions": 30, "success": 28}
    ]
  }
}

10. Workflow-Specific Analytics

GET /workflows/:activityId/analytics

Get detailed analytics for a specific workflow.

curl http://localhost:1337/api/workflows/123/analytics \
  -H "Authorization: Bearer $JWT"

Response:

{
  "data": {
    "activity_id": 123,
    "workflow_name": "Invoice Processing",
    "total_executions": 89,
    "last_30_days": {
      "executions": 89,
      "success_rate": 95.5,
      "avg_execution_time": 32.1,
      "total_cost": 5.67,
      "step_performance": [
        {
          "step_name": "classify-1",
          "avg_duration": 8.5,
          "success_rate": 98.0,
          "most_common_error": null
        },
        {
          "step_name": "extract-1",
          "avg_duration": 22.3,
          "success_rate": 94.2,
          "most_common_error": "Timeout waiting for LLM"
        }
      ],
      "failure_analysis": {
        "total_failures": 4,
        "by_step": {
          "extract-1": 3,
          "classify-1": 1
        },
        "by_error_type": {
          "timeout": 2,
          "validation_error": 1,
          "llm_error": 1
        }
      }
    }
  }
}

11. Execution Timeline

GET /workflow-executions/:executionId/timeline

Get step-by-step timeline of execution.

curl http://localhost:1337/api/workflow-executions/456/timeline \
  -H "Authorization: Bearer $JWT"

Response:

{
  "data": {
    "execution_id": 456,
    "status": "completed",
    "total_duration_seconds": 45.6,
    "timeline": [
      {
        "step_name": "classify-1",
        "step_type": "classification_task",
        "status": "completed",
        "started_at": "2025-11-17T10:00:00Z",
        "completed_at": "2025-11-17T10:00:08Z",
        "duration_seconds": 8.5,
        "cost": 0.045,
        "result_summary": {
          "prompt_title": "Invoice",
          "confidence": 0.92
        }
      },
      {
        "step_name": "extract-1",
        "step_type": "extraction_task",
        "status": "completed",
        "started_at": "2025-11-17T10:00:08Z",
        "completed_at": "2025-11-17T10:00:30Z",
        "duration_seconds": 22.3,
        "cost": 0.234,
        "result_summary": {
          "fields_extracted": 15,
          "confidence": 0.88
        }
      }
    ],
    "inputs": {
      "file_url": "gs://bucket/invoice.pdf"
    },
    "outputs": {
      "extraction_result": {...}
    }
  }
}

12. Recent Errors

GET /workflows/errors/recent

Get recent workflow failures.

curl "http://localhost:1337/api/workflows/errors/recent?limit=50" \
  -H "Authorization: Bearer $JWT"

Response:

{
  "data": {
    "errors": [
      {
        "execution_id": 789,
        "activity_id": 123,
        "workflow_name": "Invoice Processing",
        "failed_step": "extract-1",
        "error_type": "timeout",
        "error_message": "LLM request timed out after 60 seconds",
        "occurred_at": "2025-11-17T09:30:45Z",
        "retry_count": 3,
        "inputs_summary": {
          "document_name": "complex_invoice.pdf",
          "file_url": "gs://..."
        }
      }
    ],
    "total": 15,
    "by_error_type": {
      "timeout": 8,
      "validation_error": 4,
      "llm_error": 3
    }
  }
}

Complete Workflow Example

Scenario: Build and Execute Invoice Processing Workflow

This example shows the complete lifecycle from discovery to analytics.

#!/bin/bash

# Configuration
API_URL="http://localhost:1337/api"
JWT="your-jwt-token-here"

# Step 1: Discover available task types
echo "Step 1: Discovering task types..."
TASK_TYPES=$(curl -s "$API_URL/workflows/task-types" \
  -H "Authorization: Bearer $JWT")

echo "Available task types: $(echo $TASK_TYPES | jq '.data.task_types | length')"

# Step 2: Build workflow definition
echo "Step 2: Building workflow definition..."
WORKFLOW_DEF='{
  "nodes": [
    {
      "id": "start-1",
      "type": "start",
      "data": {
        "label": "Start",
        "task_type": "custom_task",
        "config": {}
      }
    },
    {
      "id": "classify-1",
      "type": "task",
      "data": {
        "label": "Classify Document",
        "task_type": "classification_task",
        "config": {
          "classification_type": "document_type",
          "include_confidence": true
        }
      }
    },
    {
      "id": "extract-1",
      "type": "task",
      "data": {
        "label": "Extract Data",
        "task_type": "extraction_task",
        "config": {
          "use_classification_prompt": true,
          "output_format": "json"
        }
      }
    },
    {
      "id": "end-1",
      "type": "end",
      "data": {
        "label": "End",
        "task_type": "custom_task",
        "config": {}
      }
    }
  ],
  "edges": [
    {"id": "e1", "source": "start-1", "target": "classify-1"},
    {"id": "e2", "source": "classify-1", "target": "extract-1"},
    {"id": "e3", "source": "extract-1", "target": "end-1"}
  ],
  "input_schema": {
    "type": "object",
    "required": ["file_url"],
    "properties": {
      "file_url": {"type": "string"}
    }
  }
}'

# Step 3: Validate workflow
echo "Step 3: Validating workflow..."
VALIDATION=$(curl -s -X POST "$API_URL/workflows/validate" \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d "{\"workflow_definition\": $WORKFLOW_DEF}")

IS_VALID=$(echo $VALIDATION | jq '.data.valid')

if [ "$IS_VALID" != "true" ]; then
  echo "Validation failed:"
  echo $VALIDATION | jq '.data.errors'
  exit 1
fi

echo "Workflow is valid!"

# Step 4: Save as activity
echo "Step 4: Saving workflow as activity..."
ACTIVITY=$(curl -s -X POST "$API_URL/activities" \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d "{
    \"data\": {
      \"name\": \"Invoice Processing Pipeline\",
      \"description\": \"Classify and extract invoice data\",
      \"workflow_type\": \"document_processing\",
      \"status\": \"active\",
      \"workflow_definition\": $WORKFLOW_DEF
    }
  }")

ACTIVITY_ID=$(echo $ACTIVITY | jq '.data.id')
echo "Created activity ID: $ACTIVITY_ID"

# Step 5: Create execution instance
echo "Step 5: Creating execution instance..."
EXECUTION=$(curl -s -X POST "$API_URL/workflow-executions" \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d "{
    \"data\": {
      \"activity\": $ACTIVITY_ID,
      \"workflow_definition\": $WORKFLOW_DEF,
      \"inputs\": {
        \"file_url\": \"gs://my-bucket/invoice.pdf\"
      }
    }
  }")

EXECUTION_ID=$(echo $EXECUTION | jq '.data.id')
echo "Created execution ID: $EXECUTION_ID"

# Step 6: Execute workflow
echo "Step 6: Executing workflow..."
curl -s -X POST "$API_URL/workflow-executions/$EXECUTION_ID/execute" \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{
    "inputs": {
      "file_url": "gs://my-bucket/invoice.pdf",
      "document_name": "invoice.pdf"
    }
  }' > /dev/null

echo "Workflow started. Polling status..."

# Step 7: Poll status until complete
while true; do
  STATUS=$(curl -s "$API_URL/workflow-executions/$EXECUTION_ID/status" \
    -H "Authorization: Bearer $JWT")

  CURRENT_STATUS=$(echo $STATUS | jq -r '.data.status')
  echo "Status: $CURRENT_STATUS"

  if [ "$CURRENT_STATUS" == "completed" ] || [ "$CURRENT_STATUS" == "failed" ]; then
    break
  fi

  sleep 2
done

# Step 8: Get execution timeline
echo "Step 8: Getting execution timeline..."
TIMELINE=$(curl -s "$API_URL/workflow-executions/$EXECUTION_ID/timeline" \
  -H "Authorization: Bearer $JWT")

echo "Timeline:"
echo $TIMELINE | jq '.data.timeline'

# Step 9: Get workflow analytics
echo "Step 9: Getting workflow analytics..."
ANALYTICS=$(curl -s "$API_URL/workflows/$ACTIVITY_ID/analytics" \
  -H "Authorization: Bearer $JWT")

echo "Analytics:"
echo $ANALYTICS | jq '.data.last_30_days'

echo "Complete!"

Error Handling

Common Error Responses

401 Unauthorized:

{
  "error": {
    "status": 401,
    "name": "UnauthorizedError",
    "message": "Invalid token"
  }
}

400 Bad Request:

{
  "error": {
    "status": 400,
    "name": "BadRequestError",
    "message": "workflow_definition is required",
    "details": {
      "errors": ["workflow_definition is required"]
    }
  }
}

404 Not Found:

{
  "error": {
    "status": 404,
    "name": "NotFoundError",
    "message": "Activity not found"
  }
}

Retry Strategy

For execution polling, implement exponential backoff:

async function pollWithBackoff(executionId, jwt) {
  let delay = 1000; // Start with 1 second
  const maxDelay = 10000; // Max 10 seconds

  while (true) {
    const status = await getExecutionStatus(executionId, jwt);

    if (status.status === 'completed' || status.status === 'failed') {
      return status;
    }

    await sleep(delay);
    delay = Math.min(delay * 1.5, maxDelay);
  }
}

Best Practices

1. Workflow Design

✅ DO:

Validate workflows before saving
Use descriptive node names
Include error handling nodes
Set reasonable timeouts

❌ DON'T:

Create circular dependencies
Use hardcoded values (use inputs instead)
Skip validation

2. Execution Monitoring

✅ DO:

Poll status every 2-5 seconds
Handle awaiting_user_input status
Implement timeout limits (e.g., 5 minutes)
Log execution IDs for debugging

❌ DON'T:

Poll too frequently (< 1 second)
Assume instant completion
Ignore error states

3. Performance

✅ DO:

Reuse activity IDs for repeated workflows
Batch analytics requests
Cache task type registry
Use pagination for large result sets

❌ DON'T:

Create new activity for every execution
Fetch all executions without pagination
Poll analytics endpoints too frequently

Next Steps

Explore Examples: Check /examples directory for working scripts
Test Interactive Script: Run examples/test-complete-workflow.sh
Build Your Integration: Use this guide to build your workflow builder
Join Community: GitHub Discussions

Support

Documentation: Full API Docs
Issues: GitHub Issues
Email: support@turfai.com

Happy Building! 🚀

TurfAI Workflow Integration Guide

Table of Contents

Overview

Key Features

Architecture

Authentication

Getting a JWT Token

Using JWT in Requests

API Reference

Base URL

Endpoints

1. Task Types Discovery

2. Workflow Validation

3. Create Activity (Save Workflow)

4. List Activities

5. Create Workflow Execution

6. Execute Workflow

7. Get Execution Status

8. Resume Workflow

9. Workflow Metrics Overview

10. Workflow-Specific Analytics

11. Execution Timeline

12. Recent Errors

Complete Workflow Example

Scenario: Build and Execute Invoice Processing Workflow

Error Handling

Common Error Responses

Retry Strategy

Best Practices

1. Workflow Design

2. Execution Monitoring

3. Performance

Next Steps

Support

On this page