Analyze flow definition and extract schemas

curl --request POST \
  --url https://flows.super.ai/api/task-data \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --header 'X-API-Key: <api-key>' \
  --data '
{
  "flow_definition": {}
}
'

import requests

url = "https://flows.super.ai/api/task-data"

payload = { "flow_definition": {} }
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({flow_definition: {}})
};

fetch('https://flows.super.ai/api/task-data', 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/task-data",
  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([
    'flow_definition' => [
        
    ]
  ]),
  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/task-data"

	payload := strings.NewReader("{\n  \"flow_definition\": {}\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/task-data")
  .header("X-API-Key", "<api-key>")
  .header("Authorization", "Bearer <token>")
  .header("Content-Type", "application/json")
  .body("{\n  \"flow_definition\": {}\n}")
  .asString();

require 'uri'
require 'net/http'

url = URI("https://flows.super.ai/api/task-data")

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  \"flow_definition\": {}\n}"

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

{
  "dynamic_data": {},
  "input_schemas": {},
  "output_schemas": {},
  "validation_errors": [
    "<string>"
  ]
}

task-data

Analyze flow definition and extract schemas

Analyze a flow definition to extract task output schemas, dynamic data, and validation errors.

Overview This endpoint validates a flow definition and returns metadata needed for flow building, execution planning, and UI generation. It processes all tasks in the flow definition to extract their output schemas (for type validation), dynamic configuration data (for UI dropdowns and conditional fields), and any structural or semantic validation errors.

Use Cases

Flow Builder UI: Get task output schemas for downstream task input validation
Dynamic Configuration: Retrieve runtime-dependent options (e.g., available spreadsheets, databases)
Flow Validation: Check for errors before saving or executing a flow
Schema Discovery: Understand what data each task will produce
Custom Integrations: Programmatically validate and analyze flow definitions

How It Works

Parses and validates the flow definition structure (basic validation)
For each task in the flow:
- Loads the task executor implementation
- Calls the executor’s get_output_model() to get the output schema
- Calls the executor’s dynamic data methods (if available) to get runtime options
- Collects any errors encountered during schema/data extraction
If full_validation=true, performs additional semantic validation (checks task parameter values, validates connections between tasks, etc.)
Returns all schemas, dynamic data, and validation errors in a single response

Dynamic Data Explained Dynamic data represents configuration options that depend on external state or user credentials. For example:

Google Sheets task: List of spreadsheets accessible with user’s credentials
Database task: List of available tables/columns
API task: Available endpoints or data models from connected service

Validation Levels

Basic (default): Structure validation only (required fields, data types)
Full (full_validation=true): Includes semantic validation (valid references, parameter constraints, external resource availability)

Performance Notes

May make external API calls to fetch dynamic data (e.g., Google Sheets API)
Response time varies based on number of tasks and external service latency
Typical response time: 100ms-2s depending on task executors used

Related Endpoints

POST /flows - Create flow (uses this endpoint’s validation internally)
PUT /flows/{flow_id} - Update flow (uses this endpoint’s validation internally)
GET /task-executors - List available task executors

POST

api

task-data

Analyze flow definition and extract schemas

curl --request POST \
  --url https://flows.super.ai/api/task-data \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --header 'X-API-Key: <api-key>' \
  --data '
{
  "flow_definition": {}
}
'

import requests

url = "https://flows.super.ai/api/task-data"

payload = { "flow_definition": {} }
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({flow_definition: {}})
};

