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.

Base URL

https://builder-api.synthreo.ai

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}	Fetch a job’s status or result. Consumes a finished job
Peek at job status	GET	/job/{job_id}/check	Read job status without consuming the job
Cancel a job	DELETE	/job/{job_id}	Cancel 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://builder-api.synthreo.ai/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://builder-api.synthreo.ai/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://builder-api.synthreo.ai/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://builder-api.synthreo.ai/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 json
import requests

def initiate_job(token, diagram_id, user_input):
    url = f"https://builder-api.synthreo.ai/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)

    # No blanket raise_for_status(): a diagram that ran and failed answers 400 with the
    # reason in the body, and that is exactly what a caller needs to see.
    if response.status_code == 400:
        raise RuntimeError((response.json() or {}).get('error') or 'Job failed')
    if response.status_code not in (200, 202):
        response.raise_for_status()

    data = response.json()

    # Two shapes are possible. A job that was still running answers 202 and carries a
    # `job` object to poll; a job that finished immediately answers 200 with the result
    # and no job id at all - so never read data['job']['id'] unconditionally.
    if response.status_code == 202:
        job = data.get('job') if isinstance(data, dict) else None
        if not (job and job.get('id')):
            raise RuntimeError(f'202 response carried no job id: {data}')
        return {'job_id': job['id'], 'result': None}

    return {'job_id': None, 'result': data}

Job Completion Contract

The HTTP status code is the signal to branch on. The response body differs per status, so a client that only inspects the body will misread at least one case:

Status	Meaning	Body
200 OK	Finished successfully	The result only, exactly as the synchronous /Execute endpoint returns it. There is no job object and therefore no isCompleted field.
202 Accepted	Still queued or running	A job envelope: job (with isCompleted: false), result: null, error: null. A Location header points at the polling URL.
400 Bad Request	Finished and failed	The job envelope with job.isFaulted: true and a human-readable error.

Two consequences worth designing for:

• Do not call a blanket “raise on non-2xx” helper (raise_for_status, EnsureSuccessStatusCode, and similar). A failed job answers 400, and its body carries the reason.
• Reading a finished job consumes it. A 200 or 400 from GET /job/{job_id} removes the job from the server, so a second poll for the same id will not return the result again. Store the response. Use GET /job/{job_id}/check when you want to inspect status without consuming the job.

Job Initiation Response

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

Quick Completion (roughly half a second or less)

If the agent finishes almost immediately, the job is never registered and you receive a regular response object similar to the synchronous execute endpoint. Handle this case explicitly: there is no job id to poll.

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

Long-Running Job

If the agent is still running after roughly half a second, the response is 202 Accepted with a Location header and a job object that you 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 it by job ID until it finishes. Branch on the status code, as described in Job Completion Contract above.

Endpoint

GET https://builder-api.synthreo.ai/job/{job_id}

Request Headers

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

Example Status Polling (Python)

import time

import requests

def poll_job_status(get_token, job_id, interval_seconds=30, timeout_seconds=3600):
    """Poll one job to completion. Returns the result payload, raises on failure.

    `get_token` is a callable that returns a CURRENT access token. Tokens live 15
    minutes, which is shorter than a long job, so a single captured token would start
    answering 401 mid-poll - fetch (or reuse a cached, unexpired) token each pass.
    """
    url = f"https://builder-api.synthreo.ai/job/{job_id}"
    deadline = time.monotonic() + timeout_seconds

    while True:
        headers = {
            'Accept': '*/*',
            'Authorization': f'Bearer {get_token()}'
        }
        response = requests.get(url, headers=headers)

        # 202 = still queued or running. Nothing to read yet.
        if response.status_code == 202:
            if time.monotonic() > deadline:
                raise TimeoutError(f"Job {job_id} did not finish within {timeout_seconds}s")
            print("Job still running...")
            time.sleep(interval_seconds)
            continue

        # 200 = finished successfully. The body IS the result - there is no job envelope,
        # so do not look for isCompleted here. This read consumes the job.
        # An EMPTY 200 body means the id is unknown or was already read, NOT success.
        if response.status_code == 200:
            if not response.content.strip():
                raise RuntimeError(f"Job {job_id} is unknown, already read, or expired")
            print("Job completed!")
            return response.json()

        # 400 = the job ran and failed. The envelope carries the reason.
        if response.status_code == 400:
            payload = response.json()
            raise RuntimeError(payload.get('error') or 'Job failed')

        response.raise_for_status()
        raise RuntimeError(f"Unexpected status code: {response.status_code}")

Job Status Responses

Job In Progress (202 Accepted)

{
  "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 (200 OK)

The body is the diagram result on its own. The job envelope is gone, and this read consumes the job.

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

Job Failed (400 Bad Request)

A job that ran and faulted answers 400 with the envelope, not 200. Read error for the reason.

{
  "job": {
    "id": "7ea49160-e58a-4fde-a9ed-d442ec0d3820",
    "created": "2025-01-15T01:57:35.6048354Z",
    "started": "2025-01-15T01:57:35.605016Z",
    "finished": "2025-01-15T01:59:02.113400Z",
    "status": 7,
    "isCanceled": false,
    "isCompleted": true,
    "isCompletedSuccessfully": false,
    "isFaulted": true,
    "running": 86508
  },
  "result": null,
  "expand": null,
  "error": "Job failed to execute successfully"
}

Peeking Without Consuming

GET /job/{job_id}/check answers 200 with the job status object itself (the same fields listed above, not nested under job) and does not remove a finished job. Use it for dashboards or health checks, then call GET /job/{job_id} once you are ready to take the result.

To stop a job that is still running, call DELETE /job/{job_id}.

An unknown job id answers 200, not 404

Jobs are held in memory by the API and are dropped once their result has been read, once they have been cancelled, or after they sit unclaimed for long enough. Polling an id that no longer exists (or that belongs to another user) answers 200 with an empty body rather than 404. Treat “a 200 with no result payload” as “this job is gone”, not as success, and do not rely on a job result staying retrievable indefinitely.

---

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://builder-api.synthreo.ai/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://builder-api.synthreo.ai/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://builder-api.synthreo.ai/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://builder-api.synthreo.ai'
    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