Real-time Iot Anomaly Detection with Mqtt, Gpt-4o-mini Ai, and Multi-channel…

The workflow JSON

Copy or download the full n8n JSON below. Paste it into a new n8n workflow, add your credentials, activate. Full import guide →

{
  "id": "GL82b7tNiQBPG2Kc",
  "meta": {
    "templateCredsSetupCompleted": true
  },
  "name": "IoT Sensor Data Aggregation and Real-Time Anomaly Detection Pipeline",
  "tags": [],
  "nodes": [
    {
      "id": "e3ef930d-8e6a-405e-8cae-c4bf5157b28d",
      "name": "MQTT Sensor Data Ingestion",
      "type": "n8n-nodes-base.mqttTrigger",
      "position": [
        704,
        720
      ],
      "parameters": {
        "topics": "sensors/+/data",
        "options": {}
      },
      "typeVersion": 1
    },
    {
      "id": "97c5fa23-3713-4810-be77-22047e65244f",
      "name": "Workflow Configuration",
      "type": "n8n-nodes-base.set",
      "position": [
        880,
        720
      ],
      "parameters": {
        "options": {},
        "assignments": {
          "assignments": [
            {
              "id": "id-1",
              "name": "anomalyThreshold",
              "type": "number",
              "value": 2.5
            },
            {
              "id": "id-2",
              "name": "dashboardApiUrl",
              "type": "string",
              "value": "<__PLACEHOLDER_VALUE__Dashboard API endpoint URL__>"
            },
            {
              "id": "id-3",
              "name": "slackChannel",
              "type": "string",
              "value": "<__PLACEHOLDER_VALUE__Slack channel name__>"
            },
            {
              "id": "id-4",
              "name": "alertEmail",
              "type": "string",
              "value": "<__PLACEHOLDER_VALUE__Alert recipient email address__>"
            },
            {
              "id": "id-5",
              "name": "mlModelApiUrl",
              "type": "string",
              "value": "<__PLACEHOLDER_VALUE__ML model training API endpoint__>"
            }
          ]
        },
        "includeOtherFields": true
      },
      "typeVersion": 3.4
    },
    {
      "id": "60ee40bb-703a-42c5-b031-fc8a5e63ccf1",
      "name": "Edge Preprocessing & Validation",
      "type": "n8n-nodes-base.code",
      "position": [
        1072,
        720
      ],
      "parameters": {
        "mode": "runOnceForEachItem",
        "jsCode": "// Edge Preprocessing & Validation\n// Validate and preprocess incoming sensor data\n\nconst item = $input.item.json;\n\n// Initialize result object\nlet result = {\n  valid: false,\n  errors: [],\n  warnings: [],\n  dataQualityScore: 0\n};\n\ntry {\n  // Parse JSON payload if it's a string\n  let sensorData;\n  if (typeof item === 'string') {\n    try {\n      sensorData = JSON.parse(item);\n    } catch (e) {\n      result.errors.push('Invalid JSON payload');\n      return result;\n    }\n  } else {\n    sensorData = item;\n  }\n\n  // Validate required fields\n  const requiredFields = ['sensor_id', 'timestamp', 'value', 'unit'];\n  const missingFields = [];\n  \n  for (const field of requiredFields) {\n    if (!sensorData.hasOwnProperty(field) || sensorData[field] === null || sensorData[field] === undefined) {\n      missingFields.push(field);\n    }\n  }\n\n  if (missingFields.length > 0) {\n    result.errors.push(`Missing required fields: ${missingFields.join(', ')}`);\n    return result;\n  }\n\n  // Validate sensor_id\n  if (typeof sensorData.sensor_id !== 'string' || sensorData.sensor_id.trim() === '') {\n    result.errors.push('Invalid sensor_id: must be a non-empty string');\n  }\n\n  // Validate timestamp\n  const timestamp = new Date(sensorData.timestamp);\n  if (isNaN(timestamp.getTime())) {\n    result.errors.push('Invalid timestamp format');\n  } else {\n    // Check if timestamp is in the future or too old (more than 7 days)\n    const now = new Date();\n    const sevenDaysAgo = new Date(now.getTime() - 7 * 24 * 60 * 60 * 1000);\n    \n    if (timestamp > now) {\n      result.warnings.push('Timestamp is in the future');\n    } else if (timestamp < sevenDaysAgo) {\n      result.warnings.push('Timestamp is older than 7 days');\n    }\n  }\n\n  // Validate value\n  const value = parseFloat(sensorData.value);\n  if (isNaN(value)) {\n    result.errors.push('Invalid value: must be a number');\n  } else {\n    // Check for extreme values (potential sensor malfunction)\n    if (value < -1000 || value > 10000) {\n      result.warnings.push('Value is outside typical range');\n    }\n  }\n\n  // Validate unit\n  if (typeof sensorData.unit !== 'string' || sensorData.unit.trim() === '') {\n    result.errors.push('Invalid unit: must be a non-empty string');\n  }\n\n  // If there are errors, mark as invalid\n  if (result.errors.length > 0) {\n    result.valid = false;\n    result.dataQualityScore = 0;\n    return result;\n  }\n\n  // Calculate data quality score (0-100)\n  let qualityScore = 100;\n  \n  // Deduct points for warnings\n  qualityScore -= result.warnings.length * 10;\n  \n  // Deduct points for missing optional fields\n  const optionalFields = ['location', 'device_type', 'metadata'];\n  const missingOptional = optionalFields.filter(field => !sensorData.hasOwnProperty(field));\n  qualityScore -= missingOptional.length * 5;\n  \n  // Ensure score is between 0 and 100\n  qualityScore = Math.max(0, Math.min(100, qualityScore));\n  \n  result.dataQualityScore = qualityScore;\n  result.valid = true;\n\n  // Return cleaned data\n  return {\n    ...result,\n    cleanedData: {\n      sensor_id: sensorData.sensor_id.trim(),\n      timestamp: timestamp.toISOString(),\n      value: value,\n      unit: sensorData.unit.trim(),\n      location: sensorData.location || null,\n      device_type: sensorData.device_type || null,\n      metadata: sensorData.metadata || {},\n      processed_at: new Date().toISOString()\n    }\n  };\n\n} catch (error) {\n  result.errors.push(`Processing error: ${error.message}`);\n  result.valid = false;\n  return result;\n}"
      },
      "typeVersion": 2
    },
    {
      "id": "3ffe9a91-920a-4c33-a24c-9cdce677dc45",
      "name": "Normalize Sensor Data",
      "type": "n8n-nodes-base.set",
      "position": [
        1248,
        720
      ],
      "parameters": {
        "options": {},
        "assignments": {
          "assignments": [
            {
              "id": "id-1",
              "name": "sensor_id",
              "type": "string",
              "value": "={{ $json.sensor_id }}"
            },
            {
              "id": "id-2",
              "name": "timestamp",
              "type": "string",
              "value": "={{ new Date($json.timestamp).toISOString() }}"
            },
            {
              "id": "id-3",
              "name": "value",
              "type": "number",
              "value": "={{ parseFloat($json.value) }}"
            },
            {
              "id": "id-4",
              "name": "unit",
              "type": "string",
              "value": "={{ $json.unit }}"
            },
            {
              "id": "id-5",
              "name": "normalized_value",
              "type": "number",
              "value": "={{ $json.sensor_type === 'temperature' && $json.unit === 'F' ? (parseFloat($json.value) - 32) * 5/9 : parseFloat($json.value) }}"
            },
            {
              "id": "id-6",
              "name": "data_quality",
              "type": "string",
              "value": "={{ $json.data_quality }}"
            },
            {
              "id": "id-7",
              "name": "ingestion_time",
              "type": "string",
              "value": "={{ new Date().toISOString() }}"
            }
          ]
        }
      },
      "typeVersion": 3.4
    },
    {
      "id": "b84df6d0-a417-4141-9f8e-788b12cfa255",
      "name": "Store Raw Data in Time-Series DB",
      "type": "n8n-nodes-base.postgres",
      "position": [
        1440,
        720
      ],
      "parameters": {
        "table": {
          "__rl": true,
          "mode": "name",
          "value": "sensor_readings"
        },
        "schema": {
          "__rl": true,
          "mode": "list",
          "value": "public"
        },
        "columns": {
          "value": {
            "unit": "={{ $json.unit }}",
            "value": "={{ $json.value }}",
            "sensor_id": "={{ $json.sensor_id }}",
            "timestamp": "={{ $json.timestamp }}",
            "data_quality": "={{ $json.data_quality }}",
            "ingestion_time": "={{ $json.ingestion_time }}",
            "normalized_value": "={{ $json.normalized_value }}"
          },
          "mappingMode": "defineBelow"
        },
        "options": {}
      },
      "typeVersion": 2.6
    },
    {
      "id": "f99ffdf4-1fdf-42fd-b357-c43699c88f9b",
      "name": "Aggregate Sensor Readings",
      "type": "n8n-nodes-base.aggregate",
      "position": [
        1616,
        720
      ],
      "parameters": {
        "options": {},
        "aggregate": "aggregateAllItemData"
      },
      "typeVersion": 1
    },
    {
      "id": "d2944511-e678-4927-a98f-a6c16b6292b0",
      "name": "AI Anomaly Detection Agent",
      "type": "@n8n/n8n-nodes-langchain.agent",
      "position": [
        1952,
        720
      ],
      "parameters": {
        "text": "You are an IoT anomaly detection expert. Analyze the sensor data with statistical metrics provided. The data includes sensor readings with Z-scores and statistical analysis. Determine if the patterns indicate genuine anomalies or false positives. Consider temporal patterns, sensor correlations, and domain knowledge. Return a JSON object with: {is_anomaly: boolean, confidence: number 0-1, anomaly_type: string, explanation: string, recommended_action: string}",
        "options": {},
        "promptType": "define"
      },
      "typeVersion": 3
    },
    {
      "id": "b5f8a212-99f7-430f-ae6b-0f04dce8560f",
      "name": "OpenAI Chat Model",
      "type": "@n8n/n8n-nodes-langchain.lmChatOpenAi",
      "position": [
        1984,
        896
      ],
      "parameters": {
        "model": {
          "__rl": true,
          "mode": "id",
          "value": "gpt-4o-mini"
        },
        "options": {}
      },
      "credentials": {
        "openAiApi": {
          "name": "<your credential>"
        }
      },
      "typeVersion": 1.2
    },
    {
      "id": "bda65b6e-2128-449e-9c47-047e09bd2213",
      "name": "Check for Anomalies",
      "type": "n8n-nodes-base.if",
      "position": [
        2256,
        720
      ],
      "parameters": {
        "options": {},
        "conditions": {
          "options": {
            "leftValue": "",
            "caseSensitive": true,
            "typeValidation": "strict"
          },
          "combinator": "and",
          "conditions": [
            {
              "id": "id-1",
              "operator": {
                "type": "boolean",
                "operation": "equals"
              },
              "leftValue": "={{ $('AI Anomaly Detection Agent').item.json.is_anomaly }}",
              "rightValue": true
            }
          ]
        }
      },
      "typeVersion": 2.2
    },
    {
      "id": "c7fd2e6c-cd66-4c0a-b427-b545643a311d",
      "name": "Store Anomaly Records",
      "type": "n8n-nodes-base.postgres",
      "position": [
        2480,
        720
      ],
      "parameters": {
        "table": {
          "__rl": true,
          "mode": "name",
          "value": "anomaly_records"
        },
        "schema": {
          "__rl": true,
          "mode": "list",
          "value": "public"
        },
        "columns": {
          "value": {
            "sensor_id": "={{ $json.sensor_id }}",
            "timestamp": "={{ $json.timestamp }}",
            "confidence": "={{ $json.confidence }}",
            "detected_at": "={{ $json.detected_at }}",
            "explanation": "={{ $json.explanation }}",
            "anomaly_type": "={{ $json.anomaly_type }}",
            "recommended_action": "={{ $json.recommended_action }}"
          },
          "mappingMode": "defineBelow"
        },
        "options": {}
      },
      "typeVersion": 2.6
    },
    {
      "id": "749e8125-b313-495f-a13d-e06918a1a8ac",
      "name": "Send Alert to Dashboard API",
      "type": "n8n-nodes-base.httpRequest",
      "position": [
        2688,
        720
      ],
      "parameters": {
        "url": "={{ $('Workflow Configuration').first().json.dashboardApiUrl }}",
        "method": "POST",
        "options": {},
        "sendBody": true,
        "sendHeaders": true,
        "bodyParameters": {
          "parameters": [
            {
              "name": "sensor_id",
              "value": "={{ $json.sensor_id }}"
            },
            {
              "name": "timestamp",
              "value": "={{ $json.timestamp }}"
            },
            {
              "name": "anomaly_type",
              "value": "={{ $json.anomaly_type }}"
            },
            {
              "name": "confidence",
              "value": "={{ $json.confidence }}"
            },
            {
              "name": "explanation",
              "value": "={{ $json.explanation }}"
            }
          ]
        },
        "headerParameters": {
          "parameters": [
            {
              "name": "Content-Type",
              "value": "application/json"
            }
          ]
        }
      },
      "typeVersion": 4.3
    },
    {
      "id": "68e0d2a7-0a0a-405c-88fa-3f4e9b2ee2a8",
      "name": "Send Slack Alert",
      "type": "n8n-nodes-base.slack",
      "position": [
        2944,
        720
      ],
      "parameters": {
        "text": "=\ud83d\udea8 ANOMALY DETECTED \ud83d\udea8\n\n*Sensor ID:* {{ $('AI Anomaly Detection Agent').first().json.sensorId }}\n*Anomaly Type:* {{ $('AI Anomaly Detection Agent').first().json.anomalyType }}\n*Confidence:* {{ $('AI Anomaly Detection Agent').first().json.confidence }}%\n*Recommended Action:* {{ $('AI Anomaly Detection Agent').first().json.recommendedAction }}\n\n*Timestamp:* {{ $now.format('yyyy-MM-dd HH:mm:ss') }}",
        "select": "channel",
        "channelId": {
          "__rl": true,
          "mode": "id",
          "value": "={{ $('Workflow Configuration').first().json.slackChannel }}"
        },
        "otherOptions": {},
        "authentication": "oAuth2"
      },
      "credentials": {
        "slackOAuth2Api": {
          "name": "<your credential>"
        }
      },
      "typeVersion": 2.3
    },
    {
      "id": "75f75fad-8109-415b-a54d-b45e5903c157",
      "name": "Send Email Alert",
      "type": "n8n-nodes-base.emailSend",
      "position": [
        3104,
        720
      ],
      "parameters": {
        "html": "=<!DOCTYPE html>\n<html>\n<head>\n  <style>\n    body { font-family: Arial, sans-serif; line-height: 1.6; color: #333; }\n    .container { max-width: 600px; margin: 0 auto; padding: 20px; }\n    .header { background-color: #ff6b6b; color: white; padding: 20px; text-align: center; border-radius: 5px; }\n    .content { background-color: #f9f9f9; padding: 20px; margin-top: 20px; border-radius: 5px; }\n    .detail-row { margin: 10px 0; padding: 10px; background-color: white; border-left: 4px solid #4CAF50; }\n    .label { font-weight: bold; color: #555; }\n    .value { color: #333; }\n    .confidence { font-size: 24px; font-weight: bold; color: #ff6b6b; }\n    .actions { background-color: #fff3cd; padding: 15px; margin-top: 20px; border-radius: 5px; border-left: 4px solid #ffc107; }\n    .footer { text-align: center; margin-top: 20px; color: #777; font-size: 12px; }\n  </style>\n</head>\n<body>\n  <div class=\"container\">\n    <div class=\"header\">\n      <h1>\ud83d\udea8 IoT Anomaly Detected</h1>\n      <p>High Confidence Detection Alert</p>\n    </div>\n    \n    <div class=\"content\">\n      <h2>Anomaly Details</h2>\n      \n      <div class=\"detail-row\">\n        <span class=\"label\">Sensor ID:</span>\n        <span class=\"value\">{{ $('AI Anomaly Detection Agent').first().json.sensorId }}</span>\n      </div>\n      \n      <div class=\"detail-row\">\n        <span class=\"label\">Sensor Type:</span>\n        <span class=\"value\">{{ $('AI Anomaly Detection Agent').first().json.sensorType }}</span>\n      </div>\n      \n      <div class=\"detail-row\">\n        <span class=\"label\">Location:</span>\n        <span class=\"value\">{{ $('AI Anomaly Detection Agent').first().json.location }}</span>\n      </div>\n      \n      <div class=\"detail-row\">\n        <span class=\"label\">Detected Value:</span>\n        <span class=\"value\">{{ $('AI Anomaly Detection Agent').first().json.value }} {{ $('AI Anomaly Detection Agent').first().json.unit }}</span>\n      </div>\n      \n      <div class=\"detail-row\">\n        <span class=\"label\">Timestamp:</span>\n        <span class=\"value\">{{ $('AI Anomaly Detection Agent').first().json.timestamp }}</span>\n      </div>\n      \n      <div class=\"detail-row\">\n        <span class=\"label\">Confidence Score:</span>\n        <span class=\"confidence\">{{ $('AI Anomaly Detection Agent').first().json.confidenceScore }}%</span>\n      </div>\n      \n      <div class=\"detail-row\">\n        <span class=\"label\">Anomaly Type:</span>\n        <span class=\"value\">{{ $('AI Anomaly Detection Agent').first().json.anomalyType }}</span>\n      </div>\n      \n      <div class=\"detail-row\">\n        <span class=\"label\">Description:</span>\n        <span class=\"value\">{{ $('AI Anomaly Detection Agent').first().json.description }}</span>\n      </div>\n    </div>\n    \n    <div class=\"actions\">\n      <h3>\u26a0\ufe0f Recommended Actions</h3>\n      <ul>\n        <li>Verify sensor calibration and physical condition</li>\n        <li>Check for environmental factors affecting readings</li>\n        <li>Review historical data for similar patterns</li>\n        <li>Consider maintenance or replacement if anomaly persists</li>\n        <li>Monitor related sensors in the same zone</li>\n      </ul>\n    </div>\n    \n    <div class=\"footer\">\n      <p>This is an automated alert from the IoT Monitoring System</p>\n      <p>For support, contact: support@iot-monitoring.com</p>\n    </div>\n  </div>\n</body>\n</html>",
        "options": {},
        "subject": "IoT Anomaly Alert: High Confidence Detection",
        "toEmail": "={{ $('Workflow Configuration').first().json.alertEmail }}",
        "fromEmail": "user@example.com"
      },
      "typeVersion": 2.1
    },
    {
      "id": "73602f06-11ce-41de-80b1-88ed5af0583c",
      "name": "Model Retraining Schedule",
      "type": "n8n-nodes-base.scheduleTrigger",
      "position": [
        704,
        1088
      ],
      "parameters": {
        "rule": {
          "interval": [
            {
              "daysInterval": 7,
              "triggerAtHour": 2
            }
          ]
        }
      },
      "typeVersion": 1.2
    },
    {
      "id": "e735b054-b179-4d33-a43a-af2dc0c0f530",
      "name": "Fetch Historical Data for Retraining",
      "type": "n8n-nodes-base.postgres",
      "position": [
        928,
        1088
      ],
      "parameters": {
        "query": "SELECT sensor_id, timestamp, value, normalized_value, data_quality FROM sensor_readings WHERE timestamp > NOW() - INTERVAL '30 days' ORDER BY timestamp DESC",
        "options": {},
        "operation": "executeQuery"
      },
      "typeVersion": 2.6
    },
    {
      "id": "03aa1ea6-a2d4-4775-a77c-d98648f439ae",
      "name": "Retrain ML Model via API",
      "type": "n8n-nodes-base.httpRequest",
      "position": [
        1152,
        1088
      ],
      "parameters": {
        "url": "={{ $('Workflow Configuration').first().json.mlModelApiUrl }}",
        "method": "POST",
        "options": {},
        "jsonBody": "={{ {\n  \"training_data\": $('Fetch Historical Data for Retraining').all(),\n  \"parameters\": {\n    \"epochs\": 100,\n    \"batch_size\": 32,\n    \"learning_rate\": 0.001\n  }\n} }}",
        "sendBody": true,
        "sendHeaders": true,
        "specifyBody": "json",
        "headerParameters": {
          "parameters": [
            {
              "name": "Content-Type",
              "value": "application/json"
            }
          ]
        }
      },
      "typeVersion": 4.3
    },
    {
      "id": "efda13f8-33a0-4dae-99a8-92b28db0b587",
      "name": "Statistical Anomaly Detection",
      "type": "n8n-nodes-base.code",
      "position": [
        1792,
        688
      ],
      "parameters": {
        "jsCode": "// Statistical Anomaly Detection using Z-score method\n\n// Get all input items\nconst items = $input.all();\n\n// Get threshold from workflow config (default to 3 if not set)\nconst threshold = $('Workflow Configuration').item.json.anomaly_threshold || 3;\n\n// Extract all sensor readings for statistical analysis\nconst readings = items.map(item => item.json.value || item.json.reading || 0);\n\n// Calculate mean\nconst mean = readings.reduce((sum, val) => sum + val, 0) / readings.length;\n\n// Calculate standard deviation\nconst variance = readings.reduce((sum, val) => sum + Math.pow(val - mean, 2), 0) / readings.length;\nconst stdDev = Math.sqrt(variance);\n\n// Process each item and calculate Z-score\nconst enrichedItems = items.map(item => {\n  const value = item.json.value || item.json.reading || 0;\n  \n  // Calculate Z-score\n  const zScore = stdDev === 0 ? 0 : (value - mean) / stdDev;\n  \n  // Flag as anomaly if Z-score exceeds threshold\n  const isAnomaly = Math.abs(zScore) > threshold;\n  \n  return {\n    json: {\n      ...item.json,\n      anomaly_score: Math.abs(zScore),\n      is_anomaly: isAnomaly,\n      statistical_analysis: {\n        z_score: zScore,\n        mean: mean,\n        std_dev: stdDev,\n        threshold: threshold,\n        deviation_from_mean: value - mean\n      }\n    }\n  };\n});\n\nreturn enrichedItems;"
      },
      "typeVersion": 2
    },
    {
      "id": "e6697ad0-d5a2-4fd7-b74b-6b77c55ddc86",
      "name": "Sticky Note",
      "type": "n8n-nodes-base.stickyNote",
      "position": [
        608,
        336
      ],
      "parameters": {
        "width": 576,
        "height": 240,
        "content": "## How It Works\nMQTT ingests real-time sensor data from connected devices. The workflow normalizes the values and trains or retrains machine learning models on a defined schedule. An AI agent detects anomalies, validates the results for accuracy, and ensures reliable alerts. Detected issues are then routed to dashboards for visualization and sent via email notifications to relevant stakeholders, enabling timely monitoring and response.\n\n"
      },
      "typeVersion": 1
    },
    {
      "id": "c915c806-f59a-43b5-822f-b8b7313c02ad",
      "name": "Sticky Note1",
      "type": "n8n-nodes-base.stickyNote",
      "position": [
        1232,
        336
      ],
      "parameters": {
        "width": 560,
        "height": 240,
        "content": "\n## Setup Steps\n1. **MQTT:** Configure broker connection, set topic subscriptions, and verify data flow.\n2. **ML Model:** Define retraining schedule and specify historical data sources for model updates.\n3. **AI Agent:** Connect Claude or OpenAI APIs and configure anomaly validation prompts.\n4. **Alerts:** Set dashboard URL and email recipients to receive real-time notifications."
      },
      "typeVersion": 1
    },
    {
      "id": "81bb97c5-8c2b-42c3-974e-db1ba8998154",
      "name": "Sticky Note2",
      "type": "n8n-nodes-base.stickyNote",
      "position": [
        1808,
        336
      ],
      "parameters": {
        "color": 4,
        "width": 608,
        "height": 224,
        "content": "## Prerequisites\nMQTT broker credentials; historical training data; OpenAI/Claude API key; dashboard access; email service\n\n## Use Cases\nIoT sensor monitoring; server performance tracking; network traffic anomalies; application log analysis; predictive maintenance alerts\n\n"
      },
      "typeVersion": 1
    },
    {
      "id": "6964732e-ab63-43be-897e-54468329706c",
      "name": "Sticky Note3",
      "type": "n8n-nodes-base.stickyNote",
      "position": [
        2432,
        336
      ],
      "parameters": {
        "color": 6,
        "width": 656,
        "height": 224,
        "content": "## Customization\nAdjust sensitivity thresholds; swap ML models; modify notification channels; add Slack/Teams integration; customize validation rules\n\n## Benefits\nReduces detection latency 95%; eliminates manual monitoring; prevents false alerts; enables rapid incident response; improves system reliability"
      },
      "typeVersion": 1
    },
    {
      "id": "608b8e15-445c-4733-9dcf-8052c2ecba98",
      "name": "Sticky Note4",
      "type": "n8n-nodes-base.stickyNote",
      "position": [
        608,
        592
      ],
      "parameters": {
        "color": 7,
        "width": 592,
        "height": 320,
        "content": "## MQTT Ingests Sensor Stream\n\nContinuously receives metrics from sensors or apps via MQTT.\n**Why:** Lightweight and reliable for high-volume data, enabling anomaly detection within seconds."
      },
      "typeVersion": 1
    },
    {
      "id": "5a28cecd-5c04-4432-84d9-62cb2a84cea8",
      "name": "Sticky Note5",
      "type": "n8n-nodes-base.stickyNote",
      "position": [
        1216,
        592
      ],
      "parameters": {
        "color": 7,
        "width": 528,
        "height": 320,
        "content": "## Normalize Sensor Readings & Storage\n\nStandardizes units and scales across sensor types.\n**Why:** Ensures fair comparison so AI interprets all metrics consistently.\n"
      },
      "typeVersion": 1
    },
    {
      "id": "15649d44-6d37-4996-933c-dcc6d07e748d",
      "name": "Sticky Note6",
      "type": "n8n-nodes-base.stickyNote",
      "position": [
        608,
        928
      ],
      "parameters": {
        "color": 7,
        "width": 720,
        "height": 336,
        "content": "## Schedule ML Retraining\n\nRetrains models on historical data at set intervals.\n**Why:** Sensor behavior shifts over time; retraining keeps detection accurate."
      },
      "typeVersion": 1
    },
    {
      "id": "cc98f529-4756-4caa-884d-511824af4310",
      "name": "Sticky Note7",
      "type": "n8n-nodes-base.stickyNote",
      "position": [
        1760,
        592
      ],
      "parameters": {
        "color": 7,
        "width": 166,
        "height": 432,
        "content": "## Statistical Detection\n\n\n\n\n\n\n\n\n\n\n\n\nUses deviation, moving averages, and rate-of-change to flag outliers.\n**Why:** Fast first-pass detection of obvious spikes with minimal compute."
      },
      "typeVersion": 1
    },
    {
      "id": "4c7848c7-ef97-467a-a19c-572636f995d5",
      "name": "Sticky Note8",
      "type": "n8n-nodes-base.stickyNote",
      "position": [
        1936,
        592
      ],
      "parameters": {
        "color": 7,
        "width": 464,
        "height": 432,
        "content": "## AI Anomaly Agent\n\nValidates statistical flags using context and cross-metric signals.\n**Why:** Reduces noise by distinguishing harmless blips from real issues."
      },
      "typeVersion": 1
    },
    {
      "id": "100e1969-938a-4509-b342-47574dd249de",
      "name": "Sticky Note9",
      "type": "n8n-nodes-base.stickyNote",
      "position": [
        2416,
        592
      ],
      "parameters": {
        "color": 7,
        "width": 448,
        "height": 432,
        "content": "## Monitoring Dashboard\n\nPublishes verified anomalies with context and severity.\n**Why:** Centralized view accelerates root-cause analysis.\n"
      },
      "typeVersion": 1
    },
    {
      "id": "8b9c3d9e-0b6d-43a1-9303-e437b3b69bb0",
      "name": "Sticky Note10",
      "type": "n8n-nodes-base.stickyNote",
      "position": [
        2880,
        592
      ],
      "parameters": {
        "color": 7,
        "width": 448,
        "height": 432,
        "content": "## Notifications\n\nSends summaries with suggested investigation steps.\n**Why:** Gives on-call engineers immediate, actionable details."
      },
      "typeVersion": 1
    }
  ],
  "active": false,
  "settings": {
    "executionOrder": "v1"
  },
  "versionId": "42a24dc3-5180-4ac1-b1c1-4ca77dbc4373",
  "connections": {
    "Send Slack Alert": {
      "main": [
        [
          {
            "node": "Send Email Alert",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "OpenAI Chat Model": {
      "ai_languageModel": [
        [
          {
            "node": "AI Anomaly Detection Agent",
            "type": "ai_languageModel",
            "index": 0
          }
        ]
      ]
    },
    "Check for Anomalies": {
      "main": [
        [
          {
            "node": "Store Anomaly Records",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Normalize Sensor Data": {
      "main": [
        [
          {
            "node": "Store Raw Data in Time-Series DB",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Store Anomaly Records": {
      "main": [
        [
          {
            "node": "Send Alert to Dashboard API",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Workflow Configuration": {
      "main": [
        [
          {
            "node": "Edge Preprocessing & Validation",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Aggregate Sensor Readings": {
      "main": [
        [
          {
            "node": "Statistical Anomaly Detection",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Model Retraining Schedule": {
      "main": [
        [
          {
            "node": "Fetch Historical Data for Retraining",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "AI Anomaly Detection Agent": {
      "main": [
        [
          {
            "node": "Check for Anomalies",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "MQTT Sensor Data Ingestion": {
      "main": [
        [
          {
            "node": "Workflow Configuration",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Send Alert to Dashboard API": {
      "main": [
        [
          {
            "node": "Send Slack Alert",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Statistical Anomaly Detection": {
      "main": [
        [
          {
            "node": "AI Anomaly Detection Agent",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Edge Preprocessing & Validation": {
      "main": [
        [
          {
            "node": "Normalize Sensor Data",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Store Raw Data in Time-Series DB": {
      "main": [
        [
          {
            "node": "Aggregate Sensor Readings",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Fetch Historical Data for Retraining": {
      "main": [
        [
          {
            "node": "Retrain ML Model via API",
            "type": "main",
            "index": 0
          }
        ]
      ]
    }
  }
}

Credentials you'll need

Each integration node will prompt for credentials when you import. We strip credential IDs before publishing — you'll add your own.

• openAiApi
• slackOAuth2Api

About this workflow

MQTT ingests real-time sensor data from connected devices. The workflow normalizes the values and trains or retrains machine learning models on a defined schedule. An AI agent detects anomalies, validates the results for accuracy, and ensures reliable alerts. Detected issues are…

Source: https://n8n.io/workflows/10759/ — original creator credit. Request a take-down →

More AI & RAG workflows → · Browse all categories →

Related workflows

Workflows that share integrations, category, or trigger type with this one. All free to copy and import.