configuring-topics

Original：🇺🇸 English

Not Translated

Configure Celigo Event Streams topics -- durable message channels that let one flow publish records and any number of other flows consume them, each from its own position. Covers creating and registering topics, retention and licensing, publisher and listener steps, publishing, reading and purging messages, listener positions, and delete rules. Use when creating or registering a topic, wiring a topic publisher or listener, publishing or inspecting messages, diagnosing a listener that never runs, or deleting a topic.

5installs

Sourceceligo/ai

Added on2026-10-21

NPX Install

npx skill4agent add celigo/ai configuring-topics

SKILL.md Content

Configuring Topics (Event Streams)

A topic is a named, durable message channel. Publishers append messages; every listener receives all of them, reading from its own position at its own pace; and a message stays readable until the topic's retention window ends, whether or not anything read it. A topic decouples the flow that produces an event from the flows that consume it: add a consumer without touching the producer, pause one consumer while the others carry on, and let it catch up from where it stopped.

A topic is an account resource that lives in one environment, outside any integration -- like a lookup cache. Flows use it through two steps, neither of which takes a connection:

Publisher -- an import with
```
adaptorType: "TopicImport"
```
. Every record it receives becomes one message.
Listener -- an export with
```
adaptorType: "TopicExport"
```
and
```
type: "stream"
```
, the page generator of a flow that runs when messages arrive.

Concerns when working with topics:

The topic -- name, retention, the license that gates it, and registration on the integrations whose flows use it
Publishing -- a publisher step in a flow, API, or Tool, or a direct publish call
Listening -- a listener step, and the flow-enable transition that subscribes it
Messages -- reading what is retained (never moves a listener) and purging it (irreversible)
Positions -- each listener's offset on the topic: where its next run resumes

How to Run the Steps

This skill names each step by what it reads or changes; the exact calls depend on how you reach the account. Read the one reference that matches your tools -- both follow the step numbers of How to Build an Event Stream below, then the headings of Messages and Positions, Diagnosing a Listener That Never Runs, and Delete, Restore, and Copies:

Celigo MCP tools (
```
upsert_export
```
,
```
upsert_import
```
,
```
upsert_flow
```
,
```
patch_flow
```
, ...) -- references/mcp.md. The deployed server has no topic tools: it writes the listener and publisher steps and their flows, while creating, registering, publishing to, reading, purging, and deleting topics take the CLI, the Celigo UI, or the REST API.
The
celigo
CLI -- references/cli.md

Listing topics and reading messages, positions, dependencies, and the audit log only read. Creating, updating, registering, deregistering, and deleting a topic write to the account. Publishing reaches every flow subscribed to the topic -- each one processes the message -- and purging deletes every retained message for good, so confirm both with the user first.

When to Use a Topic

Need	Use
One event, several independent consumers	A topic -- each listener flow receives every message
Hand records from one flow to others	A topic -- chaining with `_runNextFlowIds` passes no records (building-flows > Configure chaining)
A consumer that can be paused and catch up	A topic -- a listener whose flow is disabled keeps its position and resumes from it, as long as the messages are still retained
Enrich records mid-flow from published data	Not a topic -- a listener can't be a lookup step. Listen in its own flow, or keep the reference data in a lookup cache
React to events in an outside system	A listener on that system (webhook, NetSuite or Salesforce listener, CDC) -- configuring-exports > Listeners. An outside system can also publish to a topic directly, with an API token scoped to it

Quick Reference

Topic Fields

Field	Rules
`name`	Send one: the API saves a topic without a name, and it can't be told apart in lists. Up to 200 characters. Names aren't unique -- steps bind the `_id` , so a rename never breaks one
`description`	Up to 5,120 characters; display only
`retentionMs`	Milliseconds a message stays readable, read or not. Default `604800000` (7 days); at least `3600000` (1 hour); at most `7776000000` (90 days) and the license's `topics.retentionCeilingDays` . Shortening it drops messages older than the new window
`externalId`	Your own key, to find the topic again without storing its id (a list filter)
`draft`	Set only on create: the topic deletes itself 48 hours later unless an update sets `draft` to `false`
`__integrationIds`	Create only: registers the new topic on these integrations. Never stored or echoed back; a registration that fails is reported in `__failedIntegrationRegistrations` , and the topic is still created

Only

name

and

description

can be changed on their own. Any other change is a full update, and a full update clears

description

and

externalId

and resets

retentionMs

to 7 days when the body leaves them out -- always send the complete topic.

License and Quota

Event Streams is licensed. The account license carries

topics.enabled

topics.topicsAllowed

(how many topics the account may hold), and

topics.retentionCeilingDays

