Cognitive Diagrams Management API

Cognitive Diagrams API reference - execute AI agents, manage async jobs, and trigger RAG training via REST API.

The Cognitive Diagrams API allows you to programmatically execute and interact with your AI cognitive diagrams (workflows) built within the Synthreo platform. This is the primary way to integrate your AI agents into external applications.

Base URL

https://api.botx.cloud

Quick Reference

Operation	Method	Endpoint	Description
Execute synchronously	POST	`/CognitiveDiagram/{id}/Execute`	Run a diagram and wait for the result
Execute as async job	POST	`/CognitiveDiagram/{id}/ExecuteAsJob`	Start a long-running job
Poll job status	GET	`/job/{job_id}`	Check the status of a running job
Trigger training	PATCH	`/CognitiveDiagram/{id}/TrainNode`	Start RAG model training
Get agent status	GET	`/CognitiveDiagram/{id}`	Check agent state (idle vs training)

Execute Cognitive Diagram (Synchronous)

This endpoint allows you to send user input to a specific cognitive diagram and receive an AI-generated response synchronously. Use this for quick operations that complete within a few seconds.

Endpoint

POST https://api.botx.cloud/CognitiveDiagram/{diagram_id}/Execute

Replace {diagram_id} with the unique identifier of your cognitive diagram.

Request Headers

{
  "Content-Type": "application/json",
  "Accept": "application/json",
  "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}

Request Body

The request body is a JSON object that typically includes the Action and UserSays fields. The UserSays field contains a JSON string that can include various data points, which are then mapped to template variables within your cognitive diagram.

Field	Type	Required	Description
`Action`	string	Yes	Always set to `"Execute"`
`UserSays`	string	Yes	A JSON-encoded array containing one object with your input variables
`RobotSays`	string	No	Optional field for passing previous AI response context
`CallerSource`	integer	No	Optional identifier for the request source (e.g., `1` for API)

Basic Message

To send a simple text message to your AI agent:

{
  "Action": "Execute",
  "UserSays": "[{\"userSays\":\"Your message here\"}]"
}

With Conversation ID

To maintain context across multiple calls, you can include any variable name you choose. There is no built-in method of maintaining context - the person designing the builder agent needs to manually create a method to “maintain context” by looking at these variables within the cognitive diagram. You can use any variable name, not just conversationId:

{
  "Action": "Execute",
  "UserSays": "[{\"userSays\":\"Follow up question\", \"conversationId\":\"conv-123\"}]"
}

With Custom Fields

You can include any custom fields that your cognitive diagram template variables are designed to reference. These fields will be populated as variables within your AI agent.

{
  "Action": "Execute",
  "UserSays": "[{\"userMessage\":\"Hello\", \"userId\":\"user-456\", \"sessionId\":\"sess-789\"}]",
  "RobotSays": "",
  "CallerSource": 1
}

Example Request (cURL)

curl -X POST 'https://api.botx.cloud/CognitiveDiagram/12345/Execute' \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json' \
  -H 'Authorization: Bearer YOUR_ACCESS_TOKEN' \
  -d '{
    "Action": "Execute",
    "UserSays": "[{\"userSays\":\"How can you help me today?\"}]"
  }'

Example Request (Python)

import requests
import json

def execute_cognitive_diagram(token, diagram_id, message):
    url = f"https://api.botx.cloud/CognitiveDiagram/{diagram_id}/Execute"

    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Authorization': f'Bearer {token}'
    }

    payload = {
        "Action": "Execute",
        "UserSays": json.dumps([{"userSays": message}])
    }

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

    if not response.ok:
        raise Exception(f"HTTP {response.status_code}: {response.text}")

    return response.json()

Response Format

Successful Response (200 OK)

A successful execution will return a JSON object containing outputData, errorData, and debugData.

{
  "result": "OK",
  "outputData": "[{\"response\":\"AI generated response here\"}]",
  "errorData": "[]",
  "debugData": "{\"nodes\":[...]}"
}

Field	Type	Description
`result`	string	Internal dispatch status. Returns `"OK"` when the job was dispatched successfully, even if the agent later encounters errors
`outputData`	string	JSON-encoded string containing the AI-generated response. The most important field to parse
`errorData`	string	JSON-encoded array of errors or informational messages from within the cognitive diagram. An empty array (`"[]"`) means no issues
`debugData`	string	JSON-encoded debugging information about the execution flow through diagram nodes

