AWS IoT Rules Engine for MCP Servers — Topic SQL, Rule Actions, and Substitution Templates

Creating a rule with CreateTopicRuleCommand

import { IoTClient, CreateTopicRuleCommand } from '@aws-sdk/client-iot';

const iot = new IoTClient({ region: 'us-east-1' });

await iot.send(new CreateTopicRuleCommand({
  ruleName: 'HighTemperatureAlert',  // alphanumeric + underscore; no hyphens
  topicRulePayload: {
    // SQL statement — equivalent to: SELECT all fields FROM messages on this topic filter
    // WHERE the temperature field exceeds the threshold
    sql: "SELECT *, topic() AS sourceTopic, timestamp() AS arrivedAt FROM 'sensors/+/temperature' WHERE temperature > 35",
    // SQL version — always use '2016-03-23'; the older '2015-10-08' version
    // has different function availability and type handling
    awsIotSqlVersion: '2016-03-23',
    description: 'Route high-temperature events to Lambda alert handler',
    ruleDisabled: false,

    // Rule actions — multiple actions run in parallel (not sequential)
    actions: [
      {
        // Lambda action — invokes the function; passes the transformed payload as the event
        lambda: {
          functionArn: 'arn:aws:lambda:us-east-1:123456789012:function:TemperatureAlertHandler',
          // No IAM role here — Lambda action uses a resource-based policy on the function
          // Add permission: aws lambda add-permission --principal iot.amazonaws.com ...
        },
      },
      {
        // Firehose action — streams the event to S3 via Firehose
        firehose: {
          deliveryStreamName: 'iot-temperature-events',
          // roleArn — IoT Core needs permission to call firehose:PutRecord
          roleArn: 'arn:aws:iam::123456789012:role/IoTFirehoseRole',
          // separator — appended after each record; newline produces NDJSON in S3
          separator: '\n',
        },
      },
    ],

    // Error action — fires when ANY action above fails; at most one error action per rule
    errorAction: {
      sqs: {
        queueUrl: 'https://sqs.us-east-1.amazonaws.com/123456789012/iot-rule-dlq',
        roleArn: 'arn:aws:iam::123456789012:role/IoTSQSRole',
        useBase64: false,
      },
    },
  },
}));

Rule name constraints: alphanumeric characters and underscores only — no hyphens, spaces, or slashes. Rule names must be unique within an account and region.

Lambda action vs other actions: the Lambda action does not use a role ARN because Lambda permissions are grant-based (resource-based policies). You must call aws lambda add-permission or use AddPermissionCommand to allow iot.amazonaws.com to invoke the function. For all other action types (SQS, SNS, Firehose, DynamoDB, S3, Republish, Timestream), you provide a role ARN; the role needs a trust policy for iot.amazonaws.com as the principal and permission to perform the target action.

Topic SQL — SELECT, FROM, and WHERE

The IoT Rules Engine SQL dialect resembles standard SQL but operates on JSON payloads arriving on MQTT topics. The FROM clause is a topic filter, not a table name.

-- Select all fields from payloads on any device's temperature topic
SELECT * FROM 'sensors/+/temperature'

-- Select specific fields and add computed fields
SELECT
  deviceId,
  temperature,
  humidity,
  temperature - humidity AS heatIndex,
  topic(2) AS deviceId2,     -- topic(2) extracts segment 2 from topic path (0-indexed)
  topic() AS fullTopic,      -- topic() = full topic string
  timestamp() AS arrivedAtMs, -- epoch milliseconds
  newuuid() AS eventId       -- random UUID for idempotency keys
FROM 'sensors/+/environment'
WHERE temperature > 30 AND humidity > 70

-- Wildcard topics and nested JSON
SELECT
  state.temperature AS temp,  -- dot notation for nested fields
  state.location.floor AS floor,
  get_dynamodb_item('DeviceMetadata', 'deviceId', deviceId).ownerEmail AS ownerEmail
FROM 'devices/#'              -- # matches all remaining topic levels
WHERE state.temperature > 35