. Listing topics works on every account; creating or updating one, and every message operation, needs the entitlement -- without it they fail with

topics_subscription_required

(422 on topic writes, 403 on message calls). The quota is checked when a topic is created or restored from the recycle bin (

topics_limit_exceeded

), nothing warns as the account approaches it, and topics in the recycle bin don't count.

The Two Steps

Step	Resource	Required	Defaults	Schema
Publisher	import, `adaptorType: "TopicImport"`	`topic._topicId`	`topic.batchSize` 1000	configuring-imports topic.yml
Listener	export, `adaptorType: "TopicExport"` , `type: "stream"`	`topic._topicId`	`topic.startPosition` `latest` , `topic.maxWaitTimeMs` 30000	configuring-exports topic.yml

topic._topicId

is the topic's

_id

, never its name. Neither step takes a

_connectionId

: with one, the save validates the step as that connection's adaptor and fails. The step rules live in configuring-exports > Listeners and configuring-imports > Topic Imports; this skill covers the topic they share.

Schema Index

The listener and publisher step schemas are the

topic.yml

files linked under The Two Steps. The topic's own request and response shapes -- the topic, publish and its per-message results, a page of messages, a listener's position, and the purge result -- are on the API reference's Topics page (agent-readable markdown).

Related Skills

configuring-exports > Listeners -- the topic listener step (
```
TopicExport
```
)
configuring-imports > Topic Imports -- the publisher step (
```
TopicImport
```
)
building-flows > Create disabled, test, enable -- enabling a flow is what subscribes its listener
configuring-imports > Writing Exactly Once -- idempotent writes in listener flows
managing-integrations -- registration on integrations, cloning, and promotion across environments
managing-api-tokens > Decide the scope -- tokens that can publish to and read only named topics
configuring-lookup-caches -- the other environment resource registered on integrations; use it for enrichment

How to Build an Event Stream

1. Plan publishers and listeners

The message. What one message stands for (an order, an order line), and its shape: any JSON object up to 1 MiB serialized. A listener's flow receives each message as its record, exactly as it was published -- no envelope, no timestamp.
Who publishes. A publisher step in a flow, an API, or a Tool, or a direct publish call -- from outside Celigo too, with an API token scoped to the topic.
Who listens. One listener flow per consumer. Every listener receives every message, from its own position.
Retention. At least the longest time a listener may be paused or failing before someone fixes it. Messages older than the window are dropped whether or not anything read them; a listener paused longer than that never sees them and resumes from the earliest message left.
Duplicates. The same message can be processed more than once (step 7), so plan an idempotent write for every listener flow.

2. Check for an existing topic

List the account's topics, filter by

externalId

, or list what one integration has registered. Names aren't unique: before reusing a topic, confirm it by

_id

and description, and see what already publishes to and listens on it (its dependencies).

3. Create the topic and register it

Create the topic with a name, a description of what it carries, and a retention period; add an

externalId

if something must find it again by key. Flows publish and listen on topics registered on their integration. Register the topic on create (

__integrationIds

) or afterwards through the integration's topic registration -- or let the flow do it: saving a flow with a step bound to a topic registers that topic on the flow's integration. Registration adds no alias, unlike a lookup cache's.

4. Build the publisher

An import with

adaptorType: "TopicImport"

and

topic._topicId

, with no connection and no create, update, or upsert logic: every record becomes one message.

Without a mapping, the message is the record as the step received it; a mapping reshapes it.
With
```
oneToMany: true
```
and
```
pathToMany
```
, each element of that array is published as its own message, carrying the rest of the parent record (without the array) under
```
_PARENT
```
.
```
topic.batchSize
```
(default 1000) is how many records go per publish request; lower it when messages are large.

The step can publish from a flow, from an API, from a Tool reached through a

ToolImport

step, or on its own through an import invoke. A flow or API test run publishes nothing -- import steps are mocked there -- while an invoke publishes for real to every subscribed flow.

5. Build the listener and its flow

An export with

adaptorType: "TopicExport"

type: "stream"

, and

topic._topicId

, with no connection, no

cdc

object, and no

isLookup

. Make it the flow's page generator and create the flow with

disabled: true

and no schedule.

topic.startPosition
--
```
latest
```
(the default) delivers only messages published after the listener subscribes;
```
earliest
```
first replays every message the topic still holds, the moment it subscribes. It's read when the listener is created and whenever its
```
_topicId
```
changes, so to reprocess retained history create a new listener with
```
earliest
```
.
topic.maxWaitTimeMs
-- how long the listener collects messages before handing the flow a batch, at least 1000 (default 30000). The first run after a publish starts roughly that long after it; lower it for latency on a quiet topic. The export's
```
pageSize
```
caps the records per batch.

