Add task status updates (Internal API)

curl --request POST \
  --url https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --header 'X-API-Key: <api-key>' \
  --data '
{
  "statuses": [
    {
      "status": "completed",
      "task_name": "extract_data"
    },
    {
      "status": "running",
      "task_name": "validate_results"
    }
  ]
}
'

import requests

url = "https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status"

payload = { "statuses": [
        {
            "status": "completed",
            "task_name": "extract_data"
        },
        {
            "status": "running",
            "task_name": "validate_results"
        }
    ] }
headers = {
    "X-API-Key": "<api-key>",
    "Authorization": "Bearer <token>",
    "Content-Type": "application/json"
}

response = requests.post(url, json=payload, headers=headers)

print(response.text)

const options = {
  method: 'POST',
  headers: {
    'X-API-Key': '<api-key>',
    Authorization: 'Bearer <token>',
    'Content-Type': 'application/json'
  },
  body: JSON.stringify({
    statuses: [
      {status: 'completed', task_name: 'extract_data'},
      {status: 'running', task_name: 'validate_results'}
    ]
  })
};

fetch('https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status', options)
  .then(res => res.json())
  .then(res => console.log(res))
  .catch(err => console.error(err));

<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "POST",
  CURLOPT_POSTFIELDS => json_encode([
    'statuses' => [
        [
                'status' => 'completed',
                'task_name' => 'extract_data'
        ],
        [
                'status' => 'running',
                'task_name' => 'validate_results'
        ]
    ]
  ]),
  CURLOPT_HTTPHEADER => [
    "Authorization: Bearer <token>",
    "Content-Type: application/json",
    "X-API-Key: <api-key>"
  ],
]);

$response = curl_exec($curl);
$err = curl_error($curl);

curl_close($curl);

if ($err) {
  echo "cURL Error #:" . $err;
} else {
  echo $response;
}

package main

import (
	"fmt"
	"strings"
	"net/http"
	"io"
)

func main() {

	url := "https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status"

	payload := strings.NewReader("{\n  \"statuses\": [\n    {\n      \"status\": \"completed\",\n      \"task_name\": \"extract_data\"\n    },\n    {\n      \"status\": \"running\",\n      \"task_name\": \"validate_results\"\n    }\n  ]\n}")

	req, _ := http.NewRequest("POST", url, payload)

	req.Header.Add("X-API-Key", "<api-key>")
	req.Header.Add("Authorization", "Bearer <token>")
	req.Header.Add("Content-Type", "application/json")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(string(body))

}

HttpResponse<String> response = Unirest.post("https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status")
  .header("X-API-Key", "<api-key>")
  .header("Authorization", "Bearer <token>")
  .header("Content-Type", "application/json")
  .body("{\n  \"statuses\": [\n    {\n      \"status\": \"completed\",\n      \"task_name\": \"extract_data\"\n    },\n    {\n      \"status\": \"running\",\n      \"task_name\": \"validate_results\"\n    }\n  ]\n}")
  .asString();

require 'uri'
require 'net/http'

url = URI("https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Post.new(url)
request["X-API-Key"] = '<api-key>'
request["Authorization"] = 'Bearer <token>'
request["Content-Type"] = 'application/json'
request.body = "{\n  \"statuses\": [\n    {\n      \"status\": \"completed\",\n      \"task_name\": \"extract_data\"\n    },\n    {\n      \"status\": \"running\",\n      \"task_name\": \"validate_results\"\n    }\n  ]\n}"

response = http.request(request)
puts response.read_body

[
  {
    "flow_execution_id": "3c90c3cc-0d44-4b50-8888-8dd25736052a",
    "id": "3c90c3cc-0d44-4b50-8888-8dd25736052a",
    "organization_id": "3c90c3cc-0d44-4b50-8888-8dd25736052a",
    "status": "<string>",
    "task_name": "<string>",
    "created_at": "2023-11-07T05:31:56Z"
  }
]

flow-executions

Add task status updates (Internal API)

Add task status updates to a flow execution (Internal API).

This endpoint is used by system workers to report task execution status changes. Multiple task statuses can be added in a single request for batch efficiency. Status updates trigger state transitions and WebSocket notifications.

⚠️ Internal API: This endpoint is designed for internal system communication. Direct customer usage is not recommended. Use GET /task-executions instead.

Context:

Called by system workers during task execution
Updates task status records in bulk for performance
Triggers task output state calculations
Sends WebSocket notifications asynchronously
Worker receives immediate response without blocking on notifications

Behavior:

Validates flow execution exists and is accessible
Creates status records for each task in batch
Updates related task output states in parallel
Sends WebSocket notifications in background task
Returns immediately to worker for low latency

Performance Optimizations:

Batch inserts for multiple status updates (~50ms for 10 tasks)
WebSocket notifications processed asynchronously
Task output state updates use batch operations
Response returns before notification delivery completes

Use Cases:

System workers reporting task start/completion
Bulk status updates during execution
Real-time execution monitoring via WebSocket

Important Notes:

This endpoint does NOT retrieve task status (use GET /task-executions)
Requires service-level authentication token
Not intended for direct customer API integration
WebSocket notifications may arrive after HTTP response

Business Logic:

Task output states recalculated based on new statuses
Multiple status updates for same task create history
Execution-level status not automatically updated
Status transitions not validated (allows any status value)

Related Internal Endpoints:

POST /flow-executions - Creates parent execution
GET /task-executions - Retrieves task execution details
WebSocket /ws/flows/ - Receives status notifications

POST

api

flow-executions

{flow_execution_id}

task-status

Add task status updates (Internal API)

curl --request POST \
  --url https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --header 'X-API-Key: <api-key>' \
  --data '
{
  "statuses": [
    {
      "status": "completed",
      "task_name": "extract_data"
    },
    {
      "status": "running",
      "task_name": "validate_results"
    }
  ]
}
'

import requests

url = "https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status"

payload = { "statuses": [
        {
            "status": "completed",
            "task_name": "extract_data"
        },
        {
            "status": "running",
            "task_name": "validate_results"
        }
    ] }
headers = {
    "X-API-Key": "<api-key>",
    "Authorization": "Bearer <token>",
    "Content-Type": "application/json"
}

response = requests.post(url, json=payload, headers=headers)

print(response.text)

const options = {
  method: 'POST',
  headers: {
    'X-API-Key': '<api-key>',
    Authorization: 'Bearer <token>',
    'Content-Type': 'application/json'
  },
  body: JSON.stringify({
    statuses: [
      {status: 'completed', task_name: 'extract_data'},
      {status: 'running', task_name: 'validate_results'}
    ]
  })
};

fetch('https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status', options)
  .then(res => res.json())
  .then(res => console.log(res))
  .catch(err => console.error(err));

<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "POST",
  CURLOPT_POSTFIELDS => json_encode([
    'statuses' => [
        [
                'status' => 'completed',
                'task_name' => 'extract_data'
        ],
        [
                'status' => 'running',
                'task_name' => 'validate_results'
        ]
    ]
  ]),
  CURLOPT_HTTPHEADER => [
    "Authorization: Bearer <token>",
    "Content-Type: application/json",
    "X-API-Key: <api-key>"
  ],
]);