-- Functions available in 2016-03-23 version
-- topic()          — full MQTT topic string
-- topic(n)         — nth topic segment (0-indexed)
-- timestamp()      — epoch milliseconds (bigint)
-- newuuid()        — random UUID string
-- rand()           — random float 0.0-1.0
-- floor(), ceil(), round(), abs(), sqrt(), ln(), log()
-- concat(a, b, ...) — string concatenation
-- md5(str)         — MD5 hash as hex string
-- startsWith(str, prefix) / endsWith(str, suffix)
-- parse_time(format, time [, timezone])
-- get_dynamodb_item(table, key, value [, roleArn])  — lookup enrichment

WHERE clause: the WHERE condition is evaluated against each incoming message. If the condition is false, the rule does not fire for that message — no actions run, no error action runs. The WHERE clause can reference any field in the JSON payload, topic segments via topic(n), and most SQL functions. Comparisons follow JavaScript-like coercion rules: "35" > 30 is true because numeric comparison is attempted first.

Nested JSON access: use dot notation for nested fields (state.temperature, sensors[0].value). Array indexing with [n] is supported. If a field does not exist in the payload, it evaluates to undefined, which typically causes WHERE conditions referencing it to be false (the rule does not fire for that message).

Topic wildcard rules

Extracting device ID from wildcard topics: use the topic(n) function in SELECT to capture the wildcard segment. For FROM 'sensors/+/temperature', topic(1) returns the device ID (the segment matched by +, 0-indexed). Include it in the SELECT to pass device context to downstream actions:

SELECT *, topic(1) AS deviceId FROM 'sensors/+/temperature'
-- For message on topic 'sensors/device-abc-123/temperature':
-- topic(1) = 'device-abc-123'