Without

type: "stream"

the export still saves but runs as a batch export that never starts on a publish -- see configuring-exports > Listeners for the repair.

6. Enable the flow to subscribe

The listener subscribes when its flow goes from disabled to enabled; within about half a minute its position appears (step 3 of Diagnosing a Listener That Never Runs). A flow created already enabled never subscribes: it doesn't run when messages are published, and its listener has no position until the flow is disabled and enabled again. Under

latest

, anything published before the subscription formed is skipped -- enable first, then publish. After repairing a listener, disable and re-enable its flow.

Then publish a test message (or invoke the publisher with one record) and confirm the flow ran.

7. Make every listener flow's writes idempotent

A message can reach the destination more than once, so key each listener flow's write on the message's business key (configuring-imports > Writing Exactly Once):

An
```
earliest
```
listener replays every retained message when it subscribes.
Every listener processes every message -- including a listener in a copy of the flow cloned within the same environment, which keeps the topic.
A retried error re-sends its record (building-flows > Duplicate Records).
A listener whose saved position expired or was purged resumes from the earliest message still retained.

Messages and Positions

Publish. The body is

{"messages": [ {...}, ... ]}

-- each message one JSON object, at most 1 MiB (1,048,576 bytes) serialized; a bare array fails with 400

invalid_messages

, and an empty array publishes nothing. A 200 means the body was valid, not that every message was published: read each

responses[].statusCode

. An oversized message fails on its own with 422

topic_publish_event_too_large

while the rest of the batch publishes. The first publish to a new topic provisions its stream and takes a few seconds longer than later ones.

Read. Messages come back oldest first, at most 100 per page (default 50), each as

{timestamp, payload}

, where

payload

is the message as published. Follow

nextPageURL

until it's absent, and narrow by publish time with

time_gt

and

time_lte

(epoch milliseconds). There's no search. Reading never moves any listener's position.

Purge. Deletes every retained message and keeps the topic, its retention, its registrations, and its subscriptions; it can't be undone, and the topic's audit log records it as a

purge

messages

. Offsets keep counting after a purge, and enabled listeners receive the next messages as usual. Purge empties the topic; deleting it removes the topic -- choose as with a lookup cache.

Positions. For one listener on one topic the position is

{hasCursor, offset}

, where

offset

is the next message the listener will read, counted from the start of the stream.

```
hasCursor: false
```
(offset
```
-1
```
) -- the listener has never subscribed: its flow hasn't gone from disabled to enabled since it was saved, it was saved without
```
type: "stream"
```
, or it doesn't listen to this topic.
A disabled flow keeps its listener's position, and on re-enable the flow receives everything published meanwhile.
While no message has expired or been purged, the retained message count minus
```
offset
```
is the listener's backlog.
When the message at a listener's position is no longer retained -- it expired or was purged -- the listener still reports
```
hasCursor: true
```
with that offset, and its next run resumes from the earliest message still retained, whatever its
```
startPosition
```
.

Diagnosing a Listener That Never Runs

Read the listener export. It needs
```
adaptorType: "TopicExport"
```
,
```
type: "stream"
```
, and the
```
_id
```
of the topic the publisher uses in
```
topic._topicId
```
. Without the type, repair the export as in step 5, then toggle the flow.
Read the flow. It must be enabled. A flow created enabled has never subscribed: disable it, then enable it.
Read the listener's position.
```
hasCursor: false
```
means it never subscribed (steps 1 and 2).
```
hasCursor: true
```
means it's subscribed.
Read the topic's messages. While nothing has expired or been purged, the retained count minus
```
offset
```
is what the listener hasn't read yet. Nothing unread, or no message with a
```
timestamp
```
after the flow was enabled, means nothing was published since it subscribed: check the publisher. Flow and API test runs publish nothing, and a publish answers 200 even when a message was rejected.
Allow for the batch window. A run starts about
```
maxWaitTimeMs
```
(default 30 seconds) after a publish. Then read the flow's runs and errors as for any flow (troubleshooting-flows).
Records missing from runs. Under
```
latest
```
, messages published before the subscription formed were skipped; messages that expired past
```
retentionMs
```
(or were purged) before the listener read them are gone, and it carried on from the earliest message left. To reprocess what is still retained, add a new listener with
```
startPosition: "earliest"
```
.

Delete, Restore, and Copies