Response with Internal Errors

Even if the HTTP status code is 200 OK, the errorData field might contain errors generated within the cognitive diagram itself.

{
  "result": "OK",
  "outputData": "",
  "errorData": "[{\"message\":\"General error: variable not populated\",\"node_name\":\"Azure OpenAI\"}]"
}

Note: Always check errorData even when the HTTP response is 200 OK. The result field only indicates that the job was dispatched - not that the cognitive diagram executed without errors.

Execute Cognitive Diagram as Job (Asynchronous)

For long-running operations that may take several minutes (such as data processing, web scraping, or complex AI training tasks), use the asynchronous job execution pattern. This involves initiating a job and then polling for its completion status.

Initiate a Job

POST https://api.botx.cloud/CognitiveDiagram/{diagram_id}/ExecuteAsJob

Request Headers

{
  "Content-Type": "application/json",
  "Accept": "*/*",
  "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}

Request Body

{
  "Action": "Execute",
  "UserSays": "[{\"userSays\":\"start\"}]"
}

Field	Type	Required	Description
`Action`	string	Yes	Always set to `"Execute"`
`UserSays`	string	Yes	JSON-encoded array with your input variables

Example Request (Python)

import requests

def initiate_job(token, diagram_id, user_input):
    url = f"https://api.botx.cloud/CognitiveDiagram/{diagram_id}/ExecuteAsJob"

    headers = {
        'Content-Type': 'application/json',
        'Accept': '*/*',
        'Authorization': f'Bearer {token}'
    }

    payload = {
        "Action": "Execute",
        "UserSays": json.dumps([{"userSays": user_input}])
    }

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

    return response.json()

Job Initiation Response

The response can be one of two formats depending on how quickly the agent completes:

Quick Completion (under 1 second)

If the agent finishes quickly, you’ll receive a regular response object similar to the synchronous execute endpoint:

{
  "result": "OK",
  "outputData": "[{\"response\":\"Quick response here\"}]",
  "errorData": "[]"
}

Long-Running Job (over 1 second)

If the agent takes more than one second to run, you’ll receive a job object that you need to poll:

{
  "job": {
    "id": "7ea49160-e58a-4fde-a9ed-d442ec0d3820",
    "created": "2025-01-15T01:57:35.6048354Z",
    "started": null,
    "finished": null,
    "status": 1,
    "isCanceled": false,
    "isCompleted": false,
    "isCompletedSuccessfully": false,
    "isFaulted": false,
    "running": 0
  },
  "result": null,
  "expand": null,
  "error": null
}

Field	Type	Description
`job.id`	string	UUID used to poll job status
`job.created`	string	ISO 8601 timestamp when the job was created
`job.started`	string or null	Timestamp when execution began (null if still queued)
`job.finished`	string or null	Timestamp when execution ended (null if still running)
`job.status`	integer	Current status code. `1` indicates queued or starting
`job.isCompleted`	boolean	`true` when the job has finished (either success or failure)
`job.isCompletedSuccessfully`	boolean	`true` only when the job finished without errors
`job.isFaulted`	boolean	`true` when the job encountered an unrecoverable error

Poll Job Status

After initiating a job that returns a job object, poll its status using the job ID until completion.

Endpoint

GET https://api.botx.cloud/job/{job_id}

Request Headers

{
  "Accept": "*/*",
  "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}

Example Status Polling (Python)

import time

def poll_job_status(token, job_id, interval_seconds=30):
    url = f"https://api.botx.cloud/job/{job_id}"

    headers = {
        'Accept': '*/*',
        'Authorization': f'Bearer {token}'
    }

    while True:
        response = requests.get(url, headers=headers)
        response.raise_for_status()

        data = response.json()

        # Check if job is completed using the isCompleted property
        if data.get('job', {}).get('isCompleted', False):
            print("Job completed!")
            return data

        # Alternative: check the result property
        if data.get('result') is not None:
            print("Job completed!")
            return data

        print("Job still running...")
        time.sleep(interval_seconds)

Job Status Responses

Job In Progress

{
  "job": {
    "id": "7ea49160-e58a-4fde-a9ed-d442ec0d3820",
    "created": "2025-01-15T01:57:35.6048354Z",
    "started": "2025-01-15T01:57:35.605016Z",
    "finished": null,
    "status": 1,
    "isCanceled": false,
    "isCompleted": false,
    "isCompletedSuccessfully": false,
    "isFaulted": false,
    "running": 96806
  },
  "result": null,
  "expand": null,
  "error": null
}

Job Completed

{
  "result": "OK",
  "outputData": "Task completed successfully",
  "errorData": "[{\"message\":\"Processing completed\",\"type\":\"INFO\"}]"
}

Note: The errorData field can contain informational messages and doesn’t necessarily indicate errors occurred.

Agent Training

Use these endpoints to trigger and monitor AI agent training processes, such as updating RAG models or retraining cognitive diagrams with new data.

Trigger Training

PATCH https://api.botx.cloud/CognitiveDiagram/{diagram_id}/TrainNode

Request Body

{
  "nodeId": "your-training-node-uuid",
  "repositoryNodeId": 59,
  "finishedFlag": false,
  "logText": "Training started by API user"
}

Field	Type	Required	Description
`nodeId`	string	Yes	UUID of the specific training node within your cognitive diagram
`repositoryNodeId`	integer	Yes	Repository identifier for the training data source
`finishedFlag`	boolean	Yes	Always set to `false` when initiating training
`logText`	string	No	Descriptive message written to training logs for audit purposes

Request Headers

{
  "Content-Type": "application/json",
  "Accept": "*/*",
  "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}

Example Training Request (Python)

import requests

def trigger_training(token, diagram_id, node_id):
    url = f"https://api.botx.cloud/CognitiveDiagram/{diagram_id}/TrainNode"

    headers = {
        'Content-Type': 'application/json',
        'Accept': '*/*',
        'Authorization': f'Bearer {token}'
    }

    payload = {
        "nodeId": node_id,
        "repositoryNodeId": 59,
        "finishedFlag": False,
        "logText": "Training started by API user"
    }

    response = requests.patch(url, headers=headers, json=payload)
    response.raise_for_status()

    return response.json()

Get Agent Status

Monitor the state of a cognitive diagram agent to determine whether it is idle, training, or in another operational state.

Endpoint

GET https://api.botx.cloud/CognitiveDiagram/{diagram_id}

Training Status Response

Agent Training (stateId = 6)

{
  "id": 8139,
  "name": "Your AI Agent",
  "stateId": 6,
  "typeId": 4,
  "langId": 1,
  "online": true,
  "createdOn": "2024-11-27T22:02:14.483",
  "primaryModelId": 4829,
  "deleted": false,
  "draftId": 8140,
  "isDraft": false,
  "cognitiveModels": []
}

Agent Idle/Ready (stateId = 2)

{
  "id": 8139,
  "name": "Your AI Agent",
  "stateId": 2,
  "typeId": 4,
  "langId": 1,
  "online": true,
  "createdOn": "2024-11-27T22:02:14.483",
  "primaryModelId": 4829,
  "deleted": false,
  "draftId": 8140,
  "isDraft": false,
  "cognitiveModels": []
}

Agent State IDs

State ID	Meaning	Action
`2`	Idle - agent is ready to accept requests	Execute normally
`6`	Training - agent is processing new training data	Wait before executing

Other state IDs may indicate transitional or error states. Contact Synthreo support if you observe persistent unexpected states.

Example Training Workflow (Python)

import requests
import time

def trigger_and_monitor_training(token, diagram_id, node_id):
    base_url = 'https://api.botx.cloud'
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json',
        'Accept': '*/*'
    }

    # Step 1: Trigger training
    training_payload = {
        "nodeId": node_id,
        "repositoryNodeId": 59,
        "finishedFlag": False,
        "logText": "Training started by API user"
    }

    response = requests.patch(
        f'{base_url}/CognitiveDiagram/{diagram_id}/TrainNode',
        headers=headers,
        json=training_payload
    )
    response.raise_for_status()

    print("Training initiated")

    # Step 2: Monitor training status
    while True:
        status_response = requests.get(
            f'{base_url}/CognitiveDiagram/{diagram_id}',
            headers=headers
        )
        status_response.raise_for_status()

        agent_data = status_response.json()
        state_id = agent_data.get('stateId')

        if state_id == 6:
            print("Agent is training...")
            time.sleep(60)  # Check every minute
        elif state_id == 2:
            print("Training completed! Agent is ready.")
            break
        else:
            print(f"Unexpected state ID: {state_id}")
            break

Parsing Responses

To extract the AI response from the outputData field, you typically need to parse the JSON string contained within it. The structure of outputData can vary based on your cognitive diagram’s configuration.

import json

def parse_response(api_response):
    # Check for successful output
    if api_response.get('outputData'):
        try:
            output_data = json.loads(api_response['outputData'])

            # Handle array response
            if isinstance(output_data, list) and len(output_data) > 0:
                return output_data[0]

            # Handle object response
            return output_data
        except json.JSONDecodeError:
            # Return raw string if JSON parsing fails
            return api_response['outputData']

    # Check for errors
    if api_response.get('errorData') and api_response['errorData'] != "[]":
        errors = json.loads(api_response['errorData'])
        raise Exception(errors[0].get('message', 'Unknown error occurred'))

    # No output data
    raise Exception('No response generated')

def extract_ai_response(response):
    parsed = parse_response(response)

    # Common response formats
    if isinstance(parsed, str):
        return parsed

    if isinstance(parsed, dict):
        if 'response' in parsed:
            return parsed['response']

        if 'gpt_response' in parsed:
            return parsed['gpt_response']

        if 'answer' in parsed:
            return parsed['answer']

    # Return as formatted JSON if no standard field found
    return json.dumps(parsed, indent=2)

Best Practices

Choosing Between Synchronous and Asynchronous Execution

Use synchronous execution (/Execute) for:

Quick AI responses (under 30 seconds)
Real-time chat interactions
Simple data queries

Use asynchronous execution (/ExecuteAsJob) for:

Long-running processes (web scraping, data processing)
Training operations
Batch processing tasks
Operations that may take several minutes

Polling Guidelines

Polling Intervals: Start with 30-second intervals for most jobs. Use 60-second intervals for training operations. Adjust based on expected job duration.
Timeout Handling: Implement a maximum polling duration (e.g., 1 hour) to avoid infinite loops.
Error Handling: Always check HTTP status codes and handle both network errors and job failures.

Training Best Practices

Training Frequency: Only trigger training when new data is available to avoid unnecessary load.
State Monitoring: Always verify the agent returns to idle state (stateId = 2) before using it in production.
Concurrent Training: Avoid triggering multiple training sessions simultaneously on the same agent.

Finding Your Diagram ID

Your cognitive diagram ID is essential for executing your AI agents. You can find it in your AI Agents Dashboard within the Synthreo platform by opening the agent and checking the URL or the agent details panel.

Finding Node IDs

For training operations, you’ll need specific node UUIDs. These can be found in the Synthreo Builder when configuring your cognitive diagram nodes - open a training node and locate the node UUID in its settings.

Related pages:

Authentication - obtain your access token before calling this API
Template Variables - pass dynamic data in the UserSays field
Error Handling - handle errorData and HTTP error codes
Best Practices - polling strategies and production patterns

Cognitive Diagrams Management API

Base URL

Quick Reference

Execute Cognitive Diagram (Synchronous)

Endpoint

Request Headers

Request Body

Basic Message

With Conversation ID

With Custom Fields

Example Request (cURL)

Example Request (Python)

Response Format

Successful Response (200 OK)

Response with Internal Errors

Execute Cognitive Diagram as Job (Asynchronous)

Initiate a Job

Request Headers

Request Body

Example Request (Python)

Job Initiation Response

Quick Completion (under 1 second)

Long-Running Job (over 1 second)

Poll Job Status

Endpoint

Request Headers

Example Status Polling (Python)

Job Status Responses

Job In Progress

Job Completed

Agent Training

Trigger Training

Request Body

Request Headers

Example Training Request (Python)

Get Agent Status

Endpoint

Training Status Response

Agent Training (stateId = 6)

Agent Idle/Ready (stateId = 2)

Agent State IDs

Example Training Workflow (Python)

Parsing Responses

Best Practices

Choosing Between Synchronous and Asynchronous Execution

Polling Guidelines

Training Best Practices

Finding Your Diagram ID

Finding Node IDs

ThreoAI

Builder

Tenant Management