Performance note: rules with broad wildcards (devices/#) match more messages and invoke more action targets. For large fleets sending high-frequency telemetry, prefer narrow topic filters and push filtering into the WHERE clause rather than matching everything and filtering in Lambda.

Rule actions reference

Multiple actions: a rule can have multiple actions (maximum 10). All actions run in parallel when the rule fires — there is no ordering. If one action fails, the error action fires, but the other actions still run. The error action receives information about which action failed.

Substitution templates

Substitution templates inject dynamic values into rule action parameters at the time the rule fires. They are enclosed in ${...} and can appear in strings like S3 keys, DynamoDB table names, Republish topic names, and SQS message group IDs.

// S3 key with substitution templates — creates date-partitioned paths
// For a message from device-abc on 2026-11-09:
// key = 'raw/2026/11/09/device-abc/1731147600000.json'
{
  s3: {
    bucketName: 'my-iot-archive',
    key: "raw/${parse_time('yyyy', timestamp(), 'UTC')}/${parse_time('MM', timestamp(), 'UTC')}/${parse_time('dd', timestamp(), 'UTC')}/${topic(1)}/${timestamp()}.json",
    roleArn: 'arn:aws:iam::123456789012:role/IoTToS3Role',
  }
}

// DynamoDB key with substitution templates
{
  dynamoDBv2: {
    roleArn: 'arn:aws:iam::123456789012:role/IoTToDynamoRole',
    putItem: {
      tableName: 'DeviceTelemetry',
      // tableOperation + key expressions are fixed strings in DynamoDB v2 action
      // The entire JSON message is written as an item; use SELECT to shape the payload
    },
  }
}

// Republish with dynamic topic — route each device's enriched payload to its own topic
{
  republish: {
    // Publishes to: enriched/device-abc/temperature (where topic(1) = device-abc)
    topic: "enriched/${topic(1)}/${topic(2)}",
    roleArn: 'arn:aws:iam::123456789012:role/IoTRepublishRole',
    qos: 0,
  }
}

Template functions: ${timestamp()} returns epoch milliseconds; ${topic()} returns the full topic; ${topic(n)} returns the nth segment; ${newuuid()} generates a UUID; ${parse_time('pattern', timestamp(), 'timezone')} formats a timestamp using Java SimpleDateFormat patterns. Field references like ${temperature} access the top-level field named temperature from the message payload.

Template vs SELECT: substitution templates are evaluated at the action level using the original message payload, not the SELECT-transformed payload. If you use SELECT temperature AS temp FROM ..., the action template ${temperature} still refers to the original field name, not temp. Use ${temp} in templates only if the original message had a field named temp. This is a common source of unexpected undefined values in S3 keys and DynamoDB items.

Error action mechanics

Each rule has at most one error action. The error action fires when any primary action fails — a Lambda invocation returns an error, an SQS send is rejected, a Firehose stream is throttled, etc. It does not fire when the WHERE clause evaluates to false (no action was attempted).

// Error action — route failed rule executions to a DLQ for diagnosis
errorAction: {
  sqs: {
    queueUrl: 'https://sqs.us-east-1.amazonaws.com/123456789012/iot-rule-dlq',
    roleArn: 'arn:aws:iam::123456789012:role/IoTDLQRole',
    useBase64: false,
  },
}

// Error action — route to CloudWatch Logs for visibility
errorAction: {
  cloudwatchLogs: {
    logGroupName: '/aws/iot/rules/errors',
    // logStream can use substitution templates
    logStreamName: '${rule()}/${timestamp()}',
    roleArn: 'arn:aws:iam::123456789012:role/IoTCWLogsRole',
  },
}

Always configure an error action in production rules. Without one, failed rule executions produce no observable signal — your IoT messages silently fail to reach their targets. The CloudWatch Logs error action is the simplest to debug with; the SQS DLQ pattern enables replay if the downstream system recovers.

Error action payload: the error action receives a payload that includes the original message, the rule name, the error code, and the name of the action that failed. This allows the DLQ consumer or CloudWatch Logs analysis to identify which action caused the failure without re-querying IoT Core.

Managing rules from an MCP server

import {
  IoTClient,
  DisableTopicRuleCommand,
  EnableTopicRuleCommand,
  DeleteTopicRuleCommand,
  GetTopicRuleCommand,
  ListTopicRulesCommand,
} from '@aws-sdk/client-iot';

// Disable a rule without deleting it — useful for maintenance windows
await iot.send(new DisableTopicRuleCommand({ ruleName: 'HighTemperatureAlert' }));

// Re-enable
await iot.send(new EnableTopicRuleCommand({ ruleName: 'HighTemperatureAlert' }));

// Inspect a rule definition
const { rule } = await iot.send(new GetTopicRuleCommand({ ruleName: 'HighTemperatureAlert' }));
console.log(rule?.topicRulePayload?.sql);
console.log(rule?.topicRulePayload?.actions);

// List all rules (paginated)
async function* listAllRules() {
  let nextToken: string | undefined;
  do {
    const { rules, nextToken: next } = await iot.send(new ListTopicRulesCommand({
      maxResults: 250,
      nextToken,
      // Optional filters:
      topic: 'sensors/#',  // filter by topic pattern
      ruleDisabled: false, // only active rules
    }));
    for (const rule of rules ?? []) yield rule;
    nextToken = next;
  } while (nextToken);
}

// Delete — cannot be undone; recreate from infrastructure-as-code if needed
await iot.send(new DeleteTopicRuleCommand({ ruleName: 'ObsoleteRule' }));

Infrastructure as code: treat IoT rules as infrastructure, not application logic. Store rule definitions in CDK, Terraform, or CloudFormation and deploy them via CI/CD. MCP tools that create rules dynamically (e.g., a create_device_routing_rule tool) should track created rules in a database (device ID → rule name) so they can be cleaned up when the device is deregistered. Orphaned rules are invisible in cost but consume IoT Core rule evaluation quota.

IoT Rules vs EventBridge

Use IoT Rules for high-throughput device telemetry fan-out (thousands of events/second per device type). Use EventBridge for IoT lifecycle events (device connected, job succeeded, certificate expiring) and for routing to cross-account targets. Many IoT architectures use both: an IoT Rule fans out to Firehose for telemetry archiving, and EventBridge handles device lifecycle events for fleet management workflows.