Delete is refused while anything binds the topic: any export or import with its
```
topic._topicId
```
-- even one in no flow -- or a flow that contains one. The 422
```
dependencies_not_deleted
```
names each blocker; read the topic's dependencies, then delete or repoint each one.
Deregistering is refused while a step in that integration binds the topic (400
```
not_allowed
```
). Deleting the topic needs no deregister first: the delete drops it from every integration it was registered on.
A deleted topic sits in the recycle bin for 30 days, missing from lists and answering 404 on reads. Restoring it brings back its retained messages but not its registrations -- register it again. Purging it from the bin deletes it and its messages for good, and steps that bound it must be purged from the bin first.
A clone within the same environment keeps the topic. A cloned export, import, or whole flow keeps
```
topic._topicId
```
, so the clone of a listener flow is one more consumer once enabled. A copy in another environment needs that environment's topic: after cloning into another environment, installing a template, or pulling changes, check each topic step's
```
topic._topicId
```
.

Pre-Submit Checklist

The license has Event Streams enabled, and a create leaves the account within
```
topicsAllowed
```
```
retentionMs
```
is between one hour and 90 days and within the license's
```
retentionCeilingDays
```
A full update of the topic carries
```
description
```
,
```
externalId
```
, and
```
retentionMs
```
The topic is registered on every integration whose flows publish or listen (or a flow save registers it)

Listener:

adaptorType: "TopicExport"

type: "stream"

topic._topicId

set to the topic's

_id

; no

_connectionId

cdc

, or

isLookup

Publisher:

adaptorType: "TopicImport"

topic._topicId

set to the topic's

_id

; no

_connectionId

```
startPosition
```
chosen deliberately --
```
earliest
```
replays everything retained
The listener flow was created with
```
disabled: true
```
and no schedule, and is enabled only when ready
Every listener flow's write is idempotent
Every publish result was checked per message

Gotchas

A listener flow created enabled never subscribes. It doesn't run on a publish and its listener has no position until the flow goes from disabled to enabled.
latest
skips what came before the subscription. Messages published before the flow was enabled -- or while a flow created enabled sat unsubscribed -- never reach that listener.
earliest
replays everything retained the moment the listener subscribes, without any new publish. Make sure the flow's writes are idempotent first.
A listener saved without
type: "stream"
saves without error and never runs on a publish; a PATCH can't add the type.
A full update erases what it omits.
```
description
```
and
```
externalId
```
are cleared and
```
retentionMs
```
goes back to 7 days.
A publish 200 isn't per-message success. Read every
```
responses[].statusCode
```
; a message over 1 MiB fails alone.
Test runs publish nothing; an invoke publishes for real, to every subscribed flow.
Purge is irreversible. It keeps subscriptions and offsets keep counting -- there's no rewind.
Names aren't unique. Bind and look up by
```
_id
```
(or
```
externalId
```
).
Any binding blocks the delete, even an export or import that's in no flow -- and once the topic is in the recycle bin, the deleted steps that bound it block its purge until they're purged.
Restore doesn't re-register. Register the restored topic on each integration again.
A cloned listener flow is another consumer. Same-environment clones keep
```
topic._topicId
```
.

Common Errors

Error	Cause	Fix
`topics_subscription_required` (422 on create or update, 403 on message calls)	The license doesn't include Event Streams	Check the license's `topics.enabled` ; the account owner contacts Celigo
422 `topics_limit_exceeded`	The account holds `topicsAllowed` topics	Delete or reuse an unused topic (topics in the recycle bin don't count), or have the account owner ask Celigo to raise the limit
422 `min` on `retentionMs`	Under one hour ( `3600000` )	Use at least `3600000`
422 `topics_retention_exceeds_entitlement` and `max`	Over the license ceiling or 90 days	Stay within `retentionCeilingDays` and `7776000000`
422 `... is not a whitelisted property` on a PATCH	Only `/name` and `/description` patch	Send a full update with every field
400 `invalid_messages` on publish	The body isn't `{"messages": [...]}`	Wrap the messages in a `messages` array
`topic_publish_event_too_large` in `responses[]`	One message over 1 MiB serialized	Trim or split that message; the rest were published
422 `invalid_ref` "... is not a valid topic._topicId"	Not a topic `_id` in this environment -- often the name	Look the id up by name or `externalId`
422 `missing_required_field` "http subschema not defined"	A `_connectionId` on a topic step	Remove the connection
422 `not_supported` "Lookup functionality is not available for Stream Export"	`isLookup: true` on a listener	Listen in a separate flow; enrich from a lookup cache
422 `dependencies_not_deleted` on delete	An export, import, or flow still binds the topic	Delete or repoint each one named, then retry
400 `not_allowed` on deregister	A step in the integration binds the topic, or it isn't registered there	Repoint or delete the steps first
404 "already in the Recycle Bin"	The topic was deleted	Restore it, then register it again
Listener flow never runs	Not subscribed, nothing published, or the batch window	Diagnosing a Listener That Never Runs