Aggregator Node

The Aggregator node batches multiple events together before forwarding them as a single payload. Events are collected until a batch size threshold is reached and optionally grouped by a payload field — ideal for reducing API calls and building summary notifications.

Overview

The Aggregator node collects incoming events and groups them into batches. When the batch reaches batchSize, the aggregator flushes and emits a single event containing all collected payloads. Optionally, events can be grouped by a specific payload field (groupByField) so that events with different key values produce separate batches.

This dramatically reduces the number of downstream API calls and enables summary-style processing — for example, batching 50 individual order events into a single batch payload for bulk processing.

Configuration

Field	Type	Default	Description
name	string	—	Friendly label for the aggregator
batchSize	number	10	Flush when N events have been accumulated
windowSeconds	number	60	Flush after N seconds even if batch is not full
groupByField	string	""	Dot-notation payload field to group events by (empty = all grouped together)
outputFormat	'array' \| 'keyed'	array	'array' (flat list) or 'keyed' (grouped by key values)
inputNodes	{ nodeType, nodeId }[]	[]	Upstream nodes that feed this one. nodeType is the node kind (webhook, scheduledWorkflow, filter, …)
inputNodes	{ nodeType, nodeId }[]	[]	Upstream nodes that feed this one. nodeType is the node kind (webhook, scheduledWorkflow, filter, …)
inputNodes	string[]	[]	Generic node IDs that feed events into this aggregator
outputWebhookIds	string[]	[]	Webhook IDs to forward batched events to
outputTransformIds	string[]	[]	Transform node IDs to forward batched events to
outputNodes	string[]	[]	Generic node IDs to forward batched events to
filters	PayloadFilter[]	[]	Optional payload filters applied before aggregation
filterMode	'and' \| 'or'	and	How multiple filters combine — all must match (and) or any (or)
isActive	boolean	true	Whether the aggregator is enabled

ℹWhen groupByField is set, events with different key values are batched separately. For example, grouping by data.customerId produces one batch per customer. When empty, all events are batched together regardless of content.

Create an Aggregator

bashCreate Aggregator

curl -X POST /api/aggregator-nodes \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Order Batch for Inventory",
    "batchSize": 10,
    "windowSeconds": 60,
    "groupByField": "data.customerId",
    "outputFormat": "array",
    "isActive": true
  }'

Response

json201 Created

{
  "_id": "6655f2a3b4c5d6e7f8a9b0c1",
  "name": "Order Batch for Inventory",
  "batchSize": 10,
  "windowSeconds": 60,
  "groupByField": "data.customerId",
  "outputFormat": "array",
  "inputNodes": [],
  "inputNodes": [],
  "inputNodes": [],
  "outputWebhookIds": [],
  "outputTransformIds": [],
  "outputNodes": [],
  "filters": [],
  "filterMode": "and",
  "isActive": true,
  "createdAt": "2025-05-01T12:00:00.000Z",
  "updatedAt": "2025-05-01T12:00:00.000Z"
}

Update an Aggregator

bashUpdate Aggregator

curl -X PATCH /api/aggregator-nodes/6655f2a3b4c5d6e7f8a9b0c1 \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{
    "batchSize": 100,
    "outputFormat": "keyed"
  }'

Canvas Integration

The Aggregator is available as a canvas node. Connect webhooks or scheduled webhooks as inputs, and webhooks or transforms as outputs. Events are buffered in the aggregator until the batch threshold is reached, then forwarded as a single aggregated payload.

Payload Examples

Individual events entering the aggregator

jsonEvent 1

{
  "type": "order.created",
  "data": { "orderId": "ORD-001", "customerId": "cus_abc", "amount": 5000 }
}

jsonEvent 2

{
  "type": "order.created",
  "data": { "orderId": "ORD-002", "customerId": "cus_abc", "amount": 3000 }
}

jsonEvent 3

{
  "type": "order.created",
  "data": { "orderId": "ORD-003", "customerId": "cus_xyz", "amount": 7500 }
}

Batched output (keyed format, grouped by customerId)

When the batch flushes, the aggregator emits a single payload with all collected events:

jsonAggregated payload

{
  "type": "aggregated_batch",
  "groupByField": "data.customerId",
  "count": 3,
  "groups": {
    "cus_abc": [
      { "type": "order.created", "data": { "orderId": "ORD-001", "customerId": "cus_abc", "amount": 5000 } },
      { "type": "order.created", "data": { "orderId": "ORD-002", "customerId": "cus_abc", "amount": 3000 } }
    ],
    "cus_xyz": [
      { "type": "order.created", "data": { "orderId": "ORD-003", "customerId": "cus_xyz", "amount": 7500 } }
    ]
  }
}

Batched output (array format)

jsonAggregated payload (array)

{
  "type": "aggregated_batch",
  "count": 3,
  "items": [
    { "type": "order.created", "data": { "orderId": "ORD-001", "customerId": "cus_abc", "amount": 5000 } },
    { "type": "order.created", "data": { "orderId": "ORD-002", "customerId": "cus_abc", "amount": 3000 } },
    { "type": "order.created", "data": { "orderId": "ORD-003", "customerId": "cus_xyz", "amount": 7500 } }
  ]
}

Use Cases

Inventory bulk updates — Aggregate 10 order events then send a single bulk update to your inventory system instead of 10 individual API calls.
Bulk API calls — Batch individual webhook events into a single API call for services that support bulk operations (e.g., batch database inserts).
Digest notifications — Collect events and send a single summary notification instead of one per event.
Analytics aggregation — Group events by customer or product and forward batched analytics data to your data warehouse.
Cost optimization — Reduce the number of outbound API calls to paid services by batching events.
Report generation — Collect all events within a batch and generate a batch report for downstream processing.

API Reference

Method	Webhook	Description
GET	/api/aggregator-nodes	List all aggregator nodes
POST	/api/aggregator-nodes	Create a new aggregator
GET	/api/aggregator-nodes/:id	Get a specific aggregator
PATCH	/api/aggregator-nodes/:id	Update an aggregator
DELETE	/api/aggregator-nodes/:id	Delete an aggregator