Cognitive Diagrams Management 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.

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

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
}

Headers

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

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

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

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

• result: Internal field indicating whether the job was dispatched successfully. Note that even if the agent failed immediately, this might still return "OK" if the job was dispatched properly.
• outputData: A JSON string containing the AI-generated response. This is typically the most important field.
• errorData: A JSON string containing any errors encountered during execution within the cognitive diagram. If empty ("[]"), no internal errors occurred.
• debugData: A JSON string with debugging information about the cognitive diagram execution flow.

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\"}]"
}

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 Body

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

Parameters:

• Action: Always set to "Execute"
• UserSays: JSON string containing the input parameters for your agent

Headers

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

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
}

Response Fields:

• job.id: Unique identifier for tracking the job
• job.created: Timestamp when the job was created
• job.isCompleted: Boolean indicating if the job has finished
• job.status: Current status code (1 = queued/starting)
• Other fields provide additional job state information

Poll Job Status

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

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

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"
}

Parameters:

• nodeId: UUID of the specific training node within your cognitive diagram
• repositoryNodeId: Repository identifier for the training data
• finishedFlag: Always set to false when initiating training
• logText: Descriptive message for training logs

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()

Monitor Training Status

After triggering training, monitor the agent's state to determine when training completes.

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": []
}

State IDs:

• stateId = 6: Agent is currently training
• stateId = 2: Agent is idle and ready for requests
• Other state IDs may indicate different operational 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 maximum polling duration 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
• State Monitoring: Always verify the agent returns to idle state (stateId = 2) before using it
• 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.

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.