fetch('https://flows.super.ai/api/task-data', 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/task-data",
  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([
    'flow_definition' => [
        
    ]
  ]),
  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/task-data"

	payload := strings.NewReader("{\n  \"flow_definition\": {}\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/task-data")
  .header("X-API-Key", "<api-key>")
  .header("Authorization", "Bearer <token>")
  .header("Content-Type", "application/json")
  .body("{\n  \"flow_definition\": {}\n}")
  .asString();

require 'uri'
require 'net/http'

url = URI("https://flows.super.ai/api/task-data")

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  \"flow_definition\": {}\n}"

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

{
  "dynamic_data": {},
  "input_schemas": {},
  "output_schemas": {},
  "validation_errors": [
    "<string>"
  ]
}

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...

Query Parameters

full_validation

boolean

default:false

Enable comprehensive semantic validation in addition to basic structural validation. When false (default): Only validates flow structure (required fields, data types, task executor existence). When true: Performs additional validation including: - Task parameter value validation (correct types, valid enums, etc.). - Task connection validation (input/output type compatibility). - Circular dependency detection. - External resource validation (e.g., checking if referenced files exist). Full validation may take longer and make external API calls. Use during flow save/publish operations. Use false (default) for faster validation during editing.

include_input_schema

boolean

default:false

Include JSON Schema definitions for input task executors' input structure. Calls get_input_model() to generate dynamic input schemas for form building. Adds ~10-50ms latency for schema generation.

Body

application/json

Request model for task data analysis endpoint.

Accepts a flow definition in the SuperAI Flows DSL format for validation and schema extraction. The flow definition should include all tasks, connections, and configurations needed to execute the flow.

flow_definition

Flow Definition · object

required

Complete flow definition in SuperAI Flows DSL format. Must include: - 'name': Flow name (string, required). - 'tasks': List of task definitions with task_executor_name, parameters, etc. - 'connections': Optional list of task dependencies and data flows. - 'config': Optional flow-level configuration. The definition will be validated and analyzed to extract task output schemas and dynamic configuration data. Validation errors will be returned in the response if the definition is invalid.

Example:

{
  "name": "my-workflow",
  "tasks": [
    {
      "name": "fetch_data",
      "parameters": {
        "method": "GET",
        "url": "https://api.example.com/data"
      },
      "task_executor_name": "http_request"
    },
    {
      "name": "process_data",
      "parameters": {
        "code": "result = input_data['value'] * 2"
      },
      "task_executor_name": "python_code"
    }
  ]
}

Response

Flow definition successfully analyzed

Response model containing task output schemas, dynamic data, and validation results.

This response provides all metadata needed to build UIs for flow configuration, validate task connections, and understand what data each task will produce. Each field serves a specific purpose in the flow building and validation process.

dynamic_data

Dynamic Data · object

Runtime-dependent configuration options for tasks, keyed by task name. Contains dynamic values that depend on user credentials, external services, or other runtime state. Format: {task_name: {field_name: options}}. Common use cases: - Dropdown options (e.g., available Google Sheets, database tables). - Conditional field visibility based on other parameters. - Resource lists requiring authentication (e.g., Slack channels). - Default values computed from external state. The structure varies by task executor - check task executor documentation for specific formats. Empty dict if no tasks provide dynamic data or if fetching dynamic data failed (check validation_errors).

Example:

{
  "read_sheet": {
    "spreadsheet_id": {
      "options": [
        {
          "label": "Sales Data 2024",
          "value": "abc123"
        },
        {
          "label": "Customer List",
          "value": "def456"
        }
      ]
    }
  }
}

input_schemas

Input Schemas · object

JSON schemas for input task executors' inputs, keyed by task name. Each schema describes the structure and types of data the task expects as input. Only included when include_input_schema=true is specified. Format: {task_name: json_schema_object}. JSON schemas follow JSON Schema Draft 7 specification and include: - 'type': Root type (usually 'object'). - 'properties': Object properties with types and descriptions. - 'required': List of required property names. - 'title': Human-readable schema name. Use these schemas to generate input forms for flow execution. Empty dict if include_input_schema=false or no input tasks found.

Example:

{
  "custom_form_input": {
    "properties": {
      "document_url": {
        "description": "URL to uploaded document",
        "type": "string"
      },
      "filename": {
        "description": "Name of uploaded file",
        "type": "string"
      },
      "metadata": {
        "description": "File metadata",
        "type": "object"
      },
      "mime_type": {
        "description": "MIME type of uploaded file",
        "type": "string"
      }
    },
    "required": ["filename", "mime_type", "document_url"],
    "title": "CustomFormInput",
    "type": "object"
  }
}

output_schemas

Output Schemas · object

JSON schemas for task outputs, keyed by task name. Each schema describes the structure and types of data the task will produce. Format: {task_name: json_schema_object}. JSON schemas follow JSON Schema Draft 7 specification and include: - 'type': Root type (usually 'object'). - 'properties': Object properties with types and descriptions. - 'required': List of required property names. - 'title': Human-readable schema name. Use these schemas to validate downstream task inputs and generate UI forms. Empty dict if no tasks have output schemas or if all tasks failed to load.

Example:

{
  "fetch_data": {
    "properties": {
      "body": {
        "description": "Response body as JSON",
        "type": "object"
      },
      "status_code": {
        "description": "HTTP response status",
        "type": "integer"
      }
    },
    "required": ["status_code", "body"],
    "title": "HttpRequestOutput",
    "type": "object"
  }
}

validation_errors

string[]

Human-readable validation error messages describing problems with the flow definition. Errors are returned even if the request succeeds (HTTP 200) - check this field to determine if the flow definition is valid. Empty list indicates a valid flow. Error categories: - Structural errors: Missing required fields, invalid task references. - Semantic errors (full_validation=true only): Invalid parameter values, unreachable tasks, circular dependencies. - Task executor errors: Unknown task executor, failed to load executor. - Schema extraction errors: Failed to determine output schema for task. Each error is a descriptive string suitable for display to users. Example: 'Task "process_data" references unknown task "fetch_data" in input mapping'.

Examples:

[
  "Task 'invalid_task' uses unknown task executor 'nonexistent_executor'"
]

[]

Proxy SAML metadata to avoid CORS issues List task executions with filtering