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\"}]"
}
```text
#### 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`:

```json
{
  "Action": "Execute",
  "UserSays": "[{\"userSays\":\"Follow up question\", \"conversationId\":\"conv-123\"}]"
}
```text
#### 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.

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

```json
{
  "Content-Type": "application/json",
  "Accept": "application/json",
  "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}
```text
### Example Request (Python)

```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()
```text
### Response Format

#### Successful Response

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

```json
{
  "result": "OK",
  "outputData": "[{\"response\":\"AI generated response here\"}]",
  "errorData": "[]",
  "debugData": "{\"nodes\":[...]}"
}
```text
- `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.

```json
{
  "result": "OK",
  "outputData": "",
  "errorData": "[{\"message\":\"General error: variable not populated\",\"node_name\":\"Azure OpenAI\"}]"
}
```text
## 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

```json
{
  "Action": "Execute",
  "UserSays": "[{\"userSays\":\"start\"}]"
}
```text
**Parameters:**

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

### Headers

```json
{
  "Content-Type": "application/json",
  "Accept": "*/*",
  "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}
```text
### Example Request (Python)

```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()
```text
### 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:

```json
{
  "result": "OK",
  "outputData": "[{\"response\":\"Quick response here\"}]",
  "errorData": "[]"
}
```text
#### 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:

```json
{
  "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
}
```text
**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

```json
{
  "Accept": "*/*",
  "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}
```text
### Example Status Polling (Python)

```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)
```text
### Job Status Responses

#### Job In Progress

```json
{
  "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
}
```text
#### Job Completed

```json
{
  "result": "OK",
  "outputData": "Task completed successfully",
  "errorData": "[{\"message\":\"Processing completed\",\"type\":\"INFO\"}]"
}
```text
**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

```json
{
  "nodeId": "your-training-node-uuid",
  "repositoryNodeId": 59,
  "finishedFlag": false,
  "logText": "Training started by API user"
}
```text
**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

```json
{
  "Content-Type": "application/json",
  "Accept": "*/*",
  "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}
```text
### Example Training Request (Python)

```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()
```text
### 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)

```json
{
  "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": []
}
```text
#### Agent Idle/Ready (stateId = 2)

```json
{
  "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": []
}
```text
**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)

```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
```text
## 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.

```python
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)
```text
## 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.