$response = curl_exec($curl);
$err = curl_error($curl);

curl_close($curl);

if ($err) {
  echo "cURL Error #:" . $err;
} else {
  echo $response;
}

package main

import (
	"fmt"
	"strings"
	"net/http"
	"io"
)

func main() {

	url := "https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status"

	payload := strings.NewReader("{\n  \"statuses\": [\n    {\n      \"status\": \"completed\",\n      \"task_name\": \"extract_data\"\n    },\n    {\n      \"status\": \"running\",\n      \"task_name\": \"validate_results\"\n    }\n  ]\n}")

	req, _ := http.NewRequest("POST", url, payload)

	req.Header.Add("X-API-Key", "<api-key>")
	req.Header.Add("Authorization", "Bearer <token>")
	req.Header.Add("Content-Type", "application/json")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(string(body))

}

HttpResponse<String> response = Unirest.post("https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status")
  .header("X-API-Key", "<api-key>")
  .header("Authorization", "Bearer <token>")
  .header("Content-Type", "application/json")
  .body("{\n  \"statuses\": [\n    {\n      \"status\": \"completed\",\n      \"task_name\": \"extract_data\"\n    },\n    {\n      \"status\": \"running\",\n      \"task_name\": \"validate_results\"\n    }\n  ]\n}")
  .asString();

require 'uri'
require 'net/http'

url = URI("https://flows.super.ai/api/flow-executions/{flow_execution_id}/task-status")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Post.new(url)
request["X-API-Key"] = '<api-key>'
request["Authorization"] = 'Bearer <token>'
request["Content-Type"] = 'application/json'
request.body = "{\n  \"statuses\": [\n    {\n      \"status\": \"completed\",\n      \"task_name\": \"extract_data\"\n    },\n    {\n      \"status\": \"running\",\n      \"task_name\": \"validate_results\"\n    }\n  ]\n}"

response = http.request(request)
puts response.read_body

[
  {
    "flow_execution_id": "3c90c3cc-0d44-4b50-8888-8dd25736052a",
    "id": "3c90c3cc-0d44-4b50-8888-8dd25736052a",
    "organization_id": "3c90c3cc-0d44-4b50-8888-8dd25736052a",
    "status": "<string>",
    "task_name": "<string>",
    "created_at": "2023-11-07T05:31:56Z"
  }
]

Authorizations

X-API-Key

string

header

required

API key authentication. Include your API key in the X-API-Key header as: X-API-Key YOUR_API_KEY

Example:

X-API-Key: saf_1234567890

Authorization

string

header

required

JWT Bearer token authentication. Include your access token in the Authorization header as: Bearer YOUR_ACCESS_TOKEN

Example:

Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

Path Parameters

flow_execution_id

string<uuid>

required

UUID of the flow execution receiving task status updates. Must be an active (non-deleted) execution.

Body

application/json

Batch request to add multiple task status updates.

Allows system workers to report status changes for multiple tasks in a single API call for improved performance.

statuses

AddTaskStatusRequestItem · object[]

required

Array of task status updates to apply. Minimum: 1 update, Maximum: 100 updates per request. Updates are processed in order provided.

Required array length: 1 - 100 elements

Show child attributes

Example:

[
  {
    "status": "completed",
    "task_name": "extract_data"
  },
  {
    "status": "running",
    "task_name": "validate_results"
  }
]

Response

Task status updates successfully recorded

flow_execution_id

string<uuid>

required

string<uuid>

required

organization_id

string<uuid>

required

status

string

required

task_name

string

required

created_at

string<date-time> | null

Patch Flow Execution Status List all flows